MB-54729: DCP disk backfill a CDC stream

Backfill of a CDC stream has 3 possible outcomes.

If no history is retained, the backfill behaves no differently than it
would on 7.1. This case can occur because "history_retention_bytes=0"
or the vbucket has explicitly "dropped" any retained history (corner
case of history replication).

If history is retained the backfill can behave in two ways depending on
where the backfill start seqno is relative to
ScanContext::historyStartSeqno.

Case1: start >= ScanContext::historyStartSeqno

The backfill starts inside the retained history range, the backfill
will generate a single DCP snapshot marker followed by the mutations.
The DCP snapshot will state "history" and "may contain duplicates".

Case2: start < ScanContext::historyStartSeqno

The backfill starts below the retained history range. In this case the
disk snapshot is used to generate two DCP snapshots.

The first DCP snapshot is the non-history range, a KVStore::scan from
the requested start upto, but not including
ScanContext::historyStartSeqno.

The second DCP snapshot is the history range, a
KVStore::scanAllVersions from ScanContext::historyStartSeqno to the end
of the disk snapshots sequence index.

In case 2, both DCP snapshot markers represent the entire disk snapshot
That is that they will both state the full disk range and the MVS/HCS
come from the entire range.

For example

The disk seqno index range is from 1 to 20, and this is then split into
two sub-ranges non-history (nh) and history (h)

    nh{1,10} h{11, 20}

If a backfill occurs and the requested stream start is any of 0 to 10
DCP will transmit two ranges using two markers as follows:

snapshot marker 1:
    snapshot-range{start, 20}
    mvs = 20, hcs = 20
    flags = disk | checkpoint

mutations start to 10

snapshot marker 2:
    snapshot range{start, 20}
    mvs = 20, hcs = 20
    flags = disk | checkpoint | history | may_contain_duplicates

mutations 11 to 20

The implementation of the "history" range adds a new optional phase to
the DCP backfill state machine. When the backfill transitions into
backfill_state_scanning the variations of the backfill are checked for.

From here the following paths exist.

No history;

    backfill_state_scanning -> backfill_state_complete

In this case the full snapshot is delivered from backfill_state_scanning
phase.

Only history:
    backfill_state_scanning -> backfill_state_scanning_history_snapshot

In this case the full snapshot is delivered from
backfill_state_scanning_history_snapshot phase. The

The backfill_state_scanning phase has only inspected the ScanContext
and skipped to history.

Two snapshots
    backfill_state_scanning -> backfill_state_scanning_history_snapshot

In this case both backfill phases are delivering "snapshots", but the
same magma snapshot is the source.

Change-Id: I5a6df7ed929d99187a74a071c1d523d904cd6f7e

Author: jimwwalker

engines/ep/src/dcp/active_stream.cc

@@ -279,7 +279,8 @@ bool ActiveStream::markDiskSnapshot(uint64_t startSeqno,
                                     uint64_t endSeqno,
                                     std::optional<uint64_t> highCompletedSeqno,
                                     uint64_t maxVisibleSeqno,
-                                    std::optional<uint64_t> timestamp) {
+                                    std::optional<uint64_t> timestamp,
+                                    SnapshotSource source) {
     {
         std::unique_lock<std::mutex> lh(streamMutex);
 
@@ -322,7 +323,8 @@ bool ActiveStream::markDiskSnapshot(uint64_t startSeqno,
         /* 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 */
-        startSeqno = adjustStartIfFirstSnapshot(startSeqno);
+        startSeqno = adjustStartIfFirstSnapshot(
+                startSeqno, source != SnapshotSource::NoHistoryPrologue);
 
         VBucketPtr vb = engine->getVBucket(vb_);
         if (!vb) {
@@ -361,24 +363,32 @@ bool ActiveStream::markDiskSnapshot(uint64_t startSeqno,
         auto mvsToSend = supportSyncReplication()
                                  ? std::make_optional(maxVisibleSeqno)
                                  : std::nullopt;
+
+        auto flags = MARKER_FLAG_DISK | MARKER_FLAG_CHK;
+
+        if (source == SnapshotSource::History) {
+            flags |= (MARKER_FLAG_HISTORY |
+                      MARKER_FLAG_MAY_CONTAIN_DUPLICATE_KEYS);
+        }
+
         log(spdlog::level::level_enum::info,
             "{} ActiveStream::markDiskSnapshot: Sending disk snapshot with "
-            "start {}, end {}, and high completed {}, max visible {}",
+            "start:{}, end:{}, flags:0x{:x}, hcs:{}, mvs:{}",
             logPrefix,
             startSeqno,
             endSeqno,
+            flags,
             to_string_or_none(hcsToSend),
             to_string_or_none(mvsToSend));
-        pushToReadyQ(std::make_unique<SnapshotMarker>(
-                opaque_,
-                vb_,
-                startSeqno,
-                endSeqno,
-                MARKER_FLAG_DISK | MARKER_FLAG_CHK,
-                hcsToSend,
-                mvsToSend,
-                timestamp,
-                sid));
+        pushToReadyQ(std::make_unique<SnapshotMarker>(opaque_,
+                                                      vb_,
+                                                      startSeqno,
+                                                      endSeqno,
+                                                      flags,
+                                                      hcsToSend,
+                                                      mvsToSend,
+                                                      timestamp,
+                                                      sid));
         lastSentSnapEndSeqno.store(endSeqno, std::memory_order_relaxed);
 
         if (!isDiskOnly()) {
@@ -2343,7 +2353,7 @@ bool ActiveStream::isSeqnoGapAtEndOfSnapshot(uint64_t streamSeqno) const {
 void ActiveStream::sendSnapshotAndSeqnoAdvanced(CheckpointType checkpointType,
                                                 uint64_t start,
                                                 uint64_t end) {
-    start = adjustStartIfFirstSnapshot(start);
+    start = adjustStartIfFirstSnapshot(start, true);
 
     const auto isCkptTypeDisk = isDiskCheckpointType(checkpointType);
     uint32_t flags = isCkptTypeDisk ? MARKER_FLAG_DISK : MARKER_FLAG_MEMORY;
@@ -2364,9 +2374,12 @@ void ActiveStream::sendSnapshotAndSeqnoAdvanced(CheckpointType checkpointType,
     queueSeqnoAdvanced();
 }
 
-uint64_t ActiveStream::adjustStartIfFirstSnapshot(uint64_t start) {
+uint64_t ActiveStream::adjustStartIfFirstSnapshot(uint64_t start,
+                                                  bool isCompleteSnapshot) {
     if (!firstMarkerSent) {
-        firstMarkerSent = true;
+        if (isCompleteSnapshot) {
+            firstMarkerSent = true;
+        }
         return std::min(snap_start_seqno_, start);
     }
     return start;

engines/ep/src/dcp/active_stream.h

@@ -190,6 +190,23 @@ class ActiveStream : public Stream,
 
     void setBackfillRemaining_UNLOCKED(size_t value);
 
+    // The source of the snapshot marker
+    //
+    // History - This is a range which has history, all updates to keys will be
+    //           returned.
+    // NoHistory - A range which does not have history, all keys are the most
+    //             recent updates.
+    // NoHistoryPrologue - This is the NoHistory range from a disk snapshot also
+    //                  also contains History. The backfill will cross from
+    //                  the no-history to history ranges.
+    //
+    // NoHistoryPrologue exists to indicate the case when a disk snapshot
+    // has both History and NoHistory ranges - in this case markDiskSnapshot
+    // for example will get invoked twice by the same source backfill. First
+    // NoHistoryPrologue, second History. This allows ActiveStream to
+    // distinguish from NoHistory which will not transition to History.
+    enum SnapshotSource { History, NoHistory, NoHistoryPrologue };
+
     /**
      * Queues a snapshot marker to be sent - only if there are items in
      * the backfill range which will be sent.
@@ -206,14 +223,16 @@ class ActiveStream : public Stream,
      * @param maxVisibleSeqno seqno of last visible (commit/mutation/system
      * event) item
      * @param timestamp of the disk snapshot (if available)
+     * @param source if the snapshot is a history or non-history snapshot
      * @return If the stream has queued a snapshot marker. If this is false, the
      *         stream determined none of the items in the backfill would be sent
      */
     bool markDiskSnapshot(uint64_t startSeqno,
                           uint64_t endSeqno,
                           std::optional<uint64_t> highCompletedSeqno,
                           uint64_t maxVisibleSeqno,
-                          std::optional<uint64_t> timestamp);
+                          std::optional<uint64_t> timestamp,
+                          SnapshotSource source);
 
     /**
      * Queues a single "Out of Seqno Order" marker with the 'start' flag
@@ -759,9 +778,16 @@ class ActiveStream : public Stream,
      * the stream.
      * If firstMarkerSent is false this call will set it to true.
      * @param start a seqno we think should be the snapshot start
+     * @param isCompleteSnapshot a boolean which was added by the History/CDC
+     *        work. This bool should be true for when the snapshot is not spread
+     *        over a >1 markers - which is what CDC can do when it has to send
+     *        a disk snapshot as NoHistory{a,b} followed by History{c,d}. If
+     *        this bool is true, the stream can state that the first snapshot
+     *        has been fully processed (the marker of the first snapshot).
      * @return the snapshot start to use
      */
-    uint64_t adjustStartIfFirstSnapshot(uint64_t start);
+    uint64_t adjustStartIfFirstSnapshot(uint64_t start,
+                                        bool isCompleteSnapshot);
 
     /* The last sequence number queued from memory, but is yet to be
        snapshotted and put onto readyQ */

engines/ep/src/dcp/backfill_by_id_disk.cc

@@ -93,6 +93,9 @@ backfill_status_t DCPBackfillByIdDisk::create() {
         stream->setDead(cb::mcbp::DcpStreamEndStatus::BackfillFail);
         transitionState(backfill_state_done);
     } else {
+        // Will check if a history scan is required.
+        setupForHistoryScan(*stream, *scanCtx, 0);
+
         bool markerSent = stream->markOSODiskSnapshot(scanCtx->maxSeqno);
         if (markerSent) {
             transitionState(backfill_state_scanning);
@@ -125,11 +128,29 @@ backfill_status_t DCPBackfillByIdDisk::scan() {
         return backfill_success;
     }
 
-    transitionState(backfill_state_completing);
+    if (historyScan) {
+        complete(*stream, false);
+        transitionState(backfill_state_scanning_history_snapshot);
+    } else {
+        transitionState(backfill_state_completing);
+    }
 
     return backfill_success;
 }
 
+void DCPBackfillByIdDisk::complete(ActiveStream& stream, bool cancelled) {
+    stream.completeOSOBackfill(
+            scanCtx->maxSeqno, runtime, scanCtx->diskBytesRead);
+
+    auto severity = cancelled ? spdlog::level::level_enum::info
+                              : spdlog::level::level_enum::debug;
+    stream.log(severity,
+               "({}) Backfill task cid:{} {}",
+               vbid,
+               cid.to_string(),
+               cancelled ? "cancelled" : "finished");
+}
+
 void DCPBackfillByIdDisk::complete(bool cancelled) {
     auto stream = streamPtr.lock();
     if (!stream) {
@@ -143,16 +164,7 @@ void DCPBackfillByIdDisk::complete(bool cancelled) {
         return;
     }
 
-    stream->completeOSOBackfill(
-            scanCtx->maxSeqno, runtime, scanCtx->diskBytesRead);
-
-    auto severity = cancelled ? spdlog::level::level_enum::info
-                              : spdlog::level::level_enum::debug;
-    stream->log(severity,
-                "({}) Backfill task cid:{} {}",
-                vbid,
-                cid.to_string(),
-                cancelled ? "cancelled" : "finished");
+    complete(*stream, cancelled);
 
     transitionState(backfill_state_done);
 }

engines/ep/src/dcp/backfill_by_id_disk.h

@@ -50,6 +50,8 @@ class DCPBackfillByIdDisk : public DCPBackfillDisk {
      */
     void complete(bool cancelled) override;
 
+    void complete(ActiveStream& stream, bool cancelled);
+
     /// collection to scan for
     CollectionID cid;
 };

engines/ep/src/dcp/backfill_by_seqno_disk.cc

@@ -141,6 +141,13 @@ backfill_status_t DCPBackfillBySeqnoDisk::create() {
 
         stream->setDead(cb::mcbp::DcpStreamEndStatus::Rollback);
         transitionState(backfill_state_done);
+        return backfill_success;
+    }
+
+    // Check if a history scan is required or if only a history scan is required
+    if (setupForHistoryScan(*stream, *scanCtx, startSeqno)) {
+        // The scan is completely inside the history window
+        transitionState(backfill_state_scanning_history_snapshot);
     } else {
         bool markerSent = markDiskSnapshot(*stream, *scanCtx, *kvstore);
 
@@ -177,8 +184,12 @@ backfill_status_t DCPBackfillBySeqnoDisk::scan() {
     auto& bySeqnoCtx = dynamic_cast<BySeqnoScanContext&>(*scanCtx);
     switch (kvstore->scan(bySeqnoCtx)) {
     case scan_success:
-        stream->setBackfillScanLastRead(scanCtx->lastReadSeqno);
-        transitionState(backfill_state_completing);
+        if (historyScan) {
+            transitionState(backfill_state_scanning_history_snapshot);
+        } else {
+            stream->setBackfillScanLastRead(scanCtx->lastReadSeqno);
+            transitionState(backfill_state_completing);
+        }
         return backfill_success;
     case scan_again:
         // Scan should run again (e.g. was paused by callback)
@@ -269,11 +280,22 @@ bool DCPBackfillBySeqnoDisk::markDiskSnapshot(ActiveStream& stream,
     if (stream.getFilter().isLegacyFilter()) {
         return markLegacyDiskSnapshot(stream, scanCtx, kvs);
     }
-    return stream.markDiskSnapshot(startSeqno,
-                                   scanCtx.maxSeqno,
-                                   scanCtx.persistedCompletedSeqno,
-                                   scanCtx.maxVisibleSeqno,
-                                   scanCtx.timestamp);
+    // HistoryScan: If a disk snapshot is being "split" into non-history and
+    // history ranges, then the endSeqno of this first range should show the
+    // entire snapshot. E.g.
+    // disk snapshot is [a...d], but split
+    // no-history[a..b]
+    // history[c..d]
+    // Then all of the markers from this backfill stats start:a, end:d and mvs
+    // hcs can only be valid once d is reached.
+    return stream.markDiskSnapshot(
+            startSeqno,
+            historyScan ? historyScan->snapshotMaxSeqno : scanCtx.maxSeqno,
+            scanCtx.persistedCompletedSeqno,
+            scanCtx.maxVisibleSeqno,
+            scanCtx.timestamp,
+            historyScan ? ActiveStream::SnapshotSource::NoHistoryPrologue
+                        : ActiveStream::SnapshotSource::NoHistory);
 }
 
 // This function is used for backfills where the stream is configured as a
@@ -348,7 +370,8 @@ bool DCPBackfillBySeqnoDisk::markLegacyDiskSnapshot(ActiveStream& stream,
                                        scanCtx.maxSeqno,
                                        scanCtx.persistedCompletedSeqno,
                                        scanCtx.maxVisibleSeqno,
-                                       scanCtx.timestamp);
+                                       scanCtx.timestamp,
+                                       ActiveStream::SnapshotSource::NoHistory);
     }
 
     // Need to figure out the maxSeqno/maxVisibleSeqno for calling
@@ -433,7 +456,12 @@ bool DCPBackfillBySeqnoDisk::markLegacyDiskSnapshot(ActiveStream& stream,
         if (gv.item->isCommitted()) {
             // Step 3. If this is a committed item, done.
             return stream.markDiskSnapshot(
-                    startSeqno, stats.highSeqno, {}, stats.highSeqno, {});
+                    startSeqno,
+                    stats.highSeqno,
+                    {},
+                    stats.highSeqno,
+                    {},
+                    ActiveStream::SnapshotSource::NoHistory);
         }
     } else if (gv.getStatus() != cb::engine_errc::no_such_key) {
         stream.log(spdlog::level::level_enum::warn,
@@ -546,8 +574,12 @@ bool DCPBackfillBySeqnoDisk::markLegacyDiskSnapshot(ActiveStream& stream,
             cb.maxVisibleSeqno < backfillRangeEndSeqno) {
             stream.setEndSeqno(cb.maxVisibleSeqno);
         }
-        return stream.markDiskSnapshot(
-                startSeqno, cb.maxVisibleSeqno, {}, cb.maxVisibleSeqno, {});
+        return stream.markDiskSnapshot(startSeqno,
+                                       cb.maxVisibleSeqno,
+                                       {},
+                                       cb.maxVisibleSeqno,
+                                       {},
+                                       ActiveStream::SnapshotSource::NoHistory);
     } else {
         endStreamIfNeeded();
         // Found nothing committed at all