MB-54553: Propagate CheckpointHistorical to magma::saveDocs()

paolococchi · paolococchi · commit 5629c4ea37d3 · 2023-01-22T21:24:17.000Z
Passed when committing the flush-batch. Primary usage is to inform magma in the case where the historical data stream has been interrupted. * Note * Magma needs that the flag is always passed to the storage. 7.2 active nodes always create checkpoints with Historical::Yes. While 7.2 replicas set the Historical flag in checkpoints depending on what is received in snapshot Markers. Change-Id: Ia1a3ff07130a77e2cc919334adee1876c0092c9e Reviewed-on: https://review.couchbase.org/c/kv_engine/+/185333 Well-Formed: Restriction Checker Reviewed-by: Jim Walker <jim@couchbase.com> Tested-by: Paolo Cocchi <paolo.cocchi@couchbase.com>
diff --git a/engines/ep/src/checkpoint_manager.cc b/engines/ep/src/checkpoint_manager.cc
@@ -65,7 +65,7 @@ CheckpointManager::CheckpointManager(EPStats& st,
                       {},
                       maxPrepareSeqno,
                       CheckpointType::Memory,
-                      CheckpointHistorical::No);
+                      CheckpointHistorical::Yes);
 
     if (checkpointConfig.isPersistenceEnabled()) {
         // Register the persistence cursor
@@ -137,7 +137,7 @@ void CheckpointManager::addNewCheckpoint(
                      maxVisibleSeqno,
                      {},
                      CheckpointType::Memory,
-                     CheckpointHistorical::No);
+                     CheckpointHistorical::Yes);
 }
 
 void CheckpointManager::addNewCheckpoint(
@@ -1170,7 +1170,7 @@ void CheckpointManager::clear(const std::lock_guard<std::mutex>& lh,
                       {},
                       0, // HPS=0 because we have correct val on disk and in PDM
                       CheckpointType::Memory,
-                      CheckpointHistorical::No);
+                      CheckpointHistorical::Yes);
     resetCursors();
 }
 
diff --git a/engines/ep/src/ep_bucket.cc b/engines/ep/src/ep_bucket.cc
@@ -488,7 +488,8 @@ EPBucket::FlushResult EPBucket::flushVBucket_UNLOCKED(LockedVBucketPtr vb) {
         writeOp = WriteOperation::Insert;
     }
 
-    VB::Commit commitData(vb->getManifest(), writeOp, vbstate, callback);
+    VB::Commit commitData(
+            vb->getManifest(), writeOp, vbstate, callback, toFlush.historical);
 
     vbucket_state& proposedVBState = commitData.proposedVBState;
 
diff --git a/engines/ep/src/kvstore/magma-kvstore/magma-kvstore.cc b/engines/ep/src/kvstore/magma-kvstore/magma-kvstore.cc
@@ -750,7 +750,8 @@ bool MagmaKVStore::commit(std::unique_ptr<TransactionContext> txnCtx,
     preFlushHook();
 
     // Flush all documents to disk
-    auto errCode = saveDocs(ctx, commitData, kvctx);
+    auto errCode = saveDocs(
+            ctx, commitData, kvctx, toHistoryMode(commitData.historical));
     if (errCode != static_cast<int>(cb::engine_errc::success)) {
         logger->warn("MagmaKVStore::commit: saveDocs {} errCode:{}",
                      txnCtx->vbid,
@@ -1290,7 +1291,8 @@ GetValue MagmaKVStore::makeGetValue(Vbid vb,
 
 int MagmaKVStore::saveDocs(MagmaKVStoreTransactionContext& txnCtx,
                            VB::Commit& commitData,
-                           kvstats_ctx& kvctx) {
+                           kvstats_ctx& kvctx,
+                           magma::Magma::HistoryMode historyMode) {
     uint64_t ninserts = 0;
     uint64_t ndeletes = 0;
 
@@ -1541,7 +1543,8 @@ int MagmaKVStore::saveDocs(MagmaKVStoreTransactionContext& txnCtx,
                                    writeOps,
                                    kvstoreRevList[getCacheSlot(vbid)],
                                    writeDocsCB,
-                                   postWriteDocsCB);
+                                   postWriteDocsCB,
+                                   historyMode);
 
     saveDocsPostWriteDocsHook();
 
diff --git a/engines/ep/src/kvstore/magma-kvstore/magma-kvstore.h b/engines/ep/src/kvstore/magma-kvstore/magma-kvstore.h
@@ -593,7 +593,9 @@ class MagmaKVStore : public KVStore {
 
     virtual int saveDocs(MagmaKVStoreTransactionContext& txnCtx,
                          VB::Commit& commitData,
-                         kvstats_ctx& kvctx);
+                         kvstats_ctx& kvctx,
+                         magma::Magma::HistoryMode historyMode =
+                                 magma::Magma::HistoryMode::Disabled);
 
     void commitCallback(MagmaKVStoreTransactionContext& txnCtx,
                         int status,
@@ -768,6 +770,17 @@ class MagmaKVStore : public KVStore {
             const magma::Slice& valSlice,
             std::function<magma::Status(magma::Slice&)> valueRead) const;
 
+    static magma::Magma::HistoryMode toHistoryMode(
+            CheckpointHistorical historical) {
+        switch (historical) {
+        case CheckpointHistorical::No:
+            return magma::Magma::HistoryMode::Disabled;
+        case CheckpointHistorical::Yes:
+            return magma::Magma::HistoryMode::Enabled;
+        }
+        folly::assume_unreachable();
+    }
+
     MagmaKVStoreConfig& configuration;
 
     /**
diff --git a/engines/ep/src/kvstore/magma-kvstore/magma-memory-tracking-proxy.cc b/engines/ep/src/kvstore/magma-kvstore/magma-memory-tracking-proxy.cc
@@ -463,7 +463,8 @@ magma::Status MagmaMemoryTrackingProxy::WriteDocs(
         const std::vector<magma::Magma::WriteOperation>& docOperations,
         const magma::Magma::KVStoreRevision kvsRev,
         const magma::Magma::WriteDocsCallback docCallback,
-        const magma::Magma::PostWriteDocsCallback postCallback) {
+        const magma::Magma::PostWriteDocsCallback postCallback,
+        const magma::Magma::HistoryMode historyMode) {
     magma::Magma::WriteDocsCallback wrappedDocCallback = nullptr;
     magma::Magma::PostWriteDocsCallback wrappedPostCallback = nullptr;
 
@@ -491,7 +492,8 @@ magma::Status MagmaMemoryTrackingProxy::WriteDocs(
                             docOperations,
                             kvsRev,
                             wrappedDocCallback,
-                            wrappedPostCallback);
+                            wrappedPostCallback,
+                            historyMode);
 }
 
 magma::Status MagmaMemoryTrackingProxy::NewCheckpoint(
diff --git a/engines/ep/src/kvstore/magma-kvstore/magma-memory-tracking-proxy.h b/engines/ep/src/kvstore/magma-kvstore/magma-memory-tracking-proxy.h
@@ -207,12 +207,16 @@ class MagmaMemoryTrackingProxy {
     void SetNumThreads(magma::Magma::ThreadType threadType, size_t nThreads);
     magma::Status Sync(bool flushAll);
     magma::Status SyncKVStore(const magma::Magma::KVStoreID kvID);
+
     magma::Status WriteDocs(
             const magma::Magma::KVStoreID kvID,
             const std::vector<magma::Magma::WriteOperation>& docOperations,
             const magma::Magma::KVStoreRevision kvsRev = 1,
             const magma::Magma::WriteDocsCallback docCallback = nullptr,
-            const magma::Magma::PostWriteDocsCallback postCallback = nullptr);
+            const magma::Magma::PostWriteDocsCallback postCallback = nullptr,
+            const magma::Magma::HistoryMode historyMode =
+                    magma::Magma::HistoryMode::Disabled);
+
     magma::Status NewCheckpoint(const magma::Magma::KVStoreID kvID);
     magma::Status StopBGCompaction(const magma::Magma::KVStoreID kvID);
     magma::Status ResumeBGCompaction(const magma::Magma::KVStoreID kvID);
diff --git a/engines/ep/src/vb_commit.cc b/engines/ep/src/vb_commit.cc
@@ -17,10 +17,12 @@ namespace VB {
 Commit::Commit(Collections::VB::Manifest& manifest,
                WriteOperation writeOp,
                vbucket_state vbs,
-               SysErrorCallback sysErrorCallback)
+               SysErrorCallback sysErrorCallback,
+               CheckpointHistorical historical)
     : collections(manifest),
       writeOp(writeOp),
       proposedVBState(std::move(vbs)),
-      sysErrorCallback(std::move(sysErrorCallback)) {
+      sysErrorCallback(std::move(sysErrorCallback)),
+      historical(historical) {
 }
 } // namespace VB
diff --git a/engines/ep/src/vb_commit.h b/engines/ep/src/vb_commit.h
@@ -39,7 +39,8 @@ class Commit {
     explicit Commit(Collections::VB::Manifest& manifest,
                     WriteOperation writeOp = WriteOperation::Upsert,
                     vbucket_state vbs = {},
-                    SysErrorCallback sysErrorCallback = {});
+                    SysErrorCallback sysErrorCallback = {},
+                    CheckpointHistorical historical = CheckpointHistorical::No);
 
     /// Object for updating the collection's meta-data during commit
     Collections::VB::Flush collections;
@@ -60,6 +61,12 @@ class Commit {
      * Allows EP to take decisions on how to react to the failure.
      */
     SysErrorCallback sysErrorCallback;
+
+    /**
+     * Tells the storage whether the flush-batch is part of a seamless sequence
+     * of historical data.
+     */
+    CheckpointHistorical historical;
 };
 
 } // end namespace VB
diff --git a/engines/ep/tests/mock/mock_magma_kvstore.cc b/engines/ep/tests/mock/mock_magma_kvstore.cc
@@ -35,12 +35,13 @@ KVStoreIface::ReadVBStateResult MockMagmaKVStore::readVBStateFromDisk(
 
 int MockMagmaKVStore::saveDocs(MagmaKVStoreTransactionContext& txnCtx,
                                VB::Commit& commitData,
-                               kvstats_ctx& kvctx) {
+                               kvstats_ctx& kvctx,
+                               magma::Magma::HistoryMode historyMode) {
     if (saveDocsErrorInjector) {
         return saveDocsErrorInjector(commitData, kvctx);
     }
 
-    return MagmaKVStore::saveDocs(txnCtx, commitData, kvctx);
+    return MagmaKVStore::saveDocs(txnCtx, commitData, kvctx, historyMode);
 }
 
 bool MockMagmaKVStore::snapshotVBucket(Vbid vbid,
diff --git a/engines/ep/tests/mock/mock_magma_kvstore.h b/engines/ep/tests/mock/mock_magma_kvstore.h
@@ -54,7 +54,9 @@ class MockMagmaKVStore : public MagmaKVStore {
 
     int saveDocs(MagmaKVStoreTransactionContext& txnCtx,
                  VB::Commit& commitData,
-                 kvstats_ctx& kvctx) override;
+                 kvstats_ctx& kvctx,
+                 magma::Magma::HistoryMode historyMode =
+                         magma::Magma::HistoryMode::Disabled) override;
 
     bool snapshotVBucket(Vbid vbid, const vbucket_state& newVBState) override;
 
diff --git a/engines/ep/tests/module_tests/collections/collections_kvstore_test.cc b/engines/ep/tests/module_tests/collections/collections_kvstore_test.cc
@@ -73,17 +73,21 @@ class CollectionsKVStoreTestBase : public KVStoreBackend, public KVStoreTest {
                         std::make_shared<Collections::Manager>()));
     }
 
-    void getEventsFromCheckpoint(std::vector<queued_item>& events) {
+    CheckpointManager::ItemsForCursor getEventsFromCheckpoint(
+            std::vector<queued_item>& events) {
         std::vector<queued_item> items;
-        vbucket->checkpointManager->getNextItemsForPersistence(items);
+        auto res =
+                vbucket->checkpointManager->getNextItemsForPersistence(items);
         for (const auto& qi : items) {
             if (qi->getOperation() == queue_op::system_event) {
                 events.push_back(qi);
             }
         }
 
-        ASSERT_FALSE(events.empty())
+        EXPECT_FALSE(events.empty())
                 << "getEventsFromCheckpoint: no events in " << vbucket->getId();
+
+        return res;
     }
 
     void applyEvents(TransactionContext& txnCtx,
@@ -92,7 +96,8 @@ class CollectionsKVStoreTestBase : public KVStoreBackend, public KVStoreTest {
         manifest.update(*vbucket, makeManifest(cm));
 
         std::vector<queued_item> events;
-        getEventsFromCheckpoint(events);
+        auto res = getEventsFromCheckpoint(events);
+        commitData.historical = res.historical;
 
         for (auto& ev : events) {
             commitData.collections.recordSystemEvent(*ev);

Original file line number	Diff line number	Diff line change
`@@ -488,7 +488,8 @@ EPBucket::FlushResult EPBucket::flushVBucket_UNLOCKED(LockedVBucketPtr vb) {`
`488`	`488`	`writeOp = WriteOperation::Insert;`
`489`	`489`	`}`
`490`	`490`
`491`		`- VB::Commit commitData(vb->getManifest(), writeOp, vbstate, callback);`
	`491`	`+ VB::Commit commitData(`
	`492`	`+ vb->getManifest(), writeOp, vbstate, callback, toFlush.historical);`
`492`	`493`
`493`	`494`	`vbucket_state& proposedVBState = commitData.proposedVBState;`
`494`	`495`
Original file line number	Diff line number	Diff line change
`@@ -35,12 +35,13 @@ KVStoreIface::ReadVBStateResult MockMagmaKVStore::readVBStateFromDisk(`
`35`	`35`
`36`	`36`	`int MockMagmaKVStore::saveDocs(MagmaKVStoreTransactionContext& txnCtx,`
`37`	`37`	`VB::Commit& commitData,`
`38`		`- kvstats_ctx& kvctx) {`
	`38`	`+ kvstats_ctx& kvctx,`
	`39`	`+ magma::Magma::HistoryMode historyMode) {`
`39`	`40`	`if (saveDocsErrorInjector) {`
`40`	`41`	`return saveDocsErrorInjector(commitData, kvctx);`
`41`	`42`	`}`
`42`	`43`
`43`		`- return MagmaKVStore::saveDocs(txnCtx, commitData, kvctx);`
	`44`	`+ return MagmaKVStore::saveDocs(txnCtx, commitData, kvctx, historyMode);`
`44`	`45`	`}`
`45`	`46`
`46`	`47`	`bool MockMagmaKVStore::snapshotVBucket(Vbid vbid,`