feat(replication): make the full-sync checkpoint alive time configurable

kvrocks.conf

@@ -415,6 +415,22 @@ slave-empty-db-before-fullsync no
 # Default: no
 purge-backup-on-fullsync no
 
+# When replicas need full synchronization, the master creates a RocksDB checkpoint
+# to feed them. The checkpoint is purged by the server cron task in two cases:
+#
+# 1. It has been idle (no replica is fetching its files) for longer than
+#    'checkpoint-idle-time-secs' seconds.
+# 2. It has existed for longer than 'checkpoint-alive-time-secs' seconds, even if
+#    some replicas are still fetching its files. This is a hard limit to avoid a
+#    slow or stuck replica keeping the checkpoint (and the SST files it references)
+#    alive forever.
+#
+# Default: 30 (idle seconds before purging an unused checkpoint)
+checkpoint-idle-time-secs 30
+
+# Default: 86400 (i.e. 24 hours) the maximum lifetime of a full sync checkpoint
+checkpoint-alive-time-secs 86400
+
 # The maximum allowed rate (in MB/s) that should be used by replication.
 # If the rate exceeds max-replication-mb, replication will slow down.
 # Default: 0 (i.e. no limit)

src/config/config.cc

@@ -239,6 +239,8 @@ Config::Config() {
       {"purge-backup-on-fullsync", false, new YesNoField(&purge_backup_on_fullsync, false)},
       {"rename-command", true, new MultiStringField(&rename_command_, std::vector<std::string>{})},
       {"fullsync-recv-file-delay", false, new IntField(&fullsync_recv_file_delay, 0, 0, INT_MAX)},
+      {"checkpoint-alive-time-secs", false, new IntField(&checkpoint_alive_time_secs, 24 * 60 * 60, 1, INT_MAX)},
+      {"checkpoint-idle-time-secs", false, new IntField(&checkpoint_idle_time_secs, 30, 1, INT_MAX)},
       {"cluster-enabled", true, new YesNoField(&cluster_enabled, false)},
       {"migrate-sequence-gap", false, new IntField(&sequence_gap, 10000, 1, INT_MAX)},
       {"migrate-batch-size-kb", false, new IntField(&migrate_batch_size_kb, 16, 1, INT_MAX)},

src/config/config.h

@@ -158,6 +158,12 @@ struct Config {
   bool master_use_repl_port = false;
   bool purge_backup_on_fullsync = false;
   int fullsync_recv_file_delay = 0;
+  // The maximum lifetime (in seconds) of the checkpoint used by full sync; once it's
+  // exceeded, the checkpoint will be purged even if replicas are still fetching files.
+  int checkpoint_alive_time_secs = 24 * 60 * 60;
+  // The idle time (in seconds) to purge the checkpoint used by full sync when no
+  // replica is fetching its files.
+  int checkpoint_idle_time_secs = 30;
   bool use_rsid_psync = false;
   bool replication_group_sync = false;
   bool replication_no_slowdown = false;

src/storage/storage.cc

@@ -1201,12 +1201,11 @@ Status Storage::TryPurgeCheckpoint(int fetch_file_threads) {
       return Status::OK();
     }
 
-    // TODO(shooterit): support to config the alive time of checkpoint
     int64_t create_time_secs = checkpoint_info_.create_time_secs;
     int64_t access_time_secs = checkpoint_info_.access_time_secs;
     int64_t now_secs = util::GetTimeStamp<std::chrono::seconds>();
-    bool should_purge =
-        (fetch_file_threads == 0 && now_secs - access_time_secs > 30) || (now_secs - create_time_secs > 24 * 60 * 60);
+    bool should_purge = (fetch_file_threads == 0 && now_secs - access_time_secs > config_->checkpoint_idle_time_secs) ||
+                        (now_secs - create_time_secs > config_->checkpoint_alive_time_secs);
     if (!should_purge) {
       return Status::OK();
     }

src/storage/storage.h

@@ -357,6 +357,7 @@ class Storage {
   // Rename checkpoint to "*.trash" under checkpoint_mu_, then DestroyDB outside the lock.
   Status TryPurgeCheckpoint(int fetch_file_threads);
   int64_t GetCheckpointCreateTimeSecs() const { return checkpoint_info_.create_time_secs; }
+  void SetCheckpointCreateTimeSecs(int64_t t) { checkpoint_info_.create_time_secs = t; }
   void SetCheckpointAccessTimeSecs(int64_t t) { checkpoint_info_.access_time_secs = t; }
   int64_t GetCheckpointAccessTimeSecs() const { return checkpoint_info_.access_time_secs; }
   void SetDBInRetryableIOError(bool yes_or_no) { db_in_retryable_io_error_ = yes_or_no; }

tests/cppunit/config_test.cc

@@ -90,6 +90,8 @@ TEST(Config, GetAndSet) {
       {"rocksdb.sst_file_delete_rate_bytes_per_sec", "0"},
       {"max-replication-lag", "50000000"},
       {"replication-send-timeout-ms", "60000"},
+      {"checkpoint-alive-time-secs", "3600"},
+      {"checkpoint-idle-time-secs", "10"},
   };
   std::vector<std::string> values;
   for (const auto &iter : mutable_cases) {

tests/cppunit/storage_test.cc

@@ -31,6 +31,7 @@
 #include <vector>
 
 #include "test_base.h"
+#include "time_util.h"
 
 TEST(Storage, CreateBackup) {
   std::error_code ec;
@@ -258,6 +259,53 @@ TEST(Storage, TryPurgeCheckpoint) {
   EXPECT_TRUE(storage->ExistCheckpoint());
 }
 
+TEST(Storage, TryPurgeCheckpointRespectsConfiguredAliveTime) {
+  std::error_code ec;
+
+  const std::string test_dir = "test_purge_checkpoint_alive_time";
+  Config config;
+  config.db_dir = test_dir + "/db";
+  config.checkpoint_dir = test_dir + "/checkpoint";
+  config.slot_id_encoded = false;
+  config.checkpoint_alive_time_secs = 1;
+  config.checkpoint_idle_time_secs = 10000;
+
+  std::filesystem::remove_all(test_dir, ec);
+  ASSERT_FALSE(ec);
+  std::filesystem::create_directory(test_dir, ec);
+  ASSERT_FALSE(ec);
+
+  auto cleanup = MakeDBDirCleanup(test_dir);
+  auto storage = std::make_unique<engine::Storage>(&config);
+  auto s = storage->Open();
+  ASSERT_TRUE(s.IsOK()) << s.Msg();
+
+  auto ctx = engine::Context(storage.get());
+  rocksdb::WriteBatch batch;
+  batch.Put("k", "v");
+  ASSERT_TRUE(storage->Write(ctx, rocksdb::WriteOptions(), &batch).ok());
+
+  std::string files;
+  s = engine::Storage::ReplDataManager::GetFullReplDataInfo(storage.get(), &files);
+  ASSERT_TRUE(s.IsOK()) << s.Msg();
+
+  // The configured idle time is far away, so the checkpoint should not be purged
+  // even though no replica is fetching its files.
+  s = storage->TryPurgeCheckpoint(/*fetch_file_threads=*/0);
+  ASSERT_TRUE(s.IsOK()) << s.Msg();
+  EXPECT_TRUE(storage->ExistCheckpoint());
+
+  // Simulate that the checkpoint was created long before the configured alive time,
+  // it should be purged even if replicas are still fetching its files.
+  storage->SetCheckpointCreateTimeSecs(util::GetTimeStamp<std::chrono::seconds>() - 10);
+
+  s = storage->TryPurgeCheckpoint(/*fetch_file_threads=*/1);
+  ASSERT_TRUE(s.IsOK()) << s.Msg();
+  EXPECT_FALSE(storage->ExistCheckpoint());
+  EXPECT_EQ(storage->GetCheckpointCreateTimeSecs(), 0);
+  EXPECT_EQ(storage->GetCheckpointAccessTimeSecs(), 0);
+}
+
 TEST(Storage, ReplDataManagerFileExistsCRC) {
   std::error_code ec;
   const std::string test_dir = "test_repl_crc_check";