MB-34017 [SR]: Include the prepared-seqno in Commit queued items

When we complete a Prepare, we queue a Commit or Abort item into the
CheckpointManager.
At persistence of Commit/Abort, we don't know what is the Prepare
associated with that Commit/Abort.
While that information has never been necessary so far, now it is for
persisting the High Completed Seqno to disk.

This patch ensures that both the Commit and Abort queued items in CM
carry the seqno of the associated Prepare (ie, the prepared-seqno).

In follow-up patches, the prepared-seqno will be used at persistence for
computing the HCS and writing it to disk.

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

Author: paolococchi

engines/ep/src/item.h

@@ -515,10 +515,10 @@ class Item : public RCValue {
     std::atomic<int64_t> bySeqno;
 
     // @todo: Try to avoid this, as it increases mem_usage of 8 bytes per Item.
-    // This is added for Durability items Commit and Abort, which both carry
-    // 2 seqnos:
-    // 1) the seqno of this (Commit/Abort) Item (encoded in bySeqno)
-    // 2) the seqno of the Committed/Aborted Prepare
+    // This is added for Durability items Commit and Abort, for which need the
+    // associated Prepare's seqno at persistence for writing the correct High
+    // Completed Seqno (HCS) to disk. Note that the HCS is the seqno of the
+    // last Committed or Aborted Prepare.
     cb::uint48_t prepareSeqno;
 
     uint32_t queuedTime;

engines/ep/src/vbucket.cc

@@ -798,6 +798,9 @@ ENGINE_ERROR_CODE VBucket::commit(
     // it may have been set explicitly.
     queueItmCtx.genCas = GenerateCas::No;
 
+    queueItmCtx.durability =
+            DurabilityItemCtx{res.pending->getBySeqno(), nullptr /*cookie*/};
+
     auto notify = commitStoredValue(res, queueItmCtx, commitSeqno);
 
     notifyNewSeqno(notify);
@@ -1145,17 +1148,26 @@ VBNotifyCtx VBucket::queueDirty(const HashTable::HashBucketLock& hbl,
         setMaxCasAndTrackDrift(v.getCas());
     }
 
-    // If we are queuing a SyncWrite StoredValue; extract the durability
+    // If we are queueing a SyncWrite StoredValue; extract the durability
     // requirements to use to create the Item.
     boost::optional<cb::durability::Requirements> durabilityReqs;
-    if (ctx.durability) {
-        durabilityReqs = ctx.durability->requirements;
+    if (v.isPending()) {
+        Expects(ctx.durability.is_initialized());
+        durabilityReqs = boost::get<cb::durability::Requirements>(
+                ctx.durability->requirementsOrPreparedSeqno);
     }
+
     queued_item qi(v.toItem(getId(),
                             StoredValue::HideLockedCas::No,
                             StoredValue::IncludeValue::Yes,
                             durabilityReqs));
 
+    if (qi->isCommitSyncWrite()) {
+        Expects(ctx.durability.is_initialized());
+        qi->setPrepareSeqno(boost::get<int64_t>(
+                ctx.durability->requirementsOrPreparedSeqno));
+    }
+
     // MB-27457: Timestamp deletes only when they don't already have a timestamp
     // assigned. This is here to ensure all deleted items have a timestamp which
     // our tombstone purger can use to determine which tombstones to purge. A

engines/ep/src/vbucket.h

@@ -28,6 +28,8 @@
 #include "vbucket_bgfetch_item.h"
 #include "vbucket_fwd.h"
 
+#include "boost/variant.hpp"
+
 #include <folly/Synchronized.h>
 #include <memcached/engine.h>
 #include <nlohmann/json.hpp>
@@ -78,8 +80,13 @@ struct VBNotifyCtx {
  * has Durability requirements.
  */
 struct DurabilityItemCtx {
-    /// The durability requirements for this item.
-    cb::durability::Requirements requirements;
+    /**
+     * Stores:
+     * - the Durability Requirements, if we queue a Prepare
+     * - the prepared-seqno, if we queue a Commit
+     */
+    boost::variant<cb::durability::Requirements, int64_t>
+            requirementsOrPreparedSeqno;
     /**
      * The client cookie associated with the Durability operation. If non-null
      * then notifyIOComplete will be called on it when operation is committed.
@@ -1795,6 +1802,7 @@ class VBucket : public std::enable_shared_from_this<VBucket> {
      *
      * @param hbl The lock to the HashTable-bucket containing the StoredValue
      * @param v The StoredValue of the Pending being aborted.
+     * @param prepareSeqno The seqno of the Prepare being aborted
      * @param ctx The VBQueueItemCtx. Holds info needed to enqueue the item,
      *     look at the structure for details.
      * @return the notification context used for notifying the Flusher and

engines/ep/tests/module_tests/vbucket_durability_test.cc

@@ -491,7 +491,7 @@ TEST_P(VBucketDurabilityTest, Active_Commit_MultipleReplicas) {
             CheckpointManagerTestIntrospector::public_getCheckpointList(
                     *ckptMgr);
 
-    auto checkPending = [this, &key, &ckptList]() -> void {
+    auto checkPending = [this, &key, &ckptList, preparedSeqno]() -> void {
         EXPECT_EQ(nullptr, ht->findForRead(key).storedValue);
         const auto sv = ht->findForWrite(key).storedValue;
         ASSERT_NE(nullptr, sv);
@@ -501,11 +501,13 @@ TEST_P(VBucketDurabilityTest, Active_Commit_MultipleReplicas) {
         for (const auto& qi : *ckptList.front()) {
             if (!qi->isCheckPointMetaItem()) {
                 EXPECT_EQ(queue_op::pending_sync_write, qi->getOperation());
+                EXPECT_EQ(preparedSeqno, qi->getBySeqno());
+                EXPECT_EQ("value", qi->getValue()->to_s());
             }
         }
     };
 
-    auto checkCommitted = [this, &key, &ckptList]() -> void {
+    auto checkCommitted = [this, &key, &ckptList, preparedSeqno]() -> void {
         const auto sv = ht->findForRead(key).storedValue;
         ASSERT_NE(nullptr, sv);
         EXPECT_NE(nullptr, ht->findForWrite(key).storedValue);
@@ -515,6 +517,9 @@ TEST_P(VBucketDurabilityTest, Active_Commit_MultipleReplicas) {
         for (const auto& qi : *ckptList.front()) {
             if (!qi->isCheckPointMetaItem()) {
                 EXPECT_EQ(queue_op::commit_sync_write, qi->getOperation());
+                EXPECT_GT(qi->getBySeqno() /*commitSeqno*/, preparedSeqno);
+                EXPECT_EQ(preparedSeqno, qi->getPrepareSeqno());
+                EXPECT_EQ("value", qi->getValue()->to_s());
             }
         }
     };
@@ -679,7 +684,8 @@ TEST_P(VBucketDurabilityTest, NonPendingKeyAtAbort) {
  * 3) the abort_sync_write is not added to the DurabilityMonitor
  */
 TEST_P(VBucketDurabilityTest, Active_AbortSyncWrite) {
-    storeSyncWrites({1} /*seqno*/);
+    const int64_t preparedSeqno = 1;
+    storeSyncWrites({preparedSeqno});
     ASSERT_EQ(1,
               VBucketTestIntrospector::public_getActiveDM(*vbucket)
                       .getNumTracked());
@@ -695,9 +701,9 @@ TEST_P(VBucketDurabilityTest, Active_AbortSyncWrite) {
         // Visible at write
         auto storedItem = ht->findForWrite(key);
         ASSERT_TRUE(storedItem.storedValue);
-        // item pending
         EXPECT_EQ(CommittedState::Pending,
                   storedItem.storedValue->getCommitted());
+        EXPECT_EQ(preparedSeqno, storedItem.storedValue->getBySeqno());
     }
 
     const auto& ckptList =
@@ -719,6 +725,7 @@ TEST_P(VBucketDurabilityTest, Active_AbortSyncWrite) {
     it++;
     ASSERT_EQ(1, ckpt->getNumItems());
     EXPECT_EQ(queue_op::pending_sync_write, (*it)->getOperation());
+    EXPECT_EQ(preparedSeqno, (*it)->getBySeqno());
     EXPECT_EQ("value", (*it)->getValue()->to_s());
 
     // The Pending is tracked by the DurabilityMonitor
@@ -765,6 +772,8 @@ TEST_P(VBucketDurabilityTest, Active_AbortSyncWrite) {
     EXPECT_EQ(queue_op::abort_sync_write, (*it)->getOperation());
     EXPECT_TRUE((*it)->isDeleted());
     EXPECT_FALSE((*it)->getValue());
+    EXPECT_GT((*it)->getBySeqno(), preparedSeqno);
+    EXPECT_EQ(preparedSeqno, (*it)->getPrepareSeqno());
 
     // The Aborted item is not added for tracking.
     // Note: The Pending has not been removed as we are testing at VBucket
@@ -1026,8 +1035,9 @@ void VBucketDurabilityTest::testHTSyncDeleteCommit() {
     auto* writeView = ht->findForWrite(key).storedValue;
     ASSERT_TRUE(writeView);
     VBQueueItemCtx ctx;
-    ctx.durability =
-            DurabilityItemCtx{{cb::durability::Level::Majority, {}}, cookie};
+    ctx.durability = DurabilityItemCtx{
+            cb::durability::Requirements{cb::durability::Level::Majority, {}},
+            cookie};
     ASSERT_EQ(MutationStatus::WasDirty,
               public_processSoftDelete(key, ctx).first);
 
@@ -1462,14 +1472,28 @@ void VBucketDurabilityTest::testCompleteSWInPassiveDM(vbucket_state_t state,
     const auto& ckptList =
             CheckpointManagerTestIntrospector::public_getCheckpointList(
                     *ckptMgr);
+    // 1 checkpoint
     ASSERT_EQ(1, ckptList.size());
-    EXPECT_EQ(writes.size(), ckptList.front()->getNumItems());
+    // empty-item
+    const auto& ckpt = *ckptList.front();
+    auto it = ckpt.begin();
+    ASSERT_EQ(queue_op::empty, (*it)->getOperation());
+    // 1 metaitem (checkpoint-start)
+    it++;
+    ASSERT_EQ(1, ckpt.getNumMetaItems());
+    EXPECT_EQ(queue_op::checkpoint_start, (*it)->getOperation());
+    // 3 non-metaitem are Committed or Aborted
+    ASSERT_EQ(writes.size(), ckpt.getNumItems());
     const auto expectedOp =
             (res == Resolution::Commit ? queue_op::commit_sync_write
                                        : queue_op::abort_sync_write);
-    for (const auto& qi : *ckptList.front()) {
-        if (!qi->isCheckPointMetaItem()) {
-            EXPECT_EQ(expectedOp, qi->getOperation());
+    for (const auto& prepare : writes) {
+        it++;
+        EXPECT_EQ(expectedOp, (*it)->getOperation());
+        EXPECT_GT((*it)->getBySeqno() /*commitSeqno*/, prepare.seqno);
+        EXPECT_EQ(prepare.seqno, (*it)->getPrepareSeqno());
+        if (expectedOp == queue_op::commit_sync_write) {
+            EXPECT_EQ("value", (*it)->getValue()->to_s());
         }
     }