MB-31327: Ephemeral backfill does not respect purge seqno.

There exists a bug in Ephemeral backfill_memory such that we can
start a backfill for a startSeqno > purgeSeqno of a given vBucket.
This leads to data inconsistencies as we may not replciate a delete.

Fix the bug by checking that the purgeSeqno < startSeqno of a
backfill when startSeqno != 1.

Only fix for buffered memory backfill as the non-buffered backfill
is not used and will be deleted in master.

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

Author: BenHuddleston

engines/ep/src/dcp/backfill_memory.cc

@@ -204,6 +204,22 @@ backfill_status_t DCPBackfillMemoryBuffered::create() {
         return backfill_snooze;
     }
 
+    /* Check startSeqno against the purge-seqno of the vb.
+     * If the startSeqno != 1 (a 0 to n request) then startSeqno must be
+     * greater than purgeSeqno. */
+    if (startSeqno != 1 && (startSeqno <= evb->getPurgeSeqno())) {
+        LOG(EXTENSION_LOG_WARNING,
+            "DCPBackfillMemoryBuffered::create(): "
+            "(vb:%" PRIu16
+            ") running backfill failed because the startSeqno:%" PRIu64
+            " is < purgeSeqno:%" PRIu64,
+            getVBucketId(),
+            startSeqno,
+            evb->getPurgeSeqno());
+        stream->setDead(END_STREAM_ROLLBACK);
+        return backfill_finished;
+    }
+
     /* Advance the cursor till start, mark snapshot and update backfill
        remaining count */
     while (rangeItr.curr() != rangeItr.end()) {

engines/ep/tests/mock/mock_stream.h

@@ -158,6 +158,8 @@ class MockActiveStream : public ActiveStream {
 
     /// A callback to allow tests to inject code before we access the checkpoint
     std::function<void()> preGetOutstandingItemsCallback = [] { return; };
+
+    bool isDead() { return ActiveStream::getState() == StreamState::Dead; };
 };
 
 /**

engines/ep/tests/module_tests/dcp_test.cc

@@ -1119,6 +1119,23 @@ TEST_P(StreamTest, backfillGetsNoItems) {
     }
 }
 
+TEST_P(StreamTest, bufferedMemoryBackfillPurgeGreaterThanStart) {
+    if (engine->getConfiguration().getBucketType() == "ephemeral") {
+        setup_dcp_stream(0, IncludeValue::No, IncludeXattrs::No);
+        auto evb = std::shared_ptr<EphemeralVBucket>(
+                std::dynamic_pointer_cast<EphemeralVBucket>(vb0));
+
+        // Force the purgeSeqno because it's easier than creating and
+        // deleting items
+        evb->setPurgeSeqno(3);
+
+        // Backfill with start != 1 and start != end and start < purge
+        DCPBackfillMemoryBuffered dcpbfm (evb, stream, 2, 4);
+        dcpbfm.run();
+        EXPECT_TRUE(stream->isDead());
+    }
+}
+
 /* Regression test for MB-17766 - ensure that when an ActiveStream is preparing
  * queued items to be sent out via a DCP consumer, that nextCheckpointItem()
  * doesn't incorrectly return false (meaning that there are no more checkpoint

engines/ep/tests/module_tests/evp_store_single_threaded_test.cc

@@ -26,6 +26,7 @@
 #include "bgfetcher.h"
 #include "checkpoint.h"
 #include "dcp/dcpconnmap.h"
+#include "ephemeral_tombstone_purger.h"
 #include "ep_time.h"
 #include "evp_store_test.h"
 #include "failover-table.h"
@@ -42,6 +43,7 @@
 #include <xattr/utils.h>
 
 #include <thread>
+#include <engines/ep/src/ephemeral_vb.h>
 
 ProcessClock::time_point SingleThreadedKVBucketTest::runNextTask(
         TaskQueue& taskQ, const std::string& expectedTaskName) {
@@ -290,6 +292,111 @@ void SingleThreadedKVBucketTest::runCompaction(uint64_t purgeBeforeTime,
                 "Compact DB file 0");
 }
 
+/*
+ * MB-31175
+ * The following test checks to see that when we call handleSlowStream in an
+ * in memory state and drop the cursor/schedule a backfill as a result, the
+ * resulting backfill checks the purgeSeqno and tells the stream to rollback
+ * if purgeSeqno > startSeqno.
+ */
+TEST_P(STParameterizedBucketTest, SlowStreamBackfillPurgeSeqnoCheck) {
+    // Make vbucket active.
+    setVBucketStateAndRunPersistTask(vbid, vbucket_state_active);
+    auto vb = store->getVBuckets().getBucket(vbid);
+    ASSERT_TRUE(vb.get());
+
+    // Store two items
+    std::array<std::string, 2> initialKeys = {{"k1", "k2"}};
+    for (const auto& key : initialKeys) {
+        store_item(vbid, makeStoredDocKey(key), key);
+    }
+    flushVBucketToDiskIfPersistent(vbid, initialKeys.size());
+
+    // Delete the items so that we can advance the purgeSeqno using
+    // compaction later
+    for (const auto& key : initialKeys) {
+        delete_item(vbid, makeStoredDocKey(key));
+    }
+    flushVBucketToDiskIfPersistent(vbid, initialKeys.size());
+
+    auto& ckpt_mgr = *vb->checkpointManager;
+
+    // Create a Mock Dcp producer
+    // Create the Mock Active Stream with a startSeqno of 1
+    // as a startSeqno is always valid
+    auto producer = std::make_shared<MockDcpProducer>(
+            *engine,
+            cookie,
+            "test_producer",
+            /*notifyOnly*/ false,
+            cb::const_byte_buffer() /*no json*/);
+            // Create a Mock Active Stream
+            auto mock_stream = std::make_shared<MockActiveStream>(
+            static_cast<EventuallyPersistentEngine*>(engine.get()),
+            producer,
+            /*flags*/ 0,
+            /*opaque*/ 0,
+            *vb,
+            /*st_seqno*/ 1,
+            /*en_seqno*/ ~0,
+            /*vb_uuid*/ 0xabcd,
+            /*snap_start_seqno*/ 0,
+            /*snap_end_seqno*/ ~0,
+            IncludeValue::Yes,
+            IncludeXattrs::Yes);
+
+    producer->createCheckpointProcessorTask();
+    producer->scheduleCheckpointProcessorTask();
+
+    mock_stream->transitionStateToBackfilling();
+    ASSERT_TRUE(mock_stream->isInMemory())
+    << "stream state should have transitioned to InMemory";
+
+    // Check number of expected cursors (might not have persistence cursor)
+    int expectedCursors = persistent() ? 2 : 1;
+    EXPECT_EQ(expectedCursors, ckpt_mgr.getNumOfCursors());
+
+    EXPECT_TRUE(mock_stream->handleSlowStream());
+    EXPECT_TRUE(mock_stream->public_getPendingBackfill());
+
+    // Might not have persistence cursor
+    expectedCursors = persistent() ? 1 : 0;
+    EXPECT_EQ(expectedCursors, ckpt_mgr.getNumOfCursors())
+    << "stream cursor should have been dropped";
+
+    // This will schedule the backfill
+    mock_stream->transitionStateToBackfilling();
+    ASSERT_TRUE(mock_stream->isBackfilling());
+
+    // Advance the purgeSeqno
+    if (persistent()) {
+        runCompaction(~0, 3);
+    } else {
+        EphemeralVBucket::HTTombstonePurger purger(0);
+        auto vbptr = store->getVBucket(vbid);
+        EphemeralVBucket* evb = dynamic_cast<EphemeralVBucket*>(vbptr.get());
+        purger.setCurrentVBucket(*evb);
+        evb->ht.visit(purger);
+        evb->purgeStaleItems();
+    }
+
+    ASSERT_EQ(3, vb->getPurgeSeqno());
+
+    // Run the backfill we scheduled when we transitioned to the backfilling
+    // state
+    auto& bfm = producer->getBFM();
+    bfm.backfill();
+
+    // The backfill should have set the stream state to dead because
+    // purgeSeqno > startSeqno
+    EXPECT_TRUE(mock_stream->isDead());
+
+    // Stop Producer checkpoint processor task
+    producer->cancelCheckpointCreatorTask();
+
+    cancelAndPurgeTasks();
+}
+
 /*
  * The following test checks to see if we call handleSlowStream when in a
  * backfilling state, but the backfillTask is not running, we
@@ -3380,3 +3487,13 @@ INSTANTIATE_TEST_CASE_P(XattrCompressedTest,
                         XattrCompressedTest,
                         ::testing::Combine(::testing::Bool(),
                                            ::testing::Bool()), );
+
+static auto allConfigValues = ::testing::Values(
+        std::make_tuple(std::string("ephemeral"), std::string("auto_delete")),
+        std::make_tuple(std::string("ephemeral"), std::string("fail_new_data")),
+        std::make_tuple(std::string("persistent"), std::string{}));
+
+// Test cases which run for persistent and ephemeral buckets
+INSTANTIATE_TEST_CASE_P(EphemeralOrPersistent,
+                        STParameterizedBucketTest,
+                        allConfigValues, );

engines/ep/tests/module_tests/evp_store_single_threaded_test.h

@@ -160,3 +160,34 @@ class SingleThreadedEPBucketTest : public SingleThreadedKVBucketTest {
         return dynamic_cast<EPBucket&>(*store);
     }
 };
+
+/**
+ * Test fixture for KVBucket tests running in single-threaded mode.
+ *
+ * Parameterised on a pair of:
+ * - bucket_type (ephemeral of persistent)
+ * - ephemeral_full_policy (for specifying ephemeral auto-delete & fail_new_data
+ *   eviction modes). If empty then unused (persistent buckets).
+ */
+class STParameterizedBucketTest
+        : public SingleThreadedEPBucketTest,
+          public ::testing::WithParamInterface<
+                  std::tuple<std::string, std::string>> {
+public:
+    bool persistent() {
+        return std::get<0>(GetParam()) == "persistent";
+    }
+
+protected:
+    void SetUp() {
+        if (!config_string.empty()) {
+            config_string += ";";
+        }
+        config_string += "bucket_type=" + std::get<0>(GetParam());
+        auto ephFullPolicy = std::get<1>(GetParam());
+        if (!ephFullPolicy.empty()) {
+            config_string += ";ephemeral_full_policy=" + ephFullPolicy;
+        }
+        SingleThreadedEPBucketTest::SetUp();
+    }
+};

engines/ep/tests/module_tests/item_pager_test.cc

@@ -36,34 +36,6 @@
 #include <xattr/blob.h>
 #include <xattr/utils.h>
 
-/**
- * Test fixture for KVBucket tests running in single-threaded mode.
- *
- * Parameterised on a pair of:
- * - bucket_type (ephemeral of persistent)
- * - ephemeral_full_policy (for specifying ephemeral auto-delete & fail_new_data
- *   eviction modes). If empty then unused (persistent buckets).
- */
-class STParameterizedBucketTest
-        : public SingleThreadedEPBucketTest,
-          public ::testing::WithParamInterface<
-                  std::tuple<std::string, std::string>> {
-protected:
-    void SetUp() override;
-};
-
-void STParameterizedBucketTest::SetUp() {
-    if (!config_string.empty()) {
-        config_string += ";";
-    }
-    config_string += "bucket_type=" + std::get<0>(GetParam());
-    auto ephFullPolicy = std::get<1>(GetParam());
-    if (!ephFullPolicy.empty()) {
-        config_string += ";ephemeral_full_policy=" + ephFullPolicy;
-    }
-    SingleThreadedEPBucketTest::SetUp();
-}
-
 /**
  * Test fixture for bucket quota tests. Sets quota (max_size) to 200KB and
  * enables the MemoryTracker.