bug server: fix lns mismatch in replication

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

Author: adiholden

src/server/db_slice.cc

@@ -1167,7 +1167,6 @@ DbSlice::PrimeItAndExp DbSlice::ExpireIfNeeded(const Context& cntx, PrimeIterato
   }
 
   auto& db = db_arr_[cntx.db_index];
-
   auto expire_it = db->expire.Find(it->first);
 
   if (IsValid(expire_it)) {
@@ -1184,6 +1183,9 @@ DbSlice::PrimeItAndExp DbSlice::ExpireIfNeeded(const Context& cntx, PrimeIterato
                 << ", prime table size: " << db->prime.size() << util::fb2::GetStacktrace();
   }
 
+  DCHECK(shard_owner()->shard_lock()->Check(IntentLock::Mode::EXCLUSIVE))
+      << util::fb2::GetStacktrace();
+
   string scratch;
   string_view key = it->first.GetSlice(&scratch);
 

src/server/transaction.cc

@@ -150,11 +150,17 @@ 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) {
+    namespaces->GetDefaultNamespace().GetDbSlice(shard->shard_id()).SetExpireAllowed(false);
+  });
 }
 
 Transaction::Guard::~Guard() {
   tx->Conclude();
   tx->Refurbish();
+  shard_set->RunBriefInParallel([](EngineShard* shard) {
+    namespaces->GetDefaultNamespace().GetDbSlice(shard->shard_id()).SetExpireAllowed(true);
+  });
 }
 
 void Transaction::Init(unsigned num_shards) {

tests/dragonfly/replication_test.py

@@ -2982,6 +2982,7 @@ async def test_bug_in_json_memory_tracking(df_factory: DflyInstanceFactory):
 
     seeder = SeederV2(key_target=50_000)
     fill_task = asyncio.create_task(seeder.run(master.client()))
+    await asyncio.sleep(0.2)
 
     for replica in c_replicas:
         await replica.execute_command(f"REPLICAOF LOCALHOST {master.port}")