MB-34412: Hoist isPending check to VBucket::process* methods

We need to be able to overwrite a prepare unconditionally on a
replica so pull these functions up to processSet/processDelete so
that we do not have to add new HashTable functions.

Change-Id: If36a92f8ef2d13d251df7895d3280cf484b72244
Reviewed-on: http://review.couchbase.org/110145
Tested-by: Build Bot <[email protected]>
Reviewed-by: Dave Rigby <[email protected]>

Author: BenHuddleston

engines/ep/src/ephemeral_vb.cc

@@ -392,16 +392,6 @@ EphemeralVBucket::updateStoredValue(const HashTable::HashBucketLock& hbl,
                                     const Item& itm,
                                     const VBQueueItemCtx& queueItmCtx,
                                     bool justTouch) {
-    // Don't make any update if we know that this is a pending sync write that
-    // has not yet been completed. We need to return early before we make any
-    // modification to the seqList. This is typically handled in the EP case by
-    // the HashTable functions, but for ephemeral we update the seqList first so
-    // we need an earlier check.
-    if (v.isPending()) {
-        return std::make_tuple(
-                nullptr, MutationStatus::IsPendingSyncWrite, VBNotifyCtx{});
-    }
-
     VBNotifyCtx notifyCtx;
     StoredValue* newSv = nullptr;
     MutationStatus status(MutationStatus::WasClean);

engines/ep/src/hash_table.cc

@@ -346,10 +346,6 @@ HashTable::UpdateResult HashTable::unlocked_updateStoredValue(
                 "call on a non-active HT object");
     }
 
-    if (v.isPending()) {
-        return {MutationStatus::IsPendingSyncWrite, nullptr};
-    }
-
     // Logically /can/ update a non-Pending StoredValue with a Pending Item;
     // however internally this is implemented as a separate (new)
     // StoredValue object for the Pending item. We can replace a completed
@@ -548,16 +544,14 @@ HashTable::DeleteResult HashTable::unlocked_softDelete(
         bool onlyMarkDeleted,
         DeleteSource delSource) {
     switch (v.getCommitted()) {
-    case CommittedState::Pending:
-    case CommittedState::PreparedMaybeVisible:
-        // Cannot update a SV if it's a Pending item.
-        return {DeletionStatus::IsPendingSyncWrite, nullptr};
     case CommittedState::PrepareAborted:
     case CommittedState::PrepareCommitted:
         // We shouldn't be trying to use PrepareCompleted states yet
         throw std::logic_error(
                 "HashTable::unlocked_softDelete attempting"
                 " to delete a completed prepare");
+    case CommittedState::Pending:
+    case CommittedState::PreparedMaybeVisible:
     case CommittedState::CommittedViaMutation:
     case CommittedState::CommittedViaPrepare:
         const auto preProps = valueStats.prologue(&v);

engines/ep/src/vbucket.cc

@@ -2956,6 +2956,10 @@ std::pair<MutationStatus, boost::optional<VBNotifyCtx>> VBucket::processSet(
                 getId().to_string());
     }
 
+    if (v && v->isPending()) {
+        return {MutationStatus::IsPendingSyncWrite, {}};
+    }
+
     if (!hasMemoryForStoredValue(stats, itm)) {
         return {MutationStatus::NoMem, {}};
     }
@@ -3140,6 +3144,10 @@ VBucket::processSoftDelete(const HashTable::HashBucketLock& hbl,
                            uint64_t bySeqno,
                            DeleteSource deleteSource) {
     boost::optional<VBNotifyCtx> empty;
+    if (v.isPending()) {
+        return {MutationStatus::IsPendingSyncWrite, &v, empty};
+    }
+
     if (v.isTempInitialItem() && eviction == EvictionPolicy::Full) {
         return std::make_tuple(MutationStatus::NeedBgFetch, &v, empty);
     }
@@ -3174,11 +3182,6 @@ VBucket::processSoftDelete(const HashTable::HashBucketLock& hbl,
 
     // SyncDeletes are special cases. We actually want to add a new prepare.
     if (queueItmCtx.durability) {
-        if (v.isPending()) {
-            return std::make_tuple(
-                    MutationStatus::IsPendingSyncWrite, &v, empty);
-        }
-
         auto requirements = boost::get<cb::durability::Requirements>(
                 queueItmCtx.durability->requirementsOrPreparedSeqno);
 

engines/ep/tests/module_tests/hash_table_perspective_test.cc

@@ -126,28 +126,6 @@ TEST_P(HashTablePerspectiveTest, CorrectItemForEachPersisective) {
     del(ht, key);
 }
 
-// Test that the normal set() method cannot be used to change a pending item
-// to committed - commit() must be used.
-TEST_P(HashTablePerspectiveTest, DenyReplacePendingWithCommitted) {
-    auto pending = makePendingItem(key, "pending"s);
-    ASSERT_EQ(MutationStatus::WasClean, ht.set(*pending));
-
-    // Attempt setting the item again with a committed value.
-    auto committed = makeCommittedItem(key, "committed"s);
-    ASSERT_EQ(MutationStatus::IsPendingSyncWrite, ht.set(*committed));
-}
-
-// Test that the normal set() method cannot be used to change a pending item
-// to another pending - commit() must be used.
-TEST_P(HashTablePerspectiveTest, DenyReplacePendingWithPending) {
-    auto pending = makePendingItem(key, "pending"s);
-    ASSERT_EQ(MutationStatus::WasClean, ht.set(*pending));
-
-    // Attempt setting the item again with a committed value.
-    auto pending2 = makePendingItem(key, "pending2"s);
-    ASSERT_EQ(MutationStatus::IsPendingSyncWrite, ht.set(*pending2));
-}
-
 // Check that if a pending SyncWrite is added _before_ a Committed one (to the
 // same key), then findforWrite finds the pending one.
 // (While normally pending is added _after_ the existing Committed; during

engines/ep/tests/module_tests/vbucket_durability_test.cc

@@ -1434,6 +1434,40 @@ void VBucketDurabilityTest::setupPendingDelete(StoredDocKey key) {
     EXPECT_EQ(2, ht->getNumItems());
 }
 
+// Test that we cannot do a normal set on top of a pending SyncWrite
+TEST_P(VBucketDurabilityTest, DenyReplacePendingWithCommitted) {
+    auto key = makeStoredDocKey("key");
+    auto pending = makePendingItem(key, "pending"s);
+    VBQueueItemCtx ctx;
+    ctx.durability = DurabilityItemCtx{
+            cb::durability::Requirements{cb::durability::Level::Majority, {}},
+            cookie};
+    ASSERT_EQ(MutationStatus::WasClean,
+              public_processSet(*pending, 0 /*cas*/, ctx));
+
+    // Attempt setting the item again with a committed value.
+    auto committed = makeCommittedItem(key, "committed"s);
+    ASSERT_EQ(MutationStatus::IsPendingSyncWrite,
+              public_processSet(*committed, 0 /*cas*/, ctx));
+}
+
+// Test that we cannot do a pending SyncWrite on top of a pending SyncWrite
+TEST_P(VBucketDurabilityTest, DenyReplacePendingWithPending) {
+    auto key = makeStoredDocKey("key");
+    auto pending = makePendingItem(key, "pending"s);
+    VBQueueItemCtx ctx;
+    ctx.durability = DurabilityItemCtx{
+            cb::durability::Requirements{cb::durability::Level::Majority, {}},
+            cookie};
+    ASSERT_EQ(MutationStatus::WasClean,
+              public_processSet(*pending, 0 /*cas*/, ctx));
+
+    // Attempt setting the item again with a committed value.
+    auto pending2 = makePendingItem(key, "pending2"s);
+    ASSERT_EQ(MutationStatus::IsPendingSyncWrite,
+              public_processSet(*pending2, 0 /*cas*/, ctx));
+}
+
 // Positive test - check that an item can have a pending delete added
 // (SyncDelete).
 TEST_P(VBucketDurabilityTest, SyncDeletePending) {
@@ -1444,14 +1478,37 @@ TEST_P(VBucketDurabilityTest, SyncDeletePending) {
 
 // Negative test - check that if a key has a pending SyncDelete it cannot
 // otherwise be modified.
-TEST_P(VBucketDurabilityTest, PendingSyncWriteToPendingDeleteFails) {
+TEST_P(VBucketDurabilityTest, PendingSyncDeleteToPendingWriteFails) {
     auto key = makeStoredDocKey("key");
     setupPendingDelete(key);
 
     // Test - attempt to mutate a key which has a pending SyncDelete against it
     // with a pending SyncWrite.
     auto pending = makePendingItem(key, "pending"s);
-    ASSERT_EQ(MutationStatus::IsPendingSyncWrite, ht->set(*pending));
+    VBQueueItemCtx ctx;
+    ctx.genBySeqno = GenerateBySeqno::No;
+    ctx.durability = DurabilityItemCtx{pending->getDurabilityReqs(), cookie};
+    ASSERT_EQ(MutationStatus::IsPendingSyncWrite,
+              public_processSet(*pending, 0 /*cas*/, ctx));
+}
+
+// Negative test - check that if a key has a pending SyncWrite it cannot
+// be SyncDeleted
+TEST_P(VBucketDurabilityTest, PendingSyncWriteToPendingDeleteFails) {
+    auto key = makeStoredDocKey("key");
+
+    // Test - attempt to mutate a key which has a pending SyncWrite against it
+    // with a pending SyncDelete.
+    auto pending = makePendingItem(key, "pending"s);
+    VBQueueItemCtx ctx;
+    ctx.durability = DurabilityItemCtx{
+            cb::durability::Requirements{cb::durability::Level::Majority, {}},
+            cookie};
+    ASSERT_EQ(MutationStatus::WasClean,
+              public_processSet(*pending, 0 /*cas*/, ctx));
+
+    ASSERT_EQ(MutationStatus::IsPendingSyncWrite,
+              public_processSoftDelete(key, ctx).first);
 }
 
 // Negative test - check that if a key has a pending SyncDelete it cannot