[BP] MB-42610: Do not expire prepares when committed items exist

Currently if a committed item exists then it's possible for the
expiry code to attempt to expire a prepare. This obviously breaks
durability and it can lead to us having two committed items in the
HashTable for the same key which will throw assertions the next time
we lookup that key.

This can only happen in a very specific and limited set of
circumstances. The vBucket in question must go down having only
partially persisted a Disk snapshot. The vBucket must then flip from
replica to active (a situation that would lead to data loss) then
we must attempt to expire an item (via compaction or the pager).
This expiration could find a prepare in the HashTable and attempt to
expire it if the item has the same cas (this should only be possible
when we load a completed prepare).

Fix this issue by skipping expiration if we find a prepare with the
same cas of our item in the HashTable.

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

Author: BenHuddleston

engines/ep/src/vbucket.cc

@@ -2412,6 +2412,16 @@ void VBucket::deleteExpiredItem(const Item& it,
             return;
         }
 
+        if (v->isPending()) {
+            // If cas is the same (above statement) and the HashTable has
+            // returned a prepare then we must have loaded a logically complete
+            // prepare (as we remove them from the HashTable at completion) for
+            // some reason. The prepare should be in a maybe visible state but
+            // it probably isn't a good idea to assert that here. In this case
+            // we must do nothing as we MUST commit any maybe visible prepares.
+            return;
+        }
+
         if (v->isTempNonExistentItem() || v->isTempDeletedItem()) {
             bool deleted = deleteStoredValue(hbl, *v);
             if (!deleted) {

engines/ep/tests/module_tests/evp_store_single_threaded_test.cc

@@ -51,6 +51,7 @@
 #include "tests/module_tests/thread_gate.h"
 #include "tests/test_fileops.h"
 #include "vbucket_state.h"
+#include "vbucket_bgfetch_item.h"
 
 #include "../couchstore/src/internal.h"
 
@@ -4911,6 +4912,83 @@ TEST_P(STParamPersistentBucketTest,
     testAbortDoesNotIncrementOpsDelete(false /*flusherDedup*/);
 }
 
+// @TODO move to EPBucketFullEvictionTest on merge forward to master
+TEST_P(STParamPersistentBucketTest, ExpiryFindsPrepareWithSameCas) {
+    if (!fullEviction()) {
+        return;
+    }
+    setVBucketStateAndRunPersistTask(
+            vbid,
+            vbucket_state_active,
+            {{"topology", nlohmann::json::array({{"active", "replica"}})}});
+
+    // 1) Store prepare with expiry
+    auto key = makeStoredDocKey("a");
+    using namespace cb::durability;
+    auto pre = makePendingItem(key, "value", Requirements{Level::Majority, {}});
+    pre->setVBucketId(vbid);
+    pre->setExpTime(1);
+
+    EXPECT_EQ(ENGINE_SYNC_WRITE_PENDING, store->set(*pre, cookie));
+    flushVBucketToDiskIfPersistent(vbid, 1);
+
+    auto vb = store->getVBucket(vbid);
+
+    // 2) Seqno ack and commit the prepare
+    vb->seqnoAcknowledged(folly::SharedMutex::ReadHolder(vb->getStateLock()),
+                          "replica",
+                          1 /*prepareSeqno*/);
+    vb->processResolvedSyncWrites();
+
+    // 3) Fudge the current snapshot so that when we warmup we scan the entire
+    //    snapshot for prepares (i.e. incomplete disk snapshot)
+    vb->checkpointManager->updateCurrentSnapshot(3, 3, CheckpointType::Disk);
+    flushVBucketToDiskIfPersistent(vbid, 1);
+
+    // 4) Restart and warmup
+    vb.reset();
+    resetEngineAndWarmup();
+    vb = store->getVBucket(vbid);
+
+    {
+        // Verify that the prepare is there and it's "MaybeVisible"
+        auto ret = vb->ht.findForUpdate(key);
+        ASSERT_TRUE(ret.pending);
+        ASSERT_TRUE(ret.pending->isPreparedMaybeVisible());
+
+        // And that the commit is there too
+        ASSERT_TRUE(ret.committed);
+    }
+
+    // 5) Grab the item from disk just like the compactor would
+    vb_bgfetch_queue_t q;
+    vb_bgfetch_item_ctx_t ctx;
+    ctx.isMetaOnly = GetMetaOnly::No;
+    auto diskDocKey = makeDiskDocKey("a");
+    q[diskDocKey] = std::move(ctx);
+    store->getRWUnderlying(vbid)->getMulti(vbid, q);
+    EXPECT_EQ(ENGINE_SUCCESS, q[diskDocKey].value.getStatus());
+
+    // 6) Callback from the "compactor" with the item to try and expire it. We
+    //    could also pretend to be the pager here.
+    ASSERT_EQ(0, vb->numExpiredItems);
+    vb->deleteExpiredItem(*q[diskDocKey].value.item, 2, ExpireBy::Compactor);
+
+    // Item expiry cannot take place if the MaybeVisible prepare exists.
+    EXPECT_EQ(0, vb->numExpiredItems);
+    {
+        // Verify that the prepare is there and it's "MaybeVisible". Before the
+        // fix deleteExpiredItem would select and replace the prepare which is
+        // incorrect and causes us to have two committed items in the HashTable.
+        auto ret = vb->ht.findForUpdate(key);
+        ASSERT_TRUE(ret.pending);
+        ASSERT_TRUE(ret.pending->isPreparedMaybeVisible());
+
+        // And that the commit is there too
+        ASSERT_TRUE(ret.committed);
+    }
+}
+
 INSTANTIATE_TEST_CASE_P(Persistent,
                         STParamPersistentBucketTest,
                         STParameterizedBucketTest::persistentConfigValues(),