[BP] MB-46439: Don't remove resolved sync-writes when getting cookies

Currently ActiveDurabilityMonitor::getCookiesForInFlightSyncWrites()
isn't idempotent with respect to the sync-writes stored in
resolvedQueue. As it removes ActiveSyncWrite when trying to get hold of
their cookie so we can give an ambiguous status to the client.

This is problematic as before this patch the DurabilityCompletionTask
and DurabilityTimeoutTask could run after
getCookiesForInFlightSyncWrites() is called. This could mean that we end
up completing and persisting sync-writes to disk out of order. Due to
the sync-writes that where in resolvedQueue being removed before they
had be process by the DurabilityCompletionTask.

To fix this we've modified getCookiesForInFlightSyncWrites() to leave
the resolvedQueue in a valid state, with all the sync-writes that it had
before still there and in the same order, just with their cookie
removed. To do this we take hold of a write lock to the ADM state and
also the resolvedQueue consumer lock. This should give use exclusive
access so both the resolvedQueue and trackedWrites. Then we go though
the resolvedQueue removing ActiveSyncWrites and cache them in a vector.
To restore the state of the resolvedQueue we reset the queue after its
empty and re-push all the ActiveSyncWrites.

Change-Id: If4529c6e4074ef2e332e196a728919a26ba65c98
Reviewed-on: http://review.couchbase.org/c/kv_engine/+/154128
Tested-by: Build Bot <[email protected]>
Well-Formed: Build Bot <[email protected]>
Reviewed-by: Richard de Mellow <[email protected]>

Author: rdemellow

engines/ep/src/durability/active_durability_monitor.cc

@@ -894,19 +894,30 @@ ActiveDurabilityMonitor::prepareTransitionAwayFromActive() {
 std::vector<const void*>
 ActiveDurabilityMonitor::getCookiesForInFlightSyncWrites() {
     auto vec = std::vector<const void*>();
-    {
-        std::lock_guard<ResolvedQueue::ConsumerLock> lock(
-                resolvedQueue->getConsumerLock());
-        while (auto write = resolvedQueue->try_dequeue(lock)) {
-            auto* cookie = write->getCookie();
-            if (cookie) {
-                vec.push_back(cookie);
-                write->clearCookie();
-            }
+
+    std::lock_guard<ResolvedQueue::ConsumerLock> lock(
+            resolvedQueue->getConsumerLock());
+    // Take a write lock on the state now as we don't want trackedWrites being
+    // completed and being placed on the resolvedQueue while we pop all the
+    // SyncWrites from the resolvedQueue and then re-push them.
+    auto s = state.wlock();
+    std::vector<ActiveSyncWrite> resolvedSwToBeRePushed;
+    while (auto write = resolvedQueue->try_dequeue(lock)) {
+        auto* cookie = write->getCookie();
+        if (cookie) {
+            vec.push_back(cookie);
+            write->clearCookie();
         }
+        resolvedSwToBeRePushed.push_back(*write);
+    }
+    // "reset" the queue, this just sets the value of highEnqueuedSeqno back
+    // to 0 and ensures that the queue is empty. No memory management is
+    // performed by the method.
+    resolvedQueue->reset(lock);
+    for (auto& sw : resolvedSwToBeRePushed) {
+        resolvedQueue->enqueue(*s, std::move(sw));
     }
 
-    auto s = state.wlock();
     for (auto& write : s->trackedWrites) {
         auto* cookie = write.getCookie();
         if (cookie) {

engines/ep/tests/module_tests/evp_store_durability_test.cc

@@ -4404,6 +4404,63 @@ TEST_P(DurabilityBucketTest, PrepareDoesNotExpire) {
     EXPECT_EQ(DeleteSource::TTL, res.storedValue->getDeletionSource());
 }
 
+TEST_P(DurabilityBucketTest, MB_46272) {
+    using namespace cb::durability;
+    setVBucketToActiveWithValidTopology();
+
+    auto vb = store->getVBucket(vbid);
+    auto& adm = VBucketTestIntrospector::public_getActiveDM(*vb);
+
+    // 1. Two writes, one to abort after 5 seconds, one to abort after 10
+    auto keyA = makeStoredDocKey("keyA");
+    ASSERT_EQ(
+            ENGINE_SYNC_WRITE_PENDING,
+            store->set(*makePendingItem(
+                               keyA, "value", {Level::Majority, Timeout(4000)}),
+                       cookie));
+    auto keyB = makeStoredDocKey("keyB");
+    ASSERT_EQ(
+            ENGINE_SYNC_WRITE_PENDING,
+            store->set(*makePendingItem(
+                               keyB, "value", {Level::Majority, Timeout(9000)}),
+                       cookie));
+    // 2. Flush the two prepares
+    flushVBucketToDiskIfPersistent(vbid, 2);
+
+    // 3. timeout keyA which should generate an abort
+    adm.processTimeout(std::chrono::steady_clock::now() +
+                       std::chrono::seconds(5));
+    // 4. cancel the inflight sync-writes returning to the client as ambiguous
+    engine->cancel_all_operations_in_ewb_state();
+    // Now check that logically we would be in a consistent state if the
+    // durability timeout or completion tasks where to run
+
+    // 5. Timeout keyB which should generate an abort
+    adm.processTimeout(std::chrono::steady_clock::now() +
+                       std::chrono::seconds(10));
+    // 6. Process resolved sync-writes this should cause us to add two aborts
+    //    to the checkpoint manager first for keyA then keyB
+    vb->processResolvedSyncWrites();
+    const auto& ckptList =
+            CheckpointManagerTestIntrospector::public_getCheckpointList(
+                    *vb->checkpointManager);
+    const auto& ckpt = *ckptList.back();
+    auto it = ckpt.begin();
+    it++; // skip the empty-item
+    it++; // skip checkpoint_start
+    // 7. Check keyA's abort
+    EXPECT_EQ(keyA, (*it)->getKey());
+    EXPECT_EQ(queue_op::abort_sync_write, (*it)->getOperation());
+    EXPECT_EQ(3, (*it)->getBySeqno());
+    it++; // move to keyB's abort
+    // 8. Check keyB's abort
+    EXPECT_EQ(keyB, (*it)->getKey());
+    EXPECT_EQ(4, (*it)->getBySeqno());
+    EXPECT_EQ(queue_op::abort_sync_write, (*it)->getOperation());
+    // 9. Flush the two aborts to disk
+    flushVBucketToDiskIfPersistent(vbid, 2);
+}
+
 // Test cases which run against couchstore
 INSTANTIATE_TEST_CASE_P(AllBackends,
                         DurabilityCouchstoreBucketTest,