MB-34946: Use HashTable::FindCommitResult in processSoftDelete

In a following change, we need to make use of the FindCommitResult
in VBucket::processSoftDelete as it contains both the pending and
committed StoredValues. To keep a single processSoftDelete
interface, update all callers to pass a FindCommitResult instead of
a single StoredValue&. Also, add a processSoftDeleteInner function
as we will need to modify which StoredValue& v is deleted but
assignment is not allowed.

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

Author: BenHuddleston

engines/ep/src/vbucket.cc

@@ -2077,10 +2077,9 @@ ENGINE_ERROR_CODE VBucket::deleteItem(
     ENGINE_ERROR_CODE ret = durability ? ENGINE_EWOULDBLOCK : ENGINE_SUCCESS;
 
     { // HashBucketLock scope
-        auto htRes = durability ? ht.findForSyncWrite(cHandle.getKey())
-                                : ht.findForWrite(cHandle.getKey());
-        auto* v = htRes.storedValue;
-        auto& hbl = htRes.lock;
+        auto htRes = ht.findForCommit(cHandle.getKey());
+        auto* v = htRes.selectSVToModify(durability.is_initialized());
+        auto& hbl = htRes.getHBL();
 
         if (!v || v->isDeleted() || v->isTempItem() ||
             cHandle.isLogicallyDeleted(v->getBySeqno())) {
@@ -2135,7 +2134,7 @@ ENGINE_ERROR_CODE VBucket::deleteItem(
                 queueItmCtx.durability = DurabilityItemCtx{*durability, cookie};
             }
             std::tie(delrv, v, notifyCtx) =
-                    processSoftDelete(hbl,
+                    processSoftDelete(htRes,
                                       *v,
                                       cas,
                                       metadata,
@@ -2215,9 +2214,9 @@ ENGINE_ERROR_CODE VBucket::deleteWithMeta(
         const Collections::VB::Manifest::CachingReadHandle& cHandle,
         DeleteSource deleteSource) {
     const auto& key = cHandle.getKey();
-    auto htRes = ht.findForWrite(key);
-    auto* v = htRes.storedValue;
-    auto& hbl = htRes.lock;
+    auto htRes = ht.findForCommit(key);
+    auto* v = htRes.pending ? htRes.pending.getSV() : htRes.committed;
+    auto& hbl = htRes.pending.getHBL();
 
     if (v && cHandle.isLogicallyDeleted(v->getBySeqno())) {
         return ENGINE_KEY_ENOENT;
@@ -2317,7 +2316,7 @@ ENGINE_ERROR_CODE VBucket::deleteWithMeta(
             // this is a replication call (i.e. not to an active vbucket),
             // the active has done this and we must just store what we're
             // given.
-            std::tie(delrv, v, notifyCtx) = processSoftDelete(hbl,
+            std::tie(delrv, v, notifyCtx) = processSoftDelete(htRes,
                                                               *v,
                                                               cas,
                                                               itemMeta,
@@ -3376,15 +3375,14 @@ std::pair<AddStatus, boost::optional<VBNotifyCtx>> VBucket::processAdd(
 }
 
 std::tuple<MutationStatus, StoredValue*, boost::optional<VBNotifyCtx>>
-VBucket::processSoftDelete(const HashTable::HashBucketLock& hbl,
+VBucket::processSoftDelete(HashTable::FindCommitResult& htRes,
                            StoredValue& v,
                            uint64_t cas,
                            const ItemMetaData& metadata,
                            const VBQueueItemCtx& queueItmCtx,
                            bool use_meta,
                            uint64_t bySeqno,
                            DeleteSource deleteSource) {
-    boost::optional<VBNotifyCtx> empty;
     if (v.isPending()) {
         // It is not valid for an active vBucket to attempt to overwrite an
         // in flight SyncWrite. If this vBucket is not active, we are
@@ -3395,14 +3393,40 @@ VBucket::processSoftDelete(const HashTable::HashBucketLock& hbl,
         // missing a prepare. This code allows this mutation to be accepted
         // and overwrites the existing prepare.
         if (getState() == vbucket_state_active || !isReceivingDiskSnapshot()) {
-            return {MutationStatus::IsPendingSyncWrite, &v, empty};
+            return {MutationStatus::IsPendingSyncWrite, &v, boost::none};
         }
 
         getPassiveDM().completeSyncWrite(
                 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();
     }
 
+    return processSoftDeleteInner(htRes.getHBL(),
+                                  v,
+                                  cas,
+                                  metadata,
+                                  queueItmCtx,
+                                  use_meta,
+                                  bySeqno,
+                                  deleteSource);
+}
+
+std::tuple<MutationStatus, StoredValue*, boost::optional<VBNotifyCtx>>
+VBucket::processSoftDeleteInner(const HashTable::HashBucketLock& hbl,
+                                StoredValue& v,
+                                uint64_t cas,
+                                const ItemMetaData& metadata,
+                                const VBQueueItemCtx& queueItmCtx,
+                                bool use_meta,
+                                uint64_t bySeqno,
+                                DeleteSource deleteSource) {
+    boost::optional<VBNotifyCtx> empty;
     if (v.isTempInitialItem() && eviction == EvictionPolicy::Full) {
         return std::make_tuple(MutationStatus::NeedBgFetch, &v, empty);
     }

engines/ep/src/vbucket.h

@@ -1838,8 +1838,8 @@ class VBucket : public std::enable_shared_from_this<VBucket> {
      * in-memory structure like HT, and checkpoint mgr.
      * Assumes that HT bucket lock is grabbed.
      *
-     * @param hbl Hash table bucket lock that must be held
-     * @param v Reference to the StoredValue to be soft deleted
+     * @param htRes Committed and Pending StoredValues
+     * @param v Reference to the StoredValue to delete (in the general case)
      * @param cas the expected CAS of the item (or 0 to override)
      * @param metadata ref to item meta data
      * @param queueItmCtx holds info needed to queue an item in chkpt or vb
@@ -1855,15 +1855,27 @@ class VBucket : public std::enable_shared_from_this<VBucket> {
      *         notification info, if status was successful.
      */
     std::tuple<MutationStatus, StoredValue*, boost::optional<VBNotifyCtx>>
-    processSoftDelete(const HashTable::HashBucketLock& hbl,
+    processSoftDelete(HashTable::FindCommitResult& htRes,
                       StoredValue& v,
                       uint64_t cas,
                       const ItemMetaData& metadata,
                       const VBQueueItemCtx& queueItmCtx,
                       bool use_meta,
                       uint64_t bySeqno,
                       DeleteSource deleteSource);
-
+    /**
+     * Inner function for processSoftDelete. Allows overwriting of in-flight
+     * prepares.
+     */
+    std::tuple<MutationStatus, StoredValue*, boost::optional<VBNotifyCtx>>
+    processSoftDeleteInner(const HashTable::HashBucketLock& hbl,
+                           StoredValue& v,
+                           uint64_t cas,
+                           const ItemMetaData& metadata,
+                           const VBQueueItemCtx& queueItmCtx,
+                           bool use_meta,
+                           uint64_t bySeqno,
+                           DeleteSource deleteSource);
     /**
      * Delete a key (associated StoredValue) from ALL in-memory data structures
      * like HT.

engines/ep/tests/module_tests/vbucket_test.cc

@@ -228,27 +228,27 @@ VBucketTestBase::public_processSoftDelete(const DocKey& key,
         VBQueueItemCtx ctx) {
     // Need to take the collections read handle before the hbl
     auto cHandle = vbucket->lockCollections(key);
-    auto res = ctx.durability ? vbucket->ht.findForSyncWrite(key)
-                              : vbucket->ht.findForWrite(key);
-    if (!res.storedValue) {
+    auto htRes = vbucket->ht.findForCommit(key);
+    auto* v = htRes.selectSVToModify(ctx.durability.is_initialized());
+    if (!v) {
         return {MutationStatus::NotFound, nullptr};
     }
-    if (res.storedValue->isDeleted() && !res.storedValue->isPending()) {
+    if (v->isDeleted() && !v->isPending()) {
         return {MutationStatus::NotFound, nullptr};
     }
-    return public_processSoftDelete(res.lock, *res.storedValue, ctx);
+    return public_processSoftDelete(htRes, *v, ctx);
 }
 
 std::pair<MutationStatus, StoredValue*>
-VBucketTestBase::public_processSoftDelete(const HashTable::HashBucketLock& hbl,
+VBucketTestBase::public_processSoftDelete(HashTable::FindCommitResult& htRes,
                                           StoredValue& v,
                                           VBQueueItemCtx ctx) {
     ItemMetaData metadata;
     metadata.revSeqno = v.getRevSeqno() + 1;
     MutationStatus status;
     StoredValue* deletedSV;
     std::tie(status, deletedSV, std::ignore) =
-            vbucket->processSoftDelete(hbl,
+            vbucket->processSoftDelete(htRes,
                                        v,
                                        /*cas*/ 0,
                                        metadata,

engines/ep/tests/module_tests/vbucket_test.h

@@ -112,7 +112,9 @@ class VBucketTestBase {
     std::pair<MutationStatus, StoredValue*> public_processSoftDelete(
             const DocKey& key, VBQueueItemCtx ctx = {});
     std::pair<MutationStatus, StoredValue*> public_processSoftDelete(
-            const HashTable::HashBucketLock& hbl, StoredValue& v, VBQueueItemCtx ctx = {});
+            HashTable::FindCommitResult& htRes,
+            StoredValue& v,
+            VBQueueItemCtx ctx = {});
 
     bool public_deleteStoredValue(const DocKey& key);