MB-35001: Consumer: Avoid duplicate PREs after disk/mem transition

Summary:

When transitioning to the inMemory phase of an Active DCP stream,
ensure that the initial in-memory Snapshot has the CKPT flag set, to
avoid the replica potentially seeing duplicate keys in the same
Checkpoint.

Details:

Consider the following scenario of items on disk and in memory
(checkpoint manager):

    Disk:
    1:PRE(a), 2:CMT(a), 3:SET(b)

    Memory:
                        3:CKPT_START
                        3:SET(b),     4:PRE(a), 5:SET(c)

(items 1..2 were in a removed checkpoint and no longer in-memory.)

An ep-engine replica attempting to stream all of this (0..infinity)
will result in a backfill of items 1..3, with a checkpoint cursor
being placed at seqno:4. Note this isn't the start of the Checkpoint
(which is 3) and hence not pointing at a checkpoint_start item.

As such when this is streamed over DCP (up to seqno:4) the consumer
will see:

    SNAPSHOT_MARKER(start=1, end=3, flags=DISK|CKPT)
    1:PRE(a)
    2:CMT(a)
    3:SET(b)
    SNAPSHOT_MARKER(start=4, end=5, flags=MEM)
    4:PRE(a),
    [[[missing seqno 5]]

If the consumer puts all of these mutations in the same Checkpoint,
then it will result in duplicate PRE(a) items (which breaks Checkpoint
invariant).

Address this by adding the CKPT flag to the Snapshot marker when
transitioning between disk and memory snapshots.

Note there is still an oustanding issue with the calculation of the
failover table branch point as a consequence of this change; however
that is being deferred for a second MB - see MB-35003.

Change-Id: Idfb3b676136df0a3a6af7e626d793b6e696d7dfe
Reviewed-on: http://review.couchbase.org/111908
Reviewed-by: Jim Walker <[email protected]>
Tested-by: Dave Rigby <[email protected]>

Author: daverigby

engines/ep/src/dcp/active_stream.cc

@@ -1616,11 +1616,23 @@ void ActiveStream::transitionState(StreamState newState) {
             endStream(END_STREAM_OK);
             notifyStreamReady();
         } else {
+            // Starting a new in-memory snapshot which could contain duplicate
+            // keys compared to the previous backfill snapshot. Therefore set
+            // the Checkpoint flag on the next snapshot so the Consumer will
+            // know to create a new Checkpoint.
+            nextSnapshotIsCheckpoint = true;
             nextCheckpointItem();
         }
         break;
     case StreamState::TakeoverSend:
         takeoverStart = ep_current_time();
+
+        // Starting a new in-memory (takeover) snapshot which could contain
+        // duplicate keys compared to the previous Backfill snapshot. Therefore
+        // set the Checkpoint flag on the next snapshot so the Consumer will
+        // know to create a new Checkpoint.
+        nextSnapshotIsCheckpoint = true;
+
         if (!nextCheckpointItem()) {
             notifyStreamReady(true);
         }

engines/ep/src/dcp/response.cc

@@ -108,6 +108,41 @@ uint32_t MutationResponse::getMessageSize() const {
     return header + body;
 }
 
+MutationConsumerMessage::MutationConsumerMessage(MutationResponse& response)
+    : MutationResponse(response.getItem(),
+                       response.getOpaque(),
+                       response.getIncludeValue(),
+                       response.getIncludeXattrs(),
+                       response.getIncludeDeleteTime(),
+                       response.getDocKeyEncodesCollectionId(),
+                       response.getEnableExpiryOutput(),
+                       response.getStreamId()),
+      emd(nullptr) {
+    // If the item is actually a commit_sync_write which had to be transmitted
+    // as a DCP_MUTATION (i.e. MutationConsumerResponse), we need
+    // to recreate the Item as operation==mutation otherwise the Consumer cannot
+    // handle it.
+    if (item_->getOperation() == queue_op::commit_sync_write) {
+        auto* mutation = new Item(item_->getKey(),
+                                  item_->getFlags(),
+                                  item_->getExptime(),
+                                  item_->getValue()->getData(),
+                                  item_->getValue()->valueSize(),
+                                  item_->getDataType(),
+                                  item_->getCas(),
+                                  item_->getBySeqno(),
+                                  item_->getVBucketId(),
+                                  item_->getRevSeqno(),
+                                  item_->getNRUValue(),
+                                  item_->getNRUValue());
+        // Using const_cast here as item_ is const; alternative would be a
+        // complex initialiser_list with ternaries to move the object
+        // constuction there. Given the object isn't created until after this
+        // method returns, this seems a reasonable compromise.
+        const_cast<queued_item&>(item_).reset(mutation);
+    }
+}
+
 uint32_t MutationConsumerMessage::getMessageSize() const {
     uint32_t body = MutationResponse::getMessageSize();
 

engines/ep/src/dcp/response.h

@@ -358,19 +358,19 @@ class SnapshotMarker : public DcpResponse {
           flags_(flags) {
     }
 
-    Vbid getVBucket() {
+    Vbid getVBucket() const {
         return vbucket_;
     }
 
-    uint64_t getStartSeqno() {
+    uint64_t getStartSeqno() const {
         return start_seqno_;
     }
 
-    uint64_t getEndSeqno() {
+    uint64_t getEndSeqno() const {
         return end_seqno_;
     }
 
-    uint32_t getFlags() {
+    uint32_t getFlags() const {
         return flags_;
     }
 
@@ -554,18 +554,13 @@ class MutationConsumerMessage : public MutationResponse {
           emd(e) {
     }
 
-    // Used in test code for wiring a producer/consumer directly
-    MutationConsumerMessage(MutationResponse& response)
-        : MutationResponse(response.getItem(),
-                           response.getOpaque(),
-                           response.getIncludeValue(),
-                           response.getIncludeXattrs(),
-                           response.getIncludeDeleteTime(),
-                           response.getDocKeyEncodesCollectionId(),
-                           response.getEnableExpiryOutput(),
-                           response.getStreamId()),
-          emd(nullptr) {
-    }
+    /**
+     * Used in test code for wiring a producer/consumer directly.
+     * @param response MutationResponse (from Producer) to create message from.
+     *        Note this is non-const as the item needs to be modified for some
+     *        Event types.
+     */
+    MutationConsumerMessage(MutationResponse& response);
 
     /**
      * @return size of message on the wire

engines/ep/tests/mock/mock_dcp_producer.h

@@ -84,6 +84,9 @@ class MockDcpProducer : public DcpProducer {
         return enableExpiryOpcode;
     }
 
+    void setSyncReplication(bool value) {
+        supportsSyncReplication = value;
+    }
     /**
      * Create the ActiveStreamCheckpointProcessorTask and assign to
      * checkpointCreator->task