fix: partial replication data loss

src/server/cluster/incoming_slot_migration.cc

@@ -149,7 +149,9 @@ class ClusterShardMigration {
     }
 
     if (!tx_data.IsGlobalCmd()) {
-      return executor_.Execute(tx_data.dbid, tx_data.command);
+      auto res = executor_.Execute(tx_data.dbid, tx_data.command);
+      return res == facade::DispatchResult::OOM ? make_error_code(errc::not_enough_memory)
+                                                : error_code();
     } else {
       // TODO check which global commands should be supported
       std::string error =

src/server/journal/executor.cc

@@ -11,6 +11,7 @@
 #include <memory>
 
 #include "base/logging.h"
+#include "facade/service_interface.h"
 #include "server/main_service.h"
 
 using namespace std;
@@ -48,33 +49,24 @@ JournalExecutor::JournalExecutor(Service* service)
 JournalExecutor::~JournalExecutor() {
 }
 
-void JournalExecutor::Execute(DbIndex dbid, absl::Span<journal::ParsedEntry::CmdData> cmds) {
-  SelectDb(dbid);
-  for (auto& cmd : cmds) {
-    Execute(cmd);
-  }
-}
-
-std::error_code JournalExecutor::Execute(DbIndex dbid, journal::ParsedEntry::CmdData& cmd) {
+facade::DispatchResult JournalExecutor::Execute(DbIndex dbid, journal::ParsedEntry::CmdData& cmd) {
   SelectDb(dbid);
   return Execute(cmd);
 }
 
 void JournalExecutor::FlushAll() {
   auto cmd = BuildFromParts("FLUSHALL");
-  Execute(cmd);
+  std::ignore = Execute(cmd);
 }
 
 void JournalExecutor::FlushSlots(const cluster::SlotRange& slot_range) {
   auto cmd = BuildFromParts("DFLYCLUSTER", "FLUSHSLOTS", slot_range.start, slot_range.end);
-  Execute(cmd);
+  std::ignore = Execute(cmd);
 }
 
-std::error_code JournalExecutor::Execute(journal::ParsedEntry::CmdData& cmd) {
+facade::DispatchResult JournalExecutor::Execute(journal::ParsedEntry::CmdData& cmd) {
   auto span = CmdArgList{cmd.cmd_args.data(), cmd.cmd_args.size()};
-  auto res = service_->DispatchCommand(span, &reply_builder_, &conn_context_);
-  return res == facade::DispatchResult::OOM ? make_error_code(errc::not_enough_memory)
-                                            : error_code();
+  return service_->DispatchCommand(span, &reply_builder_, &conn_context_);
 }
 
 void JournalExecutor::SelectDb(DbIndex dbid) {
@@ -83,7 +75,7 @@ void JournalExecutor::SelectDb(DbIndex dbid) {
 
   if (!ensured_dbs_[dbid]) {
     auto cmd = BuildFromParts("SELECT", dbid);
-    Execute(cmd);
+    std::ignore = Execute(cmd);
     ensured_dbs_[dbid] = true;
 
     // TODO: This is a temporary fix for #4146.

src/server/journal/executor.h

@@ -7,6 +7,7 @@
 #include <absl/types/span.h>
 
 #include "facade/reply_capture.h"
+#include "facade/service_interface.h"
 #include "server/cluster/cluster_defs.h"
 #include "server/journal/types.h"
 
@@ -22,8 +23,8 @@ class JournalExecutor {
 
   JournalExecutor(JournalExecutor&&) = delete;
 
-  void Execute(DbIndex dbid, absl::Span<journal::ParsedEntry::CmdData> cmds);
-  std::error_code Execute(DbIndex dbid, journal::ParsedEntry::CmdData& cmd);
+  // Return true is command executed without error
+  facade::DispatchResult Execute(DbIndex dbid, journal::ParsedEntry::CmdData& cmd);
 
   void FlushAll();  // Execute FLUSHALL.
   void FlushSlots(const cluster::SlotRange& slot_range);
@@ -33,7 +34,7 @@ class JournalExecutor {
   }
 
  private:
-  std::error_code Execute(journal::ParsedEntry::CmdData& cmd);
+  facade::DispatchResult Execute(journal::ParsedEntry::CmdData& cmd);
 
   // Select database. Ensure it exists if accessed for first time.
   void SelectDb(DbIndex dbid);

src/server/replica.cc

@@ -6,6 +6,7 @@
 #include <chrono>
 
 #include "absl/strings/match.h"
+#include "facade/service_interface.h"
 
 extern "C" {
 #include "redis/rdb.h"
@@ -913,8 +914,17 @@ void DflyShardReplica::StableSyncDflyReadFb(ExecutionState* cntx) {
       force_ping_ = true;
       journal_rec_executed_.fetch_add(1, std::memory_order_relaxed);
     } else {
-      ExecuteTx(std::move(*tx_data), cntx);
-      journal_rec_executed_.fetch_add(1, std::memory_order_relaxed);
+      auto exe_res = ExecuteTx(std::move(*tx_data), cntx);
+      if (exe_res) {
+        // We only increment upon successful execution of the transaction.
+        // The reason for this is that during partial sync we sent this
+        // number as the lsn number to resume from. However, if for example
+        // we increment this when a command fails (because the context
+        // got cancelled, e.g, replication connection broke), we will get
+        // inconsistent data because the replica will resume from the next
+        // lsn of the master and this lsn entry will be lost.
+        journal_rec_executed_.fetch_add(1, std::memory_order_relaxed);
+      }
     }
     shard_replica_waker_.notifyAll();
   }
@@ -994,15 +1004,14 @@ DflyShardReplica::~DflyShardReplica() {
   JoinFlow();
 }
 
-void DflyShardReplica::ExecuteTx(TransactionData&& tx_data, ExecutionState* cntx) {
+bool DflyShardReplica::ExecuteTx(TransactionData&& tx_data, ExecutionState* cntx) {
   if (!cntx->IsRunning()) {
-    return;
+    return false;
   }
 
   if (!tx_data.IsGlobalCmd()) {
     VLOG(3) << "Execute cmd without sync between shards. txid: " << tx_data.txid;
-    executor_->Execute(tx_data.dbid, tx_data.command);
-    return;
+    return executor_->Execute(tx_data.dbid, tx_data.command) == facade::DispatchResult::OK;
   }
 
   bool inserted_by_me =
@@ -1017,26 +1026,27 @@ void DflyShardReplica::ExecuteTx(TransactionData&& tx_data, ExecutionState* cntx
   multi_shard_data.block->Wait();
   // Check if we woke up due to cancellation.
   if (!exec_st_.IsRunning())
-    return;
+    return false;
   VLOG(2) << "Execute txid: " << tx_data.txid << " block wait finished";
 
   VLOG(2) << "Execute txid: " << tx_data.txid << " global command execution";
   // Wait until all shards flows get to execution step of this transaction.
   multi_shard_data.barrier.Wait();
   // Check if we woke up due to cancellation.
   if (!exec_st_.IsRunning())
-    return;
+    return false;
   // Global command will be executed only from one flow fiber. This ensure corectness of data in
   // replica.
+  bool execution_res = false;
   if (inserted_by_me) {
-    executor_->Execute(tx_data.dbid, tx_data.command);
+    execution_res = executor_->Execute(tx_data.dbid, tx_data.command) == facade::DispatchResult::OK;
   }
   // Wait until exection is done, to make sure we done execute next commands while the global is
   // executed.
   multi_shard_data.barrier.Wait();
   // Check if we woke up due to cancellation.
   if (!exec_st_.IsRunning())
-    return;
+    return false;
 
   // Erase from map can be done only after all flow fibers executed the transaction commands.
   // The last fiber which will decrease the counter to 0 will be the one to erase the data from
@@ -1046,6 +1056,7 @@ void DflyShardReplica::ExecuteTx(TransactionData&& tx_data, ExecutionState* cntx
   if (val == 1) {
     multi_shard_exe_->Erase(tx_data.txid);
   }
+  return inserted_by_me ? execution_res : true;
 }
 
 error_code Replica::ParseReplicationHeader(base::IoBuf* io_buf, PSyncResponse* dest) {

src/server/replica.h

@@ -207,7 +207,7 @@ class DflyShardReplica : public ProtocolClient {
 
   void StableSyncDflyAcksFb(ExecutionState* cntx);
 
-  void ExecuteTx(TransactionData&& tx_data, ExecutionState* cntx);
+  bool ExecuteTx(TransactionData&& tx_data, ExecutionState* cntx);
 
   uint32_t FlowId() const;