MB-54729: DCP disk backfill a CDC stream

Backfill of a CDC stream has 4 possible outcomes.

1) No History.

If no history is retained by the KVStore, the backfill behaves no
differently than it would on 7.1. ByID or BySeqno backfilling will
produce a single snapshot (normal or OSO) for start to end of the
disk range.

2) History Available.

If the KVStore indicates that history is available the following 3
outcomes are possible.

2.1) start >= ScanContext::historyStartSeqno => single snapshot marker

Note this case cannot be encountered with OSO. OSO requires a start
of 0.

This case occurs because the start seqno is within the retained history.
A single snapshot is produced that includes all seqnos from start to the
end of the seqno index. This will be "scanAllVersions" scan and the
snapshot gets tagged with the "History" flag and tagged that it "may
contain duplicates".

2.2) start < ScanContext::historyStartSeqno => two snapshots

Note that this case has two outcomes. If an OSO snapshot is possible.
An OSO snapshot is followed by a regular snapshot (tagged as History).
If OSO is not possible, two regular snapshots occur (differentiated by
flags).

When the backfill starts below the retained history, DCP will backfill
all of the available sequence data, non-history followed by history.
This requires two DCP snapshots produced from the same disk snapshot.

The first DCP snapshot is the non-history range, a KVStore::scan from
the requested start up to, but not including
ScanContext::historyStartSeqno. Or this will be an OSO snapshot of the
requested collection - the entire collection ByID range.

The second DCP snapshot is the history range, a
KVStore::scanAllVersions from ScanContext::historyStartSeqno to the end
of the disk snapshots sequence index. This snapshot is tagged to
indicate that it's History and that it "may contain duplicates".

Note that in the case when two DCP snapshots are generated (no OSO),
both markers represent full disk range, including the MVS/HCS. The
intention of this is to ensure that a failure before reaching the end
of the disk snapshot, the client knows it's somewhere in the middle of
the full disk range and can correctly resume DCP with the correct start
and snapshot range.

For example:

The disk snapshot has a range of sequence numbers from 1 to 20. This is
split into two sub-ranges. Non-history (nh) and history (h).

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

If a full backfill occurs of this disk snapshot (DCP stream start is 0)
two markers will prefix the transmission of the ranges.

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

mutations 0 to 10

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

mutations 11 to 20

If the client disconnected after mutation 10, they would resume with
start=10, range={0,20}, indicating they have a partial snapshot.

Implementation Notes:

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.

1) No History.

    backfill_state_scanning -> backfill_state_complete

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

2.1) 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.

2.2) Two snapshots (OSO and regular).
    backfill_state_scanning -> backfill_state_scanning_history_snapshot

In this case both backfill phases are delivering "snapshots" from the
same magma disk snapshot.

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