MB-34850: Consumer: Accept (but ignore) messages after Close Stream

As identified during Sync Replication testing, it is possible for the
ns_server DCP proxy to forward DCP messages from the Producer to the
Consumer even after the DCP proxy has sent Close Stream to the
Consumer.

Currently these messages are treated as erroneous by the DCP Consumer,
causing it to send KEY_ENOENT responses back to the DCP Producer
(which hasn't started the Stream close process), and hence the DCP
Producer treats this as an error and tears down the connection,
ultimately resulting in rebalance failing.

Ideally these 'extra' messages would never be forwarded by ns_server,
however (a) the DCP Documentation[1] _does_ permit this behaviour, and
(b) it's non-trivial for ns_server to fix this issue in the DCP proxy
as two different processes handle the producer and consumer proxying,
and do not share state.

Instead this is addressed in ep-engine - any stream-level messages
received between the Close Stream request and the later Stream End
request should be silently ignored.

[1]: https://github.com/couchbase/kv_engine/blob/master/docs/dcp/documentation/commands/close-stream.md

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

Author: daverigby

engines/ep/src/dcp/consumer.cc

@@ -350,7 +350,9 @@ ENGINE_ERROR_CODE DcpConsumer::closeStream(uint32_t opaque,
 
     uint32_t bytesCleared = stream->setDead(END_STREAM_CLOSED);
     flowControl.incrFreedBytes(bytesCleared);
-    removeStream(vbucket);
+    // Note the stream is not yet removed from the `streams` map; as we need to
+    // handle (but ignore) any in-flight messages from the Producer until
+    // STREAM_END is received.
     scheduleNotifyIfNecessary();
 
     return ENGINE_SUCCESS;
@@ -374,12 +376,6 @@ ENGINE_ERROR_CODE DcpConsumer::streamEnd(uint32_t opaque,
         return ENGINE_KEY_ENOENT;
     }
 
-    if (!stream->isActive()) {
-        logger->warn("({}) End stream received but stream is not active",
-                     vbucket);
-        return ENGINE_KEY_ENOENT;
-    }
-
     if (stream->getOpaque() != opaque) {
         logger->warn("({}) End stream received with opaque {} but expected {}",
                      vbucket,
@@ -395,7 +391,16 @@ ENGINE_ERROR_CODE DcpConsumer::streamEnd(uint32_t opaque,
             static_cast<end_stream_status_t>(flags),
             vbucket,
             cb::mcbp::DcpStreamId{});
-    return lookupStreamAndDispatchMessage(ufc, vbucket, opaque, std::move(msg));
+    auto res = lookupStreamAndDispatchMessage(
+            ufc, vbucket, opaque, std::move(msg));
+
+    if (res == ENGINE_SUCCESS) {
+        // Stream End message successfully passed to stream. Can now remove
+        // the stream from the streams map as it has completed its lifetime.
+        removeStream(vbucket);
+    }
+
+    return res;
 }
 
 ENGINE_ERROR_CODE DcpConsumer::processMutationOrPrepare(
@@ -540,64 +545,66 @@ ENGINE_ERROR_CODE DcpConsumer::deletion(uint32_t opaque,
         return ENGINE_EINVAL;
     }
 
-    ENGINE_ERROR_CODE err = ENGINE_KEY_ENOENT;
     auto stream = findStream(vbucket);
-    if (stream && stream->getOpaque() == opaque && stream->isActive()) {
-        queued_item item(Item::makeDeletedItem(deletionCause,
-                                               key,
-                                               0,
-                                               deleteTime,
-                                               value.data(),
-                                               value.size(),
-                                               datatype,
-                                               cas,
-                                               bySeqno,
-                                               vbucket,
-                                               revSeqno));
-
-        // MB-29040: Producer may send deleted doc with value that still has
-        // the user xattrs and the body. Fix up that mistake by running the
-        // expiry hook which will correctly process the document
-        if (value.size()) {
-            if (mcbp::datatype::is_xattr(datatype)) {
-                auto vb = engine_.getVBucket(vbucket);
-                if (vb) {
-                    engine_.getKVBucket()->runPreExpiryHook(*vb, *item);
-                }
-            } else {
-                // MB-31141: Deletes cannot have a value
-                item->replaceValue(Blob::New(0));
-                item->setDataType(PROTOCOL_BINARY_RAW_BYTES);
-            }
-        }
+    if (!stream || (stream->getOpaque() != opaque)) {
+        // No stream for this vBucket / opaque - return ENOENT to indicate this.
+        return ENGINE_KEY_ENOENT;
+    }
 
-        std::unique_ptr<ExtendedMetaData> emd;
-        if (meta.size() > 0) {
-            emd = std::make_unique<ExtendedMetaData>(meta.data(), meta.size());
-            if (emd->getStatus() == ENGINE_EINVAL) {
-                err = ENGINE_EINVAL;
+    queued_item item(Item::makeDeletedItem(deletionCause,
+                                           key,
+                                           0,
+                                           deleteTime,
+                                           value.data(),
+                                           value.size(),
+                                           datatype,
+                                           cas,
+                                           bySeqno,
+                                           vbucket,
+                                           revSeqno));
+
+    // MB-29040: Producer may send deleted doc with value that still has
+    // the user xattrs and the body. Fix up that mistake by running the
+    // expiry hook which will correctly process the document
+    if (value.size()) {
+        if (mcbp::datatype::is_xattr(datatype)) {
+            auto vb = engine_.getVBucket(vbucket);
+            if (vb) {
+                engine_.getKVBucket()->runPreExpiryHook(*vb, *item);
             }
+        } else {
+            // MB-31141: Deletes cannot have a value
+            item->replaceValue(Blob::New(0));
+            item->setDataType(PROTOCOL_BINARY_RAW_BYTES);
         }
+    }
 
-        try {
-            err = stream->messageReceived(
-                    std::make_unique<MutationConsumerMessage>(
-                            item,
-                            opaque,
-                            IncludeValue::Yes,
-                            IncludeXattrs::Yes,
-                            includeDeleteTime,
-                            key.getEncoding(),
-                            emd.release(),
-                            cb::mcbp::DcpStreamId{}));
-        } catch (const std::bad_alloc&) {
-            err = ENGINE_ENOMEM;
+    ENGINE_ERROR_CODE err;
+    std::unique_ptr<ExtendedMetaData> emd;
+    if (meta.size() > 0) {
+        emd = std::make_unique<ExtendedMetaData>(meta.data(), meta.size());
+        if (emd->getStatus() == ENGINE_EINVAL) {
+            err = ENGINE_EINVAL;
         }
+    }
 
-        // The item was buffered and will be processed later
-        if (err == ENGINE_TMPFAIL) {
-            notifyVbucketReady(vbucket);
-        }
+    try {
+        err = stream->messageReceived(std::make_unique<MutationConsumerMessage>(
+                item,
+                opaque,
+                IncludeValue::Yes,
+                IncludeXattrs::Yes,
+                includeDeleteTime,
+                key.getEncoding(),
+                emd.release(),
+                cb::mcbp::DcpStreamId{}));
+    } catch (const std::bad_alloc&) {
+        err = ENGINE_ENOMEM;
+    }
+
+    // The item was buffered and will be processed later
+    if (err == ENGINE_TMPFAIL) {
+        notifyVbucketReady(vbucket);
     }
 
     return err;
@@ -1648,12 +1655,6 @@ ENGINE_ERROR_CODE DcpConsumer::lookupStreamAndDispatchMessage(
         return ENGINE_KEY_ENOENT;
     }
 
-    if (!stream->isActive()) {
-        // Stream is not active - also uses KEY_ENOENT to indicate no valid
-        // stream.
-        return ENGINE_KEY_ENOENT;
-    }
-
     // Pass the message to the associated stream.
     ENGINE_ERROR_CODE err;
     try {

engines/ep/src/dcp/passive_stream.cc

@@ -225,7 +225,12 @@ ENGINE_ERROR_CODE PassiveStream::messageReceived(
     }
 
     if (!isActive()) {
-        return ENGINE_KEY_ENOENT;
+        // If the Stream isn't active, *but* the object is still receiving
+        // messages from the DcpConsumer that means the stream is still
+        // registered in the streams map and hence we should ignore any
+        // messages (until STREAM_END is received and the stream is removed form
+        // the map).
+        return ENGINE_SUCCESS;
     }
 
     auto seqno = dcpResponse->getBySeqno();

engines/ep/tests/ep_testsuite_dcp.cc

@@ -4882,9 +4882,6 @@ static enum test_result test_dcp_consumer_end_stream(EngineIface* h) {
             dcp->stream_end(cookie, stream_opaque, vbucket, end_flag),
             "Expected success");
 
-    wait_for_stat_to_be(
-            h, "eq_dcpq:unittest:stream_0_state", std::string{"dead"}, "dcp");
-
     testHarness->destroy_cookie(cookie);
     return SUCCESS;
 }

engines/ep/tests/module_tests/evp_store_single_threaded_test.cc

@@ -3821,6 +3821,109 @@ TEST_F(SingleThreadedEPBucketTest, testValidTombstonePurgeOnRetainErroneousTombs
     EXPECT_EQ(2, store->getVBucket(vbid)->getPurgeSeqno());
 }
 
+// MB-34850: Check that a consumer correctly handles (and ignores) stream-level
+// messages (Mutation/Deletion/Prepare/Commit/Abort/...) received after
+// CloseStream response but *before* the Producer sends STREAM_END.
+TEST_F(SingleThreadedEPBucketTest,
+       MB_34850_ConsumerRecvMessagesAfterCloseStream) {
+    // Setup: Create replica VB and create stream for vbid.
+    // Have the consumer receive a snapshot marker(1..10), and then close the
+    // stream .
+    setVBucketStateAndRunPersistTask(vbid, vbucket_state_replica);
+    auto consumer = std::make_shared<MockDcpConsumer>(*engine, cookie, "conn");
+    int opaque = 1;
+    ASSERT_EQ(ENGINE_SUCCESS, consumer->addStream(opaque, vbid, /*flags*/ 0));
+    ASSERT_EQ(ENGINE_SUCCESS,
+              consumer->snapshotMarker(opaque, vbid, 1, 10, MARKER_FLAG_CHK));
+    ASSERT_EQ(ENGINE_SUCCESS, consumer->closeStream(opaque, vbid));
+
+    // Test: Have the producer send further messages on the stream (before the
+    // final STREAM_END. These should all be accepted (but discarded) by the
+    // replica.
+    auto testAllStreamLevelMessages = [&consumer, this, opaque](
+                                              ENGINE_ERROR_CODE expected) {
+        auto key = makeStoredDocKey("key");
+        auto dtype = PROTOCOL_BINARY_RAW_BYTES;
+        EXPECT_EQ(expected,
+                  consumer->mutation(opaque,
+                                     key,
+                                     {},
+                                     0,
+                                     dtype,
+                                     {},
+                                     vbid,
+                                     {},
+                                     1,
+                                     {},
+                                     {},
+                                     {},
+                                     {},
+                                     {}));
+
+        EXPECT_EQ(expected,
+                  consumer->deletion(
+                          opaque, key, {}, 0, dtype, {}, vbid, 2, {}, {}));
+
+        EXPECT_EQ(expected,
+                  consumer->deletionV2(
+                          opaque, key, {}, 0, dtype, {}, vbid, 3, {}, {}));
+
+        EXPECT_EQ(expected,
+                  consumer->expiration(
+                          opaque, key, {}, 0, dtype, {}, vbid, 4, {}, {}));
+
+        EXPECT_EQ(
+                expected,
+                consumer->setVBucketState(opaque, vbid, vbucket_state_active));
+        auto vb = engine->getKVBucket()->getVBucket(vbid);
+        EXPECT_EQ(vbucket_state_replica, vb->getState());
+
+        EXPECT_EQ(expected,
+                  consumer->systemEvent(opaque,
+                                        vbid,
+                                        mcbp::systemevent::id::CreateCollection,
+                                        5,
+                                        mcbp::systemevent::version::version1,
+                                        {},
+                                        {}));
+
+        EXPECT_EQ(expected,
+                  consumer->prepare(opaque,
+                                    key,
+                                    {},
+                                    0,
+                                    dtype,
+                                    {},
+                                    vbid,
+                                    {},
+                                    6,
+                                    {},
+                                    {},
+                                    {},
+                                    {},
+                                    {},
+                                    {}));
+
+        EXPECT_EQ(expected, consumer->commit(opaque, vbid, key, 6, 7));
+
+        EXPECT_EQ(expected, consumer->abort(opaque, vbid, key, 6, 7));
+
+        EXPECT_EQ(expected,
+                  consumer->snapshotMarker(
+                          opaque, vbid, 11, 11, MARKER_FLAG_CHK));
+    };
+    testAllStreamLevelMessages(ENGINE_SUCCESS);
+
+    // Setup (phase 2): Receive a STREAM_END message - after which all of the
+    // above stream-level messages should be rejected as ENOENT.
+    ASSERT_EQ(ENGINE_SUCCESS,
+              consumer->streamEnd(opaque, vbid, END_STREAM_CLOSED));
+
+    // Test (phase 2): Have the producer send all the above stream-level
+    // messages to the consumer. Should all be rejected this time.
+    testAllStreamLevelMessages(ENGINE_KEY_ENOENT);
+}
+
 TEST_P(STParameterizedBucketTest, produce_delete_times) {
     setVBucketStateAndRunPersistTask(vbid, vbucket_state_active);
     auto t1 = ep_real_time();