MB-48368: Generate correct seqno advance for in-memory stream

Another case missed in ActiveStream::processItems when a checkpoint
contains only a prepare/abort for the stream's collection.

This is fixed with an extra check in path when the output mutations
is empty - we now detect that the cursor visited a non-visible
item for the stream and generate an empty snapshot to catch the
client up to the collection high-seqno (so they can no be happy
they have a complete 'copy' of the collection).

Testing is enhanced in this patch to be much more data-driven
and simpler - targetting just ActiveStream::processItems with
different sequences of inputs and checking the outputs.
Without the fixes in ActiveStream a number of the new tests
detect problems with the stream's responses.

Change-Id: Id0fb4e83e8be00cbf1ffc4db4fccdf6dfd959abb
Reviewed-on: http://review.couchbase.org/c/kv_engine/+/163468
Tested-by: Build Bot <build@couchbase.com>
Reviewed-by: Dave Rigby <daver@couchbase.com>

Author: jimwwalker

engines/ep/src/dcp/active_stream.cc

@@ -289,13 +289,10 @@ bool ActiveStream::markDiskSnapshot(uint64_t startSeqno,
             }
         }
 
-        /* We need to send the requested 'snap_start_seqno_' as the snapshot
+        /* We may need to send the requested 'snap_start_seqno_' as the snapshot
            start when we are sending the first snapshot because the first
            snapshot could be resumption of a previous snapshot */
-        if (!firstMarkerSent) {
-            startSeqno = std::min(snap_start_seqno_, startSeqno);
-            firstMarkerSent = true;
-        }
+        startSeqno = adjustStartIfFirstSnapshot(startSeqno);
 
         VBucketPtr vb = engine->getVBucket(vb_);
         if (!vb) {
@@ -1279,55 +1276,53 @@ void ActiveStream::processItems(
                      outstandingItemsResult.highCompletedSeqno,
                      visibleSeqno,
                      highNonVisibleSeqno);
-        } else if (!firstMarkerSent && isSeqnoAdvancedEnabled() &&
-                   lastReadSeqno < snap_end_seqno_) {
-            // MB-47009: This first snapshot has been completely filtered away.
-            // The remaining items must not of been for this client. We must
-            // still send a snapshot marker so that the client is moved to their
-            // end seqno - so a snapshot + seqno advance is needed.
-            firstMarkerSent = true;
-
+        } else if (isSeqnoAdvancedEnabled()) {
             // Note that we cannot enter this case if supportSyncReplication()
             // returns true (see isSeqnoAdvancedEnabled). This means that we
             // do not need to set the HCS/MVS or timestamp parameters of the
             // snapshot marker. MB-47877 tracks enabling sync-writes+filtering
-            pushToReadyQ(std::make_unique<SnapshotMarker>(opaque_,
-                                                          vb_,
-                                                          snap_start_seqno_,
-                                                          snap_end_seqno_,
-                                                          MARKER_FLAG_MEMORY,
-                                                          std::nullopt,
-                                                          std::nullopt,
-                                                          std::nullopt,
-                                                          sid));
-
-            lastSentSnapEndSeqno.store(snap_end_seqno_,
-                                       std::memory_order_relaxed);
-            nextSnapshotIsCheckpoint = false;
-
-            queueSeqnoAdvanced();
-        } else if (isSeqnoAdvancedEnabled() &&
-                   isSeqnoGapAtEndOfSnapshot(curChkSeqno)) {
-            auto vb = engine->getVBucket(getVBucket());
-            if (vb) {
-                if (vb->getState() == vbucket_state_replica) {
-                    /*
-                     * If this is a collection stream and we're not sending any
-                     * mutations from memory and we haven't queued a snapshot
-                     * shot and we're a replica. Then our snapshot covers
-                     * backfill and in memory. So we have one snapshot marker
-                     * for both items on disk and in memory. Thus, we need to
-                     * send a SeqnoAdvanced to push the consumer's seqno to the
-                     * end of the snapshot. This is need when no items for the
-                     * collection we're streaming are present in memory.
-                     */
-                    queueSeqnoAdvanced();
+            if (!firstMarkerSent && lastReadSeqno < snap_end_seqno_) {
+                // MB-47009: This first snapshot has been completely filtered
+                // away. The remaining items must not of been for this client.
+                // We must still send a snapshot marker so that the client is
+                // moved to their end seqno - so a snapshot + seqno advance is
+                // needed.
+                sendSnapshotAndSeqnoAdvanced(
+                        outstandingItemsResult.checkpointType,
+                        snap_start_seqno_,
+                        snap_end_seqno_);
+                firstMarkerSent = true;
+            } else if (isSeqnoGapAtEndOfSnapshot(curChkSeqno)) {
+                auto vb = engine->getVBucket(getVBucket());
+                if (vb) {
+                    if (vb->getState() == vbucket_state_replica) {
+                        /*
+                         * If this is a collection stream and we're not sending
+                         * any mutations from memory and we haven't queued a
+                         * snapshot and we're a replica. Then our snapshot
+                         * covers backfill and in memory. So we have one
+                         * snapshot marker for both items on disk and in memory.
+                         * Thus, we need to send a SeqnoAdvanced to push the
+                         * consumer's seqno to the end of the snapshot. This is
+                         * needed when no items for the collection we're
+                         * streaming are present in memory.
+                         */
+                        queueSeqnoAdvanced();
+                    }
+                } else {
+                    log(spdlog::level::level_enum::warn,
+                        "{} processItems() for vbucket which does not "
+                        "exist",
+                        logPrefix);
                 }
-            } else {
-                log(spdlog::level::level_enum::warn,
-                    "{} processItems() for vbucket which does not "
-                    "exist",
-                    logPrefix);
+            } else if (highNonVisibleSeqno &&
+                       curChkSeqno >= highNonVisibleSeqno.value()) {
+                // MB-48368: Nothing directly available for the stream, but a
+                // non-visible item was available - bring the client up-to-date
+                sendSnapshotAndSeqnoAdvanced(
+                        outstandingItemsResult.checkpointType,
+                        highNonVisibleSeqno.value(),
+                        highNonVisibleSeqno.value());
             }
         }
     }
@@ -2223,3 +2218,35 @@ bool ActiveStream::isSeqnoGapAtEndOfSnapshot(uint64_t streamSeqno) const {
     return (lastSentSnapEndSeqno.load() > lastReadSeqno.load()) &&
            lastSentSnapEndSeqno.load() == streamSeqno;
 }
+
+void ActiveStream::sendSnapshotAndSeqnoAdvanced(CheckpointType checkpointType,
+                                                uint64_t start,
+                                                uint64_t end) {
+    start = adjustStartIfFirstSnapshot(start);
+
+    const auto isCkptTypeDisk = checkpointType == CheckpointType::Disk;
+    uint32_t flags = isCkptTypeDisk ? MARKER_FLAG_DISK : MARKER_FLAG_MEMORY;
+
+    pushToReadyQ(std::make_unique<SnapshotMarker>(opaque_,
+                                                  vb_,
+                                                  start,
+                                                  end,
+                                                  flags,
+                                                  std::nullopt,
+                                                  std::nullopt,
+                                                  std::nullopt,
+                                                  sid));
+
+    lastSentSnapEndSeqno.store(end, std::memory_order_relaxed);
+    nextSnapshotIsCheckpoint = false;
+
+    queueSeqnoAdvanced();
+}
+
+uint64_t ActiveStream::adjustStartIfFirstSnapshot(uint64_t start) {
+    if (!firstMarkerSent) {
+        firstMarkerSent = true;
+        return std::min(snap_start_seqno_, start);
+    }
+    return start;
+}

engines/ep/src/dcp/active_stream.h

@@ -669,6 +669,26 @@ class ActiveStream : public Stream,
      */
     void queueSeqnoAdvanced();
 
+    /**
+     * Enqueue a single snapshot + seqno advance
+     * @param checkpointType type of checkpoint which triggered this snapshot
+     * @param start value of snapshot start
+     * @param end value of snapshot end
+     */
+    void sendSnapshotAndSeqnoAdvanced(CheckpointType checkpointType,
+                                      uint64_t start,
+                                      uint64_t end);
+
+    /**
+     * If firstMarkerSent is false then the startSeqno of a snapshot may need
+     * adjusting to match the snap_start_seqno the caller used when creating
+     * the stream.
+     * If firstMarkerSent is false this call will set it to true.
+     * @param start a seqno we think should be the snapshot start
+     * @return the snapshot start to use
+     */
+    uint64_t adjustStartIfFirstSnapshot(uint64_t start);
+
     /* The last sequence number queued from memory, but is yet to be
        snapshotted and put onto readyQ */
     std::atomic<uint64_t> lastReadSeqnoUnSnapshotted;

engines/ep/tests/CMakeLists.txt

@@ -41,6 +41,7 @@ cb_add_test_executable(ep-engine_ep_unit_tests
         module_tests/collections/collections_dcp_producers.cc
         module_tests/collections/collections_kvstore_test.cc
         module_tests/collections/collections_oso_dcp_test.cc
+        module_tests/collections/collections_seqno_advance.cc
         module_tests/collections/evp_store_collections_dcp_test.cc
         module_tests/collections/evp_store_collections_legacy_dcp_tests.cc
         module_tests/collections/evp_store_durability_collections_dcp_test.h