chore: Add extra debug log during replication

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

Author: adiholden

src/server/engine_shard.cc

@@ -369,7 +369,8 @@ static void RunFPeriodically(std::function<void()> f, std::chrono::milliseconds
 
     int64_t now_ms = fb2::ProactorBase::GetMonotonicTimeNs() / 1000000;
     if (now_ms - 5 * period_ms.count() > last_heartbeat_ms) {
-      VLOG(1) << "This " << error_msg << " step took " << now_ms - last_heartbeat_ms << "ms";
+      VLOG(1) << "This " << error_msg << " step was stalled for " << now_ms - last_heartbeat_ms
+              << "ms";
     }
     f();
     last_heartbeat_ms = fb2::ProactorBase::GetMonotonicTimeNs() / 1000000;

src/server/journal/streamer.cc

@@ -293,6 +293,7 @@ void JournalStreamer::AsyncWrite(bool force_send) {
     v[i] = IoVec(io::Bytes(uptr, cur_buf.buf[i].size()));
   }
 
+  DVLOG(3) << "calling AsyncWrite with buff size:" << v.size();
   dest_->AsyncWrite(v.data(), v.size(),
                     [this, len = in_flight_bytes_](std::error_code ec) { OnCompletion(ec, len); });
 }

src/server/replica.cc

@@ -984,7 +984,7 @@ void DflyShardReplica::StableSyncDflyReadFb(ExecutionState* cntx) {
   acks_fb_ = fb2::Fiber("shard_acks", &DflyShardReplica::StableSyncDflyAcksFb, this, cntx);
   std::optional<TransactionData> tx_data;
   while ((tx_data = tx_reader.NextTxData(&reader, cntx))) {
-    DVLOG(3) << "Lsn: " << tx_data->lsn;
+    DVLOG(3) << "Lsn: " << tx_data->lsn << " flowid: " << flow_id_;
 
     last_io_time_ = Proactor()->GetMonotonicTimeNs();
     if (tx_data->opcode == journal::Op::LSN) {
@@ -1060,7 +1060,8 @@ void DflyShardReplica::StableSyncDflyAcksFb(ExecutionState* cntx) {
     // Handle ACKs with the master. PING opcodes from the master mean we should immediately
     // answer.
     current_offset = journal_rec_executed_.load(std::memory_order_relaxed);
-    VLOG(1) << "Sending an ACK with offset=" << current_offset << " forced=" << force_ping_;
+    VLOG(1) << "Sending an ACK with offset=" << current_offset << " forced=" << force_ping_
+            << " flowid=" << flow_id_;
     ack_cmd = absl::StrCat("REPLCONF ACK ", current_offset);
     force_ping_ = false;
     next_ack_tp = std::chrono::steady_clock::now() + ack_time_max_interval;

tests/dragonfly/replication_test.py

@@ -1308,9 +1308,7 @@ async def delayed_takeover():
 
 @pytest.mark.exclude_epoll
 @pytest.mark.parametrize("master_threads, replica_threads", take_over_cases)
-async def test_take_over_seeder(
-    request, df_factory, df_seeder_factory, master_threads, replica_threads
-):
+async def test_take_over_seeder(df_factory, df_seeder_factory, master_threads, replica_threads):
     master = df_factory.create(
         proactor_threads=master_threads, dbfilename=f"dump_{tmp_file_name()}", admin_port=ADMIN_PORT
     )