MB-35539: Check cas of committed item when creating new prepare

Transactions identified that we were not checking the cas of the
committed item when attempting to add a new prepare. This was
causing us to lose subdoc operations. Refactor the code to hit all
of the normal checks made in VBucket::processSetInner when attempting
to add a new prepare.

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

Author: BenHuddleston

engines/ep/src/vbucket.cc

@@ -1504,8 +1504,7 @@ ENGINE_ERROR_CODE VBucket::set(
         // that requires the item's info
         if ((v == nullptr || v->isTempInitialItem()) &&
             (eviction == EvictionPolicy::Full) &&
-            ((itm.getCas() != 0) ||
-             storeIfStatus == cb::StoreIfStatus::GetItemInfo)) {
+            (cas_op || storeIfStatus == cb::StoreIfStatus::GetItemInfo)) {
             // Check Bloomfilter's prediction
             if (!maybeKeyExistsInFilter(itm.getKey())) {
                 maybeKeyExists = false;
@@ -3024,53 +3023,38 @@ std::pair<MutationStatus, boost::optional<VBNotifyCtx>> VBucket::processSet(
         const VBQueueItemCtx& queueItmCtx,
         cb::StoreIfStatus storeIfStatus,
         bool maybeKeyExists) {
-    if (v) {
-        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
-            // allowed to overwrite an in flight SyncWrite iff we are receiving
-            // a disk snapshot. This is due to disk based de-dupe that allows
-            // only 1 value per key. In this case, the active node may send a
-            // mutation instead of a commit if it knows that this replica may be
-            // 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, {}};
-            }
+    if (v && 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
+        // allowed to overwrite an in flight SyncWrite iff we are receiving
+        // a disk snapshot. This is due to disk based de-dupe that allows
+        // only 1 value per key. In this case, the active node may send a
+        // mutation instead of a commit if it knows that this replica may be
+        // 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, {}};
+        }
 
-            Expects(itm.isCommitted());
-            getPassiveDM().completeSyncWrite(
-                    itm.getKey(),
-                    PassiveDurabilityMonitor::Resolution::Commit,
-                    v->getBySeqno() /* prepareSeqno */);
+        Expects(itm.isCommitted());
+        getPassiveDM().completeSyncWrite(
+                itm.getKey(),
+                PassiveDurabilityMonitor::Resolution::Commit,
+                v->getBySeqno() /* prepareSeqno */);
 
-            // Deal with the already existing prepare
-            processImplicitlyCompletedPrepare(htRes.pending);
+        // 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
-        // still have a completed prepare (Ephemeral) which we should replace
-        // instead.
-        if (!v->isCompleted() && itm.isPending()) {
-            VBNotifyCtx notifyCtx;
-            // We have to modify the StoredValue pointer passed in or we do not
-            // return the correct cas to the client.
-            std::tie(v, notifyCtx) = addNewStoredValue(
-                    htRes.getHBL(), itm, queueItmCtx, GenerateRevSeqno::No);
-            return {MutationStatus::WasClean, notifyCtx};
-        }
+        // Add a new or overwrite the existing mutation
+        return processSetInner(htRes.pending.getHBL(),
+                               htRes.committed,
+                               itm,
+                               cas,
+                               allowExisting,
+                               hasMetaData,
+                               queueItmCtx,
+                               storeIfStatus,
+                               maybeKeyExists);
     }
 
     return processSetInner(htRes.getHBL(),
@@ -3183,9 +3167,21 @@ VBucket::processSetInner(const HashTable::HashBucketLock& hbl,
 
         MutationStatus status;
         VBNotifyCtx notifyCtx;
-        std::tie(v, status, notifyCtx) =
-                updateStoredValue(hbl, *v, itm, queueItmCtx);
+        // This is a new SyncWrite, we just want to add a new prepare unless we
+        // still have a completed prepare (Ephemeral) which we should replace
+        // instead.
+        if (v->isCommitted() && !v->isCompleted() && itm.isPending()) {
+            std::tie(v, notifyCtx) = addNewStoredValue(
+                    hbl, itm, queueItmCtx, GenerateRevSeqno::No);
+            // Add should always be clean
+            status = MutationStatus::WasClean;
+        } else {
+            std::tie(v, status, notifyCtx) =
+                    updateStoredValue(hbl, *v, itm, queueItmCtx);
+        }
+
         return {status, notifyCtx};
+
     } else if (cas != 0) {
         return {MutationStatus::NotFound, {}};
     } else {

engines/ep/tests/module_tests/dcp_durability_stream_test.cc

@@ -770,15 +770,15 @@ TEST_P(DurabilityActiveStreamTest,
     auto vb = engine->getVBucket(vbid);
     const auto key = makeStoredDocKey("key");
     const std::string value = "value";
-    auto item = makePendingItem(
-            key,
-            value,
-            cb::durability::Requirements(cb::durability::Level::Majority,
-                                         1 /*timeout*/));
-    VBQueueItemCtx ctx;
-    ctx.durability =
-            DurabilityItemCtx{item->getDurabilityReqs(), nullptr /*cookie*/};
-    {
+    { // Locking scope for collections handle
+        auto item = makePendingItem(
+                key,
+                value,
+                cb::durability::Requirements(cb::durability::Level::Majority,
+                                             1 /*timeout*/));
+        VBQueueItemCtx ctx;
+        ctx.durability = DurabilityItemCtx{item->getDurabilityReqs(),
+                                           nullptr /*cookie*/};
         auto cHandle = vb->lockCollections(item->getKey());
         EXPECT_EQ(ENGINE_EWOULDBLOCK,
                   vb->set(*item, cookie, *engine, {}, cHandle));
@@ -794,7 +794,15 @@ TEST_P(DurabilityActiveStreamTest,
     EXPECT_EQ(1, vb->getHighPreparedSeqno());
     EXPECT_EQ(1, vb->getHighCompletedSeqno());
 
-    {
+    { // Locking scope for collections handle
+        auto item = makePendingItem(
+                key,
+                value,
+                cb::durability::Requirements(cb::durability::Level::Majority,
+                                             1 /*timeout*/));
+        VBQueueItemCtx ctx;
+        ctx.durability = DurabilityItemCtx{item->getDurabilityReqs(),
+                                           nullptr /*cookie*/};
         auto cHandle = vb->lockCollections(item->getKey());
         EXPECT_EQ(ENGINE_EWOULDBLOCK,
                   vb->set(*item, cookie, *engine, {}, cHandle));

engines/ep/tests/module_tests/evp_store_durability_test.cc

@@ -2055,6 +2055,22 @@ TEST_P(DurabilityBucketTest, ActiveToReplicaAndCommit) {
     ASSERT_EQ(ENGINE_SUCCESS, vb.commit(key, 1, 4, vb.lockCollections(key)));
 }
 
+TEST_P(DurabilityBucketTest, CasCheckMadeForNewPrepare) {
+    setVBucketToActiveWithValidTopology();
+
+    auto key = makeStoredDocKey("key");
+    auto committed = makeCommittedItem(key, "committed");
+
+    ASSERT_EQ(ENGINE_SUCCESS, store->set(*committed, cookie));
+
+    auto pending = makePendingItem(key, "pending");
+    pending->setCas(123);
+    EXPECT_EQ(ENGINE_KEY_EEXISTS, store->set(*pending, cookie));
+
+    pending->setCas(committed->getCas());
+    EXPECT_EQ(ENGINE_EWOULDBLOCK, store->set(*pending, cookie));
+}
+
 // Test cases which run against couchstore
 INSTANTIATE_TEST_CASE_P(AllBackends,
                         DurabilityCouchstoreBucketTest,