MB-42918: Do not replace committed item with pending at add

Currently when we do an add we can replace certain committed items
with pending ones. One of these cases is a when we have an unpersisted
delete. If we replace this with a pending StoredValue then we allow
a subsequent get to find no committed StoredValue in the HashTable and
potentially do a BgFetch. This BgFetch may return the alive document
back to the HashTable/client which it should not do.

Fix this by creating a new StoredValue if we would have replace a
committed one with a prepared one.

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

Author: BenHuddleston

engines/ep/src/vbucket.cc

@@ -3033,19 +3033,8 @@ void VBucket::deletedOnDiskCbk(const Item& queuedItem, bool deleted) {
          * be done to ensure that the item count is accurate in the case of full
          * eviction. We should only decrement the counter for committed (via
          * mutation or commit) items as we only increment for these.
-         * Note this is done irrespective of if a StoredValue was found in the
-         * HashTable - the SV may have already been replaced (e.g. with
-         * a pending SyncWrite). We "know" the delete happened on disk thus
-         * should decrement the total number of items.
-         *
-         * @todo: About "the SV may have already been replaced (e.g. with
-         * a pending SyncWrite)" above.. Need to check if that is true. If it
-         * is, that shouldn't happen as we may end up with this at FullEviction:
-         *  - deleted key queued for persistence and removed from the HT
-         *  - deletion not persisted yet
-         *  - frontend read bg-fecthes a previous alive state (if any) from disk
          */
-        if (queuedItem.isCommitted()) {
+        if (v && queuedItem.isCommitted()) {
             decrNumTotalItems();
             ++opsDelete;
         }
@@ -3470,7 +3459,16 @@ std::pair<AddStatus, std::optional<VBNotifyCtx>> VBucket::processAdd(
 
     std::pair<AddStatus, VBNotifyCtx> rv = {AddStatus::Success, {}};
 
-    if (v) {
+    // We cannot replace a committed SV with a pending one. If we were to do
+    // so then a delete that has not yet been persisted could be replaced
+    // with a prepare. A subsequent get could trigger a bg fetch that may
+    // return the old (not deleted) document from disk if it runs before the
+    // flusher. As such, we must keep the unpersisted delete and add a new
+    // prepare for the SyncWrite. Any get will see the unpersisted delete
+    // and return KEYNOENT.
+    auto replacingCommittedItemWithPending =
+            v && v == htRes.committed && itm.isPending();
+    if (v && !replacingCommittedItemWithPending) {
         if (v->isTempInitialItem() && eviction == EvictionPolicy::Full &&
             maybeKeyExists) {
             // Need to figure out if an item exists on disk
@@ -3494,7 +3492,8 @@ std::pair<AddStatus, std::optional<VBNotifyCtx>> VBucket::processAdd(
         std::tie(v, std::ignore, rv.second) =
                 updateStoredValue(htRes.getHBL(), *v, itm, queueItmCtx);
     } else {
-        if (itm.getBySeqno() != StoredValue::state_temp_init) {
+        if (itm.getBySeqno() != StoredValue::state_temp_init &&
+            !replacingCommittedItemWithPending) {
             if (eviction == EvictionPolicy::Full && maybeKeyExists) {
                 return {AddStatus::AddTmpAndBgFetch, VBNotifyCtx()};
             }
@@ -3514,6 +3513,8 @@ std::pair<AddStatus, std::optional<VBNotifyCtx>> VBucket::processAdd(
 
         if (v->isTempItem()) {
             rv.first = AddStatus::BgFetch;
+        } else if (replacingCommittedItemWithPending) {
+            rv.first = AddStatus::UnDel;
         }
     }
 

engines/ep/tests/module_tests/evp_store_test.cc

@@ -974,6 +974,62 @@ TEST_P(EPBucketFullEvictionTest, ExpiryFindsPrepareWithSameCas) {
     }
 }
 
+TEST_P(EPBucketFullEvictionTest, UnDelWithPrepare) {
+    setVBucketStateAndRunPersistTask(
+            vbid,
+            vbucket_state_active,
+            {{"topology", nlohmann::json::array({{"active", "replica"}})}});
+
+    // 1) Store, delete, and persist the item. We need to do this to ensure that
+    //    the bloom filter tells us to go to disk when we try gets if there is
+    //    no delete in the HashTable.
+    auto key = makeStoredDocKey("key");
+    storeAndDeleteItem(vbid, key, "value");
+    auto vb = store->getVBucket(vbid);
+    EXPECT_EQ(0, vb->getNumItems());
+
+    // 1) Store the new item and persist
+    store_item(vbid, key, "value");
+    flushVBucketToDiskIfPersistent(vbid, 1);
+    EXPECT_EQ(1, vb->getNumItems());
+
+    // 2) Store the new item but don't persist it yet. We want to test what
+    //    happens when it's dirty.
+    delete_item(vbid, key);
+
+    // 1 because we haven't persisted the delete yet
+    EXPECT_EQ(1, vb->getNumItems());
+
+    // 3) Get now returns not found
+    auto options = static_cast<get_options_t>(
+            QUEUE_BG_FETCH | HONOR_STATES | TRACK_REFERENCE | DELETE_TEMP |
+            HIDE_LOCKED_CAS | TRACK_STATISTICS);
+    auto gv = getInternal(key, vbid, cookie, ForGetReplicaOp::No, options);
+    EXPECT_EQ(ENGINE_KEY_ENOENT, gv.getStatus());
+
+    // 4) Add prepare
+    auto prepare = makePendingItem(key, "value");
+    EXPECT_EQ(ENGINE_SYNC_WRITE_PENDING, addItem(*prepare, cookie));
+
+    // 1 because we haven't persisted the delete yet
+    EXPECT_EQ(1, vb->getNumItems());
+
+    // 5) Check that the HashTable state is now correct
+    {
+        auto htRes = vb->ht.findForUpdate(key);
+        ASSERT_TRUE(htRes.committed);
+        EXPECT_TRUE(htRes.committed->isDeleted());
+        EXPECT_TRUE(htRes.pending);
+    }
+
+    // @TODO RDB: Rocks item counting is broken and overcounts assuming
+    // everything is a new item
+    if (!isRocksDB()) {
+        flushVBucketToDiskIfPersistent(vbid, 2);
+        EXPECT_EQ(0, vb->getNumItems());
+    }
+}
+
 /**
  * Verify that when getIf is used it only fetches the metdata from disk for
  * the filter, and not the complete document.