chore: force shutdown connections on takeover

Signed-off-by: Kostas Kyrimis <[email protected]>

Author: kostasrim

src/server/dflycmd.cc

@@ -14,6 +14,8 @@
 #include <utility>
 
 #include "absl/cleanup/cleanup.h"
+#include "absl/debugging/stacktrace.h"
+#include "absl/flags/internal/flag.h"
 #include "absl/strings/numbers.h"
 #include "base/flags.h"
 #include "base/logging.h"
@@ -40,6 +42,8 @@ ABSL_RETIRED_FLAG(uint32_t, allow_partial_sync_with_lsn_diff, 0,
 ABSL_DECLARE_FLAG(bool, info_replication_valkey_compatible);
 ABSL_DECLARE_FLAG(uint32_t, replication_timeout);
 ABSL_DECLARE_FLAG(uint32_t, shard_repl_backlog_len);
+ABSL_FLAG(bool, experimental_force_takeover, false,
+          "Attempts to force takeover in case of stuck connections");
 
 namespace dfly {
 
@@ -466,6 +470,45 @@ std::optional<LSN> DflyCmd::ParseLsnVec(std::string_view last_master_lsn,
   return {lsn_vec[flow_id]};
 }
 
+void DflyCmd::ForceShutdownStuckConnections(uint64_t timeout) {
+  vector<facade::Connection::WeakRef> conn_refs;
+  auto cb = [&](unsigned thread_index, util::Connection* conn) {
+    facade::Connection* dcon = static_cast<facade::Connection*>(conn);
+    LOG(INFO) << dcon->DebugInfo();
+    // Kill Connection here
+    facade::Connection* dfly_conn = static_cast<facade::Connection*>(conn);
+    using Phase = facade::Connection::Phase;
+    auto phase = dfly_conn->phase();
+    if (dfly_conn->cntx() && dfly_conn->cntx()->replica_conn) {
+      return;
+    }
+
+    bool idle_read = phase == Phase::READ_SOCKET && dfly_conn->idle_time() > timeout;
+
+    bool stuck_sending = dfly_conn->IsSending() && dfly_conn->GetSendWaitTimeSec() > timeout;
+
+    if (idle_read || stuck_sending) {
+      LOG(INFO) << "Connection check: " << dfly_conn->GetClientInfo()
+                << ", phase=" << static_cast<int>(phase) << ", idle_time=" << dfly_conn->idle_time()
+                << ", is_sending=" << dfly_conn->IsSending() << ", idle_read=" << idle_read
+                << ", stuck_sending=" << stuck_sending;
+    }
+    conn_refs.push_back(dfly_conn->Borrow());
+  };
+
+  for (auto* listener : sf_->GetListeners()) {
+    listener->TraverseConnections(cb);
+  }
+
+  VLOG(1) << "Found " << conn_refs.size() << " stucked connections ";
+  for (auto& ref : conn_refs) {
+    facade::Connection* conn = ref.Get();
+    if (conn) {
+      conn->ShutdownSelfBlocking();
+    }
+  }
+}
+
 // DFLY TAKEOVER <timeout_sec> [SAVE] <sync_id>
 // timeout_sec - number of seconds to wait for TAKEOVER to converge.
 // SAVE option is used only by tests.
@@ -506,7 +549,7 @@ void DflyCmd::TakeOver(CmdArgList args, RedisReplyBuilder* rb, ConnectionContext
   LOG(INFO) << "Takeover initiated, locking down the database.";
   absl::Duration timeout_dur = absl::Seconds(timeout);
   absl::Time end_time = absl::Now() + timeout_dur;
-  AggregateStatus status;
+  OpStatus status = OpStatus::OK;
 
   // We need to await for all dispatches to finish: Otherwise a transaction might be scheduled
   // after this function exits but before the actual shutdown.
@@ -520,13 +563,22 @@ void DflyCmd::TakeOver(CmdArgList args, RedisReplyBuilder* rb, ConnectionContext
     LOG(WARNING) << "Couldn't wait for commands to finish dispatching. " << timeout_dur;
     status = OpStatus::TIMED_OUT;
 
-    auto cb = [&](unsigned thread_index, util::Connection* conn) {
-      facade::Connection* dcon = static_cast<facade::Connection*>(conn);
-      LOG(INFO) << dcon->DebugInfo();
-    };
-
-    for (auto* listener : sf_->GetListeners()) {
-      listener->TraverseConnections(cb);
+    // Force takeover on the same duration if flag is set
+    if (absl::GetFlag(FLAGS_experimental_force_takeover)) {
+      ForceShutdownStuckConnections(uint64_t(timeout));
+
+      // Safety net.
+      facade::DispatchTracker tracker{sf_->GetNonPriviligedListeners(), cntx->conn(), false, false};
+      shard_set->pool()->AwaitFiberOnAll([&](unsigned index, auto* pb) {
+        sf_->CancelBlockingOnThread();
+        tracker.TrackOnThread();
+      });
+
+      status = OpStatus::OK;
+      if (!tracker.Wait(timeout_dur)) {
+        LOG(ERROR) << "Could not force execute takeover";
+        status = OpStatus::TIMED_OUT;
+      }
     }
   }
 
@@ -540,10 +592,11 @@ void DflyCmd::TakeOver(CmdArgList args, RedisReplyBuilder* rb, ConnectionContext
   });
 
   atomic_bool catchup_success = true;
-  if (*status == OpStatus::OK) {
+  if (status == OpStatus::OK) {
     dfly::SharedLock lk{replica_ptr->shared_mu};
-    auto cb = [replica_ptr = replica_ptr, end_time, &catchup_success](EngineShard* shard) {
-      if (!WaitReplicaFlowToCatchup(end_time, replica_ptr.get(), shard)) {
+    auto time = end_time + timeout_dur;
+    auto cb = [replica_ptr = replica_ptr, time, &catchup_success](EngineShard* shard) {
+      if (!WaitReplicaFlowToCatchup(time, replica_ptr.get(), shard)) {
         catchup_success.store(false);
       }
     };
@@ -552,8 +605,9 @@ void DflyCmd::TakeOver(CmdArgList args, RedisReplyBuilder* rb, ConnectionContext
 
   VLOG(1) << "WaitReplicaFlowToCatchup done";
 
-  if (*status != OpStatus::OK || !catchup_success.load()) {
+  if (status != OpStatus::OK || !catchup_success.load()) {
     sf_->service().SwitchState(GlobalState::TAKEN_OVER, GlobalState::ACTIVE);
+    LOG(INFO) << status << " " << catchup_success.load() << " " << &status;
     return rb->SendError("Takeover failed!");
   }
 

src/server/dflycmd.h

@@ -183,6 +183,11 @@ class DflyCmd {
   // Switch to stable state replication.
   void StartStable(CmdArgList args, Transaction* tx, RedisReplyBuilder* rb);
 
+  // Helper for takeover flow. Sometimes connections get stuck on send (because of pipelines)
+  // and this causes the takeover flow to fail because checkpoint messages are not processed.
+  // This function force shuts down those connection and allows the node to complete the takeover.
+  void ForceShutdownStuckConnections(uint64_t timeout);
+
   // TAKEOVER <syncid>
   // Shut this master down atomically with replica promotion.
   void TakeOver(CmdArgList args, RedisReplyBuilder* rb, ConnectionContext* cntx);

tests/dragonfly/replication_test.py

@@ -3668,7 +3668,7 @@ async def test_replica_of_self(async_client):
         await async_client.execute_command(f"replicaof 127.0.0.1 {port}")
 
 
-@dfly_args({"proactor_threads": 2})
+@dfly_args({"proactor_threads": 2, "experimental_force_takeover": True})
 async def test_takeover_with_stuck_connections(df_factory: DflyInstanceFactory):
     master = df_factory.create()
     master.start()
@@ -3686,8 +3686,13 @@ async def test_takeover_with_stuck_connections(df_factory: DflyInstanceFactory):
 
     async def get_task():
         while True:
-            writer.write(f"GET a\n".encode())
-            await writer.drain()
+            # Will get killed by takeover because it's stucked
+            try:
+                writer.write(f"GET a\n".encode())
+                await writer.drain()
+            except:
+                return
+
             await asyncio.sleep(0.1)
 
     get = asyncio.create_task(get_task())
@@ -3715,5 +3720,7 @@ async def wait_for_stuck_on_send():
     async with async_timeout.timeout(240):
         await wait_for_replicas_state(replica_cl)
 
-    with pytest.raises(redis.exceptions.ResponseError) as e:
-        await replica_cl.execute_command("REPLTAKEOVER 5")
+    res = await replica_cl.execute_command("REPLTAKEOVER 5")
+    assert res == "OK"
+
+    await get