MB-35361: Send seqno ack on stream (re)connection

If we have a vBucket with non-zero HPS accepting a passive stream
then we should send a SeqnoAcknowledgement to the active node to
ensure that we can commit any in-flight SyncWrites.

Consider the case where we have 1 replica and a durable write:

1) Replica receives prepare from active
2) Replica disconnects from active before acking
3) Replica attempts to ack but cannot due to lack of stream
4) Replica reconnects
5) Replica sends seqno ack

If we did not do step 5 then we may never complete the given prepare.
In the general case we would just timeout, but if this prepare had
been warmed up or grandfathered into an ADM then it would have
an infinite timeout and we would forever block this key if the
replica never received a disk snapshot or any other durable writes.

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

Author: BenHuddleston

engines/ep/src/dcp/passive_stream.cc

@@ -161,13 +161,28 @@ std::string PassiveStream::getStreamTypeName() const {
 void PassiveStream::acceptStream(cb::mcbp::Status status, uint32_t add_opaque) {
     std::unique_lock<std::mutex> lh(streamMutex);
     if (isPending()) {
+        pushToReadyQ(std::make_unique<AddStreamResponse>(
+                add_opaque, opaque_, status));
         if (status == cb::mcbp::Status::Success) {
+            // Before we receive/process anything else, send a seqno ack if we
+            // are a stream for a pre-existing vBucket to ensure that the
+            // replica can commit any in-flight SyncWrites if no further
+            // SyncWrites are done and no disk snapshots processed by this
+            // replica.
+            {
+                auto consumer = consumerPtr.lock();
+                if (consumer && consumer->isSyncReplicationEnabled()) {
+                    auto vb = engine->getVBucket(vb_);
+                    if (vb && vb->getHighPreparedSeqno() != 0) {
+                        pushToReadyQ(std::make_unique<SeqnoAcknowledgement>(
+                                opaque_, vb_, vb->getHighPreparedSeqno()));
+                    }
+                }
+            }
             transitionState(StreamState::AwaitingFirstSnapshotMarker);
         } else {
             transitionState(StreamState::Dead);
         }
-        pushToReadyQ(std::make_unique<AddStreamResponse>(
-                add_opaque, opaque_, status));
         lh.unlock();
         notifyStreamReady();
     }

engines/ep/tests/mock/mock_dcp_consumer.h

@@ -133,6 +133,13 @@ class MockDcpConsumer: public DcpConsumer {
         supportsSyncReplication = true;
     }
 
+    /**
+     * Disable SyncRepl for testing
+     */
+    void disableSyncReplication() {
+        supportsSyncReplication = false;
+    }
+
     /**
      *
      * Map from the opaque used to create a stream to the internal opaque

engines/ep/tests/module_tests/dcp_durability_stream_test.cc

@@ -716,6 +716,53 @@ void DurabilityPassiveStreamTest::TearDown() {
     SingleThreadedPassiveStreamTest::TearDown();
 }
 
+TEST_P(DurabilityPassiveStreamTest, SendSeqnoAckOnStreamAcceptance) {
+    // 1) Put something in the vBucket as we won't send a seqno ack if there are
+    // no items
+    testReceiveDcpPrepare();
+
+    consumer->closeAllStreams();
+    uint32_t opaque = 0;
+    consumer->addStream(opaque, vbid, 0 /*flags*/);
+    stream = static_cast<MockPassiveStream*>(
+            (consumer->getVbucketStream(vbid)).get());
+    stream->acceptStream(cb::mcbp::Status::Success, opaque);
+
+    EXPECT_EQ(3, stream->public_readyQ().size());
+    auto resp = stream->public_popFromReadyQ();
+    EXPECT_EQ(DcpResponse::Event::StreamReq, resp->getEvent());
+    resp = stream->public_popFromReadyQ();
+    EXPECT_EQ(DcpResponse::Event::AddStream, resp->getEvent());
+    resp = stream->public_popFromReadyQ();
+    EXPECT_EQ(DcpResponse::Event::SeqnoAcknowledgement, resp->getEvent());
+    const auto& ack = static_cast<SeqnoAcknowledgement&>(*resp);
+    EXPECT_EQ(1, ack.getPreparedSeqno());
+}
+
+TEST_P(DurabilityPassiveStreamTest,
+       NoSeqnoAckOnStreamAcceptanceIfNotSupported) {
+    consumer->disableSyncReplication();
+
+    // 1) Put something in the vBucket as we won't send a seqno ack if there are
+    // no items
+    testReceiveDcpPrepare();
+
+    consumer->closeAllStreams();
+    uint32_t opaque = 0;
+    consumer->addStream(opaque, vbid, 0 /*flags*/);
+    stream = static_cast<MockPassiveStream*>(
+            (consumer->getVbucketStream(vbid)).get());
+    stream->acceptStream(cb::mcbp::Status::Success, opaque);
+
+    ASSERT_EQ(2, stream->public_readyQ().size());
+    auto resp = stream->public_popFromReadyQ();
+    EXPECT_EQ(DcpResponse::Event::StreamReq, resp->getEvent());
+    resp = stream->public_popFromReadyQ();
+    EXPECT_EQ(DcpResponse::Event::AddStream, resp->getEvent());
+    resp = stream->public_popFromReadyQ();
+    EXPECT_FALSE(resp);
+}
+
 void DurabilityPassiveStreamTest::
         testReceiveMutationOrDeletionInsteadOfCommitWhenStreamingFromDisk(
                 DocumentState docState) {