MB-34951: Ignore StreamEnd if opaque doesn't match

The DCP option 'send_stream_end_on_client_close_stream' was added via
MB-26363 to have the Producer send a STREAM_END message even when the
Consumer initiated the close. This was requested by an external DCP
client (Go SDK), however the ep-engine DcpConsumer _also_ requested
this feature.

However this can cause (benign) warnings during rebalance when the
initial (backfill) Stream is closed and the takeover stream
created. The problem is the following sequence of messages (all for
the same vbid):

    DcpConsumer            ns_server proxy                DcpProducer

    <--- 1. CloseStream Req ------
    ---- 2. CloseStream Resp ---->
                                       ---- 3. CloseStream Req --->
                                       <--- 4. CloseStream Resp ---

    <--- 5. AddStream(Takeover) --
    <--- 6. StreamEnd Request -------------------------------------
    ---- 7. StreamEnd Response [KEY_ENOENT] ---------------------->

    ---- 8. StreamRequest ---------------------------------------->

ns_server is not aware of `send_stream_end_on_client_close_stream`; so
it doesn't wait for `6. StreamEnd Request` from the Producer before
sending the AddStream request to the Consumer. As such, the Consumer
has already removed the entry in it's streams map for the given
vbucket, so when it receives the StreamEnd request from the Producer
it returns KEY_ENOENT error as the (old) opaque doesn't match the
(new) opaque for the takeover stream.

send_stream_end_on_client_close_stream is useful for ep-engine to have
an explicit window on when potential mismatched opaques could appear
(after the SteamEnd request no more should be sent from the Producer)
- and indeed it's presence is required to address MB-34850. As such we
cannot simply disable send_stream_end_on_client_close_stream.

The alternative is to simply ignore the StreamEnd if the given opaque
no longer matches (i.e. because another stream has already been
started).

Change-Id: Ib6baf414aa5b4c306f9146818db2d25ac4326f4f
Reviewed-on: http://review.couchbase.org/111746
Reviewed-by: Ben Huddleston <[email protected]>
Tested-by: Build Bot <[email protected]>

Author: daverigby

engines/ep/src/dcp/consumer.cc

@@ -370,18 +370,26 @@ ENGINE_ERROR_CODE DcpConsumer::streamEnd(uint32_t opaque,
     auto stream = findStream(vbucket);
     if (!stream) {
         logger->warn(
-                "({}) End stream received but no such stream for this "
+                "({}) End stream received with opaque:{} but no such stream "
+                "for this "
                 "vBucket",
-                vbucket);
+                vbucket,
+                opaque);
         return ENGINE_KEY_ENOENT;
     }
 
     if (stream->getOpaque() != opaque) {
-        logger->warn("({}) End stream received with opaque {} but expected {}",
-                     vbucket,
-                     opaque,
-                     stream->getOpaque());
-        return ENGINE_KEY_ENOENT;
+        // MB-34951: By the time the DcpConsumer receives the StreamEnd from
+        // the DcpProducer it is possible that ns_server has already started
+        // a new Stream (with updated opaque) for this vbucket.
+        // In which case just ignore this StreamEnd message, returning SUCCESS.
+        logger->info(
+                "({}) End stream received with opaque {} but current opaque "
+                "for that vb is {} - ignoring",
+                vbucket,
+                opaque,
+                stream->getOpaque());
+        return ENGINE_SUCCESS;
     }
 
     logger->info("({}) End stream received with reason {}", vbucket, flags);

engines/ep/tests/module_tests/evp_store_single_threaded_test.cc

@@ -3924,6 +3924,26 @@ TEST_F(SingleThreadedEPBucketTest,
     testAllStreamLevelMessages(ENGINE_KEY_ENOENT);
 }
 
+// MB-34951: Check that a consumer correctly handles (and ignores) a StreamEnd
+// request from the producer if it has already created a new stream (for the
+// same vb) with a different opaque.
+TEST_F(SingleThreadedEPBucketTest,
+       MB_34951_ConsumerRecvStreamEndAfterAddStream) {
+    // Setup: Create replica VB and create stream for vbid, then close it
+    // and add another stream (same vbid).
+    setVBucketStateAndRunPersistTask(vbid, vbucket_state_replica);
+    auto consumer = std::make_shared<MockDcpConsumer>(*engine, cookie, "conn");
+    const int opaque1 = 1;
+    ASSERT_EQ(ENGINE_SUCCESS, consumer->addStream(opaque1, vbid, {}));
+    ASSERT_EQ(ENGINE_SUCCESS, consumer->closeStream(opaque1, vbid));
+    const int opaque2 = 2;
+    ASSERT_EQ(ENGINE_SUCCESS, consumer->addStream(opaque2, vbid, {}));
+
+    // Test: Have the producer send a StreamEnd with the "old" opaque.
+    EXPECT_EQ(ENGINE_SUCCESS,
+              consumer->streamEnd(opaque1, vbid, END_STREAM_CLOSED));
+}
+
 TEST_P(STParameterizedBucketTest, produce_delete_times) {
     setVBucketStateAndRunPersistTask(vbid, vbucket_state_active);
     auto t1 = ep_real_time();