MB-34173: 1/2 Persist a valid snapshot range during backfill

If a backfill was very large and the backfill queue had grown to
be > flusher_batch_split_trigger before the flusher visits the vb,
the code path taken results in an unitialised snapshot_range_t being
persisted.

To address this we ensure the range is correctly initialised in
the path where the backfill queue is > flusher_batch_split_trigger

Change-Id: Idcdbe9fa9c7f3807d08372e3ebf24c106a449d9f
Reviewed-on: http://review.couchbase.org/109269
Well-Formed: Build Bot <[email protected]>
Reviewed-by: Dave Rigby <[email protected]>
Tested-by: Build Bot <[email protected]>

Author: jimwwalker

engines/ep/src/checkpoint.cc

@@ -1272,16 +1272,15 @@ CheckpointManager::ItemsForCursor CheckpointManager::getItemsForCursor(
             "manager on vb:%" PRIu16,
             name.c_str(),
             vbucketId);
-        return {};
+        return {0, 0};
     }
 
     auto& cursor = it->second;
 
     // Fetch whole checkpoints; as long as we don't exceed the approx item
     // limit.
-    ItemsForCursor result;
-    result.range.start = (*cursor.currentCheckpoint)->getSnapshotStartSeqno();
-    result.range.end = (*cursor.currentCheckpoint)->getSnapshotEndSeqno();
+    ItemsForCursor result((*cursor.currentCheckpoint)->getSnapshotStartSeqno(),
+                          (*cursor.currentCheckpoint)->getSnapshotEndSeqno());
     size_t itemCount = 0;
     while ((result.moreAvailable = incrCursor(cursor))) {
         queued_item& qi = *(cursor.currentPos);
@@ -1627,10 +1626,9 @@ snapshot_info_t CheckpointManager::getSnapshotInfo() {
                         "checkpointList is empty");
     }
 
-    snapshot_info_t info;
-    info.range.start = checkpointList.back()->getSnapshotStartSeqno();
-    info.start = lastBySeqno;
-    info.range.end = checkpointList.back()->getSnapshotEndSeqno();
+    snapshot_info_t info(lastBySeqno,
+                         {checkpointList.back()->getSnapshotStartSeqno(),
+                          checkpointList.back()->getSnapshotEndSeqno()});
 
     // If there are no items in the open checkpoint then we need to resume by
     // using that sequence numbers of the last closed snapshot. The exception is

engines/ep/src/checkpoint.h

@@ -628,7 +628,9 @@ class CheckpointManager {
 
     /// Return type of getItemsForCursor()
     struct ItemsForCursor {
-        snapshot_range_t range = {0, 0};
+        ItemsForCursor(uint64_t start, uint64_t end) : range(start, end) {
+        }
+        snapshot_range_t range;
         bool moreAvailable = {false};
     };
 

engines/ep/src/ep_types.h

@@ -48,15 +48,22 @@ std::string to_string(GenerateBySeqno generateBySeqno);
 std::string to_string(GenerateCas generateCas);
 std::string to_string(TrackCasDrift trackCasDrift);
 
-typedef struct {
+struct snapshot_range_t {
+    snapshot_range_t(uint64_t start, uint64_t end) : start(start), end(end) {
+        // @todo: enforce start < end in a non-maintenance release
+    }
+
     uint64_t start;
     uint64_t end;
-} snapshot_range_t;
+};
 
-typedef struct {
+struct snapshot_info_t {
+    snapshot_info_t(uint64_t start, snapshot_range_t range)
+        : start(start), range(range) {
+    }
     uint64_t start;
     snapshot_range_t range;
-} snapshot_info_t;
+};
 
 /**
  * The following options can be specified

engines/ep/src/vbucket.cc

@@ -343,25 +343,27 @@ VBucket::ItemsToFlush VBucket::getItemsToPersist(size_t approxLimit) {
     // Note that it is only valid to queue a complete checkpoint - this is where
     // the "approx" in the limit comes from.
     const auto ckptMgrLimit = approxLimit - result.items.size();
-    CheckpointManager::ItemsForCursor ckptItems;
+    bool ckptItemsAvailable = true;
     if (ckptMgrLimit > 0) {
         auto _begin_ = ProcessClock::now();
-        ckptItems = checkpointManager->getItemsForCursor(
+        auto ckptItems = checkpointManager->getItemsForCursor(
                 CheckpointManager::pCursorName, result.items, ckptMgrLimit);
         result.range = ckptItems.range;
+        ckptItemsAvailable = ckptItems.moreAvailable;
         stats.persistenceCursorGetItemsHisto.add(
                 std::chrono::duration_cast<std::chrono::microseconds>(
                         ProcessClock::now() - _begin_));
     } else {
         // We haven't got sufficient remaining capacity to read items from
         // CheckpoitnManager, therefore we must assume that there /could/
-        // more data to follow.
-        ckptItems.moreAvailable = true;
+        // more data to follow (leaving ckptItemsAvailable true). We also must
+        // ensure the valid snapshot range is returned
+        result.range = checkpointManager->getSnapshotInfo().range;
     }
 
     // Check if there's any more items remaining.
     result.moreAvailable =
-            !rejectQueue.empty() || !backfillEmpty || ckptItems.moreAvailable;
+            !rejectQueue.empty() || !backfillEmpty || ckptItemsAvailable;
 
     return result;
 }

engines/ep/src/vbucket.h

@@ -418,7 +418,7 @@ class VBucket : public std::enable_shared_from_this<VBucket> {
 
     struct ItemsToFlush {
         std::vector<queued_item> items;
-        snapshot_range_t range;
+        snapshot_range_t range{0, 0};
         bool moreAvailable = false;
     };
 

engines/ep/tests/ep_testsuite_dcp.cc

@@ -4521,6 +4521,79 @@ static enum test_result test_dcp_replica_stream_backfill(ENGINE_HANDLE *h,
     return SUCCESS;
 }
 
+// Test generates a replica VB and splits the generation with a warmup.
+// Importantly the very first batch of DCP items are marked as 'backfill' and
+// the test requires that the flusher_batch_split_trigger setting is less than
+// the size of the first batch.
+static enum test_result test_dcp_replica_stream_backfill_MB_34173(
+        ENGINE_HANDLE* h, ENGINE_HANDLE_V1* h1) {
+    check(set_vbucket_state(h, h1, 0, vbucket_state_replica),
+          "Failed to set vbucket state.");
+    const int items = 100;
+
+    // Validate that the flusher will split the items
+    checkgt(items,
+            get_int_stat(h, h1, "ep_flusher_batch_split_trigger", "config"),
+            "flusher_batch_split_trigger must be less than items");
+
+    const void *cookie = testHarness.create_cookie();
+    uint32_t opaque = 0xFFFF0000;
+    uint32_t seqno = 0;
+    uint32_t flags = 0;
+    const char* name = "MB_34173";
+
+    checkeq(ENGINE_SUCCESS,
+            h1->dcp.open(h, cookie, opaque, seqno, flags, name, {}),
+            "Failed dcp producer open connection.");
+    std::string type = get_str_stat(h, h1, "eq_dcpq:MB_34173:type", "dcp");
+    checkeq(0, type.compare("consumer"), "Consumer not found");
+
+    opaque = add_stream_for_consumer(h, h1, cookie, opaque, 0, 0,
+                                     PROTOCOL_BINARY_RESPONSE_SUCCESS);
+    // backfill items 1 to 100
+    dcp_stream_to_replica(h, h1, cookie, opaque, 0, 0x02, 1, items, 1, items);
+    wait_for_flusher_to_settle(h, h1);
+    wait_for_stat_to_be(h, h1, "vb_0:high_seqno", items, "vbucket-seqno");
+
+    testHarness.destroy_cookie(cookie);
+
+    testHarness.reload_engine(&h,
+                              &h1,
+                              testHarness.engine_path,
+                              testHarness.get_current_testcase()->cfg,
+                              true,
+                              true);
+    wait_for_warmup_complete(h, h1);
+
+    cookie = testHarness.create_cookie();
+    opaque = 0xFFFF0000;
+    checkeq(ENGINE_SUCCESS,
+            h1->dcp.open(h, cookie, opaque, seqno, flags, name, {}),
+            "Failed dcp producer open connection.");
+
+    type = get_str_stat(h, h1, "eq_dcpq:MB_34173:type", "dcp");
+    checkeq(0, type.compare("consumer"), "Consumer not found");
+
+    opaque = add_stream_for_consumer(
+            h, h1, cookie, opaque, 0, 0, PROTOCOL_BINARY_RESPONSE_SUCCESS);
+
+    // A second batch could fail if MB-34173 is not fixed, I say could because
+    // the corruption of the snapshot range may not yield a failure...
+    dcp_stream_to_replica(h,
+                          h1,
+                          cookie,
+                          opaque,
+                          0,
+                          0x02,
+                          items + 1,
+                          items + 10,
+                          items + 1,
+                          items + 10);
+
+    testHarness.destroy_cookie(cookie);
+    return SUCCESS;
+}
+
 static enum test_result test_dcp_replica_stream_in_memory(ENGINE_HANDLE *h,
                                                           ENGINE_HANDLE_V1 *h1)
 {
@@ -6083,6 +6156,12 @@ BaseTestCase testsuite_testcases[] = {
                  "chk_remover_stime=1;max_checkpoints=2",
                  prepare,
                  cleanup),
+        TestCase("test dcp replica stream backfill and warmup (MB-34173)",
+                 test_dcp_replica_stream_backfill_MB_34173, test_setup, teardown,
+                 "chk_remover_stime=1;max_checkpoints=2;"
+                 "flusher_batch_split_trigger=10",
+                 prepare_ep_bucket,
+                 cleanup),
         TestCase("test dcp replica stream in-memory",
                  test_dcp_replica_stream_in_memory, test_setup, teardown,
                  "chk_remover_stime=1;max_checkpoints=2", prepare, cleanup),

engines/ep/tests/module_tests/checkpoint_test.cc

@@ -566,10 +566,9 @@ TYPED_TEST(CheckpointTest, ItemBasedCheckpointCreation) {
     // moves the single cursor registered outside of the initial checkpoint,
     // allowing a new open checkpoint to be created.
     EXPECT_EQ(1, this->manager->getNumOfCursors());
-    snapshot_range_t range;
     std::vector<queued_item> items;
-    range = this->manager->getAllItemsForCursor(CheckpointManager::pCursorName,
-                                                items);
+    snapshot_range_t range = this->manager->getAllItemsForCursor(
+            CheckpointManager::pCursorName, items);
 
     EXPECT_EQ(0, range.start);
     EXPECT_EQ(1021, range.end);

engines/ep/tests/module_tests/vbucket_test.cc

@@ -61,9 +61,9 @@ VBucketTestBase::VBucketTestBase(item_eviction_policy_t eviction_policy) {
                                 global_stats,
                                 checkpoint_config,
                                 /*kvshard*/ nullptr,
-                                /*lastSeqno*/ 1000,
-                                /*lastSnapStart*/ 0,
-                                /*lastSnapEnd*/ 0,
+                                lastSeqno,
+                                range.start,
+                                range.end,
                                 /*table*/ nullptr,
                                 std::make_shared<DummyCB>(),
                                 /*newSeqnoCb*/ nullptr,
@@ -571,6 +571,8 @@ TEST_P(VBucketTest, GetItemsForCursor_Limit) {
     EXPECT_STREQ("1", result.items[1]->getKey().c_str());
     EXPECT_STREQ("2", result.items[2]->getKey().c_str());
     EXPECT_TRUE(result.items[3]->isCheckPointMetaItem());
+    EXPECT_EQ(range.start, result.range.start);
+    EXPECT_EQ(range.end + 2, result.range.end);
 
     // Asking for 5 items should give us all items in second checkpoint and
     // third checkpoint - 7 total
@@ -585,6 +587,8 @@ TEST_P(VBucketTest, GetItemsForCursor_Limit) {
     EXPECT_TRUE(result.items[4]->isCheckPointMetaItem());
     EXPECT_STREQ("5", result.items[5]->getKey().c_str());
     EXPECT_STREQ("6", result.items[6]->getKey().c_str());
+    EXPECT_EQ(range.end + 2, result.range.start);
+    EXPECT_EQ(range.end + 6, result.range.end);
 }
 
 // Check that getItemsToPersist() can correctly impose a limit on items fetched.
@@ -600,7 +604,8 @@ TEST_P(VBucketTest, GetItemsToPersist_Limit) {
 
     // Add 2 items to checkpoint manager (in addition to initial
     // checkpoint_start).
-    auto keys = generateKeys(2, 5);
+    const int itemsToGenerate = 2;
+    auto keys = generateKeys(itemsToGenerate, 5);
     setMany(keys, MutationStatus::WasClean);
 
     // Test - fetch items in chunks spanning the different item sources.
@@ -610,12 +615,16 @@ TEST_P(VBucketTest, GetItemsToPersist_Limit) {
     EXPECT_TRUE(result.moreAvailable);
     EXPECT_EQ(1, result.items.size());
     EXPECT_STREQ("1", result.items[0]->getKey().c_str());
+    EXPECT_EQ(range.start, result.range.start);
+    EXPECT_EQ(range.end + itemsToGenerate, result.range.end);
 
     result = this->vbucket->getItemsToPersist(2);
     EXPECT_TRUE(result.moreAvailable);
     EXPECT_EQ(2, result.items.size());
     EXPECT_STREQ("2", result.items[0]->getKey().c_str());
     EXPECT_STREQ("3", result.items[1]->getKey().c_str());
+    EXPECT_EQ(range.start, result.range.start);
+    EXPECT_EQ(range.end + itemsToGenerate, result.range.end);
 
     // Next call should read 1 item from backfill; and *all* items from
     // checkpoint (even through we only asked for 2 total), as it is not valid
@@ -627,6 +636,8 @@ TEST_P(VBucketTest, GetItemsToPersist_Limit) {
     EXPECT_TRUE(result.items[1]->isCheckPointMetaItem());
     EXPECT_STREQ("5", result.items[2]->getKey().c_str());
     EXPECT_STREQ("6", result.items[3]->getKey().c_str());
+    EXPECT_EQ(range.start, result.range.start);
+    EXPECT_EQ(range.end + itemsToGenerate, result.range.end);
 }
 
 // Check that getItemsToPersist() correctly returns `moreAvailable` if we

engines/ep/tests/module_tests/vbucket_test.h

@@ -98,5 +98,7 @@ class VBucketTestBase {
     CheckpointConfig checkpoint_config;
     Configuration config;
     const void* cookie = {};
+    const uint64_t lastSeqno{1000};
+    const snapshot_range_t range{5, lastSeqno};
 };