MB-35979: Ensure SyncReplaces check for committed values

A SyncReplace should only succeed if there is a committed value which
is not expired or deleted.

This patch fixes a bug affecting ephemeral buckets allowing
SyncReplaces even if there is no committed value in the presence of a
completed prepare. Replace previously only tested the state of the
passed stored value - this is insufficient as the passed value may be a
completed prepare, which does not provide information on whether a
_committed_ value exists.

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

Author: jameseh96

engines/ep/src/vbucket.cc

@@ -1657,7 +1657,8 @@ ENGINE_ERROR_CODE VBucket::replace(
         }
 
         if (htRes.committed) {
-            if (isLogicallyNonExistent(*htRes.committed, cHandle)) {
+            if (isLogicallyNonExistent(*htRes.committed, cHandle) ||
+                htRes.committed->isExpired(ep_real_time())) {
                 ht.cleanupIfTemporaryItem(hbl, *htRes.committed);
                 return ENGINE_KEY_ENOENT;
             }

engines/ep/tests/module_tests/evp_store_durability_test.cc

@@ -843,6 +843,76 @@ TEST_P(DurabilityBucketTest, SyncWriteComparesToCorrectCas) {
     ASSERT_EQ(ENGINE_SYNC_WRITE_PENDING, store->set(*pending, cookie));
 }
 
+TEST_P(DurabilityEphemeralBucketTest, SyncReplaceChecksCorrectSVExists) {
+    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 delete
+    mutation_descr_t delInfo;
+    uint64_t cas = 0;
+    ASSERT_EQ(ENGINE_SUCCESS,
+              store->deleteItem(key, cas, vbid, cookie, {}, nullptr, delInfo));
+
+    // now do a SyncReplace. Should FAIL as the item was deleted
+    pending = makePendingItem(key, "new_value");
+    ASSERT_EQ(ENGINE_KEY_ENOENT, store->replace(*pending, cookie));
+}
+
+TEST_P(DurabilityEphemeralBucketTest, SyncReplaceChecksCorrectExpiry) {
+    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));
+
+    // time travel to when the item has definitely expired
+    TimeTraveller abe(10);
+
+    // now do a SyncReplace. Should fail, as the item has expired.
+    pending = makePendingItem(key, "new_value");
+    ASSERT_EQ(ENGINE_KEY_ENOENT, store->replace(*pending, cookie));
+}
+
 TEST_P(DurabilityEphemeralBucketTest, SyncWriteChecksCorrectExpiry) {
     setVBucketStateAndRunPersistTask(
             vbid,