fix(replication): use correct replica pointer in takeover catchup loop (#6068)

* fix(replication): use correct replica pointer in takeover catchup loop

* fix: review comments

Author: vyavdoshenko

src/server/dflycmd.cc

@@ -567,8 +567,8 @@ void DflyCmd::TakeOver(CmdArgList args, RedisReplyBuilder* rb, ConnectionContext
         continue;
       }
 
-      auto cb = [&, end_time](EngineShard* shard) {
-        if (!WaitReplicaFlowToCatchup(end_time, replica_ptr.get(), shard)) {
+      auto cb = [repl_ptr = repl_ptr, end_time, &rest_catchup_success](EngineShard* shard) {
+        if (!WaitReplicaFlowToCatchup(end_time, repl_ptr.get(), shard)) {
           rest_catchup_success.store(false);
         }
       };

tests/dragonfly/replication_test.py

@@ -3562,3 +3562,44 @@ async def get_memory(client, field):
     line = lines[0]
     peak_bytes = extract_int_after_prefix("Serialization peak bytes: ", line)
     assert peak_bytes < value_size
+
+
+@pytest.mark.slow
+async def test_takeover_bug_wrong_replica_checked_in_logs(df_factory):
+    master = df_factory.create(proactor_threads=4, vmodule="dflycmd=1")
+    replicas = [df_factory.create(proactor_threads=2) for _ in range(3)]
+    df_factory.start_all([master] + replicas)
+
+    c_master = master.client()
+    clients = [r.client() for r in replicas]
+
+    # Connect all replicas
+    for c in clients:
+        await c.execute_command(f"REPLICAOF localhost {master.port}")
+    await asyncio.gather(*[wait_available_async(c) for c in clients])
+
+    # Disconnect replica[1] to create lag
+    await clients[1].execute_command("REPLICAOF NO ONE")
+
+    # Write data that replica[1] will miss
+    pipe = c_master.pipeline()
+    for i in range(10000):
+        pipe.set(f"k{i}", "x" * 100)
+    await pipe.execute()
+
+    # Reconnect replica[1] and immediately takeover from replica[0]
+    await clients[1].execute_command(f"REPLICAOF localhost {master.port}")
+    try:
+        await clients[0].execute_command("REPLTAKEOVER 1")
+    except Exception:
+        pass
+
+    # Check master logs
+    master.stop(kill=False)
+    timeout_logs = master.find_in_logs("Couldn't synchronize with replica")
+
+    if timeout_logs:
+        for log in timeout_logs:
+            assert (
+                str(replicas[0].port) not in log
+            ), f"BUG: Checked initiating replica {replicas[0].port} instead of others"