MB-59518: Stop PassiveStream from processing snapshot against active VB

jimwwalker · jimwwalker · commit b82526acc604 · 2023-11-21T12:52:37.000Z
Add VBucket state checks to processMarker so that the marker is not processed against an active vbucket. A unit test is updated to set the marker type as DISK. In this case and without a fix an exception would occur from within setDuplicatePrepareWindow() because it tries to get the PDM from an active VB. Change-Id: I084ec690697f9c411f80ecdb9394626294a01a41 Reviewed-on: https://review.couchbase.org/c/kv_engine/+/200569 Well-Formed: Restriction Checker Reviewed-by: Paolo Cocchi <paolo.cocchi@couchbase.com> Tested-by: Jim Walker <jim@couchbase.com>
diff --git a/engines/ep/src/dcp/passive_stream.cc b/engines/ep/src/dcp/passive_stream.cc
@@ -1134,6 +1134,15 @@ cb::engine_errc PassiveStream::processDropScope(
 
 void PassiveStream::processMarker(SnapshotMarker* marker) {
     VBucketPtr vb = engine->getVBucket(vb_);
+    if (!vb) {
+        return;
+    }
+
+    // Vbucket must be in a permitted state to apply the snapshot
+    folly::SharedMutex::ReadHolder rlh(vb->getStateLock());
+    if (!permittedVBStates.test(vb->getState())) {
+        return;
+    }
 
     cur_snapshot_start.store(marker->getStartSeqno());
     cur_snapshot_end.store(marker->getEndSeqno());
@@ -1142,117 +1151,113 @@ void PassiveStream::processMarker(SnapshotMarker* marker) {
                                     ? Snapshot::Disk
                                     : Snapshot::Memory);
 
-    if (vb) {
-        auto checkpointType = marker->getFlags() & MARKER_FLAG_DISK
-                                      ? CheckpointType::Disk
-                                      : CheckpointType::Memory;
+    auto checkpointType = marker->getFlags() & MARKER_FLAG_DISK
+                                  ? CheckpointType::Disk
+                                  : CheckpointType::Memory;
 
-        const auto historical = marker->getFlags() & MARKER_FLAG_HISTORY
-                                        ? CheckpointHistorical::Yes
-                                        : CheckpointHistorical::No;
+    const auto historical = marker->getFlags() & MARKER_FLAG_HISTORY
+                                    ? CheckpointHistorical::Yes
+                                    : CheckpointHistorical::No;
 
-        // Check whether the snapshot can be considered as an initial disk
-        // checkpoint for the replica.
-        if (checkpointType == CheckpointType::Disk && vb->getHighSeqno() == 0) {
-            checkpointType = CheckpointType::InitialDisk;
-        }
+    // Check whether the snapshot can be considered as an initial disk
+    // checkpoint for the replica.
+    if (checkpointType == CheckpointType::Disk && vb->getHighSeqno() == 0) {
+        checkpointType = CheckpointType::InitialDisk;
+    }
 
-        auto& ckptMgr = *vb->checkpointManager;
-
-        std::optional<uint64_t> hcs = marker->getHighCompletedSeqno();
-        if ((marker->getFlags() & MARKER_FLAG_DISK) &&
-            !supportsSyncReplication) {
-            // If this stream doesn't support SyncReplication (i.e. the producer
-            // is a pre-MadHatter version) then we should consider the HCS to be
-            // present but zero for disk snapshot (not possible for any
-            // SyncWrites to have completed yet). If SyncReplication is
-            // supported then use the value from the marker.
-            hcs = 0;
-        }
+    auto& ckptMgr = *vb->checkpointManager;
 
-        if (marker->getFlags() & MARKER_FLAG_DISK && !hcs) {
-            const auto msg = fmt::format(
-                    "PassiveStream::processMarker: stream:{} {}, flags:{}, "
-                    "flagsDecoded:{}, snapStart:{}, snapEnd:{}, HCS:{} - "
-                    "missing HCS",
-                    name_,
-                    vb_,
-                    marker->getFlags(),
-                    dcpMarkerFlagsToString(marker->getFlags()),
-                    marker->getStartSeqno(),
-                    marker->getEndSeqno(),
-                    to_string_or_none(hcs));
-            throw std::logic_error(msg);
-        }
+    std::optional<uint64_t> hcs = marker->getHighCompletedSeqno();
+    if ((marker->getFlags() & MARKER_FLAG_DISK) && !supportsSyncReplication) {
+        // If this stream doesn't support SyncReplication (i.e. the producer
+        // is a pre-MadHatter version) then we should consider the HCS to be
+        // present but zero for disk snapshot (not possible for any
+        // SyncWrites to have completed yet). If SyncReplication is
+        // supported then use the value from the marker.
+        hcs = 0;
+    }
 
-        if (marker->getFlags() & MARKER_FLAG_DISK) {
-            // A replica could receive a duplicate DCP prepare during a disk
-            // snapshot if it had previously received an uncompleted prepare.
-            // We can receive a disk snapshot when we either:
-            //     a) First connect
-            //     b) Get cursor dropped by the active
-            //
-            // We selectively allow these prepares to overwrite the old one by
-            // setting a duplicate prepare window in the vBucket. This will
-            // allow any currently outstanding prepares to be overwritten, but
-            // not any new ones.
-            vb->setDuplicatePrepareWindow();
-        }
+    if (marker->getFlags() & MARKER_FLAG_DISK && !hcs) {
+        const auto msg = fmt::format(
+                "PassiveStream::processMarker: stream:{} {}, flags:{}, "
+                "flagsDecoded:{}, snapStart:{}, snapEnd:{}, HCS:{} - "
+                "missing HCS",
+                name_,
+                vb_,
+                marker->getFlags(),
+                dcpMarkerFlagsToString(marker->getFlags()),
+                marker->getStartSeqno(),
+                marker->getEndSeqno(),
+                to_string_or_none(hcs));
+        throw std::logic_error(msg);
+    }
 
-        // We could be connected to a non sync-repl, so if the max-visible is
-        // not transmitted (optional is false), set visible to snap-end
-        auto visibleSeq =
-                marker->getMaxVisibleSeqno().value_or(marker->getEndSeqno());
+    if (marker->getFlags() & MARKER_FLAG_DISK) {
+        // A replica could receive a duplicate DCP prepare during a disk
+        // snapshot if it had previously received an uncompleted prepare.
+        // We can receive a disk snapshot when we either:
+        //     a) First connect
+        //     b) Get cursor dropped by the active
+        //
+        // We selectively allow these prepares to overwrite the old one by
+        // setting a duplicate prepare window in the vBucket. This will
+        // allow any currently outstanding prepares to be overwritten, but
+        // not any new ones.
+        vb->setDuplicatePrepareWindow();
+    }
 
-        if (checkpointType == CheckpointType::InitialDisk) {
-            // Case: receiving the first snapshot in a Disk snapshot.
-            // Note that replica may never execute here as the active may switch
-            // directly to in-memory and send the first snapshot in a Memory
-            // snapshot.
+    // We could be connected to a non sync-repl, so if the max-visible is
+    // not transmitted (optional is false), set visible to snap-end
+    auto visibleSeq =
+            marker->getMaxVisibleSeqno().value_or(marker->getEndSeqno());
+    if (checkpointType == CheckpointType::InitialDisk) {
+        // Case: receiving the first snapshot in a Disk snapshot.
+        // Note that replica may never execute here as the active may switch
+        // directly to in-memory and send the first snapshot in a Memory
+        // snapshot.
+
+        vb->setReceivingInitialDiskSnapshot(true);
+        ckptMgr.createSnapshot(cur_snapshot_start.load(),
+                               cur_snapshot_end.load(),
+                               hcs,
+                               checkpointType,
+                               visibleSeq,
+                               historical);
+    } else {
+        // Case: receiving any type of snapshot (Disk/Memory).
 
-            vb->setReceivingInitialDiskSnapshot(true);
+        if (marker->getFlags() & MARKER_FLAG_CHK) {
             ckptMgr.createSnapshot(cur_snapshot_start.load(),
                                    cur_snapshot_end.load(),
                                    hcs,
                                    checkpointType,
                                    visibleSeq,
                                    historical);
         } else {
-            // Case: receiving any type of snapshot (Disk/Memory).
-
-            if (marker->getFlags() & MARKER_FLAG_CHK) {
+            // MB-42780: In general we cannot merge multiple snapshots into
+            // the same checkpoint. The only exception is for when replica
+            // receives multiple Memory checkpoints in a row.
+            // Since 6.5.0 the Active behaves correctly with regard to that
+            // (ie, the Active always sets the MARKER_FLAG_CHK in a snapshot
+            // transition tha involves Disk snapshots), but older Producers
+            // may still miss the MARKER_FLAG_CHK.
+            if (prevSnapType == Snapshot::Memory &&
+                cur_snapshot_type == Snapshot::Memory) {
+                ckptMgr.extendOpenCheckpoint(cur_snapshot_end.load(),
+                                             visibleSeq);
+            } else {
                 ckptMgr.createSnapshot(cur_snapshot_start.load(),
                                        cur_snapshot_end.load(),
                                        hcs,
                                        checkpointType,
                                        visibleSeq,
                                        historical);
-            } else {
-                // MB-42780: In general we cannot merge multiple snapshots into
-                // the same checkpoint. The only exception is for when replica
-                // receives multiple Memory checkpoints in a row.
-                // Since 6.5.0 the Active behaves correctly with regard to that
-                // (ie, the Active always sets the MARKER_FLAG_CHK in a snapshot
-                // transition tha involves Disk snapshots), but older Producers
-                // may still miss the MARKER_FLAG_CHK.
-                if (prevSnapType == Snapshot::Memory &&
-                    cur_snapshot_type == Snapshot::Memory) {
-                    ckptMgr.extendOpenCheckpoint(cur_snapshot_end.load(),
-                                                 visibleSeq);
-                } else {
-                    ckptMgr.createSnapshot(cur_snapshot_start.load(),
-                                           cur_snapshot_end.load(),
-                                           hcs,
-                                           checkpointType,
-                                           visibleSeq,
-                                           historical);
-                }
             }
         }
+    }
 
-        if (marker->getFlags() & MARKER_FLAG_ACK) {
-            cur_snapshot_ack = true;
-        }
+    if (marker->getFlags() & MARKER_FLAG_ACK) {
+        cur_snapshot_ack = true;
     }
 }
 
diff --git a/engines/ep/tests/module_tests/dcp_durability_stream_test.cc b/engines/ep/tests/module_tests/dcp_durability_stream_test.cc
@@ -4585,9 +4585,16 @@ void DurabilityPassiveStreamPersistentTest::replicaToActiveBufferedResolution(
 
     auto vb = engine->getVBucket(vbid);
 
-    // Buffer the commit/abort of the prepare
+    // Buffer a snapshot with commit/abort of the prepare.
+    // Note to detect a bug when processing a buffered snapshot marker, if we
+    // set the type as DISK it would hit an exception when calling
+    // setDuplicatePrepareWindow.
     EXPECT_EQ(cb::engine_errc::success,
-              snapshot(*consumer, stream->getOpaque(), 3, 3));
+              snapshot(*consumer,
+                       stream->getOpaque(),
+                       3,
+                       3,
+                       MARKER_FLAG_DISK | MARKER_FLAG_CHK));
     if (resolutionIsCommit) {
         EXPECT_EQ(cb::engine_errc::success,
                   commit(*consumer, stream->getOpaque(), durableKey, 2, 3));
diff --git a/engines/ep/tests/module_tests/evp_store_single_threaded_test.cc b/engines/ep/tests/module_tests/evp_store_single_threaded_test.cc
@@ -1068,14 +1068,9 @@ cb::engine_errc STParameterizedBucketTest::addItem(Item& itm,
 cb::engine_errc STParameterizedBucketTest::snapshot(DcpConsumer& consumer,
                                                     uint32_t opaque,
                                                     uint64_t start,
-                                                    uint64_t end) {
-    return consumer.snapshotMarker(opaque,
-                                   vbid,
-                                   start,
-                                   end,
-                                   MARKER_FLAG_MEMORY | MARKER_FLAG_CHK,
-                                   0,
-                                   0);
+                                                    uint64_t end,
+                                                    uint32_t flags) {
+    return consumer.snapshotMarker(opaque, vbid, start, end, flags, 0, end);
 }
 
 cb::engine_errc STParameterizedBucketTest::mutation(DcpConsumer& consumer,
diff --git a/engines/ep/tests/module_tests/evp_store_single_threaded_test.h b/engines/ep/tests/module_tests/evp_store_single_threaded_test.h
@@ -832,7 +832,9 @@ class STParameterizedBucketTest
     cb::engine_errc snapshot(DcpConsumer& consumer,
                              uint32_t opaque,
                              uint64_t start,
-                             uint64_t end);
+                             uint64_t end,
+                             uint32_t flags = MARKER_FLAG_MEMORY |
+                                              MARKER_FLAG_CHK);
     cb::engine_errc mutation(DcpConsumer& consumer,
                              uint32_t opaque,
                              const DocKey& key,
