MB-42780: Make replica resilient to missing MARKER_FLAG_CHK

Since 6.5.0 the Active ensures that Disk snapshots are never merged into
an existing checkpoint. That is achieve by setting the MARKER_FLAG_CHK
in the SnapshotMarker sent to Replica.

That all works fine as long as Active behaves as expected.
In pre-6.5.0 we had issues where Active could miss the MARKER_FLAG_CHK
in markers. In the 6.0.1->6.6.1 upgrade scenario seen in MB-42780, one
of the effect is replica failure (with consequent rebalance/upgrade
failure).

With this patch, replica enforces the same logic that we had already
added at active in 6.5.0. That makes replica resilient to any DCP
Producer that misses to set the MARKER_FLAG_CHK properly.

Change-Id: I9b32dec1caa5b262f0cb524317624e4e955481b2
Reviewed-on: http://review.couchbase.org/c/kv_engine/+/140626
Well-Formed: Build Bot <[email protected]>
Reviewed-by: Dave Rigby <[email protected]>
Tested-by: Build Bot <[email protected]>

Author: paolococchi

engines/ep/src/dcp/passive_stream.cc

@@ -924,6 +924,7 @@ void PassiveStream::processMarker(SnapshotMarker* marker) {
 
     cur_snapshot_start.store(marker->getStartSeqno());
     cur_snapshot_end.store(marker->getEndSeqno());
+    const auto prevSnapType = cur_snapshot_type.load();
     cur_snapshot_type.store((marker->getFlags() & MARKER_FLAG_DISK)
                                     ? Snapshot::Disk
                                     : Snapshot::Memory);
@@ -964,14 +965,26 @@ void PassiveStream::processMarker(SnapshotMarker* marker) {
         // not transmitted (optional is false), set visible to snap-end
         auto visibleSeq =
                 marker->getMaxVisibleSeqno().value_or(marker->getEndSeqno());
+
         if (marker->getFlags() & MARKER_FLAG_DISK && vb->getHighSeqno() == 0) {
+            // 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);
         } else {
+            // Case: receiving any type of snapshot (Disk/Memory).
+
+            // @todo: The check on CheckpointId=0 is legacy from when that could
+            //   really happen (TAP). Currently CheckpointId is a positive
+            //   monotonic sequence starting from 1, so the check can be
+            //   removed. Deferring to a dedicated change.
             if (marker->getFlags() & MARKER_FLAG_CHK ||
                 vb->checkpointManager->getOpenCheckpointId() == 0) {
                 ckptMgr.createSnapshot(cur_snapshot_start.load(),
@@ -980,11 +993,24 @@ void PassiveStream::processMarker(SnapshotMarker* marker) {
                                        checkpointType,
                                        visibleSeq);
             } else {
-                // If we are reconnecting then we need to update the snap end
-                // and potentially the checkpoint type as We do not send the
-                // CHK snapshot marker flag for disk snapshots.
-                ckptMgr.updateCurrentSnapshot(
-                        cur_snapshot_end.load(), visibleSeq, checkpointType);
+                // 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.
+                if (prevSnapType == Snapshot::Memory &&
+                    cur_snapshot_type == Snapshot::Memory) {
+                    // If we are reconnecting then we need to update the snap
+                    // end and potentially the checkpoint type as We do not send
+                    // the CHK snapshot marker flag for disk snapshots.
+                    ckptMgr.updateCurrentSnapshot(cur_snapshot_end.load(),
+                                                  visibleSeq,
+                                                  checkpointType);
+                } else {
+                    ckptMgr.createSnapshot(cur_snapshot_start.load(),
+                                           cur_snapshot_end.load(),
+                                           hcs,
+                                           checkpointType,
+                                           visibleSeq);
+                }
             }
         }
 
@@ -1035,8 +1061,6 @@ void PassiveStream::handleSnapshotEnd(VBucketPtr& vb, uint64_t byseqno) {
             vb->notifyPassiveDMOfSnapEndReceived(byseqno);
             cur_snapshot_prepare.store(false);
         }
-
-        cur_snapshot_type.store(Snapshot::None);
     }
 }
 

engines/ep/src/dcp/stream.h

@@ -44,9 +44,9 @@ enum backfill_source_t {
 class Stream {
 public:
     enum class Snapshot {
-           None,
-           Disk,
-           Memory
+        None, // Used only at PassiveStream initialization
+        Disk,
+        Memory
     };
 
     Stream(const std::string& name,

engines/ep/tests/mock/mock_checkpoint_manager.h

@@ -114,4 +114,9 @@ class MockCheckpointManager : public CheckpointManager {
         LockHolder lh(queueLock);
         return getOpenCheckpoint_UNLOCKED(lh).getCheckpointType();
     }
+
+    auto getPersistenceCursorPos() const {
+        LockHolder lh(queueLock);
+        return getPersistenceCursor()->currentPos;
+    }
 };

engines/ep/tests/module_tests/checkpoint_test.cc

@@ -1394,14 +1394,18 @@ void SingleThreadedCheckpointTest::closeReplicaCheckpointOnMemorySnapshotEnd(
     store->deleteVBucket(vb->getId(), cookie);
 }
 
+// MB-42780: Test disabled as already marked as "could validly be removed now"
+// above + the test is now legally failing due to changes in the checkpoint-list
 TEST_F(SingleThreadedCheckpointTest,
-       CloseReplicaCheckpointOnMemorySnapshotEnd_HighMemDisk) {
+       DISABLED_CloseReplicaCheckpointOnMemorySnapshotEnd_HighMemDisk) {
     closeReplicaCheckpointOnMemorySnapshotEnd(
             true, dcp_marker_flag_t::MARKER_FLAG_DISK);
 }
 
+// MB-42780: Test disabled as already marked as "could validly be removed now"
+// above + the test is now legally failing due to changes in the checkpoint-list
 TEST_F(SingleThreadedCheckpointTest,
-       CloseReplicaCheckpointOnMemorySnapshotEnd_Disk) {
+       DISABLED_CloseReplicaCheckpointOnMemorySnapshotEnd_Disk) {
     closeReplicaCheckpointOnMemorySnapshotEnd(
             false, dcp_marker_flag_t::MARKER_FLAG_DISK);
 }

engines/ep/tests/module_tests/dcp_durability_stream_test.cc

@@ -1559,7 +1559,7 @@ void DurabilityPassiveStreamTest::
                           streamStartSeqno /*snapStart*/,
                           streamStartSeqno /*snapEnd*/,
                           dcp_marker_flag_t::MARKER_FLAG_DISK,
-                          {} /*HCS*/,
+                          0 /*HCS*/,
                           {} /*maxVisibleSeqno*/,
                           {} /*streamId*/);
     stream->processMarker(&marker);
@@ -1677,7 +1677,7 @@ TEST_P(DurabilityPassiveStreamTest,
                             1 /*snapStart*/,
                             ~1 /*snapEnd*/,
                             dcp_marker_flag_t::MARKER_FLAG_DISK,
-                            {} /*HCS*/,
+                            0 /*HCS*/,
                             {} /*maxVisibleSeqno*/,
                             {} /*streamId*/);
     stream->processMarker(&marker);