MB-59746 [1/2]: Restore CAS to original value after unlock()

+Problem+

When a document is locked via GetLocked, KV-Engine generates a new
"locked CAS", stores it in memory, and returns it to the client for it
to use in a subsequent Mutation(CAS=locked_CAS) or Unlock() operation.

However, when the document is unlocked without modification (lock
times out, or an explicit Unlock() call is made), the previous CAS
value is *not* restored. As such, the in-memory representation of the
document has an artificially higher CAS than the document "really"
has.

When using LWW XDCR this is problematic; as it means that a document
could have an artificially higher CAS than it "really" has, and hence
conflict resolution could incorrectly fail to update the
document. Consider the following scenario:

1. A LWW XDCR relation from source S to destination D is established,
   and is performing _optimistic_ conflict resolution. Buckets are in
   sync, assume CAS of all documents is from time T=0.

2. At time T=10, a Mutation operation occurs on the source bucket S
   against the doc X.

3. At T=11 a GetLocked operation occurs on the destination Bucket D against
the same doc.

4. At T=12 the destination bucket D document is Unlocked, leaving the
   CAS with the advanced value from T=11.

5. Due to XDCR processing lag / network latency, the SetWithMeta()
   arrives at {{D}} at T=13 - after the document has been
   unlocked. KV-Engine's conflict resolver will compare the (locked)
   CAS at the destination (11) with the incoming CAS (10), and decide
   the destination should win - and the update will not be applied.

This is incorrect - we should have used CAS value before the document
was locked (T=0) for conflict resolution.

+Solution+

To address this, ep-engine needs to keep a copy of the original
(pre-lock) CAS, and restore the documents' CAS to this when the
document is unlocked.

This patch does this in a straightforward manner - add a second
locked_CAS field to StoredValue to fix this issue. This has the
downside of increasing the size of per-document in-memory metadata by
8 Bytes. An (space) optimisated version will follow in a subsequent
patch.

Change-Id: Icc274626dd5c1269bc1df65a2e26304df4aaf554
Reviewed-on: https://review.couchbase.org/c/kv_engine/+/201510
Tested-by: Dave Rigby <[email protected]>
Well-Formed: Restriction Checker
Reviewed-by: Trond Norbye <[email protected]>

Author: daverigby

engines/ep/src/kv_bucket.cc

@@ -1958,7 +1958,7 @@ cb::engine_errc KVBucket::unlockKey(const DocKey& key,
             return cb::engine_errc::no_such_key;
         }
         if (v->isLocked(currentTime)) {
-            if (v->getCas() == cas) {
+            if (v->getCasForWrite(currentTime) == cas) {
                 v->unlock();
                 return cb::engine_errc::success;
             }

engines/ep/src/stored-value.cc

@@ -516,6 +516,7 @@ static std::string getSystemEventsValueFromStoredValue(const StoredValue& sv) {
 }
 
 std::ostream& operator<<(std::ostream& os, const StoredValue& sv) {
+    const auto now = ep_current_time();
     // type, address
     os << (sv.isOrdered() ? "OSV @" : " SV @") << &sv << " ";
 
@@ -529,7 +530,7 @@ std::ostream& operator<<(std::ostream& os, const StoredValue& sv) {
     os << (sv.isDirty() ? 'W' : '.');
     os << (sv.isDeleted() ? 'D' : '.');
     os << (sv.isResident() ? 'R' : '.');
-    os << (sv.isLocked(ep_current_time()) ? 'L' : '.');
+    os << (sv.isLocked(now) ? 'L' : '.');
     switch (sv.getCommitted()) {
     case CommittedState::CommittedViaMutation:
         os << "Cm";
@@ -573,6 +574,9 @@ std::ostream& operator<<(std::ostream& os, const StoredValue& sv) {
     os << std::dec << "seq:" << uint64_t(sv.getBySeqno())
        << " rev:" << sv.getRevSeqno();
     os << " cas:" << sv.getCas();
+    if (sv.isLocked(now)) {
+        os << " locked_cas:" << sv.locked_cas;
+    }
     os << " key:\"" << sv.getKey() << "\"";
     if (sv.isOrdered() && sv.isDeleted()) {
         os << " del_time:"

engines/ep/src/stored-value.h

@@ -454,6 +454,20 @@ class StoredValue {
         return cas;
     }
 
+    /**
+     * Get the CAS needed to modify this SV using optimistic concurrency
+     * control. This is normally the current CAS (the CAS of the last mutation
+     * to the doc), but if the document is locked then it is the special
+     * locked_CAS generated by getLocked() and returned to the caller.
+     * @param curtime The current time, used to check if lock has expired
+     */
+    uint64_t getCasForWrite(rel_time_t curtime) const {
+        if (isLocked(curtime)) {
+            return locked_cas;
+        }
+        return cas;
+    }
+
     /**
      * Set a new CAS ID.
      */
@@ -462,15 +476,17 @@ class StoredValue {
     }
 
     /**
-     * Lock this item until the given time.
+     * Lock this item until the given time, setting the locked CAS to
+     * the specified value.
      */
-    void lock(rel_time_t expiry) {
+    void lock(rel_time_t expiry, uint64_t cas) {
         if (isDeleted()) {
             // Cannot lock Deleted items.
             throw std::logic_error(
                     "StoredValue::lock: Called on Deleted item");
         }
         lock_expiry_or_delete_or_complete_time.lock_expiry = expiry;
+        locked_cas = cas;
     }
 
     /**
@@ -482,6 +498,7 @@ class StoredValue {
             return;
         }
         lock_expiry_or_delete_or_complete_time.lock_expiry = 0;
+        locked_cas = 0;
     }
 
     /**
@@ -1058,6 +1075,10 @@ class StoredValue {
     // the tag.
     UniquePtr chain_next_or_replacement; // 8 bytes (2-byte tag, 6 byte address)
     uint64_t           cas;            //!< CAS identifier.
+    // The CAS to use to modify/unlock a locked document. Stored separately from
+    // 'cas' as we must preserve the original CAS in the event the locked
+    // document is unlocked without being modified.
+    uint64_t locked_cas;
     // bySeqno is atomic primarily for TSAN, which would flag that we write/read
     // this in ephemeral backfills with different locks (which is true, but the
     // access is we believe actually safe)

engines/ep/src/vbucket.cc

@@ -3124,11 +3124,10 @@ GetValue VBucket::getLocked(rel_time_t currentTime,
         }
 
         // acquire lock and increment cas value
-        v->lock(currentTime + lockTimeout);
+        v->lock(currentTime + lockTimeout, nextHLCCas());
 
         auto it = v->toItem(getId());
-        it->setCas(nextHLCCas());
-        v->setCas(it->getCas());
+        it->setCas(v->getCasForWrite(currentTime));
 
         return GetValue(std::move(it));
     }
@@ -3491,17 +3490,18 @@ std::pair<MutationStatus, std::optional<VBNotifyCtx>> VBucket::processSetInner(
             !committed->isDeleted()) {
             return {MutationStatus::InvalidCas, {}};
         }
-        if (committed->isLocked(ep_current_time())) {
+        const auto now = ep_current_time();
+        if (committed->isLocked(now)) {
             /*
              * item is locked, deny if there is cas value mismatch
              * or no cas value is provided by the user
              */
-            if (cas != committed->getCas()) {
+            if (cas != committed->getCasForWrite(now)) {
                 return {MutationStatus::IsLocked, {}};
             }
             /* allow operation*/
             committed->unlock();
-        } else if (cas && cas != committed->getCas()) {
+        } else if (cas && cas != committed->getCasForWrite(now)) {
             if (committed->isTempNonExistentItem()) {
                 // This is a temporary item which marks a key as non-existent;
                 // therefore specifying a non-matching CAS should be exposed
@@ -3771,15 +3771,19 @@ VBucket::processSoftDeleteInner(const HashTable::HashBucketLock& hbl,
         return std::make_tuple(MutationStatus::NeedBgFetch, &v, empty);
     }
 
-    if (v.isLocked(ep_current_time())) {
-        if (cas != v.getCas()) {
+    // Perform CAS check. If document is locked then the request cas must
+    // match the locked CAS (i.e. the same CAS returned from the getLocked()
+    // operation), if not locked then check against the normal cas (last
+    // modified time) if CAS specified otherwise no CAS permitted.
+    const auto now = ep_current_time();
+    if (v.isLocked(now)) {
+        if (cas != v.getCasForWrite(now)) {
             return std::make_tuple(MutationStatus::IsLocked, &v, empty);
         }
-        v.unlock();
-    }
-
-    if (cas != 0 && cas != v.getCas()) {
-        return std::make_tuple(MutationStatus::InvalidCas, &v, empty);
+    } else {
+        if (cas != 0 && cas != v.getCas()) {
+            return std::make_tuple(MutationStatus::InvalidCas, &v, empty);
+        }
     }
 
     /* allow operation */

engines/ep/tests/ep_testsuite_xdcr.cc

@@ -3173,6 +3173,75 @@ static enum test_result test_expiration_options(EngineIface* h) {
     return SUCCESS;
 }
 
+// Regression test for MB-59744 - a SetWithMeta against a document which was
+// previously locked but has been unlocked without modification should perform
+// conflict resolution using the original CAS before the document was locked.
+static enum test_result test_mb_59744_set_meta_lww_after_locked(EngineIface* h) {
+
+    // 1. Store an initial document so we can lock it. Create with a known CAS
+    //    value so we can later perform a second SetWithMeta which _should_
+    //    win conflict resolution.
+    ItemMetaData itemMeta;
+    itemMeta.revSeqno = 1;
+    itemMeta.cas = 1000;
+
+    std::string key{"key"};
+    Vbid vb{0};
+    checkeq(cb::engine_errc::success,
+            set_with_meta(h,
+                          key.data(),
+                          key.size(),
+                          {},
+                          0,
+                          vb,
+                          &itemMeta,
+                          0,
+                          FORCE_ACCEPT_WITH_META_OPS),
+            "Expected item to be stored");
+
+    // 2. Lock the document. This will set the locked_CAS to the current HLC
+    //    time, which is nanoseconds since unix epoch, so much larger than
+    //    the CAS we just set (1000).
+    auto* cookie = testHarness->create_cookie(h);
+    auto locked = getl(h, cookie, key.data(), vb, 0);
+    checkeq(cb::engine_errc::success, locked.first, "Expected GetLocked to succeed");
+    checkgt(locked.second->getCas(), itemMeta.cas, "Expected locked CAs to be greater than current CAS");
+
+    // 3. Unlock the document. Before the fix for MB-59744, this would leave
+    //    the document having the same increased locked_CAS.
+    checkeq(cb::engine_errc::success,
+            unl(h, cookie, key.c_str(), vb, locked.second->getCas()),
+            "Unlock should have succeedd");
+
+    // 4. Test - attempt to set with meta, using confict resolution, to a
+    //    CAS value one greater than previous mutation, but crucially less than
+    //    the temporary locked CAS.
+    itemMeta.revSeqno++;
+    itemMeta.cas++;
+    checkeq(cb::engine_errc::success,
+            set_with_meta(h,
+                          key.data(),
+                          key.size(),
+                          {},
+                          0,
+                          vb,
+                          &itemMeta,
+                          0,
+                          FORCE_ACCEPT_WITH_META_OPS),
+            "Expected SetWithMeta with CAS+1 to succeed");
+
+    // 5. Check document has indeed been updated with new CAS / seqno.
+    cb::EngineErrorMetadataPair meta;
+    check(get_meta(h, key.c_str(), meta, cookie), "Expected GetMeta to succeed");
+    checkeq(itemMeta.cas, meta.second.cas, "Expected CAS to be updated");
+    checkeq(uint64_t(itemMeta.revSeqno), meta.second.seqno,
+            "Expected revid to be updated");
+
+    testHarness->destroy_cookie(cookie);
+
+    return SUCCESS;
+}
+
 // Test manifest //////////////////////////////////////////////////////////////
 
 const char* default_dbname = "./ep_testsuite_xdcr.db";
@@ -3519,4 +3588,12 @@ BaseTestCase testsuite_testcases[] = {
                  prepare,
                  cleanup),
 
-        TestCase(nullptr, nullptr, nullptr, nullptr, nullptr, prepare, cleanup)};
+        TestCase("MB-59744 SetMeta LWW after locked",
+                 test_mb_59744_set_meta_lww_after_locked,
+                 test_setup,
+                 teardown,
+                 "conflict_resolution_type=lww",
+                 prepare,
+                 cleanup),
+
+    TestCase(nullptr, nullptr, nullptr, nullptr, nullptr, prepare, cleanup)};

engines/ep/tests/module_tests/kv_bucket_test.cc

@@ -1421,11 +1421,13 @@ TEST_P(KVBucketParamTest, unlockKeyTempDeletedTest) {
     GetValue gv =
             store->getAndUpdateTtl(key, vbid, cookie, ep_real_time() + 10000);
     EXPECT_EQ(cb::engine_errc::success, gv.getStatus());
+    auto docCas = gv.item->getCas();
 
     gv = store->getLocked(key, vbid, ep_current_time(), 10, cookie);
     EXPECT_EQ(cb::engine_errc::success, gv.getStatus());
 
-    itm.setCas(gv.item->getCas());
+    // Need the "real" documents' CAS for expiry.
+    itm.setCas(docCas);
     store->processExpiredItem(itm, ep_real_time() + 10001, ExpireBy::Pager);
 
     flushVBucketToDiskIfPersistent(vbid, 1);

engines/ep/tests/module_tests/stored_value_test.cc

@@ -388,9 +388,9 @@ TYPED_TEST(ValueTest, freqCounterNotReset) {
 /// size (we've carefully crafted them to be as efficient as possible).
 TEST(StoredValueTest, expectedSize) {
 #ifdef CB_MEMORY_INEFFICIENT_TAGGED_PTR
-    const long expected_size = 64;
+    const long expected_size = 72;
 #else
-    const long expected_size = 56;
+    const long expected_size = 64;
 #endif
     EXPECT_EQ(expected_size, sizeof(StoredValue))
             << "Unexpected change in StoredValue fixed size";
@@ -419,16 +419,61 @@ TYPED_TEST(ValueTest, MB_32568) {
     EXPECT_EQ(DeleteSource::TTL, this->sv->getDeletionSource());
 }
 
+// Validate the CAS for writing changes when locked, and is restored afterwards
+// when unlocked.
+TYPED_TEST(ValueTest, LockedCas) {
+    this->sv->setCas(123);
+    ASSERT_EQ(123, this->sv->getCasForWrite(0))
+            << "CAS used for writing should be main CAS if not locked";
+
+    rel_time_t lock_expiry{10};
+    this->sv->lock(lock_expiry, 456);
+    EXPECT_EQ(456, this->sv->getCasForWrite(lock_expiry - 1))
+            << "CAS used for writing should have changed to locked CAS while "
+               "lock still valid";
+    EXPECT_EQ(123, this->sv->getCas())
+            << "main CAS should be unchanged when locked";
+
+    this->sv->unlock();
+    EXPECT_EQ(123, this->sv->getCasForWrite(lock_expiry - 1))
+            << "CAS used for writing should have reverted to previous VAS "
+               "after explicit unlock";
+    EXPECT_EQ(123, this->sv->getCas())
+            << "main CAS should be unchanged after explicit unlock";
+}
+
+// Validate the CAS for writing changes when locked, and is restored when lock
+// expires.
+TYPED_TEST(ValueTest, LockedCasExpired) {
+    this->sv->setCas(123);
+    ASSERT_EQ(123, this->sv->getCasForWrite(0))
+            << "CAS used for writing should be main CAS if not locked";
+
+    rel_time_t lock_expiry{10};
+    this->sv->lock(lock_expiry, 456);
+    EXPECT_EQ(456, this->sv->getCasForWrite(lock_expiry - 1))
+            << "CAS used for writing should have changed to locked CAS while "
+               "lock still valid";
+    EXPECT_EQ(123, this->sv->getCas())
+            << "main CAS should be unchanged when locked";
+
+    EXPECT_EQ(123, this->sv->getCasForWrite(lock_expiry + 1))
+            << "CAS used for writing should have reverted to previous VAS "
+               "after lock expires";
+    EXPECT_EQ(123, this->sv->getCas())
+            << "main CAS should be unchanged after locked expires";
+}
+
 /**
  * Test fixture for OrderedStoredValue-only tests.
  */
 class OrderedStoredValueTest : public ValueTest<OrderedStoredValueFactory> {};
 
 TEST_F(OrderedStoredValueTest, expectedSize) {
 #ifdef CB_MEMORY_INEFFICIENT_TAGGED_PTR
-    const long expected_size = 80;
+    const long expected_size = 88;
 #else
-    const long expected_size = 80;
+    const long expected_size = 88;
 #endif
 
     EXPECT_EQ(expected_size, sizeof(OrderedStoredValue))

include/memcached/protocol_binary.h

@@ -2192,6 +2192,9 @@ namespace mcbp::cas {
  * The special value used as a wildcard and match all CAS values
  */
 const uint64_t Wildcard = 0x0;
+
+/// The special value returned from Get and similar when document is locked.
+const uint64_t Locked = 0xffff'ffff'ffff'ffff;
 } // namespace mcbp::cas
 
 namespace cb::mcbp::request {