[SPR-960]: Save last seen vacuum cutoff xid in redis

include/common/redis_types.hh

@@ -159,4 +159,11 @@ namespace springtail::redis {
      * value: xid
      */
     static constexpr char SET_DB_INDEX_XIDS[] = "{}:set:db_index_xids:{}";
+
+    /**
+     * Hash of vacuum cutoff XIDs per db_id for a given db_instance_id
+     * args: <db_instance_id>
+     * key: <db_id>, value: <cutoff_xid>
+     */
+    static constexpr char VACUUM_CUTOFF_XID[] = "{}:vacuum_cutoff_xids";
 }

include/storage/vacuumer.hh

@@ -100,6 +100,14 @@ public:
      */
     void cleanup_db(uint64_t cleanup_db_id);
 
+    /**
+     * @brief Get last seen cutoff XID for the DB
+     *
+     * @param db_id Database ID
+     * @return cutoff_xid for the DB, or 0 if nothing found
+     */
+    uint64_t get_last_seen_cutoff_xid(uint64_t db_id);
+
 protected:
     /**
      * @brief Constructor, that inits the vacuumer thread
@@ -127,7 +135,7 @@ protected:
     void _internal_run() override;
 
     /**
-     * @brief Graceful shutdown of vacummer thread
+     * @brief Graceful shutdown of vacuumer thread
      */
     void _internal_shutdown() override;
 
@@ -199,6 +207,7 @@ private:
     std::filesystem::path _vacuum_data_base; ///< The base directory for vacuum directories
     std::filesystem::path _global_vacuum_file; ///< Global vacuum file
     std::filesystem::path _global_vacuum_runfile; ///< Global vacuum file for current run
+    std::string _vacuum_cutoff_xid_redis_hash;    ///< name of the redis hash holding last seen vacuum cutoff XIDs
 
     RedisDDL _redis_ddl; ///< Interface to the DDL structures in Redis.
 
@@ -358,5 +367,13 @@ private:
      */
     template <CleanupOperation op>
     void _cleanup_global_vacuum_file(uint64_t cleanup_db_id=-1);
+
+    /**
+     * @brief Save last seen cutoff xid in redis per db
+     *
+     * @param db_id      Database ID
+     * @param cutoff_xid Cutoff XID
+     */
+    void _save_last_seen_cutoff_xid(uint64_t db_id, uint64_t cutoff_xid);
 };
 }

src/storage/vacuumer.cc

@@ -7,6 +7,7 @@
 #include <storage/interval_tree.hh>
 
 #include <xid_mgr/xid_mgr_client.hh>
+#include <common/redis_types.hh>
 
 namespace springtail {
 
@@ -80,6 +81,10 @@ Vacuumer::_init()
     } else {
         _extents_tracking_enabled = false;
     }
+
+    // Initialize redis hash to save cutoff XIDs
+    uint64_t db_instance_id = Properties::get_db_instance_id();
+    _vacuum_cutoff_xid_redis_hash = fmt::format(redis::VACUUM_CUTOFF_XID, db_instance_id);
 }
 
 void
@@ -413,6 +418,25 @@ Vacuumer::hole_punch_file(const std::string& file,
     return not_punched;
 }
 
+void
+Vacuumer::_save_last_seen_cutoff_xid(uint64_t db_id, uint64_t cutoff_xid)
+{
+    RedisClientPtr client = RedisMgr::get_instance()->get_client();
+    client->hset(_vacuum_cutoff_xid_redis_hash, std::to_string(db_id), std::to_string(cutoff_xid));
+}
+
+uint64_t
+Vacuumer::get_last_seen_cutoff_xid(uint64_t db_id)
+{
+    RedisClientPtr client = RedisMgr::get_instance()->get_client();
+    auto val = client->hget(_vacuum_cutoff_xid_redis_hash, std::to_string(db_id));
+    if (val) {
+        return std::stoull(*val);
+    } else {
+        return 0;
+    }
+}
+
 uint64_t
 Vacuumer::_get_vacuum_cutoff_xid(uint64_t db_id)
 {
@@ -833,7 +857,13 @@ Vacuumer::_do_vacuum_run()
             const std::string& file = file_it->first;
             auto& xid_map = file_it->second;
             auto db_id = _get_db_id_from_path(file);
-            auto cutoff_xid = _get_vacuum_cutoff_xid(db_id);  // Get safest XID to vacuum till that point
+
+            // Get safest XID to vacuum till that point
+            // and save in the redis for stats
+            auto cutoff_xid = _get_vacuum_cutoff_xid(db_id);
+            if (cutoff_xid > 0) {
+                _save_last_seen_cutoff_xid(db_id, cutoff_xid);
+            }
 
             IntervalTree<uint64_t> itree;
             std::vector<uint64_t> xids_to_process;
@@ -910,7 +940,12 @@ Vacuumer::_do_vacuum_run()
         /* --------------- Snapshot deletion flow -----------------------------------------------------------*/
         // expire snapshots through the min XID
         for (auto db_it = expired_snapshots_map.begin(); db_it != expired_snapshots_map.end(); ) {
-            uint64_t cutoff_xid = _get_vacuum_cutoff_xid(db_it->first);  // Get safest XID to vacuum till that point
+            // Get safest XID to vacuum till that point
+            // and save in redis for stats
+            uint64_t cutoff_xid = _get_vacuum_cutoff_xid(db_it->first);
+            if (cutoff_xid > 0) {
+                _save_last_seen_cutoff_xid(db_it->first, cutoff_xid);
+            }
             auto& xid_map = db_it->second;
 
             for (auto xid_it = xid_map.begin(); xid_it != xid_map.end(); ) {