MB-35979: Ensure SyncWrites check expiry of committed item

Writes with cas respond differently if the existing item is logically
non-existent (e.g., expired) vs exists with different cas.

Prior to this patch, processSetInner checked the expiry of the stored
value which is about to be modified. However, in an ephemeral bucket
this stored value may be a completed prepare - the expiry of the
prepare is not necessarily related to the expiry of the committed item.

This patch ensures it is always the expiry time of the committed item
which is checked.

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

Author: jameseh96

engines/ep/src/vbucket.cc

@@ -3173,20 +3173,21 @@ VBucket::processSetInner(HashTable::FindCommitResult& htRes,
      * a cas operation for a key that doesn't exist is not a very cool
      * thing to do. See MB 3252
      */
-    if (v && v->isExpired(ep_real_time()) && !hasMetaData && !itm.isDeleted()) {
-        if (v->isLocked(ep_current_time())) {
-            v->unlock();
+    // need to test cas and locking against the committed value
+    // explicitly, as v may be a completed prepare (to be modified)
+    // with a cas, deleted status, expiry etc. different from the committed
+    auto* committed = htRes.committed;
+    if (committed && committed->isExpired(ep_real_time()) && !hasMetaData &&
+        !itm.isDeleted()) {
+        if (committed->isLocked(ep_current_time())) {
+            committed->unlock();
         }
         if (cas) {
             /* item has expired and cas value provided. Deny ! */
             return {MutationStatus::NotFound, {}};
         }
     }
 
-    // need to test cas and locking against the committed value
-    // explicitly, as v may be a completed prepare to be modified
-    // containing an unrelated cas, deleted status etc.
-    auto* committed = htRes.committed;
     if (committed) {
         if (!allowExisting && !committed->isTempItem() &&
             !committed->isDeleted()) {

engines/ep/tests/module_tests/evp_store_durability_test.cc

@@ -843,6 +843,49 @@ TEST_P(DurabilityBucketTest, SyncWriteComparesToCorrectCas) {
     ASSERT_EQ(ENGINE_SYNC_WRITE_PENDING, store->set(*pending, cookie));
 }
 
+TEST_P(DurabilityEphemeralBucketTest, SyncWriteChecksCorrectExpiry) {
+    setVBucketStateAndRunPersistTask(
+            vbid,
+            vbucket_state_active,
+            {{"topology", nlohmann::json::array({{"active", "replica"}})}});
+
+    auto& vb = *store->getVBucket(vbid);
+
+    // prepare SyncWrite and commit.
+    auto key = makeStoredDocKey("key");
+    auto pending = makePendingItem(key, "value");
+
+    ASSERT_EQ(ENGINE_SYNC_WRITE_PENDING, store->set(*pending, cookie));
+    ASSERT_EQ(ENGINE_SUCCESS,
+              vb.commit(key,
+                        pending->getBySeqno(),
+                        {} /*commitSeqno*/,
+                        vb.lockCollections(key)));
+
+    vb.processResolvedSyncWrites();
+
+    // Non-durable write with expiry
+    auto committed = makeCommittedItem(key, "some_other_value");
+
+    using namespace std::chrono;
+    auto expiry = system_clock::now() + seconds(1);
+    committed->setExpTime(system_clock::to_time_t(expiry));
+
+    ASSERT_EQ(ENGINE_SUCCESS, store->set(*committed, cookie));
+
+    // get cas
+    uint64_t cas = store->get(key, vbid, cookie, {}).item->getCas();
+
+    // time travel to when the item has definitely expired
+    TimeTraveller cooper(10);
+
+    // now do a SyncWrite with a cas - the item has expired so it should
+    // return not found, not invalid cas
+    pending = makePendingItem(key, "new_value");
+    pending->setCas(cas);
+    ASSERT_EQ(ENGINE_KEY_ENOENT, store->set(*pending, cookie));
+}
+
 void DurabilityEPBucketTest::verifyOnDiskItemCount(VBucket& vb,
                                                    uint64_t expectedValue) {
     // skip for rocksdb as it treats every mutation as an insertion