MB-34946: Remove prepare when receiving overwriting mutation

When a replica is backfilling from disk it will often be sent
a Mutation instead of a Commit. This Mutation logically commits the
prepare that the replica may already have. Currently the code will
overwrite the prepare with a mutation. This is not valid as the
replica may already have a mutation for this commit and a following
lookup in the HashTable would cause a precondition to fail as there
would now exist two commited StoredValues in the HashTable.

Instead, the replica should either remove the prepare (EP vBuckets)
or mark it as completed (Ephemeral vBuckets) and either add a new
mutation or overwrite the existing one. Update the setWithMeta and
deleteWithMeta functions that are called by the replica to process
this mutation instead of a commit to lookup both the prepare and
committed StoredValues and deal with them appropriately.

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

Author: BenHuddleston

engines/ep/src/ep_vb.cc

@@ -857,3 +857,19 @@ bool EPVBucket::isValidDurabilityLevel(cb::durability::Level level) {
 
     folly::assume_unreachable();
 }
+
+void EPVBucket::processImplicitlyCompletedPrepare(
+        HashTable::StoredValueProxy& v) {
+    // As we have passed a StoredValueProxy to this function (the callers need
+    // a HashTable::FindCommitResult) we need to be careful about our stats
+    // updates. The StoredValueProxy attempts to do a
+    // HashTable::Statistics::epilogue stats update when we destruct it. This is
+    // generally fine, but if we want to use any other HashTable function with
+    // a StoredValueProxy we need a way to skip the StoredValueProxy's stats
+    // update as the other HashTable function will do it's own. In this case,
+    // we can call StoredValueProxy::release to release the ownership of the
+    // pointer in the StoredValueProxy and skip any stats update. This consumes
+    // the StoredValue* and invalidates the StoredValueProxy so it should not be
+    // used after.
+    ht.unlocked_del(v.getHBL(), v.release());
+}

engines/ep/src/ep_vb.h

@@ -296,6 +296,9 @@ class EPVBucket : public VBucket {
 
     bool isValidDurabilityLevel(cb::durability::Level level) override;
 
+    void processImplicitlyCompletedPrepare(
+            HashTable::StoredValueProxy& v) override;
+
     /**
      * Total number of alive (non-deleted) items on-disk in this vBucket.
      * Initially populated during warmup as the number of items on disk;

engines/ep/src/ephemeral_vb.cc

@@ -926,3 +926,8 @@ bool EphemeralVBucket::isValidDurabilityLevel(cb::durability::Level level) {
 
     folly::assume_unreachable();
 }
+
+void EphemeralVBucket::processImplicitlyCompletedPrepare(
+        HashTable::StoredValueProxy& v) {
+    v.setCommitted(CommittedState::PrepareCommitted);
+}

engines/ep/src/ephemeral_vb.h

@@ -318,6 +318,9 @@ class EphemeralVBucket : public VBucket {
 
     bool isValidDurabilityLevel(cb::durability::Level level) override;
 
+    void processImplicitlyCompletedPrepare(
+            HashTable::StoredValueProxy& htRes) override;
+
     /**
      * (i) Repositions an already non-temp element in the sequence list (OR)
      * (ii) For a temp item that is being updated (that is, being made non-temp

engines/ep/src/vbucket.cc

@@ -3152,11 +3152,19 @@ std::pair<MutationStatus, boost::optional<VBNotifyCtx>> VBucket::processSet(
             getPassiveDM().completeSyncWrite(
                     itm.getKey(), PassiveDurabilityMonitor::Resolution::Commit);
 
-            // @TODO we must remove the prepare and overwrite the mutation if we
-            // are replacing a prepare with a mutation
-            // Release the pending SV from the SVP that is holding it to prevent
-            // a double stat update that would cause a stat underflow exception.
-            htRes.pending.release();
+            // Deal with the already existing prepare
+            processImplicitlyCompletedPrepare(htRes.pending);
+
+            // Add a new or overwrite the existing mutation
+            return processSetInner(htRes.pending.getHBL(),
+                                   htRes.committed,
+                                   itm,
+                                   cas,
+                                   allowExisting,
+                                   hasMetaData,
+                                   queueItmCtx,
+                                   storeIfStatus,
+                                   maybeKeyExists);
         }
 
         // This is a new SyncWrite, we just want to add a new prepare unless we
@@ -3400,11 +3408,24 @@ VBucket::processSoftDelete(HashTable::FindCommitResult& htRes,
                 StoredDocKey(v.getKey()),
                 PassiveDurabilityMonitor::Resolution::Commit);
 
-        // @TODO we must remove the prepare and overwrite the mutation if we
-        // are replacing a prepare with a mutation
-        // Release the pending SV from the SVP that is holding it to prevent
-        // a double stat update that would cause a stat underflow exception.
-        htRes.pending.release();
+        // Sanity - We should never delete a prepare in this way, and if we are
+        // hitting this delete path then we should have the committed SV.
+        if (!htRes.committed) {
+            return {MutationStatus::NotFound, htRes.committed, boost::none};
+        }
+
+        // Deal with the existing prepare
+        processImplicitlyCompletedPrepare(htRes.pending);
+
+        // Delete the existing committed SV.
+        return processSoftDeleteInner(htRes.getHBL(),
+                                      *htRes.committed,
+                                      cas,
+                                      metadata,
+                                      queueItmCtx,
+                                      use_meta,
+                                      bySeqno,
+                                      deleteSource);
     }
 
     return processSoftDeleteInner(htRes.getHBL(),

engines/ep/src/vbucket.h

@@ -1876,6 +1876,7 @@ class VBucket : public std::enable_shared_from_this<VBucket> {
                            bool use_meta,
                            uint64_t bySeqno,
                            DeleteSource deleteSource);
+
     /**
      * Delete a key (associated StoredValue) from ALL in-memory data structures
      * like HT.
@@ -2365,6 +2366,16 @@ class VBucket : public std::enable_shared_from_this<VBucket> {
      */
     VBNotifyCtx queueItem(queued_item& item, const VBQueueItemCtx& ctx);
 
+    /**
+     * Deal with the prepare in the HashTable in the derived class specific way
+     * as it is to be "replaced" by a mutation. Consumes the StoredValue* in the
+     * StoredValueProxy making it no longer usable.
+     *
+     * @param v StoredValueProxy of the prepare to complete
+     */
+    virtual void processImplicitlyCompletedPrepare(
+            HashTable::StoredValueProxy& v) = 0;
+
     Vbid id;
     std::atomic<vbucket_state_t>    state;
     folly::SharedMutex stateLock;

engines/ep/tests/module_tests/dcp_durability_stream_test.cc

@@ -704,8 +704,8 @@ void DurabilityPassiveStreamTest::
 
     SnapshotMarker marker(opaque,
                           vbid,
-                          1 /*snapStart*/,
-                          3 /*snapEnd*/,
+                          2 /*snapStart*/,
+                          4 /*snapEnd*/,
                           dcp_marker_flag_t::MARKER_FLAG_DISK,
                           {} /*streamId*/);
     stream->processMarker(&marker);
@@ -714,7 +714,7 @@ void DurabilityPassiveStreamTest::
     using namespace cb::durability;
     auto item = makePendingItem(
             key, "value", Requirements(Level::Majority, Timeout::Infinity()));
-    item->setBySeqno(1);
+    item->setBySeqno(2);
     item->setCas(999);
 
     // Send the prepare
@@ -730,7 +730,7 @@ void DurabilityPassiveStreamTest::
                       cb::mcbp::DcpStreamId{})));
 
     item = makeCommittedItem(key, "committed");
-    item->setBySeqno(3);
+    item->setBySeqno(4);
 
     if (docState == DocumentState::Deleted) {
         item->setDeleted(DeleteSource::Explicit);
@@ -748,6 +748,26 @@ void DurabilityPassiveStreamTest::
                       DocKeyEncodesCollectionId::No,
                       nullptr,
                       cb::mcbp::DcpStreamId{})));
+
+    auto vb = store->getVBucket(vbid);
+    ASSERT_TRUE(vb);
+    {
+        // findForCommit will return both pending and committed perspectives
+        auto res = vb->ht.findForCommit(key);
+        ASSERT_TRUE(res.committed);
+        EXPECT_EQ(4, res.committed->getBySeqno());
+        if (docState == DocumentState::Alive) {
+            EXPECT_TRUE(res.committed->getValue());
+        }
+        if (persistent()) {
+            EXPECT_FALSE(res.pending);
+        } else {
+            ASSERT_TRUE(res.pending);
+            EXPECT_EQ(2, res.pending->getBySeqno());
+            EXPECT_EQ(CommittedState::PrepareCommitted,
+                      res.pending->getCommitted());
+        }
+    }
 }
 
 TEST_P(DurabilityPassiveStreamTest,
@@ -756,9 +776,44 @@ TEST_P(DurabilityPassiveStreamTest,
             DocumentState::Alive);
 }
 
+void DurabilityPassiveStreamTest::
+        receiveMutationOrDeletionInsteadOfCommitWhenStreamingFromDiskMutationFirst(
+                DocumentState docState) {
+    uint32_t opaque = 1;
+    SnapshotMarker marker(opaque,
+                          vbid,
+                          1 /*snapStart*/,
+                          3 /*snapEnd*/,
+                          dcp_marker_flag_t::MARKER_FLAG_DISK,
+                          {} /*streamId*/);
+    stream->processMarker(&marker);
+
+    auto key = makeStoredDocKey("key");
+    auto item = makeCommittedItem(key, "mutation");
+    item->setBySeqno(1);
+
+    EXPECT_EQ(ENGINE_SUCCESS,
+              stream->messageReceived(std::make_unique<MutationConsumerMessage>(
+                      std::move(item),
+                      opaque,
+                      IncludeValue::Yes,
+                      IncludeXattrs::Yes,
+                      IncludeDeleteTime::No,
+                      DocKeyEncodesCollectionId::No,
+                      nullptr,
+                      cb::mcbp::DcpStreamId{})));
+    testReceiveMutationOrDeletionInsteadOfCommitWhenStreamingFromDisk(docState);
+}
+
 TEST_P(DurabilityPassiveStreamTest,
-       ReceiveDeletionInsteadOfCommitWhenStreamingFromDisk) {
-    testReceiveMutationOrDeletionInsteadOfCommitWhenStreamingFromDisk(
+       ReceiveMutationInsteadOfCommitOnTopOfMutation) {
+    receiveMutationOrDeletionInsteadOfCommitWhenStreamingFromDiskMutationFirst(
+            DocumentState::Alive);
+}
+
+TEST_P(DurabilityPassiveStreamTest,
+       ReceiveDeletionInsteadOfCommitOnTopOfMutation) {
+    receiveMutationOrDeletionInsteadOfCommitWhenStreamingFromDiskMutationFirst(
             DocumentState::Deleted);
 }
 
@@ -816,7 +871,14 @@ void DurabilityPassiveStreamTest::
     {
         // findForCommit will return both pending and committed perspectives
         auto res = vb->ht.findForCommit(key);
-        EXPECT_FALSE(res.pending);
+        if (persistent()) {
+            EXPECT_FALSE(res.pending);
+        } else {
+            ASSERT_TRUE(res.pending);
+            EXPECT_EQ(1, res.pending->getBySeqno());
+            EXPECT_EQ(CommittedState::PrepareCommitted,
+                      res.pending->getCommitted());
+        }
         ASSERT_TRUE(res.committed);
         EXPECT_EQ(4, res.committed->getBySeqno());
         EXPECT_EQ(CommittedState::CommittedViaMutation,
@@ -853,12 +915,6 @@ TEST_P(DurabilityPassiveStreamTest,
             DocumentState::Alive);
 }
 
-TEST_P(DurabilityPassiveStreamTest,
-       ReceiveDeletionInsteadOfCommitForReconnectWindowWithPrepareLast) {
-    testReceiveMutationOrDeletionInsteadOfCommitForReconnectWindowWithPrepareLast(
-            DocumentState::Deleted);
-}
-
 TEST_P(DurabilityPassiveStreamTest,
        ReceiveAbortOnTopOfCommittedDueToDedupedPrepare) {
     uint32_t opaque = 0;

engines/ep/tests/module_tests/dcp_durability_stream_test.h

@@ -100,6 +100,16 @@ class DurabilityPassiveStreamTest : public SingleThreadedPassiveStreamTest {
     void testReceiveMutationOrDeletionInsteadOfCommitWhenStreamingFromDisk(
             DocumentState docState);
 
+    /**
+     * Test that a mutation or deletion sent instead of a commit is accepted by
+     * the replica when backfilling from disk if it already has a mutation.
+     *
+     * @param docState Shoudl we send a mutation or a deletion?
+     */
+    void
+    receiveMutationOrDeletionInsteadOfCommitWhenStreamingFromDiskMutationFirst(
+            DocumentState docState);
+
     /**
      * Test that a mutaiton or deletion sent instead of a commit is accepted by
      * the replica when in the reconnect window for which a prepare may be