MB-41942: Persist the VB state earlier during warmup

Currently we attempt to persist the vBucket state(s) when we complete
warmup. This leads to a window during warmup in which we can write
new items (via expirations) after identifying that we need to generate
a new FailoverTable entry but before we persist it. If we then crash
or shut down in this period then we can fail to roll back any replica
nodes. This can in turn cause data loss/data corruption.

Correct this by persisting the vBucket state during warmup as soon as
we know that we need to generate a new FailoverTable entry.

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

Author: BenHuddleston

engines/ep/src/ep_bucket.cc

@@ -1869,10 +1869,6 @@ bool EPBucket::maybeEnableTraffic() {
 }
 
 void EPBucket::warmupCompleted() {
-    // Snapshot VBucket state after warmup to ensure Failover table is
-    // persisted.
-    scheduleVBStatePersist();
-
     if (engine.getConfiguration().getAlogPath().length() > 0) {
         if (engine.getConfiguration().isAccessScannerEnabled()) {
             {

engines/ep/src/warmup.cc

@@ -1063,6 +1063,12 @@ void Warmup::createVBuckets(uint16_t shardId) {
                 } else {
                     vb->failovers->createEntry(vbs.lastSnapStart);
                 }
+
+                // MB-41942: Persist the latest VB state to ensure that the
+                // FailoverTable is correct if we write any new seqnos during
+                // warmup (e.g. due to expiration)
+                vb->checkpointManager->queueSetVBState(*vb);
+
                 auto entry = vb->failovers->getLatestEntry();
                 EP_LOG_INFO(
                         "Warmup::createVBuckets: {} created new failover entry "

engines/ep/tests/module_tests/evp_store_single_threaded_test.cc

@@ -52,6 +52,7 @@
 #include "tests/test_fileops.h"
 #include "vbucket_state.h"
 #include "vbucket_bgfetch_item.h"
+#include "warmup.h"
 
 #include "../couchstore/src/internal.h"
 
@@ -4903,6 +4904,119 @@ void STParamPersistentBucketTest::testAbortDoesNotIncrementOpsDelete(
     EXPECT_EQ(0, vb.opsDelete);
 }
 
+
+void STParamPersistentBucketTest::testFailoverTableEntryPersistedAtWarmup(std::function<void()> testFunction) {
+    // 1) Store something so we can expire it later
+    engine->getKVBucket()->setVBucketState(vbid, vbucket_state_active);
+    auto vb = engine->getKVBucket()->getVBucket(vbid);
+
+    // Grab initialUuid for testing
+    auto initialUuid = vb->failovers->getLatestUUID();
+
+    auto key = makeStoredDocKey("key");
+    store_item(vbid, key, "value", 1 /*expiryTime*/);
+    flushVBucketToDiskIfPersistent(vbid, 1);
+    EXPECT_EQ(1, vb->getHighSeqno());
+
+    // 2) Restart as though we had an unclean shutdown (creating a new failover
+    //    table entry) and run the warmup up to the point of completion.
+    vb.reset();
+    resetEngineAndEnableWarmup();
+
+    auto& readerQueue = *task_executor->getLpTaskQ()[READER_TASK_IDX];
+    auto* warmup = engine->getKVBucket()->getWarmup();
+    ASSERT_TRUE(warmup);
+
+    // Warmup - load everything but don't run the complete phase which schedules
+    // persistence of the vBucket state (new failover entry)
+    while (warmup->getWarmupState() != WarmupState::State::Done) {
+        runNextTask(readerQueue);
+    }
+
+    // 3) Test
+    testFunction();
+
+    // New high seqno
+    vb = engine->getKVBucket()->getVBucket(vbid);
+    EXPECT_EQ(2, vb->getHighSeqno());
+
+    // Flush the expiry
+    flushVBucketToDiskIfPersistent(vbid, 1);
+
+    // Verify that the item has been expired
+    auto options = static_cast<get_options_t>(
+        QUEUE_BG_FETCH | HONOR_STATES | TRACK_REFERENCE | DELETE_TEMP |
+            HIDE_LOCKED_CAS | TRACK_STATISTICS);
+    auto gv = store->get(key, vbid, cookie, options);
+
+    if (gv.getStatus() == ENGINE_EWOULDBLOCK) {
+        runBGFetcherTask();
+        gv = store->get(key, vbid, cookie, options);
+    }
+
+    EXPECT_EQ(ENGINE_KEY_ENOENT, gv.getStatus());
+
+    // Get our new uuid now
+    auto secondUuid = vb->failovers->getLatestUUID();
+    ASSERT_NE(initialUuid, secondUuid);
+
+    // "Complete" the warmup or the test will get stuck shutting down, we won't
+    // actually flush the new vb state though so we're still testing as though
+    // this didn't happen
+    runNextTask(readerQueue);
+
+    // 4) Restart again
+    vb.reset();
+    resetEngineAndWarmup();
+
+    // 5) The test - uuid should have both of the previous entries
+    vb = engine->getKVBucket()->getVBucket(vbid);
+
+    auto failovers = vb->failovers->getFailoverLog();
+    auto itr = std::find_if(failovers.begin(), failovers.end(), [&initialUuid](const auto& failoverEntry) {
+        return failoverEntry.uuid == initialUuid;
+    });
+    EXPECT_NE(itr, failovers.end());
+
+    itr = std::find_if(failovers.begin(), failovers.end(), [&secondUuid](const auto& failoverEntry) {
+        return failoverEntry.uuid == secondUuid;
+    });
+    EXPECT_NE(itr, failovers.end());
+
+    EXPECT_EQ(2, vb->getHighSeqno());
+    gv = store->get(key, vbid, cookie, options);
+
+    if (gv.getStatus() == ENGINE_EWOULDBLOCK) {
+        runBGFetcherTask();
+        gv = store->get(key, vbid, cookie, options);
+    }
+}
+
+TEST_P(STParamPersistentBucketTest, TestExpiryDueToCompactionPersistsFailoverTableEntryDuringWarmup) {
+    testFailoverTableEntryPersistedAtWarmup([this]() {
+        CompactionConfig config;
+        engine->compactDB(vbid, config, cookie);
+        std::string taskDescription =
+            "Compact DB file " + std::to_string(vbid.get());
+        runNextTask(*task_executor->getLpTaskQ()[WRITER_TASK_IDX], taskDescription);
+    });
+}
+
+TEST_P(STParamPersistentBucketTest, TestExpiryDueToGetPersistsFailoverTableEntryDuringWarmup) {
+    testFailoverTableEntryPersistedAtWarmup([this]() {
+        auto options = static_cast<get_options_t>(
+            QUEUE_BG_FETCH | HONOR_STATES | TRACK_REFERENCE | DELETE_TEMP |
+                HIDE_LOCKED_CAS | TRACK_STATISTICS);
+        auto key = makeStoredDocKey("key");
+        auto gv = store->get(key, vbid, cookie, options);
+
+        if (gv.getStatus() == ENGINE_EWOULDBLOCK) {
+            runBGFetcherTask();
+            gv = store->get(key, vbid, cookie, options);
+        }
+    });
+}
+
 TEST_P(STParamPersistentBucketTest, AbortDoesNotIncrementOpsDelete) {
     testAbortDoesNotIncrementOpsDelete(true /*flusherDedup*/);
 }
@@ -4992,4 +5106,4 @@ TEST_P(STParamPersistentBucketTest, ExpiryFindsPrepareWithSameCas) {
 INSTANTIATE_TEST_CASE_P(Persistent,
                         STParamPersistentBucketTest,
                         STParameterizedBucketTest::persistentConfigValues(),
-                        STParameterizedBucketTest::PrintToStringParamName);
+                        STParameterizedBucketTest::PrintToStringParamName);

engines/ep/tests/module_tests/evp_store_single_threaded_test.h

@@ -338,4 +338,6 @@ class STParameterizedBucketTest
 class STParamPersistentBucketTest : public STParameterizedBucketTest {
 protected:
     void testAbortDoesNotIncrementOpsDelete(bool flusherDedup);
+
+    void testFailoverTableEntryPersistedAtWarmup(std::function<void()>);
 };