fix: data race when doing a snapshot in cache mode (#5336)

The issue - during the snapshotting
we tried to fetch the same delayed entry which lead to data-races and deadlocks.
The fix - to pull the entry into a stack before preempting,
ensuring each fiber handles its own entry.

Fixes #4965

Signed-off-by: Roman Gershman <[email protected]>

Author: romange

src/server/snapshot.cc

@@ -104,6 +104,7 @@ void SliceSnapshot::Start(bool stream_journal, SnapshotFlush allow_flush) {
 }
 
 void SliceSnapshot::StartIncremental(LSN start_lsn) {
+  VLOG(1) << "StartIncremental: " << start_lsn;
   serializer_ = std::make_unique<RdbSerializer>(compression_mode_);
 
   snapshot_fb_ = fb2::Fiber("incremental_snapshot",
@@ -319,7 +320,6 @@ size_t SliceSnapshot::FlushSerialized(SerializerBase::FlushState flush_state) {
   uint64_t running_cycles = ThisFiber::GetRunningTimeCycles();
 
   fb2::NoOpLock lk;
-
   // We create a critical section here that ensures that records are pushed in sequential order.
   // As a result, it is not possible for two fiber producers to push concurrently.
   // If A.id = 5, and then B.id = 6, and both are blocked here, it means that last_pushed_id_ < 4.
@@ -356,7 +356,10 @@ bool SliceSnapshot::PushSerialized(bool force) {
     // Async bucket serialization might have accumulated some delayed values.
     // Because we can finally block in this function, we'll await and serialize them
     do {
-      auto& entry = delayed_entries_.back();
+      // We may call PushSerialized from multiple fibers concurrently, so we need to
+      // ensure that we are not serializing the same entry concurrently.
+      DelayedEntry entry = std::move(delayed_entries_.back());
+      delayed_entries_.pop_back();
 
       // TODO: https://github.com/dragonflydb/dragonfly/issues/4654
       // there are a few problems with how we serialize external values.
@@ -367,7 +370,6 @@ bool SliceSnapshot::PushSerialized(bool force) {
 
       // TODO: to introduce RdbSerializer::SaveString that can accept a string value directly.
       serializer_->SaveEntry(entry.key, pv, entry.expire, entry.mc_flags, entry.dbid);
-      delayed_entries_.pop_back();
     } while (!delayed_entries_.empty());
 
     // blocking point.

src/server/snapshot.h

@@ -68,7 +68,7 @@ class SliceSnapshot : public journal::JournalConsumerInterface {
 
   // Initialize snapshot, start bucket iteration fiber, register listeners.
   // In journal streaming mode it needs to be stopped by either Stop or Cancel.
-  enum class SnapshotFlush { kAllow, kDisallow };
+  enum class SnapshotFlush : uint8_t { kAllow, kDisallow };
 
   void Start(bool stream_journal, SnapshotFlush allow_flush = SnapshotFlush::kDisallow);
 

tests/dragonfly/replication_test.py

@@ -26,14 +26,6 @@
 M_NOT_EPOLL = [pytest.mark.exclude_epoll]
 
 
-async def wait_for_replicas_state(*clients, state="online", node_role="slave", timeout=0.05):
-    """Wait until all clients (replicas) reach passed state"""
-    while len(clients) > 0:
-        await asyncio.sleep(timeout)
-        roles = await asyncio.gather(*(c.role() for c in clients))
-        clients = [c for c, role in zip(clients, roles) if role[0] != node_role or role[3] != state]
-
-
 """
 Test full replication pipeline. Test full sync with streaming changes and stable state streaming.
 """

tests/dragonfly/tiering_test.py

@@ -1,14 +1,15 @@
 import async_timeout
 import asyncio
 import itertools
+import logging
 import pytest
 import random
 import redis.asyncio as aioredis
 
 from . import dfly_args
 from .seeder import DebugPopulateSeeder
-from .utility import info_tick_timer
-
+from .utility import info_tick_timer, wait_for_replicas_state
+from .instance import DflyInstanceFactory
 
 BASIC_ARGS = {"port": 6379, "proactor_threads": 4, "tiered_prefix": "/tmp/tiered/backing"}
 
@@ -87,3 +88,35 @@ async def run(sub_ops):
     res = await p.execute()
 
     assert res == [10 * k for k in key_range]
+
+
+@pytest.mark.exclude_epoll
+@pytest.mark.opt_only
+@dfly_args(
+    {
+        "proactor_threads": 2,
+        "tiered_prefix": "/tmp/tiered/backing_master",
+        "maxmemory": "4G",
+        "cache_mode": True,
+        "tiered_offload_threshold": "0.2",
+        "tiered_storage_write_depth": 100,
+    }
+)
+async def test_full_sync(async_client: aioredis.Redis, df_factory: DflyInstanceFactory):
+    replica = df_factory.create(
+        proactor_threads=2,
+        cache_mode=True,
+        maxmemory="4G",
+        tiered_prefix="/tmp/tiered/backing_replica",
+        tiered_offload_threshold="0.2",
+        tiered_storage_write_depth=1000,
+    )
+    replica.start()
+    replica_client = replica.client()
+    await async_client.execute_command("debug", "populate", "3000000", "key", "2000")
+    await replica_client.replicaof(
+        "localhost", async_client.connection_pool.connection_kwargs["port"]
+    )
+    logging.info("Waiting for replica to sync")
+    async with async_timeout.timeout(120):
+        await wait_for_replicas_state(replica_client)

tests/dragonfly/utility.py

@@ -848,3 +848,11 @@ def extract_int_after_prefix(prefix, line):
     match = re.search(prefix + "(\\d+)", line)
     assert match
     return int(match.group(1))
+
+
+async def wait_for_replicas_state(*clients, state="online", node_role="slave", timeout=0.05):
+    """Wait until all clients (replicas) reach passed state"""
+    while len(clients) > 0:
+        await asyncio.sleep(timeout)
+        roles = await asyncio.gather(*(c.role() for c in clients))
+        clients = [c for c, role in zip(clients, roles) if role[0] != node_role or role[3] != state]