MB-33332: Give all Ephemeral backfill items Majority level

Ephemeral only supports sync writes with the Majority level so
instead of storing a durability level in each OrderedStoredValue
we can simply assume that they are all have the Majority level.

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

Author: BenHuddleston

engines/ep/src/dcp/backfill_memory.cc

@@ -21,6 +21,7 @@
 #include "ep_time.h"
 #include "ephemeral_vb.h"
 #include "seqlist.h"
+#include "stored-value.h"
 
 #include <phosphor/phosphor.h>
 
@@ -297,11 +298,19 @@ backfill_status_t DCPBackfillMemoryBuffered::scan() {
             // mutated with the HashBucketLock, so get the correct bucket lock
             // before calling StoredValue::toItem
             auto hbl = evb->ht.getLockedBucket((*rangeItr).getKey());
-            // @todo-durability: Need to record the original durability
-            // requirements in OrderedStoredValue (and then add into the
-            // item we make here); otherwise we cannot correctly form the
-            // DCP_PREPARE messages for Pending items.
-            item = (*rangeItr).toItem(getVBucketId());
+            // Ephemeral only supports a durable write level of Majority so
+            // instead of storing a durability level in our OrderedStoredValues
+            // we can just assume that all durable writes have the Majority
+            // level. Given that this is a backfill item (we will send via DCP)
+            // we also need to specify an infinite durable write timeout so that
+            // we do not lose any durable writes. We can supply these items
+            // for all stored values as they are only set if the underlying
+            // StoredValue has the CommittedState of Pending.
+            item = (*rangeItr).toItem(getVBucketId(),
+                                      StoredValue::HideLockedCas::No,
+                                      StoredValue::IncludeValue::Yes,
+                                      {{cb::durability::Level::Majority,
+                                        cb::durability::Timeout::Infinity()}});
             // A deleted ephemeral item stores the delete time under a delete
             // time field, this must be copied to the expiry time so that DCP
             // can transmit the original time of deletion

engines/ep/tests/module_tests/dcp_durability_stream_test.cc

@@ -17,6 +17,7 @@
 
 #include "dcp_durability_stream_test.h"
 
+#include "../../src/dcp/backfill-manager.h"
 #include "checkpoint_utils.h"
 #include "dcp/response.h"
 #include "dcp_utils.h"
@@ -205,6 +206,51 @@ TEST_P(DurabilityActiveStreamTest, SendDcpAbort) {
     ASSERT_FALSE(resp);
 }
 
+TEST_P(DurabilityActiveStreamEphemeralTest, BackfillDurabilityLevel) {
+    auto vb = engine->getVBucket(vbid);
+    auto& ckptMgr = *vb->checkpointManager;
+    // Get rid of set_vb_state and any other queue_op we are not interested in
+    ckptMgr.clear(*vb, 0 /*seqno*/);
+
+    const auto key = makeStoredDocKey("key");
+    const auto& value = "value";
+    auto item = makePendingItem(
+            key,
+            value,
+            cb::durability::Requirements(cb::durability::Level::Majority,
+                                         1 /*timeout*/));
+    VBQueueItemCtx ctx;
+    ctx.durability =
+            DurabilityItemCtx{item->getDurabilityReqs(), nullptr /*cookie*/};
+
+    EXPECT_EQ(MutationStatus::WasClean, public_processSet(*vb, *item, ctx));
+
+    // We don't account Prepares in VB stats
+    EXPECT_EQ(0, vb->getNumItems());
+
+    stream->transitionStateToBackfilling();
+    ASSERT_TRUE(stream->isBackfilling());
+
+    // Run the backfill we scheduled when we transitioned to the backfilling
+    // state
+    auto& bfm = producer->getBFM();
+    bfm.backfill();
+
+    const auto& readyQ = stream->public_readyQ();
+    EXPECT_EQ(2, readyQ.size());
+
+    // First item is a snapshot marker so just skip it
+    auto resp = stream->public_popFromReadyQ();
+    resp = stream->public_popFromReadyQ();
+    ASSERT_TRUE(resp);
+    EXPECT_EQ(DcpResponse::Event::Prepare, resp->getEvent());
+    const auto& prep = static_cast<MutationResponse&>(*resp);
+    const auto respItem = prep.getItem();
+    EXPECT_EQ(cb::durability::Level::Majority,
+              respItem->getDurabilityReqs().getLevel());
+    EXPECT_TRUE(respItem->getDurabilityReqs().getTimeout().isInfinite());
+}
+
 TEST_P(DurabilityActiveStreamTest, RemoveUnknownSeqnoAckAtDestruction) {
     auto vb = engine->getVBucket(vbid);
 
@@ -798,3 +844,8 @@ INSTANTIATE_TEST_CASE_P(
         DurabilityPassiveStreamPersistentTest,
         STParameterizedBucketTest::persistentAllBackendsConfigValues(),
         STParameterizedBucketTest::PrintToStringParamName);
+
+INSTANTIATE_TEST_CASE_P(AllBucketTypes,
+                        DurabilityActiveStreamEphemeralTest,
+                        STParameterizedBucketTest::ephConfigValues(),
+                        STParameterizedBucketTest::PrintToStringParamName);

engines/ep/tests/module_tests/dcp_durability_stream_test.h

@@ -56,3 +56,9 @@ class DurabilityPassiveStreamTest : public SingleThreadedPassiveStreamTest {
  */
 class DurabilityPassiveStreamPersistentTest
     : public DurabilityPassiveStreamTest {};
+
+/**
+ * ActiveStream tests for Durability against ephemeral buckets. Single-threaded.
+ */
+class DurabilityActiveStreamEphemeralTest : public DurabilityActiveStreamTest {
+};