MB-35656: Don't backfill completed prepares

We don't need to backfill any completed prepares. If we send them
to the replica then the replica will just purge them on the next
compaction (regardless of the pruge seqno) because they are not
needed.

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

Author: BenHuddleston

engines/ep/src/dcp/backfill_disk.cc

@@ -124,6 +124,13 @@ void DiskCallback::callback(GetValue& val) {
         throw std::invalid_argument("DiskCallback::callback: val is NULL");
     }
 
+    // Don't need to send any completed prepares to the replica. The compactor
+    // would just remove them on it's next run anyway.
+    if (val.item->isPending() &&
+        static_cast<uint64_t>(val.item->getBySeqno()) <= highCompletedSeqno) {
+        return;
+    }
+
     // MB-26705: Make the backfilled item cold so ideally the consumer would
     // evict this before any cached item if they get into memory pressure.
     val.item->setNRUValue(MAX_NRU_VALUE);
@@ -247,6 +254,7 @@ backfill_status_t DCPBackfillDisk::create() {
         stream->setDead(status);
         transitionState(backfill_state_done);
     } else {
+        cb->setHighCompletedSeqno(scanCtx->highCompletedSeqno);
         stream->incrBackfillRemaining(scanCtx->documentCount);
         stream->markDiskSnapshot(
                 startSeqno, scanCtx->maxSeqno, scanCtx->highCompletedSeqno);

engines/ep/src/dcp/backfill_disk.h

@@ -61,8 +61,13 @@ class DiskCallback : public StatusCallback<GetValue> {
 
     void callback(GetValue& val);
 
+    void setHighCompletedSeqno(uint64_t seqno) {
+        highCompletedSeqno = seqno;
+    };
+
 private:
     std::weak_ptr<ActiveStream> streamPtr;
+    uint64_t highCompletedSeqno;
 };
 
 /**

engines/ep/src/dcp/backfill_memory.cc

@@ -299,6 +299,15 @@ 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());
+
+            // Don't need to send any completed prepares, the purging task will
+            // just remove them on the replica anyway
+            if ((*rangeItr).getCommitted() ==
+                CommittedState::PrepareCommitted) {
+                ++rangeItr;
+                continue;
+            }
+
             // 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

engines/ep/tests/module_tests/dcp_durability_stream_test.cc

@@ -829,8 +829,9 @@ TEST_P(DurabilityActiveStreamTest,
     auto& bfm = producer->getBFM();
     bfm.backfill();
     bfm.backfill();
-    EXPECT_EQ(3, stream->public_readyQSize());
-    stream->consumeBackfillItems(3);
+    // We won't backfill any completed prepares
+    EXPECT_EQ(2, stream->public_readyQSize());
+    stream->consumeBackfillItems(2);
 
     EXPECT_EQ(ENGINE_SUCCESS, stream->seqnoAck(replica, 1 /*prepareSeqno*/));
 
@@ -876,7 +877,7 @@ TEST_P(DurabilityActiveStreamTest, DiskSnapshotSendsHCSWithSyncRepSupport) {
     ASSERT_EQ(0, producer->getBytesOutstanding());
 
     // readyQ must contain a SnapshotMarker
-    ASSERT_EQ(3, stream->public_readyQSize());
+    ASSERT_EQ(2, stream->public_readyQSize());
     auto resp = stream->public_nextQueuedItem();
     ASSERT_TRUE(resp);
     EXPECT_EQ(DcpResponse::Event::SnapshotMarker, resp->getEvent());

engines/ep/tests/module_tests/dcp_reflection_test.cc

@@ -645,7 +645,7 @@ void DCPLoopbackStreamTest::testBackfillAndInMemoryDuplicatePrepares(
     // (SNAP_MARKER, PRE, CMT, SET), (SNAP_MARKER, PRE), with a flush after the
     // first 4.
     route0_1.transferSnapshotMarker(0, 3, MARKER_FLAG_CHK | MARKER_FLAG_DISK);
-    route0_1.transferMessage(DcpResponse::Event::Prepare);
+    // We won't backfill and send a completed prepare so first item is CMT.
     route0_1.transferMutation(makeStoredDocKey("a"), 2);
     route0_1.transferMutation(makeStoredDocKey("b"), 3);
 

engines/ep/tests/module_tests/dcp_stream_sync_repl_test.cc

@@ -564,7 +564,7 @@ void DcpStreamSyncReplTest::testBackfillPrepareCommit(
     // Wait for the backfill task to have pushed all items to the Stream::readyQ
     // Note: we expect 1 SnapshotMarker + numItems in the readyQ
     std::chrono::microseconds uSleepTime(128);
-    while (stream->public_readyQSize() < 1 + 2) {
+    while (stream->public_readyQSize() < 1 + 1) {
         uSleepTime = decayingSleep(uSleepTime);
     }
 
@@ -574,9 +574,8 @@ void DcpStreamSyncReplTest::testBackfillPrepareCommit(
     EXPECT_EQ(0, dcpSnapMarker.getStartSeqno());
     EXPECT_EQ(2, dcpSnapMarker.getEndSeqno());
 
-    item = stream->public_nextQueuedItem();
-    verifyDcpPrepare(*prepared, *item);
-
+    // We won't backfill a completed prepare as the replica would just compact
+    // it away.
     item = stream->public_nextQueuedItem();
     // In general, a backfill from disk will send a mutation instead of a
     // commit as we may have de-duped the preceding prepare. The only case where