MB-34017: Send HCS in Disk snapshot

BenHuddleston · daverigby · commit 03c77b9584ce · 2019-08-16T11:10:23.000Z
Send the HCS in Disk snapshots so that the replica node can flush a correct HCS at the end of the snapshot which allows a fast warmup (HCS to HPS) if we do not do any more SyncWrites after our Disk snapshot. In a future patch, we will pass this through to the flusher via the CheckpointManager. Change-Id: I36631b53611e903791643f7ce960960eea3f548f Reviewed-on: http://review.couchbase.org/113187 Tested-by: Build Bot <build@couchbase.com> Reviewed-by: Dave Rigby <daver@couchbase.com>
diff --git a/daemon/mcbp_validators.cc b/daemon/mcbp_validators.cc
@@ -564,15 +564,28 @@ static Status dcp_stream_end_validator(Cookie& cookie) {
 }
 
 static Status dcp_snapshot_marker_validator(Cookie& cookie) {
+    auto& header = cookie.getHeader();
+
+    // Pass the extras len in because we will check it manually as it is
+    // variable length
     auto status = McbpValidator::verify_header(cookie,
-                                               20,
+                                               header.getExtlen(),
                                                ExpectedKeyLen::Zero,
                                                ExpectedValueLen::Zero,
                                                ExpectedCas::Any,
                                                PROTOCOL_BINARY_RAW_BYTES);
     if (status != Status::Success) {
         return status;
     }
+
+    // Validate our extras length is correct
+    using cb::mcbp::request::DcpSnapshotMarkerV1Payload;
+    using cb::mcbp::request::DcpSnapshotMarkerV2Payload;
+    if (!(header.getExtlen() == sizeof(DcpSnapshotMarkerV1Payload) ||
+          header.getExtlen() == sizeof(DcpSnapshotMarkerV2Payload))) {
+        return Status::Einval;
+    }
+
     return verify_common_dcp_restrictions(cookie);
 }
 
diff --git a/engines/ep/src/couch-kvstore/couch-kvstore.cc b/engines/ep/src/couch-kvstore/couch-kvstore.cc
@@ -1433,6 +1433,19 @@ ScanContext* CouchKVStore::initScanContext(
         return NULL;
     }
 
+    auto* vbState = getVBucketState(vbid);
+    if (!vbState) {
+        if (readVBState(db, vbid) == CouchKVStore::ReadVBStateStatus::Success) {
+            vbState = getVBucketState(vbid);
+        }
+        if (!vbState) {
+            EP_LOG_WARN(
+                    "CouchKVStore::initScanContext:Failed to obtain vbState for"
+                    "the highCompletedSeqno");
+            return NULL;
+        }
+    }
+
     size_t scanId = scanCounter++;
 
     auto collectionsManifest = getDroppedCollections(*db);
@@ -1452,6 +1465,7 @@ ScanContext* CouchKVStore::initScanContext(
                                         options,
                                         valOptions,
                                         count,
+                                        vbState->highCompletedSeqno,
                                         configuration,
                                         collectionsManifest);
     sctx->logger = &logger;
diff --git a/engines/ep/src/dcp/active_stream.cc b/engines/ep/src/dcp/active_stream.cc
@@ -234,7 +234,10 @@ void ActiveStream::registerCursor(CheckpointManager& chkptmgr,
     }
 }
 
-void ActiveStream::markDiskSnapshot(uint64_t startSeqno, uint64_t endSeqno) {
+void ActiveStream::markDiskSnapshot(
+        uint64_t startSeqno,
+        uint64_t endSeqno,
+        boost::optional<uint64_t> highCompletedSeqno) {
     {
         LockHolder lh(streamMutex);
         uint64_t chkCursorSeqno = endSeqno;
@@ -291,14 +294,15 @@ void ActiveStream::markDiskSnapshot(uint64_t startSeqno, uint64_t endSeqno) {
             logPrefix,
             startSeqno,
             endSeqno);
-        // @TODO push through HCS
+        // If the stream supports SyncRep then send the HCS in the
+        // SnapshotMarker
         pushToReadyQ(std::make_unique<SnapshotMarker>(
                 opaque_,
                 vb_,
                 startSeqno,
                 endSeqno,
                 MARKER_FLAG_DISK | MARKER_FLAG_CHK,
-                boost::none /*HCS*/,
+                supportSyncReplication() ? highCompletedSeqno : boost::none,
                 sid));
         lastSentSnapEndSeqno.store(endSeqno, std::memory_order_relaxed);
 
diff --git a/engines/ep/src/dcp/active_stream.h b/engines/ep/src/dcp/active_stream.h
@@ -115,7 +115,9 @@ class ActiveStream : public Stream,
         backfillRemaining.fetch_add(by, std::memory_order_relaxed);
     }
 
-    void markDiskSnapshot(uint64_t startSeqno, uint64_t endSeqno);
+    void markDiskSnapshot(uint64_t startSeqno,
+                          uint64_t endSeqno,
+                          boost::optional<uint64_t> highCompletedSeqno);
 
     bool backfillReceived(std::unique_ptr<Item> itm,
                           backfill_source_t backfill_source,
diff --git a/engines/ep/src/dcp/backfill_disk.cc b/engines/ep/src/dcp/backfill_disk.cc
@@ -248,7 +248,8 @@ backfill_status_t DCPBackfillDisk::create() {
         transitionState(backfill_state_done);
     } else {
         stream->incrBackfillRemaining(scanCtx->documentCount);
-        stream->markDiskSnapshot(startSeqno, scanCtx->maxSeqno);
+        stream->markDiskSnapshot(
+                startSeqno, scanCtx->maxSeqno, scanCtx->highCompletedSeqno);
         transitionState(backfill_state_scanning);
     }
 
diff --git a/engines/ep/src/dcp/backfill_memory.cc b/engines/ep/src/dcp/backfill_memory.cc
@@ -90,7 +90,7 @@ backfill_status_t DCPBackfillMemory::run() {
     stream->incrBackfillRemaining(items.size());
 
     /* Mark disk snapshot */
-    stream->markDiskSnapshot(startSeqno, adjustedEndSeqno);
+    stream->markDiskSnapshot(startSeqno, adjustedEndSeqno, {});
 
     /* Move every item to the stream */
     for (auto& item : items) {
@@ -249,7 +249,8 @@ backfill_status_t DCPBackfillMemoryBuffered::create() {
                     std::min(endSeqno, static_cast<uint64_t>(rangeItr.back()));
 
             /* Mark disk snapshot */
-            stream->markDiskSnapshot(startSeqno, endSeqno);
+            stream->markDiskSnapshot(
+                    startSeqno, endSeqno, evb->getHighCompletedSeqno());
 
             /* Change the backfill state */
             transitionState(BackfillState::Scanning);
diff --git a/engines/ep/src/dcp/passive_stream.cc b/engines/ep/src/dcp/passive_stream.cc
@@ -869,8 +869,6 @@ ENGINE_ERROR_CODE PassiveStream::processDropScope(VBucket& vb,
 }
 
 void PassiveStream::processMarker(SnapshotMarker* marker) {
-    // @TODO sanity check -  remove when we send the HCS
-    Expects(!marker->getHighCompletedSeqno());
     VBucketPtr vb = engine->getVBucket(vb_);
 
     cur_snapshot_start.store(marker->getStartSeqno());
diff --git a/engines/ep/src/kvstore.cc b/engines/ep/src/kvstore.cc
@@ -54,6 +54,7 @@ ScanContext::ScanContext(
         DocumentFilter _docFilter,
         ValueFilter _valFilter,
         uint64_t _documentCount,
+        uint64_t highCompletedSeqno,
         const KVStoreConfig& _config,
         const std::vector<Collections::KVStore::DroppedCollection>&
                 droppedCollections)
@@ -68,6 +69,7 @@ ScanContext::ScanContext(
       docFilter(_docFilter),
       valFilter(_valFilter),
       documentCount(_documentCount),
+      highCompletedSeqno(highCompletedSeqno),
       logger(globalBucketLogger.get()),
       config(_config),
       collectionsContext(droppedCollections) {
diff --git a/engines/ep/src/kvstore.h b/engines/ep/src/kvstore.h
@@ -211,6 +211,7 @@ class ScanContext {
                 DocumentFilter _docFilter,
                 ValueFilter _valFilter,
                 uint64_t _documentCount,
+                uint64_t highCompletedSeqno,
                 const KVStoreConfig& _config,
                 const std::vector<Collections::KVStore::DroppedCollection>&
                         droppedCollections);
@@ -227,6 +228,7 @@ class ScanContext {
     const DocumentFilter docFilter;
     const ValueFilter valFilter;
     const uint64_t documentCount;
+    const uint64_t highCompletedSeqno;
 
     BucketLogger* logger;
     const KVStoreConfig& config;
diff --git a/engines/ep/src/magma-kvstore/magma-kvstore.cc b/engines/ep/src/magma-kvstore/magma-kvstore.cc
@@ -1349,6 +1349,7 @@ class MagmaScanContext : public ScanContext {
                      DocumentFilter _docFilter,
                      ValueFilter _valFilter,
                      uint64_t _documentCount,
+                     uint64_t highCompletedSeqno,
                      const KVStoreConfig& _config,
                      const std::vector<Collections::KVStore::DroppedCollection>&
                              droppedCollections,
@@ -1363,6 +1364,7 @@ class MagmaScanContext : public ScanContext {
                       _docFilter,
                       _valFilter,
                       _documentCount,
+                      highCompletedSeqno,
                       _config,
                       droppedCollections),
           kvHandle(kvHandle) {
@@ -1386,6 +1388,7 @@ ScanContext* MagmaKVStore::initScanContext(
     uint64_t highSeqno;
     uint64_t purgeSeqno;
     uint64_t docCount;
+    uint64_t highCompletedSeqno;
     {
         std::shared_lock<std::shared_timed_mutex> lock(kvHandle->vbstateMutex);
         auto vbstate = cachedVBStates[vbid.get()].get();
@@ -1397,6 +1400,7 @@ ScanContext* MagmaKVStore::initScanContext(
         highSeqno = vbstate->highSeqno;
         purgeSeqno = vbstate->purgeSeqno;
         docCount = cachedMagmaInfo[vbid.get()]->docCount;
+        highCompletedSeqno = vbstate->highCompletedSeqno;
     }
 
     auto collectionsManifest = getDroppedCollections(vbid);
@@ -1460,6 +1464,7 @@ ScanContext* MagmaKVStore::initScanContext(
                                      options,
                                      valOptions,
                                      docCount,
+                                     highCompletedSeqno,
                                      configuration,
                                      collectionsManifest,
                                      kvHandle);
diff --git a/engines/ep/src/rocksdb-kvstore/rocksdb-kvstore.cc b/engines/ep/src/rocksdb-kvstore/rocksdb-kvstore.cc
@@ -1441,19 +1441,21 @@ ScanContext* RocksDBKVStore::initScanContext(
     // as scan is supposed to be inclusive at both ends,
     // seqnos 2 to 4 covers 3 docs not 4 - 2 = 2
 
-    uint64_t endSeqno = cachedVBStates[vbid.get()]->highSeqno;
-    return new ScanContext(cb,
-                           cl,
-                           vbid,
-                           scanId,
-                           startSeqno,
-                           endSeqno,
-                           0, /*TODO RDB: pass the real purge-seqno*/
-                           options,
-                           valOptions,
-                           /* documentCount */ endSeqno - startSeqno + 1,
-                           configuration,
-                           {/*no collections in rocksdb*/});
+    auto& state = cachedVBStates[vbid.get()];
+    return new ScanContext(
+            cb,
+            cl,
+            vbid,
+            scanId,
+            startSeqno,
+            state->highSeqno,
+            0, /*TODO RDB: pass the real purge-seqno*/
+            options,
+            valOptions,
+            /* documentCount */ state->highSeqno - startSeqno + 1,
+            state->highCompletedSeqno,
+            configuration,
+            {/*no collections in rocksdb*/});
 }
 
 scan_error_t RocksDBKVStore::scan(ScanContext* ctx) {
diff --git a/engines/ep/tests/module_tests/dcp_durability_stream_test.cc b/engines/ep/tests/module_tests/dcp_durability_stream_test.cc
@@ -825,6 +825,59 @@ TEST_P(DurabilityActiveStreamTest,
     producer->cancelCheckpointCreatorTask();
 }
 
+TEST_P(DurabilityActiveStreamTest, DiskSnapshotSendsHCSWithSyncRepSupport) {
+    auto vb = engine->getVBucket(vbid);
+    auto& ckptMgr = *vb->checkpointManager;
+    // Get rid of set_vb_state and any other queue_op we are not interested in
+    ckptMgr.clear(*vb, 0 /*seqno*/);
+
+    const auto key = makeStoredDocKey("key");
+    const std::string value = "value";
+    auto item = makePendingItem(key, value);
+
+    auto ctx = VBQueueItemCtx();
+    ctx.durability = DurabilityItemCtx{item->getDurabilityReqs()};
+
+    EXPECT_EQ(MutationStatus::WasClean, public_processSet(*vb, *item, ctx));
+    vb->notifyActiveDMOfLocalSyncWrite();
+    flushVBucketToDiskIfPersistent(vbid, 1);
+
+    vb->seqnoAcknowledged(folly::SharedMutex::ReadHolder(vb->getStateLock()),
+                          replica,
+                          1 /*prepareSeqno*/);
+
+    flushVBucketToDiskIfPersistent(vbid, 1);
+
+    producer->createCheckpointProcessorTask();
+    producer->scheduleCheckpointProcessorTask();
+
+    stream->transitionStateToBackfilling();
+    ASSERT_TRUE(stream->isBackfilling());
+
+    // Run the backfill we scheduled when we transitioned to the backfilling
+    // state.
+    auto& bfm = producer->getBFM();
+    bfm.backfill();
+    bfm.backfill();
+
+    // No message processed, BufferLog empty
+    ASSERT_EQ(0, producer->getBytesOutstanding());
+
+    // readyQ must contain a SnapshotMarker
+    ASSERT_EQ(3, stream->public_readyQSize());
+    auto resp = stream->public_nextQueuedItem();
+    ASSERT_TRUE(resp);
+    EXPECT_EQ(DcpResponse::Event::SnapshotMarker, resp->getEvent());
+
+    auto& marker = dynamic_cast<SnapshotMarker&>(*resp);
+    EXPECT_TRUE(marker.getFlags() & MARKER_FLAG_CHK);
+    ASSERT_TRUE(marker.getFlags() & MARKER_FLAG_DISK);
+    ASSERT_TRUE(marker.getHighCompletedSeqno());
+    EXPECT_EQ(1, *marker.getHighCompletedSeqno());
+
+    producer->cancelCheckpointCreatorTask();
+}
+
 void DurabilityPassiveStreamTest::SetUp() {
     SingleThreadedPassiveStreamTest::SetUp();
     consumer->enableSyncReplication();
diff --git a/engines/ep/tests/module_tests/dcp_stream_test.cc b/engines/ep/tests/module_tests/dcp_stream_test.cc
@@ -1545,6 +1545,7 @@ TEST_P(SingleThreadedActiveStreamTest, DiskSnapshotSendsChkMarker) {
     auto& marker = dynamic_cast<SnapshotMarker&>(*resp);
     EXPECT_TRUE(marker.getFlags() & MARKER_FLAG_CHK);
     EXPECT_TRUE(marker.getFlags() & MARKER_FLAG_DISK);
+    EXPECT_FALSE(marker.getHighCompletedSeqno());
 
     producer->cancelCheckpointCreatorTask();
 }
diff --git a/tests/mcbp/mcbp_test.cc b/tests/mcbp/mcbp_test.cc
@@ -2086,6 +2086,12 @@ class DcpSnapshotMarkerValidatorTest
 };
 
 TEST_P(DcpSnapshotMarkerValidatorTest, CorrectMessage) {
+    // Validate V1 Payload
+    EXPECT_EQ(cb::mcbp::Status::NotSupported, validate());
+
+    // Validate V2 Payload
+    request.message.header.request.setExtlen(28);
+    request.message.header.request.setBodylen(28);
     EXPECT_EQ(cb::mcbp::Status::NotSupported, validate());
 }
 

Original file line number	Diff line number	Diff line change
`@@ -115,7 +115,9 @@ class ActiveStream : public Stream,`
`115`	`115`	`backfillRemaining.fetch_add(by, std::memory_order_relaxed);`
`116`	`116`	`}`
`117`	`117`
`118`		`- void markDiskSnapshot(uint64_t startSeqno, uint64_t endSeqno);`
	`118`	`+ void markDiskSnapshot(uint64_t startSeqno,`
	`119`	`+ uint64_t endSeqno,`
	`120`	`+ boost::optional<uint64_t> highCompletedSeqno);`
`119`	`121`
`120`	`122`	`bool backfillReceived(std::unique_ptr<Item> itm,`
`121`	`123`	`backfill_source_t backfill_source,`
Original file line number	Diff line number	Diff line change
`@@ -248,7 +248,8 @@ backfill_status_t DCPBackfillDisk::create() {`
`248`	`248`	`transitionState(backfill_state_done);`
`249`	`249`	`} else {`
`250`	`250`	`stream->incrBackfillRemaining(scanCtx->documentCount);`
`251`		`- stream->markDiskSnapshot(startSeqno, scanCtx->maxSeqno);`
	`251`	`+ stream->markDiskSnapshot(`
	`252`	`+ startSeqno, scanCtx->maxSeqno, scanCtx->highCompletedSeqno);`
`252`	`253`	`transitionState(backfill_state_scanning);`
`253`	`254`	`}`
`254`	`255`
Original file line number	Diff line number	Diff line change
`@@ -869,8 +869,6 @@ ENGINE_ERROR_CODE PassiveStream::processDropScope(VBucket& vb,`
`869`	`869`	`}`
`870`	`870`
`871`	`871`	`void PassiveStream::processMarker(SnapshotMarker* marker) {`
`872`		`- // @TODO sanity check - remove when we send the HCS`
`873`		`- Expects(!marker->getHighCompletedSeqno());`
`874`	`872`	`VBucketPtr vb = engine->getVBucket(vb_);`
`875`	`873`
`876`	`874`	`cur_snapshot_start.store(marker->getStartSeqno());`