bug server: fix lns mismatch in replication

Signed-off-by: adi_holden <adi@dragonflydb.io>

Author: adiholden

src/server/snapshot.cc

@@ -131,6 +131,7 @@ void SliceSnapshot::FinalizeJournalStream(bool cancel) {
   journal->UnregisterOnChange(cb_id);
   if (!cancel) {
     // always succeeds because serializer_ flushes to string.
+    VLOG(1) << "FinalizeJournalStream lsn: " << journal->GetLsn();
     std::ignore = serializer_->SendJournalOffset(journal->GetLsn());
     PushSerialized(true);
   }

src/server/transaction.cc

@@ -149,18 +149,20 @@ cv_status Transaction::BatonBarrier::Wait(time_point tp) {
 
 Transaction::Guard::Guard(Transaction* tx) : tx(tx) {
   DCHECK(tx->cid_->opt_mask() & CO::GLOBAL_TRANS);
-  tx->Execute([](auto*, auto*) { return OpStatus::OK; }, false);
-  shard_set->RunBriefInParallel([](EngineShard* shard) {
+  auto cb = [&](Transaction* t, EngineShard* shard) {
     namespaces->GetDefaultNamespace().GetDbSlice(shard->shard_id()).SetExpireAllowed(false);
-  });
+    return OpStatus::OK;
+  };
+  tx->Execute(cb, false);
 }
 
 Transaction::Guard::~Guard() {
-  tx->Conclude();
-  tx->Refurbish();
-  shard_set->RunBriefInParallel([](EngineShard* shard) {
+  auto cb = [&](Transaction* t, EngineShard* shard) {
     namespaces->GetDefaultNamespace().GetDbSlice(shard->shard_id()).SetExpireAllowed(true);
-  });
+    return OpStatus::OK;
+  };
+  tx->Execute(cb, true);
+  tx->Refurbish();
 }
 
 void Transaction::Init(unsigned num_shards) {

tests/dragonfly/replication_test.py

@@ -2963,7 +2963,11 @@ async def test_bug_in_json_memory_tracking(df_factory: DflyInstanceFactory):
     """
     This test reproduces a bug in the JSON memory tracking.
     """
-    master = df_factory.create(proactor_threads=2, serialization_max_chunk_size=1)
+    master = df_factory.create(
+        proactor_threads=2,
+        serialization_max_chunk_size=1,
+        vmodule="replica=2,dflycmd=2,snapshot=1,rdb_save=1,rdb_load=1,journal_slice=2",
+    )
     replicas = [df_factory.create(proactor_threads=2) for i in range(2)]
 
     # Start instances and connect clients