fix(server): fix serialization on move with big values (#5390)

* fix server: fix serialization on move with big values

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

Author: adiholden

src/server/snapshot.cc

@@ -446,6 +446,7 @@ bool SliceSnapshot::IsPositionSerialized(DbIndex id, PrimeTable::Cursor cursor)
 }
 
 void SliceSnapshot::OnMoved(DbIndex id, const DbSlice::MovedItemsVec& items) {
+  std::lock_guard barrier(big_value_mu_);
   DCHECK(!use_snapshot_version_);
   for (const auto& item_cursors : items) {
     // If item was moved from a bucket that was serialized to a bucket that was not serialized

tests/dragonfly/replication_test.py

@@ -3201,7 +3201,7 @@ async def test_bug_5221(df_factory):
 
 @pytest.mark.parametrize("proactors", [1, 4, 6])
 @pytest.mark.parametrize("backlog_len", [1, 256, 1024, 1300])
-async def test_partial_sync(df_factory, df_seeder_factory, proactors, backlog_len):
+async def test_partial_sync(df_factory, proactors, backlog_len):
     keys = 5_000
     if proactors > 1:
         keys = 10_000
@@ -3335,8 +3335,11 @@ async def state_transitioned_stable(
 
 
 @pytest.mark.slow
-async def test_replicaiton_onmove_flow(df_factory):
-    master = df_factory.create(proactor_threads=2, cache_mode=True)
+@pytest.mark.parametrize("serialization_max_size", [1, 64000])
+async def test_replication_onmove_flow(df_factory, serialization_max_size):
+    master = df_factory.create(
+        proactor_threads=2, cache_mode=True, serialization_max_chunk_size=serialization_max_size
+    )
     replica = df_factory.create(proactor_threads=2)
 
     df_factory.start_all([master, replica])
@@ -3345,7 +3348,7 @@ async def test_replicaiton_onmove_flow(df_factory):
 
     key_target = 100000
     # Fill master with test data
-    await c_master.execute_command(f"DEBUG POPULATE {key_target} key 1048 RAND")
+    await c_master.execute_command(f"DEBUG POPULATE {key_target} key 32 RAND TYPE hash ELEMENTS 10")
     logging.debug("finished populate")
 
     stop_event = asyncio.Event()
@@ -3355,7 +3358,7 @@ async def get_keys():
             pipe = c_master.pipeline(transaction=False)
             for _ in range(50):
                 id = random.randint(0, key_target)
-                pipe.get(f"key:{id}")
+                pipe.hlen(f"key:{id}")
             await pipe.execute()
 
     get_task = asyncio.create_task(get_keys())