MB-35429: Move preparedSeqno check to ActiveStream::seqnoAck

Currently we check if we should process a seqno ack based on
whether or not the preparedSeqno is less than our ActiveStream's
lastSentSeqno. This should only be done if the stream is alive as
we should never attempt to process seqno acks for any dead Stream.

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

Author: BenHuddleston

engines/ep/src/dcp/active_stream.cc

@@ -1792,6 +1792,14 @@ ENGINE_ERROR_CODE ActiveStream::seqnoAck(const std::string& consumerName,
                 return ENGINE_SUCCESS;
             }
 
+            if (preparedSeqno > getLastSentSeqno()) {
+                throw std::logic_error(
+                        vb_.to_string() + " replica \"" + consumerName +
+                        "\" acked seqno:" + std::to_string(preparedSeqno) +
+                        " which is greater than last sent seqno:" +
+                        std::to_string(getLastSentSeqno()));
+            }
+
             return vb->seqnoAcknowledged(
                     vbStateLh, consumerName, preparedSeqno);
         } // end stream mutex lock scope

engines/ep/src/dcp/producer.cc

@@ -1067,14 +1067,6 @@ ENGINE_ERROR_CODE DcpProducer::seqno_acknowledged(uint32_t opaque,
     }
 
     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(as->getLastSentSeqno()));
-    }
-
     return as->seqnoAck(consumerName, prepared_seqno);
 }
 

engines/ep/tests/module_tests/dcp_durability_stream_test.cc

@@ -453,23 +453,11 @@ TEST_P(DurabilityActiveStreamTest, BackfillAbort) {
 }
 
 TEST_P(DurabilityActiveStreamTest, RemoveUnknownSeqnoAckAtDestruction) {
-    auto vb = engine->getVBucket(vbid);
-
-    const auto key = makeStoredDocKey("key");
-    const auto& value = "value";
-    auto item = makePendingItem(
-            key,
-            value,
-            cb::durability::Requirements(cb::durability::Level::Majority,
-                                         1 /*timeout*/));
-    VBQueueItemCtx ctx;
-    ctx.durability =
-            DurabilityItemCtx{item->getDurabilityReqs(), nullptr /*cookie*/};
-
-    EXPECT_EQ(MutationStatus::WasClean, public_processSet(*vb, *item, ctx));
+    testSendDcpPrepare();
     flushVBucketToDiskIfPersistent(vbid, 1);
 
     // We don't include prepares in the numItems stat (should not exist in here)
+    auto vb = engine->getVBucket(vbid);
     EXPECT_EQ(0, vb->getNumItems());
 
     // Our topology gives replica name as "replica" an our producer/stream has
@@ -503,23 +491,11 @@ TEST_P(DurabilityActiveStreamTest, RemoveUnknownSeqnoAckAtDestruction) {
 }
 
 TEST_P(DurabilityActiveStreamTest, RemoveCorrectQueuedAckAtStreamSetDead) {
-    auto vb = engine->getVBucket(vbid);
-
-    const auto key = makeStoredDocKey("key");
-    const auto& value = "value";
-    auto item = makePendingItem(
-            key,
-            value,
-            cb::durability::Requirements(cb::durability::Level::Majority,
-                                         1 /*timeout*/));
-    VBQueueItemCtx ctx;
-    ctx.durability =
-            DurabilityItemCtx{item->getDurabilityReqs(), nullptr /*cookie*/};
-
-    EXPECT_EQ(MutationStatus::WasClean, public_processSet(*vb, *item, ctx));
+    testSendDcpPrepare();
     flushVBucketToDiskIfPersistent(vbid, 1);
 
     // We don't include prepares in the numItems stat (should not exist in here)
+    auto vb = engine->getVBucket(vbid);
     EXPECT_EQ(0, vb->getNumItems());
 
     // Our topology gives replica name as "replica" an our producer/stream has
@@ -544,6 +520,17 @@ TEST_P(DurabilityActiveStreamTest, RemoveCorrectQueuedAckAtStreamSetDead) {
     producer->scheduleCheckpointProcessorTask();
     stream->setActive();
 
+    // Process items to ensure that lastSentSeqno is GE the seqno that we will
+    // ack
+    stream->transitionStateToBackfilling();
+    ASSERT_TRUE(stream->isBackfilling());
+
+    auto& bfm = producer->getBFM();
+    bfm.backfill();
+    bfm.backfill();
+    EXPECT_EQ(2, stream->public_readyQSize());
+    stream->consumeBackfillItems(2);
+
     // Should not throw a monotonic exception as the ack should have been
     // removed by setDead.
     stream->seqnoAck(producer->getConsumerName(), 1);

engines/ep/tests/module_tests/dcp_stream_sync_repl_test.cc

@@ -770,7 +770,8 @@ TEST_P(DcpStreamSyncReplPersistentTest, ProducerAllowsSeqnoAckLEQToLastSent) {
     try {
         producer->seqno_acknowledged(0, vbid, 4);
     } catch (const std::logic_error& e) {
-        EXPECT_TRUE(std::string(e.what()).find("Replica acked seqno") !=
+        EXPECT_TRUE(std::string(e.what()).find(
+                            "replica \"replica1\" acked seqno") !=
                     std::string::npos);
         thrown = true;
     }