MB-38444: Set Backfill SnapEndSeqno to SeqList HighSeqno

This patch fixes an inconsistency between the EndSeqno and
MaxVisibleSeqno that we send in SnapMarker at Backfill in Ephemeral.

That is, we may end up breaking the (MVS <= EndSeqno) invariant, which
could cause:
 1) a wrong  MVS > EndSeqno sent in the marker
 2) a wrong MVS and EndSeqno

(1) may happen when the connection supports SyncReplication, while (2)
when the connection doesn't support SR.

*Details*

There is a substantial difference in how we create the Backfill Range
between Persistent (where everything works fine) and Ephemeral.

That is, for Persistence we pick up the the full range of the on-disk
seqno-index (ie, all data indexed in the latest on-disk header). By
doing that, EndSeqno and MVS are consistent.

While in Ephemeral we set the EndSeqno of the Backfill Range based on
what we have in the CheckpointManager. For making it simple, apart from
some exceptions that is (EndSeqno = FirstCkptSeqno - 1).
Then we consider the MVS from the Ephemeral storage (ie, the SeqList),
which in general is not consistent with the computed EndSeqno.

With this patch Ephemeral's behaviour matches the one in Persistent, ie
Backfill picks up all data in the storage. Which makes the existing
usage of MVS correct.

Change-Id: I4c504b8e161f1d5463757414e752a77c885fe25e
Reviewed-on: http://review.couchbase.org/c/kv_engine/+/144019
Well-Formed: Build Bot <[email protected]>
Tested-by: Build Bot <[email protected]>
Reviewed-by: Dave Rigby <[email protected]>

Author: paolococchi

engines/ep/src/dcp/backfill_memory.cc

@@ -152,34 +152,20 @@ backfill_status_t DCPBackfillMemoryBuffered::create() {
        remaining count */
     while (rangeItr.curr() != rangeItr.end()) {
         if (static_cast<uint64_t>((*rangeItr).getBySeqno()) >= startSeqno) {
-            /* Determine the endSeqno of the current snapshot.
-               We want to send till requested endSeqno, but if that cannot
-               constitute a snapshot then we need to send till the point
-               which can be called as snapshot end */
-            endSeqno = std::max(
-                    endSeqno,
-                    static_cast<uint64_t>(rangeItr.getEarlySnapShotEnd()));
-
-            /* We want to send items only till the point it is necessary to do
-               so */
-            endSeqno =
-                    std::min(endSeqno, static_cast<uint64_t>(rangeItr.back()));
-
-            /* Mark disk snapshot */
+            // Backfill covers the full SeqList range.
+            endSeqno = rangeItr.back();
+
+            // Send SnapMarker
             bool markerSent =
                     stream->markDiskSnapshot(startSeqno,
                                              endSeqno,
                                              evb->getHighCompletedSeqno(),
                                              rangeItr.getMaxVisibleSeqno());
 
             if (markerSent) {
-                /* Set backfill remaining
-                   [EPHE TODO]: This will be inaccurate if do not backfill till
-                   end of the iterator Additionally, this value may be an
-                   overestimate even if backfilled to the iterator end - it
-                   includes prepares/aborts which will not be sent if the stream
-                   is not sync write aware
-                 */
+                // @todo: This value may be an overestimate, as it includes
+                //  prepares/aborts which will not be sent if the stream is not
+                //  sync write aware
                 stream->setBackfillRemaining(rangeItr.count());
 
                 /* Change the backfill state and return for next stage. */

engines/ep/tests/ep_testsuite_dcp.cc

@@ -2362,19 +2362,17 @@ static enum test_result test_dcp_producer_stream_req_full(EngineIface* h) {
 
     checkne(num_items - get_stat<uint64_t>(h, "ep_items_rm_from_checkpoints"),
             uint64_t{0},
-            "Require a non-zero number of items to still be present in "
-            "CheckpointManager to be able to get 2x snapshot markers "
-            "(1x disk, 1x memory)");
+            "Ensure a non-zero number of items to still be present in "
+            "CheckpointManager to verify that we still get all mutations in the"
+            " storage in a single Backfill snapshot");
 
     const void* cookie = testHarness->create_cookie();
 
     DcpStreamCtx ctx;
     ctx.vb_uuid = get_ull_stat(h, "vb_0:0:id", "failovers");
     ctx.seqno = {0, get_ull_stat(h, "vb_0:high_seqno", "vbucket-seqno")};
     ctx.exp_mutations = num_items;
-    /* Memory backfill sends items from checkpoint snapshots as much as possible
-       Relies on backfill only when checkpoint snapshot is cleaned up */
-    ctx.exp_markers = 2;
+    ctx.exp_markers = 1;
 
     TestDcpConsumer tdc("unittest", cookie, h);
     tdc.addStreamCtx(ctx);
@@ -2512,7 +2510,14 @@ static enum test_result test_dcp_producer_stream_req_backfill(EngineIface* h) {
     DcpStreamCtx ctx;
     ctx.vb_uuid = get_ull_stat(h, "vb_0:0:id", "failovers");
     ctx.seqno = {0, 200};
-    ctx.exp_mutations = 200;
+    // The idea here is that at backfill we get the full Disk/SeqList snapshot.
+    // Persistence has been stopped at seqno 200, while Ephemeral stores all
+    // seqnos in the SeqList.
+    if (isEphemeralBucket(h)) {
+        ctx.exp_mutations = 400;
+    } else {
+        ctx.exp_mutations = 200;
+    }
     ctx.exp_markers = 1;
 
     TestDcpConsumer tdc("unittest", cookie, h);

engines/ep/tests/module_tests/dcp_stream_test.cc

@@ -4032,6 +4032,113 @@ TEST_P(SingleThreadedPassiveStreamTest, MB42780_DiskToMemoryFromPre65) {
     ASSERT_EQ(4, manager.getHighSeqno());
 }
 
+/**
+ * MB-38444: We fix an Ephemeral-only bug, but test covers Persistent bucket too
+ */
+TEST_P(SingleThreadedActiveStreamTest, BackfillRangeCoversAllDataInTheStorage) {
+    // We need to re-create the stream in a condition that triggers a backfill
+    stream.reset();
+    producer.reset();
+
+    auto& vb = *engine->getVBucket(vbid);
+    ASSERT_EQ(0, vb.getHighSeqno());
+    auto& manager = *vb.checkpointManager;
+    const auto& list =
+            CheckpointManagerTestIntrospector::public_getCheckpointList(
+                    manager);
+    ASSERT_EQ(1, list.size());
+
+    const auto keyA = makeStoredDocKey("keyA");
+    const std::string value = "value";
+    store_item(vbid, keyA, value);
+    EXPECT_EQ(1, vb.getHighSeqno());
+    EXPECT_EQ(1, vb.getMaxVisibleSeqno());
+    const auto keyB = makeStoredDocKey("keyB");
+    store_item(vbid, keyB, value);
+    EXPECT_EQ(2, vb.getHighSeqno());
+    EXPECT_EQ(2, vb.getMaxVisibleSeqno());
+
+    // Steps to ensure backfill when we re-create the stream in the following
+    manager.createNewCheckpoint();
+    flushVBucketToDiskIfPersistent(vbid, 2 /*expected_num_flushed*/);
+    ASSERT_EQ(2, list.size());
+    bool newCkptCreated;
+    EXPECT_EQ(2, manager.removeClosedUnrefCheckpoints(vb, newCkptCreated));
+    EXPECT_FALSE(newCkptCreated);
+    ASSERT_EQ(1, list.size());
+    ASSERT_EQ(0, manager.getNumOpenChkItems());
+
+    // Move high-seqno to 4
+    const auto keyC = makeStoredDocKey("keyC");
+    store_item(vbid, keyC, value);
+    EXPECT_EQ(3, vb.getHighSeqno());
+    EXPECT_EQ(3, vb.getMaxVisibleSeqno());
+    const auto keyD = makeStoredDocKey("keyD");
+    store_item(vbid, keyD, value);
+    EXPECT_EQ(4, vb.getHighSeqno());
+    EXPECT_EQ(4, vb.getMaxVisibleSeqno());
+
+    // At this point we havehigh-seqno=4 but only seqnos 3 and 4 in the CM, so
+    // we'll backfill.
+
+    // Note: The aim here is to verify that Backfill picks up everything from
+    // the storage even in the case where some seqnos are in the CM. So, we need
+    // to ensure that all seqnos are on-disk for Persistent.
+    flushVBucketToDiskIfPersistent(vbid, 2 /*expected_num_flushed*/);
+
+    // Re-create producer and stream
+    recreateProducerAndStream(vb, 0 /*flags*/);
+    ASSERT_TRUE(producer);
+    producer->createCheckpointProcessorTask();
+    ASSERT_TRUE(stream);
+    ASSERT_TRUE(stream->isBackfilling());
+    ASSERT_TRUE(stream->public_supportSyncReplication());
+    auto resp = stream->next();
+    EXPECT_FALSE(resp);
+
+    // Drive the backfill - execute
+    auto& bfm = producer->getBFM();
+    ASSERT_EQ(1, bfm.getNumBackfills());
+
+    // Backfill::create
+    // Before the fix this steps generates SnapMarker{start:0, end:2, mvs:4},
+    // while we want SnapMarker{start:0, end:4, mvs:4}
+    ASSERT_EQ(backfill_success, bfm.backfill());
+    const auto& readyQ = stream->public_readyQ();
+    ASSERT_EQ(1, readyQ.size());
+    resp = stream->next();
+    ASSERT_TRUE(resp);
+    EXPECT_EQ(DcpResponse::Event::SnapshotMarker, resp->getEvent());
+    auto snapMarker = dynamic_cast<SnapshotMarker&>(*resp);
+    EXPECT_EQ(0, snapMarker.getStartSeqno());
+    EXPECT_EQ(4, snapMarker.getEndSeqno());
+    EXPECT_EQ(4, *snapMarker.getMaxVisibleSeqno());
+
+    // Verify that all seqnos are sent at Backfill::scan
+    ASSERT_EQ(backfill_success, bfm.backfill());
+    ASSERT_EQ(4, readyQ.size());
+    resp = stream->next();
+    ASSERT_TRUE(resp);
+    EXPECT_EQ(DcpResponse::Event::Mutation, resp->getEvent());
+    EXPECT_EQ(1, *resp->getBySeqno());
+    ASSERT_EQ(3, readyQ.size());
+    resp = stream->next();
+    ASSERT_TRUE(resp);
+    EXPECT_EQ(DcpResponse::Event::Mutation, resp->getEvent());
+    EXPECT_EQ(2, *resp->getBySeqno());
+    ASSERT_EQ(2, readyQ.size());
+    resp = stream->next();
+    ASSERT_TRUE(resp);
+    EXPECT_EQ(DcpResponse::Event::Mutation, resp->getEvent());
+    EXPECT_EQ(3, *resp->getBySeqno());
+    ASSERT_EQ(1, readyQ.size());
+    resp = stream->next();
+    ASSERT_TRUE(resp);
+    EXPECT_EQ(DcpResponse::Event::Mutation, resp->getEvent());
+    EXPECT_EQ(4, *resp->getBySeqno());
+    ASSERT_EQ(0, readyQ.size());
+}
+
 INSTANTIATE_TEST_CASE_P(AllBucketTypes,
                         SingleThreadedActiveStreamTest,
                         STParameterizedBucketTest::allConfigValues(),