fix: skip full sync during partial sync (#5580)

Signed-off-by: kostas <[email protected]>

Author: kostasrim

.pre-commit-config.yaml

@@ -31,8 +31,3 @@ repos:
     rev: 25.1.0
     hooks:
       - id: black
-
-  - repo: https://github.com/gitleaks/gitleaks
-    rev: v8.16.3
-    hooks:
-      - id: gitleaks

src/server/dflycmd.cc

@@ -273,8 +273,9 @@ void DflyCmd::Flow(CmdArgList args, RedisReplyBuilder* rb, ConnectionContext* cn
   {
     util::fb2::LockGuard lk{replica_ptr->shared_mu};
 
-    if (replica_ptr->replica_state != SyncState::PREPARATION)
+    if (replica_ptr->replica_state != SyncState::PREPARATION) {
       return rb->SendError(kInvalidState);
+    }
 
     // Set meta info on connection.
     cntx->conn()->SetName(absl::StrCat("repl_flow_", sync_id));
@@ -380,8 +381,20 @@ void DflyCmd::StartStable(CmdArgList args, Transaction* tx, RedisReplyBuilder* r
     return;
 
   util::fb2::LockGuard lk{replica_ptr->shared_mu};
-  if (!CheckReplicaStateOrReply(*replica_ptr, SyncState::FULL_SYNC, rb))
+  auto repl_state = replica_ptr->replica_state;
+  if (repl_state != SyncState::FULL_SYNC && repl_state != SyncState::PREPARATION) {
+    rb->SendError(kInvalidState);
     return;
+  }
+
+  // Check all flows are connected.
+  // This might happen if a flow abruptly disconnected before sending the SYNC request.
+  for (const FlowInfo& flow : replica_ptr->flows) {
+    if (!flow.conn) {
+      rb->SendError(kInvalidState);
+      return;
+    }
+  }
 
   {
     Transaction::Guard tg{tx};
@@ -390,10 +403,15 @@ void DflyCmd::StartStable(CmdArgList args, Transaction* tx, RedisReplyBuilder* r
     auto cb = [this, &status, replica_ptr = replica_ptr](EngineShard* shard) {
       FlowInfo* flow = &replica_ptr->flows[shard->shard_id()];
 
-      status = StopFullSyncInThread(flow, &replica_ptr->exec_st, shard);
-      if (*status != OpStatus::OK) {
-        return;
+      // We are doing partial sync. We never started FullSync so we don't need to stop it.
+      bool is_partial = flow->start_partial_sync_at.has_value();
+      if (!is_partial) {
+        status = StopFullSyncInThread(flow, &replica_ptr->exec_st, shard);
+        if (*status != OpStatus::OK) {
+          return;
+        }
       }
+
       StartStableSyncInThread(flow, &replica_ptr->exec_st, shard);
     };
     shard_set->RunBlockingInParallel(std::move(cb));
@@ -665,10 +683,7 @@ OpStatus DflyCmd::StartFullSyncInThread(FlowInfo* flow, ExecutionState* exec_st,
     return OpStatus::CANCELLED;
   }
 
-  if (flow->start_partial_sync_at.has_value())
-    saver->StartIncrementalSnapshotInShard(*flow->start_partial_sync_at, exec_st, shard);
-  else
-    saver->StartSnapshotInShard(true, exec_st, shard);
+  saver->StartSnapshotInShard(true, exec_st, shard);
 
   return OpStatus::OK;
 }
@@ -700,8 +715,10 @@ void DflyCmd::StartStableSyncInThread(FlowInfo* flow, ExecutionState* exec_st, E
   DCHECK(shard);
   DCHECK(flow->conn);
 
-  flow->streamer.reset(
-      new JournalStreamer(sf_->journal(), exec_st, JournalStreamer::SendLsn::YES, true));
+  LSN partial_lsn = flow->start_partial_sync_at.value_or(0);
+  JournalStreamer::Config config{
+      .should_sent_lsn = true, .init_from_stable_sync = true, .start_partial_sync_at = partial_lsn};
+  flow->streamer.reset(new JournalStreamer(sf_->journal(), exec_st, config));
   flow->streamer->Start(flow->conn->socket());
 
   // Register cleanup.

src/server/journal/streamer.cc

@@ -109,9 +109,9 @@ void LogTcpSocketDiagnostics(util::FiberSocketBase* dest) {
 
 }  // namespace
 
-JournalStreamer::JournalStreamer(journal::Journal* journal, ExecutionState* cntx, SendLsn send_lsn,
-                                 bool is_stable_sync)
-    : cntx_(cntx), journal_(journal), is_stable_sync_(is_stable_sync), send_lsn_(send_lsn) {
+JournalStreamer::JournalStreamer(journal::Journal* journal, ExecutionState* cntx,
+                                 JournalStreamer::Config config)
+    : cntx_(cntx), journal_(journal), config_(config) {
   // cache the flag to avoid accessing it later.
   replication_stream_output_limit_cached = absl::GetFlag(FLAGS_replication_stream_output_limit);
   migration_buckets_sleep_usec_cached = absl::GetFlag(FLAGS_migration_buckets_sleep_usec);
@@ -136,7 +136,7 @@ void JournalStreamer::ConsumeJournalChange(const JournalChangeItem& item) {
   time_t now = time(nullptr);
   last_lsn_writen_ = item.journal_item.lsn;
   // TODO: to chain it to the previous Write call.
-  if (send_lsn_ == SendLsn::YES && now - last_lsn_time_ > 3) {
+  if (config_.should_sent_lsn && now - last_lsn_time_ > 3) {
     last_lsn_time_ = now;
     io::StringSink sink;
     JournalWriter writer(&sink);
@@ -148,14 +148,19 @@ void JournalStreamer::ConsumeJournalChange(const JournalChangeItem& item) {
 void JournalStreamer::Start(util::FiberSocketBase* dest) {
   CHECK(dest_ == nullptr && dest != nullptr);
   dest_ = dest;
-  journal_cb_id_ = journal_->RegisterOnChange(this);
+  // For partial sync we first catch up from journal replication buffer and only then register.
+  if (config_.start_partial_sync_at == 0) {
+    journal_cb_id_ = journal_->RegisterOnChange(this);
+  }
   StartStalledDataWriterFiber();
 }
 
 void JournalStreamer::Cancel() {
   VLOG(1) << "JournalStreamer::Cancel " << cntx_->IsCancelled();
   waker_.notifyAll();
-  journal_->UnregisterOnChange(journal_cb_id_);
+  if (journal_cb_id_) {
+    journal_->UnregisterOnChange(journal_cb_id_);
+  }
   StopStalledDataWriterFiber();
   WaitForInflightToComplete();
 }
@@ -182,7 +187,7 @@ void JournalStreamer::Write(std::string str) {
 }
 
 void JournalStreamer::StartStalledDataWriterFiber() {
-  if (is_stable_sync_ && !stalled_data_writer_.IsJoinable()) {
+  if (config_.init_from_stable_sync && !stalled_data_writer_.IsJoinable()) {
     auto pb = fb2::ProactorBase::me();
     std::chrono::milliseconds period_us(stalled_writer_base_period_ms);
     stalled_data_writer_ = MakeFiber([this, index = pb->GetPoolIndex(), period_us]() mutable {
@@ -192,8 +197,55 @@ void JournalStreamer::StartStalledDataWriterFiber() {
   }
 }
 
+bool JournalStreamer::MaybePartialStreamLSNs() {
+  // Same algorithm as SwitchIncrementalFb. The only difference is that we don't sent
+  // the old LSN"s via a snapshot but rather as journal changes.
+  if (config_.start_partial_sync_at > 0) {
+    LSN lsn = config_.start_partial_sync_at;
+    DCHECK_LE(lsn, journal_->GetLsn()) << "The replica tried to sync from the future.";
+
+    LOG(INFO) << "Starting partial sync from lsn: " << lsn;
+    // The replica sends the LSN of the next entry is wants to receive.
+    while (cntx_->IsRunning() && journal_->IsLSNInBuffer(lsn)) {
+      JournalChangeItem item;
+      item.journal_item.data = journal_->GetEntry(lsn);
+      item.journal_item.lsn = lsn;
+      ConsumeJournalChange(item);
+      lsn++;
+    }
+
+    if (!cntx_->IsRunning()) {
+      return false;
+    }
+
+    if (journal_->GetLsn() != lsn) {
+      // We stopped but we didn't manage to send the whole stream.
+      cntx_->ReportError(
+          std::make_error_code(errc::state_not_recoverable),
+          absl::StrCat("Partial sync was unsuccessful because entry #", lsn,
+                       " was dropped from the buffer. Current lsn=", journal_->GetLsn()));
+      return false;
+    }
+
+    // We are done, register back to the journal so we don't miss any changes
+    journal_cb_id_ = journal_->RegisterOnChange(this);
+
+    LOG(INFO) << "Last LSN sent in partial sync was " << (lsn - 1);
+    // flush pending
+    if (pending_buf_.Size() != 0) {
+      AsyncWrite(true);
+    }
+  }
+  return true;
+}
+
 void JournalStreamer::StalledDataWriterFiber(std::chrono::milliseconds period_ms,
                                              util::fb2::Done* waiter) {
+  if (!MaybePartialStreamLSNs()) {
+    // Either context got cancelled, or partial sync failed because the lsn's stalled.
+    return;
+  }
+
   while (cntx_->IsRunning()) {
     if (waiter->WaitFor(period_ms)) {
       if (!cntx_->IsRunning()) {
@@ -222,7 +274,7 @@ void JournalStreamer::AsyncWrite(bool force_send) {
 
   // Writing in stable sync and outside of fiber needs to check
   // threshold before writing data.
-  if (is_stable_sync_ && !force_send &&
+  if (config_.init_from_stable_sync && !force_send &&
       pending_buf_.FrontBufSize() < replication_dispatch_threshold) {
     return;
   }
@@ -332,7 +384,7 @@ void JournalStreamer::WaitForInflightToComplete() {
 }
 
 void JournalStreamer::StopStalledDataWriterFiber() {
-  if (is_stable_sync_ && stalled_data_writer_.IsJoinable()) {
+  if (config_.init_from_stable_sync && stalled_data_writer_.IsJoinable()) {
     stalled_data_writer_done_.Notify();
     if (stalled_data_writer_.IsJoinable()) {
       stalled_data_writer_.Join();
@@ -346,9 +398,7 @@ bool JournalStreamer::IsStalled() const {
 
 RestoreStreamer::RestoreStreamer(DbSlice* slice, cluster::SlotSet slots, journal::Journal* journal,
                                  ExecutionState* cntx)
-    : JournalStreamer(journal, cntx, JournalStreamer::SendLsn::NO, false),
-      db_slice_(slice),
-      my_slots_(std::move(slots)) {
+    : JournalStreamer(journal, cntx, {}), db_slice_(slice), my_slots_(std::move(slots)) {
   DCHECK(slice != nullptr);
   migration_buckets_serialization_threshold_cached =
       absl::GetFlag(FLAGS_migration_buckets_serialization_threshold);

src/server/journal/streamer.h

@@ -16,9 +16,14 @@ namespace dfly {
 // journal listener and writes them to a destination sink in a separate fiber.
 class JournalStreamer : public journal::JournalConsumerInterface {
  public:
-  enum class SendLsn : uint8_t { NO = 0, YES = 1 };
-  JournalStreamer(journal::Journal* journal, ExecutionState* cntx, SendLsn send_lsn,
-                  bool is_stable_sync);
+  struct Config {
+    bool should_sent_lsn = false;
+    bool init_from_stable_sync = false;
+    LSN start_partial_sync_at = 0;
+  };
+
+  JournalStreamer(journal::Journal* journal, ExecutionState* cntx, Config config);
+
   virtual ~JournalStreamer();
 
   // Self referential.
@@ -69,6 +74,10 @@ class JournalStreamer : public journal::JournalConsumerInterface {
   PendingBuf pending_buf_;
 
  private:
+  // Return true if all lsn's from config_.start_partial_sync_at were sent (or if started from 0).
+  // Return false if not all lsn's were sent (stalled) in time. Cancels the context with error.
+  bool MaybePartialStreamLSNs();
+
   void AsyncWrite(bool force_send);
   void OnCompletion(std::error_code ec, size_t len);
 
@@ -82,16 +91,15 @@ class JournalStreamer : public journal::JournalConsumerInterface {
   void StopStalledDataWriterFiber();
   void StalledDataWriterFiber(std::chrono::milliseconds period_ms, util::fb2::Done* waiter);
 
+  const Config config_;
   // If we are replication in stable sync we can aggregate data before sending
-  bool is_stable_sync_;
   size_t in_flight_bytes_ = 0, total_sent_ = 0;
   // Last time that send data in milliseconds
   uint64_t last_async_write_time_ = 0;
   time_t last_lsn_time_ = 0;
   LSN last_lsn_writen_ = 0;
   util::fb2::EventCount waker_;
   uint32_t journal_cb_id_{0};
-  SendLsn send_lsn_;
 };
 
 class CmdSerializer;

src/server/rdb_save.cc

@@ -1066,7 +1066,6 @@ class RdbSaver::Impl final : public SliceSnapshot::SnapshotDataConsumerInterface
   ~Impl();
 
   void StartSnapshotting(bool stream_journal, ExecutionState* cntx, EngineShard* shard);
-  void StartIncrementalSnapshotting(LSN start_lsn, ExecutionState* cntx, EngineShard* shard);
 
   void StopSnapshotting(EngineShard* shard);
   void WaitForSnapshottingFinish(EngineShard* shard);
@@ -1244,15 +1243,6 @@ void RdbSaver::Impl::StartSnapshotting(bool stream_journal, ExecutionState* cntx
   s->Start(stream_journal, allow_flush);
 }
 
-void RdbSaver::Impl::StartIncrementalSnapshotting(LSN start_lsn, ExecutionState* cntx,
-                                                  EngineShard* shard) {
-  auto& db_slice = namespaces->GetDefaultNamespace().GetDbSlice(shard->shard_id());
-  auto& s = GetSnapshot(shard);
-
-  s = CreateSliceSnapshot(shard, &db_slice, cntx);
-  s->StartIncremental(start_lsn);
-}
-
 SnapshotPtr RdbSaver::Impl::CreateSliceSnapshot(EngineShard* shard, DbSlice* db_slice,
                                                 ExecutionState* cntx) {
   return SnapshotPtr(new SliceSnapshot(compression_mode_, db_slice, this, cntx),
@@ -1459,11 +1449,6 @@ void RdbSaver::StartSnapshotInShard(bool stream_journal, ExecutionState* cntx, E
   impl_->StartSnapshotting(stream_journal, cntx, shard);
 }
 
-void RdbSaver::StartIncrementalSnapshotInShard(LSN start_lsn, ExecutionState* cntx,
-                                               EngineShard* shard) {
-  impl_->StartIncrementalSnapshotting(start_lsn, cntx, shard);
-}
-
 error_code RdbSaver::WaitSnapshotInShard(EngineShard* shard) {
   impl_->WaitForSnapshottingFinish(shard);
   return SaveEpilog();

src/server/rdb_save.h

@@ -95,9 +95,6 @@ class RdbSaver {
   // cll allows breaking in the middle.
   void StartSnapshotInShard(bool stream_journal, ExecutionState* cntx, EngineShard* shard);
 
-  // Send only the incremental snapshot since start_lsn.
-  void StartIncrementalSnapshotInShard(LSN start_lsn, ExecutionState* cntx, EngineShard* shard);
-
   // Stops full-sync serialization for replication in the shard's thread.
   std::error_code StopFullSyncInShard(EngineShard* shard);