MB-20182: Update checkpoint snapshot correctly during TAP backfill

When we do a TAP backfill we must update checkpoint snapshot start
and end correctly. Otherwise, if we immediately proceed to DCP
after TAP backfill, the checkpoint mgr will have incorrect combination
of {snap_start, snap_end, vb_high_seqno}

Change-Id: I2b738fd3b24486dadbd2962e81e0c3820c5a8786
Reviewed-on: http://review.couchbase.org/65866
Tested-by: buildbot <build@couchbase.com>
Well-Formed: buildbot <build@couchbase.com>
Reviewed-by: Manu Dhundi <manu@couchbase.com>

diff --git a/src/ep.cc b/src/ep.cc
index 09449ae..a49602e 100644 (file)
--- a/src/ep.cc
+++ b/src/ep.cc
@@ -3338,6 +3338,14 @@ void EventuallyPersistentStore::queueDirty(RCPtr<VBucket> &vb,
                                                                  genBySeqno);
         v->setBySeqno(qi->getBySeqno());
 
                                                                  genBySeqno);
         v->setBySeqno(qi->getBySeqno());
 

+        /* During backfill on a TAP receiver we need to update the snapshot
+           range in the checkpoint. Has to be done here because in case of TAP
+           backfill, above, we use vb->queueBackfillItem() instead of
+           vb->checkpointManager.queueDirty() */
+        if (tapBackfill && genBySeqno) {
+            vb->checkpointManager.resetSnapshotRange();
+        }
+

         if (seqno) {
             *seqno = v->getBySeqno();
         }
         if (seqno) {
             *seqno = v->getBySeqno();
         }

diff --git a/tests/ep_testsuite.cc b/tests/ep_testsuite.cc
index 63b6fbd..ba9f8a5 100644 (file)
--- a/tests/ep_testsuite.cc
+++ b/tests/ep_testsuite.cc
@@ -6390,6 +6390,71 @@ static enum test_result test_tap_rcvr_mutate(ENGINE_HANDLE *h, ENGINE_HANDLE_V1
     return SUCCESS;
 }
 
     return SUCCESS;
 }
 

+/* Tests if checkpoint snapshot/highseqno are updated correctly after TAP a
+ conusmer receives items in backfill phase from a TAP producer (MB-20182) */
+static enum test_result test_tap_rcvr_backfill(ENGINE_HANDLE *h,
+                                               ENGINE_HANDLE_V1 *h1) {
+    uint16_t vbid = 1;
+
+    check(set_vbucket_state(h, h1, vbid, vbucket_state_replica),
+          "Failed to set vbucket state to replica");
+
+    const void *cookie = testHarness.create_cookie();
+    uint32_t cc = htonl(TAP_OPAQUE_ENABLE_CHECKPOINT_SYNC);
+    checkeq(ENGINE_SUCCESS, h1->tap_notify(h, cookie, &cc, sizeof(uint32_t),
+                                           1, 0, TAP_OPAQUE, 1, "", 0, 828, 0,
+                                           1, PROTOCOL_BINARY_RAW_BYTES,
+                                           nullptr, 0, vbid),
+            "Failed to enable checkpoint in tap.");
+
+    /* Start TAP backfill */
+    cc = htonl(TAP_OPAQUE_INITIAL_VBUCKET_STREAM);
+    checkeq(ENGINE_SUCCESS, h1->tap_notify(h, cookie, &cc, sizeof(uint32_t),
+                                           1, 0, TAP_OPAQUE, 1, "", 0, 828, 0,
+                                           1, PROTOCOL_BINARY_RAW_BYTES,
+                                           nullptr, 0, vbid),
+            "Failed to set tap backfill phase.");
+
+    /* Write 2 items on replica via TAP connection */
+    const int num_items = 2;
+    for (int i = 0; i < num_items; ++i) {
+        /* eng_specific should be > TapEngineSpecific::sizeRevSeqno(8) */
+        uint8_t eng_specific[9] = {0};
+        std::string value("value");
+        checkeq(ENGINE_SUCCESS, h1->tap_notify(h, cookie, eng_specific,
+                                               sizeof(eng_specific), 1, 0,
+                                               TAP_MUTATION, 1, "key", 3, 828,
+                                               0, 1, PROTOCOL_BINARY_RAW_BYTES,
+                                               value.c_str(), value.length(),
+                                               vbid),
+                "Failed to send tap muation");
+    }
+
+    /* Check if 2 items are correctly written on replica vb */
+    std::string abs_high_stat_str("vb_" + std::to_string(vbid) +
+                                  ":abs_high_seqno");
+    checkeq(num_items,
+            get_int_stat(h, h1, abs_high_stat_str.c_str(), "vbucket-seqno"),
+            "items not written on replica vb correctly");
+
+    /* close the TAP backfill */
+    cc = htonl(TAP_OPAQUE_CLOSE_BACKFILL);
+    checkeq(ENGINE_SUCCESS, h1->tap_notify(h, cookie, &cc, sizeof(uint32_t),
+                                           1, 0, TAP_OPAQUE, 1, "", 0, 828, 0,
+                                           1, PROTOCOL_BINARY_RAW_BYTES,
+                                           nullptr, 0, vbid),
+            "Failed to close tap backfill phase.");
+
+    /* Check if the snapshot end seqno is updated correctly on replica vb */
+    std::string high_stat_str("vb_" + std::to_string(vbid) + ":high_seqno");
+    checkeq(num_items,
+            get_int_stat(h, h1, high_stat_str.c_str(), "vbucket-seqno"),
+            "snapshot numbers not updated correctly after TAP backfill");
+
+    testHarness.destroy_cookie(cookie);
+    return SUCCESS;
+}
+

 static enum test_result test_tap_rcvr_checkpoint(ENGINE_HANDLE *h, ENGINE_HANDLE_V1 *h1) {
     char data;
     char eng_specific[64];
 static enum test_result test_tap_rcvr_checkpoint(ENGINE_HANDLE *h, ENGINE_HANDLE_V1 *h1) {
     char data;
     char eng_specific[64];


@@ -15106,6 +15171,8 @@ engine_test_t* get_tests(void) {
                  "tap_noop_interval=10", prepare, cleanup),
         TestCase("tap receiver mutation", test_tap_rcvr_mutate, test_setup,
                  teardown, NULL, prepare, cleanup),
                  "tap_noop_interval=10", prepare, cleanup),
         TestCase("tap receiver mutation", test_tap_rcvr_mutate, test_setup,
                  teardown, NULL, prepare, cleanup),

+        TestCase("tap receiver backfill", test_tap_rcvr_backfill,
+                 test_setup, teardown, NULL, prepare, cleanup),

         TestCase("tap receiver checkpoint start/end", test_tap_rcvr_checkpoint,
                  test_setup, teardown, NULL, prepare, cleanup),
         TestCase("tap receiver vbucket state", test_tap_rcvr_set_vbstate,
         TestCase("tap receiver checkpoint start/end", test_tap_rcvr_checkpoint,
                  test_setup, teardown, NULL, prepare, cleanup),
         TestCase("tap receiver vbucket state", test_tap_rcvr_set_vbstate,