MB-35378: Ignore seqno ack on producer without stream

A stream on a producer may be removed from the streams map if
we have processed a close stream message and then stepped the
producer to send a StreamEnd. The consumer may not yet have
received and processed this StreamEnd but may have sent us a
seqno ack. If this is the case, we should simply ignore the
seqno ack.

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

Author: BenHuddleston

engines/ep/src/dcp/producer.cc

@@ -1044,25 +1044,38 @@ ENGINE_ERROR_CODE DcpProducer::seqno_acknowledged(uint32_t opaque,
                 " but we don't have a StreamContainer for that vb");
     }
 
-    std::shared_ptr<ActiveStream> stream;
-    {
-        auto handle = rv->second->rlock();
-
-        // Producer for replication should only have one stream.
-        Expects(handle.size() == 1);
-
-        stream = dynamic_pointer_cast<ActiveStream>(handle.get());
-        Expects(stream.get());
+    // Search for an active stream with the same opaque as the response.
+    // Use find_if2 which will return the matching shared_ptr<Stream>
+    auto stream = find_if2([opaque](const StreamsMap::value_type& s) {
+        auto handle = s.second->rlock();
+        for (; !handle.end(); handle.next()) {
+            auto& stream = handle.get();
+            auto* as = dynamic_cast<ActiveStream*>(stream.get());
+            if (as && opaque == stream->getOpaque()) {
+                return stream; // return matching shared_ptr<Stream>
+            }
+        }
+        return ContainerElement{};
+    });
+
+    if (!stream) {
+        // No stream found, may be the case that we have just ended our
+        // stream and removed the stream from our map but the consumer is
+        // not yet aware and we have received a seqno ack. Just return
+        // success and ignore the ack.
+        return ENGINE_SUCCESS;
     }
 
-    if (prepared_seqno > stream->getLastSentSeqno()) {
+    ActiveStream* as = static_cast<ActiveStream*>(stream.get());
+
+    if (prepared_seqno > as->getLastSentSeqno()) {
         throw std::logic_error(
                 "Replica acked seqno:" + std::to_string(prepared_seqno) +
                 " greater than last sent seqno:" +
-                std::to_string(stream->getLastSentSeqno()));
+                std::to_string(as->getLastSentSeqno()));
     }
 
-    return stream->seqnoAck(consumerName, prepared_seqno);
+    return as->seqnoAck(consumerName, prepared_seqno);
 }
 
 bool DcpProducer::handleResponse(const protocol_binary_response_header* resp) {

engines/ep/tests/mock/mock_dcp_producer.cc

@@ -91,6 +91,11 @@ std::shared_ptr<Stream> MockDcpProducer::findStream(Vbid vbid) {
     auto rv = streams.find(vbid.get());
     if (rv != streams.end()) {
         auto handle = rv->second->rlock();
+        // An empty StreamContainer for this vbid is allowed
+        if (handle.size() == 0) {
+            return nullptr;
+        }
+
         if (handle.size() != 1) {
             throw std::logic_error(
                     "MockDcpProducer::findStream against producer with many "

engines/ep/tests/module_tests/dcp_stream_test.cc

@@ -1203,6 +1203,39 @@ TEST_P(StreamTest, ProcessItemsCheckpointStartIsLastItem) {
     EXPECT_EQ(DcpResponse::Event::Mutation, readyQ.front()->getEvent());
 }
 
+TEST_P(StreamTest, ProducerReceivesSeqnoAckForErasedStream) {
+    create_dcp_producer(0, /*flags*/
+                        IncludeValue::Yes,
+                        IncludeXattrs::Yes,
+                        {{"send_stream_end_on_client_close_stream", "true"},
+                         {"enable_synchronous_replication", "true"},
+                         {"consumer_name", "replica1"}});
+
+    // Need to do a stream request to put the stream in the producers map
+    ASSERT_EQ(ENGINE_SUCCESS, doStreamRequest(*producer).status);
+
+    // Close the stream to start the removal process
+    EXPECT_EQ(ENGINE_SUCCESS, producer->closeStream(0 /*opaque*/, vbid));
+
+    // Stream should still exist, but should be dead
+    auto stream = producer->findStream(vbid);
+    EXPECT_TRUE(stream);
+    EXPECT_FALSE(stream->isActive());
+
+    // Step the stream on, this should remove the stream from the producer's
+    // StreamsMap
+    MockDcpMessageProducers producers(engine);
+    EXPECT_EQ(ENGINE_SUCCESS, producer->step(&producers));
+    EXPECT_EQ(cb::mcbp::ClientOpcode::DcpStreamEnd, producers.last_op);
+
+    // Stream should no longer exist in the map
+    EXPECT_FALSE(producer->findStream(vbid));
+
+    EXPECT_EQ(ENGINE_SUCCESS,
+              producer->seqno_acknowledged(
+                      0 /*opaque*/, vbid, 1 /*prepareSeqno*/));
+}
+
 class CacheCallbackTest : public StreamTest {
 protected:
     void SetUp() override {