MB-60046: Ephemeral: Don't auto-delete committed and pending keys.

Prior to this change the auto-delete path was able to generate a delete
of the committed item whilst there is a pending write. The delete would
be rejected by the consumer and disconnect DCP (causing a cycle of
disconnect).

This change updates EphemeralVBucket::eligibleToPageOut to check the
current hash-table bucket for a pending key.

This approach is deliberately simple and the inefficient extra search of
the hash-chain is tolerated to provide a rapid solution to this issue.

Change-Id: If83ecb927026a4ef4ca2dff813a8729ec67da211
Reviewed-on: https://review.couchbase.org/c/kv_engine/+/202456
Well-Formed: Restriction Checker
Reviewed-by: Paolo Cocchi <[email protected]>
Tested-by: Jim Walker <[email protected]>

Author: jimwwalker

engines/ep/src/ephemeral_vb.cc

@@ -156,6 +156,15 @@ bool EphemeralVBucket::eligibleToPageOut(const HashTable::HashBucketLock& lh,
         // The system event documents must not be paged-out
         return false;
     }
+
+    // Cannot auto-delete this v if it has a pending v. Simplest is to
+    // look again for a pending v in the hash-table.
+    // @todo MB-60105: Tracks that scanning the hash-chain again is not the most
+    // efficient fix.
+    if (ht.unlocked_find(lh, v.getKey()).pendingSV) {
+        return false;
+    }
+
     return true;
 }
 

engines/ep/src/hash_table.cc

@@ -329,6 +329,12 @@ HashTable::FindInnerResult HashTable::findInner(const DocKey& key) {
                 "non-active object");
     }
     HashBucketLock hbl = getLockedBucket(key);
+    auto rv = unlocked_find(hbl, key);
+    return {std::move(hbl), rv.committedSV, rv.pendingSV};
+}
+
+HashTable::UnlockedFindResult HashTable::unlocked_find(
+        const HashBucketLock& hbl, const DocKey& key) const {
     // Scan through all elements in the hash bucket chain looking for Committed
     // and Pending items with the same key.
     StoredValue* foundCmt = nullptr;
@@ -346,7 +352,7 @@ HashTable::FindInnerResult HashTable::findInner(const DocKey& key) {
         }
     }
 
-    return {std::move(hbl), foundCmt, foundPend};
+    return {foundCmt, foundPend};
 }
 
 HashTable::RandomKeyVisitor::RandomKeyVisitor(size_t size, int random)

engines/ep/src/hash_table.h

@@ -1199,6 +1199,26 @@ class HashTable {
                                StoredValue& value,
                                CommittedState state);
 
+    struct UnlockedFindResult {
+        /// The Committed StoredValue with this key if in HashTable, else
+        /// nullptr.
+        StoredValue* committedSV{nullptr};
+        /// The Committed StoredValue with this key if in HashTable, else
+        /// nullptr.
+        StoredValue* pendingSV{nullptr};
+    };
+
+    /**
+     * Attempt to find the pending and committed versions of key in the given
+     * hbl.getBucketNum hash bucket.
+     *
+     * @param hbl The structure which holds the lock and the bucket number
+     * @param key The key to find
+     * @return pair of pointers, null when nothing found.
+     */
+    UnlockedFindResult unlocked_find(const HashBucketLock& hbl,
+                                     const DocKey& key) const;
+
     /**
      * Releases an item(StoredValue) in the hash table, but does not delete it.
      * It will pass out the removed item to the caller who can decide whether to

engines/ep/tests/module_tests/ephemeral_vb_test.cc

@@ -472,6 +472,29 @@ TEST_F(EphemeralVBucketTest, SoftDeleteDuringBackfill) {
               mockEpheVB->public_getNumListItems());
 }
 
+// MB-60046: The ephemeral auto-delete was able to delete a committed item which
+// has a pending write (disrupting DCP). Test checks eligibleToPageOut now
+// detects this case.
+TEST_F(EphemeralVBucketTest, CannotPageOutCommittedAndPendingKey) {
+    auto meta = nlohmann::json{
+            {"topology", nlohmann::json::array({{"active", "replica"}})}};
+    vbucket->setState(vbucket_state_active, &meta);
+    // Commit and item and then also make it pending.
+    auto key = makeStoredDocKey("key");
+    auto item = makeCommittedItem(key, "value");
+    ASSERT_EQ(MutationStatus::WasClean,
+              public_processSet(*item, item->getCas()));
+    item = makePendingItem(key, "value");
+    VBQueueItemCtx ctx;
+    ctx.durability =
+            DurabilityItemCtx{item->getDurabilityReqs(), nullptr /*cookie*/};
+    ASSERT_EQ(MutationStatus::WasClean,
+              public_processSet(*item, item->getCas(), ctx));
+    auto result = vbucket->ht.findForRead(key);
+    ASSERT_TRUE(result.storedValue);
+    ASSERT_FALSE(vbucket->eligibleToPageOut(result.lock, *result.storedValue));
+}
+
 // EphemeralVB Tombstone Purging //////////////////////////////////////////////
 
 class EphTombstoneTest : public EphemeralVBucketTest {

engines/ep/tests/module_tests/evp_store_single_threaded_test.h

@@ -718,6 +718,13 @@ class STParameterizedBucketTest
                 "item_eviction_policy=full_eviction"s);
     }
 
+    static auto ephAutoDeleteConfigValues() {
+        using namespace std::string_literals;
+        return ::testing::Values(
+                "bucket_type=ephemeral:"
+                "ephemeral_full_policy=auto_delete"s);
+    }
+
     bool persistent() const {
         return GetParam().find("persistent") != std::string::npos;
     }

engines/ep/tests/module_tests/item_pager_test.cc

@@ -1573,6 +1573,94 @@ TEST_P(STEphemeralItemPagerTest, ReplicaNotPaged) {
     }
 }
 
+/**
+ * Test fixture for Ephemeral item pager tests with
+ * ephemeral_full_policy=auto_delete.
+ */
+class STEphemeralAutoDeleteItemPagerTest : public STItemPagerTest {};
+
+// Test covers MB-60046. A pending and committed key must not find itself auto
+// deleted, else the delete of the commit will be rejected by the DCP replica
+// and cause disconnects.
+TEST_P(STEphemeralAutoDeleteItemPagerTest, MB_60046) {
+    setVBucketState(
+            vbid,
+            vbucket_state_active,
+            {{"topology", nlohmann::json::array({{"active", "replica"}})}});
+
+    auto setActive = [this](int count) {
+        auto item =
+                make_item(vbid,
+                          makeStoredDocKey("active_" + std::to_string(count)),
+                          std::string(512, 'V'));
+        item.setFreqCounterValue(0);
+        storeItem(item);
+    };
+
+    int count = 0;
+    do {
+        setActive(++count);
+    } while (store->getPageableMemCurrent() <
+             store->getPageableMemHighWatermark());
+
+    auto key = makeStoredDocKey("active_" + std::to_string(count));
+    // Now make the last key written in the loop also pending.
+    ASSERT_EQ(cb::engine_errc::would_block,
+              storeItem(*makePendingItem(key, "pending-value")));
+
+    std::vector<queued_item> items;
+    auto& vb = *store->getVBucket(vbid);
+    auto cursor = vb.checkpointManager->registerCursorBySeqno(
+            "test", 1, CheckpointCursor::Droppable::No);
+    ASSERT_FALSE(cursor.tryBackfill) << *vb.checkpointManager;
+    vb.checkpointManager->getNextItemsForCursor(cursor.cursor.lock().get(),
+                                                items);
+    EXPECT_NE(0, items.size()) << *vb.checkpointManager;
+    bool pendingFound{false};
+    bool committedFound{false};
+    for (const auto& item : items) {
+        if (item->getKey() == key) {
+            if (item->isPending()) {
+                ASSERT_FALSE(pendingFound);
+                pendingFound = true;
+            } else {
+                ASSERT_FALSE(committedFound);
+                committedFound = true;
+            }
+        }
+    }
+
+    // Both pending and committed must be seen before we trigger auto-delete
+    ASSERT_TRUE(pendingFound);
+    ASSERT_TRUE(committedFound);
+    items.clear();
+
+    // Now trigger the auto-delete paging
+    runHighMemoryPager();
+
+    // Iterate through the checkpoint and check all deletes.
+    vb.checkpointManager->getNextItemsForCursor(cursor.cursor.lock().get(),
+                                                items);
+    // Should be some new mutations (deletes)
+    EXPECT_NE(0, items.size()) << *vb.checkpointManager;
+
+    int deleteCount{0};
+    for (const auto& item : items) {
+        if (item->getKey() == key) {
+            // The key should not be found in the new batch of items. The
+            // original commit+pending were in the first batch and not expected
+            // to see again. Prior to the fix, a delete(key) was seen here.
+            FAIL() << *item;
+        }
+        if (item->isDeleted()) {
+            ++deleteCount;
+        }
+    }
+
+    // We certainly should see deletes.
+    ASSERT_NE(0, deleteCount);
+}
+
 /**
  * Test fixture for expiry pager tests - enables the Expiry Pager (in addition
  * to what the parent class does).
@@ -2685,6 +2773,11 @@ INSTANTIATE_TEST_SUITE_P(PersistentFullValue,
                          STParameterizedBucketTest::persistentConfigValues(),
                          STParameterizedBucketTest::PrintToStringParamName);
 
+INSTANTIATE_TEST_SUITE_P(EphemeralAutoDelete,
+                         STEphemeralAutoDeleteItemPagerTest,
+                         STParameterizedBucketTest::ephAutoDeleteConfigValues(),
+                         STParameterizedBucketTest::PrintToStringParamName);
+
 #else
 GTEST_ALLOW_UNINSTANTIATED_PARAMETERIZED_TEST(STItemPagerTest);
 GTEST_ALLOW_UNINSTANTIATED_PARAMETERIZED_TEST(MultiPagingVisitorTest);
@@ -2694,4 +2787,6 @@ GTEST_ALLOW_UNINSTANTIATED_PARAMETERIZED_TEST(STValueEvictionExpiryPagerTest);
 GTEST_ALLOW_UNINSTANTIATED_PARAMETERIZED_TEST(MB_32669);
 GTEST_ALLOW_UNINSTANTIATED_PARAMETERIZED_TEST(STEphemeralItemPagerTest);
 GTEST_ALLOW_UNINSTANTIATED_PARAMETERIZED_TEST(MB_36087);
+GTEST_ALLOW_UNINSTANTIATED_PARAMETERIZED_TEST(
+        STEphemeralAutoDeleteItemPagerTest);
 #endif