MB-54729: Add historyStartSeqno to ScanContext

Add a new member, historyStartSeqno to ScanContext for later use in
DCP backfill when "change_stream=true".

This will always be 0 for a KVStore which reports:

    StorageProperties::HistoryRetentionAvailable::No

For a KVStore which reports:

    StorageProperties::HistoryRetentionAvailable::Yes

This will be set to:

    0: If no history is configured (retention_bytes=0)
    >0: When history is available, this value represents the lowest
        seqno at which a continuous change stream could be made
        available.

This commit is not dependent upon Magma's changes to add history, but
adds the ability for MockMagmaKVStore to "inject" a history start.

Change-Id: I66db1d84af54f4a999f260d30b356a9a149ddf31

Author: jimwwalker

engines/ep/src/kvstore/kvstore.cc

@@ -54,14 +54,16 @@ ScanContext::ScanContext(
         std::unique_ptr<StatusCallback<CacheLookup>> cl,
         const std::vector<Collections::KVStore::DroppedCollection>&
                 droppedCollections,
-        uint64_t maxSeqno)
+        uint64_t maxSeqno,
+        uint64_t historyStartSeqno)
     : vbid(vbid),
       handle(std::move(handle)),
       docFilter(docFilter),
       valFilter(valFilter),
       logger(getGlobalBucketLogger().get()),
       collectionsContext(droppedCollections),
       maxSeqno(maxSeqno),
+      historyStartSeqno(historyStartSeqno),
       callback(std::move(cb)),
       lookup(std::move(cl)) {
     Expects(callback != nullptr);
@@ -82,7 +84,8 @@ BySeqnoScanContext::BySeqnoScanContext(
         const vbucket_state& vbucketState,
         const std::vector<Collections::KVStore::DroppedCollection>&
                 droppedCollections,
-        std::optional<uint64_t> timestamp)
+        std::optional<uint64_t> timestamp,
+        uint64_t historyStartSeqno)
 
     : ScanContext(vb,
                   std::move(handle),
@@ -91,7 +94,8 @@ BySeqnoScanContext::BySeqnoScanContext(
                   std::move(cb),
                   std::move(cl),
                   droppedCollections,
-                  end),
+                  end,
+                  historyStartSeqno),
       startSeqno(start),
       purgeSeqno(purgeSeqno),
       documentCount(_documentCount),
@@ -110,15 +114,17 @@ ByIdScanContext::ByIdScanContext(
         ValueFilter _valFilter,
         const std::vector<Collections::KVStore::DroppedCollection>&
                 droppedCollections,
-        uint64_t maxSeqno)
+        uint64_t maxSeqno,
+        uint64_t historyStartSeqno)
     : ScanContext(vb,
                   std::move(handle),
                   _docFilter,
                   _valFilter,
                   std::move(cb),
                   std::move(cl),
                   droppedCollections,
-                  maxSeqno),
+                  maxSeqno,
+                  historyStartSeqno),
       ranges(std::move(ranges)),
       lastReadKey(nullptr, 0) {
 }

engines/ep/src/kvstore/kvstore.h

@@ -392,7 +392,8 @@ class ScanContext {
                 std::unique_ptr<StatusCallback<CacheLookup>> cl,
                 const std::vector<Collections::KVStore::DroppedCollection>&
                         droppedCollections,
-                uint64_t maxSeqno);
+                uint64_t maxSeqno,
+                uint64_t historyStartSeqno = 0);
 
     virtual ~ScanContext() = default;
 
@@ -412,14 +413,15 @@ class ScanContext {
         return *lookup;
     }
 
-    const Vbid vbid;
+    const Vbid vbid{0};
     uint64_t lastReadSeqno{0};
     std::unique_ptr<KVFileHandle> handle;
-    const DocumentFilter docFilter;
-    const ValueFilter valFilter;
-    BucketLogger* logger;
+    const DocumentFilter docFilter{DocumentFilter::ALL_ITEMS};
+    const ValueFilter valFilter{ValueFilter::KEYS_ONLY};
+    BucketLogger* logger{nullptr};
     const Collections::VB::ScanContext collectionsContext;
-    uint64_t maxSeqno;
+    uint64_t maxSeqno{0};
+    uint64_t historyStartSeqno{0};
 
     /**
      * Cumulative count of bytes read from disk during this scan. Counts
@@ -454,7 +456,8 @@ class BySeqnoScanContext : public ScanContext {
             const vbucket_state& vbucketState,
             const std::vector<Collections::KVStore::DroppedCollection>&
                     droppedCollections,
-            std::optional<uint64_t> timestamp = {});
+            std::optional<uint64_t> timestamp = {},
+            uint64_t historyStartSeqno = 0);
 
     const uint64_t startSeqno;
     const uint64_t purgeSeqno;
@@ -516,7 +519,8 @@ class ByIdScanContext : public ScanContext {
                     ValueFilter _valFilter,
                     const std::vector<Collections::KVStore::DroppedCollection>&
                             droppedCollections,
-                    uint64_t maxSeqno);
+                    uint64_t maxSeqno,
+                    uint64_t historyStartSeqno = 0);
     std::vector<ByIdRange> ranges;
     // Key should be set by KVStore when a scan must be paused, this is where
     // a scan can resume from

engines/ep/tests/mock/mock_magma_kvstore.cc

@@ -90,3 +90,44 @@ magma::Status MockMagmaKVStore::runImplicitCompactKVStore(Vbid vbid) {
 magma::Status MockMagmaKVStore::newCheckpoint(Vbid vbid) {
     return magma->NewCheckpoint(vbid.get());
 }
+
+std::unique_ptr<BySeqnoScanContext> MockMagmaKVStore::initBySeqnoScanContext(
+        std::unique_ptr<StatusCallback<GetValue>> cb,
+        std::unique_ptr<StatusCallback<CacheLookup>> cl,
+        Vbid vbid,
+        uint64_t startSeqno,
+        DocumentFilter options,
+        ValueFilter valOptions,
+        SnapshotSource source,
+        std::unique_ptr<KVFileHandle> fileHandle) const {
+    auto scanContext =
+            MagmaKVStore::initBySeqnoScanContext(std::move(cb),
+                                                 std::move(cl),
+                                                 vbid,
+                                                 startSeqno,
+                                                 options,
+                                                 valOptions,
+                                                 source,
+                                                 std::move(fileHandle));
+
+    if (historyStartSeqno) {
+        scanContext->historyStartSeqno = historyStartSeqno.value();
+    }
+    return scanContext;
+}
+
+std::unique_ptr<ByIdScanContext> MockMagmaKVStore::initByIdScanContext(
+        std::unique_ptr<StatusCallback<GetValue>> cb,
+        std::unique_ptr<StatusCallback<CacheLookup>> cl,
+        Vbid vbid,
+        const std::vector<ByIdRange>& ranges,
+        DocumentFilter options,
+        ValueFilter valOptions) const {
+    auto scanContext = MagmaKVStore::initByIdScanContext(
+            std::move(cb), std::move(cl), vbid, ranges, options, valOptions);
+
+    if (historyStartSeqno) {
+        scanContext->historyStartSeqno = historyStartSeqno.value();
+    }
+    return scanContext;
+}

engines/ep/tests/mock/mock_magma_kvstore.h

@@ -19,6 +19,34 @@ class MockMagmaKVStore : public MagmaKVStore {
 public:
     explicit MockMagmaKVStore(MagmaKVStoreConfig& config);
 
+    /**
+     * See base-class comments for usage. However this "mock" version will
+     * override the ScanContext::historyStartSeqno if this class defines
+     * historyStartSeqno (see historyStartSeqno member below).
+     */
+    std::unique_ptr<BySeqnoScanContext> initBySeqnoScanContext(
+            std::unique_ptr<StatusCallback<GetValue>> cb,
+            std::unique_ptr<StatusCallback<CacheLookup>> cl,
+            Vbid vbid,
+            uint64_t startSeqno,
+            DocumentFilter options,
+            ValueFilter valOptions,
+            SnapshotSource source,
+            std::unique_ptr<KVFileHandle> fileHandle = nullptr) const override;
+
+    /**
+     * See base-class comments for usage. However this "mock" version will
+     * override the ScanContext::historyStartSeqno if this class defines
+     * historyStartSeqno (see historyStartSeqno member below).
+     */
+    std::unique_ptr<ByIdScanContext> initByIdScanContext(
+            std::unique_ptr<StatusCallback<GetValue>> cb,
+            std::unique_ptr<StatusCallback<CacheLookup>> cl,
+            Vbid vbid,
+            const std::vector<ByIdRange>& ranges,
+            DocumentFilter options,
+            ValueFilter valOptions) const override;
+
     ReadVBStateResult readVBStateFromDisk(Vbid vbid);
 
     ReadVBStateResult readVBStateFromDisk(
@@ -75,6 +103,13 @@ class MockMagmaKVStore : public MagmaKVStore {
     std::function<bool()> snapshotVBucketErrorInjector;
 
     StorageProperties storageProperties;
+
+    /**
+     * The historyStartSeqno in this class when set will override the value
+     * MagmaKVStore sets. This allows arbitrary placement of the history range
+     * for testing of various two-phase backfills.
+     */
+    std::optional<uint64_t> historyStartSeqno;
 };
 
 #endif