MB-31481: Don't skip STREAM_END msg after cursor drop during backfill

For DCP producers needing backfill, There's an edge-case in handling
cursor dropping:

- When a backfill is re-scheduled due to a slow stream trigger (cursor
  dropping), if the second backfill isn't necessary (as the required
  items are still in the checkpoing manager) then the second backfill
  is skipped and the stream transition to in-memory.  - If the
  in-memory stream then doesn't need to fetch anymore data (as no more
  mutations have occurred), then the stream advances directly to
  endStream - see ActiveStream::transitionState():

      case StreamState::InMemory:
          // Check if the producer has sent up till the last requested
          // sequence number already, if not - move checkpoint items into
          // the ready queue.
          if (lastSentSeqno.load() >= end_seqno_) {
              // Stream transitioning to DEAD state
              endStream(END_STREAM_OK);         /// <--- HERE
              notifyStreamReady();
          }

However, there is a bug where this END_STREAM message is never
transmitted to the DCP client, resulting in the stream hanging.

The problem is that stream end message gets stuck in the readyQ, as
the stream is never notified as ready to send it. This is due to the
expectation of Stream::next() dealing with queued items being broken -
if next() retuns null then the itemsReady flag is cleared as it
assumes a null retudn means no more items are available.  However in
this situation there /are/ items ready, but not until
transitionState() is called.

Following from this, the stream is never notified so doesnt
send a closure message, which means that DCP consumers can hang
waiting for this closure message.

This patch triggers nextQueuedItem inside the pendingBackfill case to
ensure that the end stream message is notified and executed.  The
patch also includes a test to follow this problematic sequence.

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

Author: CSubmerged

engines/ep/src/dcp/stream.cc

@@ -676,6 +676,11 @@ std::unique_ptr<DcpResponse> ActiveStream::backfillPhase(
         if (pendingBackfill) {
             scheduleBackfill_UNLOCKED(true);
             pendingBackfill = false;
+            // After scheduling a backfill we may now have items in readyQ -
+            // so re-check if we didn't already have a response.
+            if (!resp) {
+                resp = nextQueuedItem();
+            }
         } else {
             if (lastReadSeqno.load() >= end_seqno_) {
                 endStream(END_STREAM_OK);

engines/ep/tests/mock/mock_dcp_producer.h

@@ -146,6 +146,10 @@ class MockDcpProducer : public DcpProducer {
         return log.getBytesOutstanding();
     }
 
+    DcpReadyQueue& getReadyQueue() {
+        return ready;
+    }
+
     /**
      * Place a mock active stream into the producer
      */

engines/ep/tests/module_tests/evp_store_single_threaded_test.cc

@@ -3479,6 +3479,113 @@ TEST_F(SingleThreadedEPBucketTest, MB_29541) {
     producer->cancelCheckpointCreatorTask();
 }
 
+/* When a backfill is activated along with a slow stream trigger,
+ * the stream end message gets stuck in the readyQ as the stream is
+ * never notified as ready to send it. As the stream transitions state
+ * to InMemory as well as having sent all requested sequence numbers,
+ * the stream is meant to end but Stream::itemsReady can cause this
+ * to never trigger. This means that DCP consumers can hang waiting
+ * for this closure message.
+ * This test checks that the DCP stream actually sends the end stream
+ * message when triggering this problematic sequence.
+ */
+TEST_F(SingleThreadedEPBucketTest, MB_31481) {
+    setVBucketStateAndRunPersistTask(vbid, vbucket_state_active);
+
+    // 1) First store 2 keys which we will backfill
+    std::array<std::string, 2> keys = {{"k1", "k2"}};
+    store_item(vbid, makeStoredDocKey(keys[0]), keys[0]);
+    store_item(vbid, makeStoredDocKey(keys[1]), keys[1]);
+
+    flush_vbucket_to_disk(vbid, keys.size());
+
+    // Simplest way to ensure DCP has to do a backfill - 'wipe memory'
+    resetEngineAndWarmup();
+
+    // Setup DCP, 1 producer and we will do a takeover of the vbucket
+    auto producer = createDcpProducer(cookie, {}, false, IncludeDeleteTime::No);
+
+    auto producers = get_dcp_producers(
+            reinterpret_cast<ENGINE_HANDLE*>(engine.get()),
+            reinterpret_cast<ENGINE_HANDLE_V1*>(engine.get()));
+
+    ASSERT_TRUE(producer->getReadyQueue().empty());
+
+    uint64_t rollbackSeqno = 0;
+    auto vb = store->getVBuckets().getBucket(vbid);
+    ASSERT_NE(nullptr, vb.get());
+    EXPECT_EQ(ENGINE_SUCCESS,
+              producer->streamRequest(0, // flags
+                                      1, // opaque
+                                      vbid,
+                                      0, // start_seqno
+                                      vb->getHighSeqno(), // end_seqno
+                                      vb->failovers->getLatestUUID(),
+                                      0, // snap_start_seqno
+                                      vb->getHighSeqno(), // snap_end_seqno
+                                      &rollbackSeqno,
+                                      &dcpAddFailoverLog));
+
+    auto vb0Stream =
+            dynamic_cast<ActiveStream*>(producer->findStream(vbid).get());
+    ASSERT_NE(nullptr, vb0Stream);
+
+    // Manually drive the backfill (not using notifyAndStepToCheckpoint)
+    auto& lpAuxioQ = *task_executor->getLpTaskQ()[AUXIO_TASK_IDX];
+    // Trigger slow stream handle
+    ASSERT_TRUE(vb0Stream->handleSlowStream());
+    // backfill:create()
+    runNextTask(lpAuxioQ);
+    // backfill:scan()
+    runNextTask(lpAuxioQ);
+
+    ASSERT_TRUE(producer->getReadyQueue().exists(vbid));
+
+    // Now drain all items before we proceed to complete, which triggers disk
+    // snapshot.
+    ASSERT_EQ(ENGINE_WANT_MORE, producer->step(producers.get()));
+    ASSERT_EQ(PROTOCOL_BINARY_CMD_DCP_SNAPSHOT_MARKER, dcp_last_op);
+    for (const auto& key : keys) {
+        ASSERT_EQ(ENGINE_WANT_MORE, producer->step(producers.get()));
+        ASSERT_EQ(PROTOCOL_BINARY_CMD_DCP_MUTATION, dcp_last_op);
+        ASSERT_EQ(key, dcp_last_key);
+    }
+
+    // Another producer step should report SUCCESS (no more data) as all items
+    // have been backfilled.
+    EXPECT_EQ(ENGINE_SUCCESS, producer->step(producers.get()));
+    // Also the readyQ should be empty
+    EXPECT_TRUE(producer->getReadyQueue().empty());
+
+    // backfill:complete()
+    runNextTask(lpAuxioQ);
+
+    // Notified to allow stream to transition to in-memory phase.
+    EXPECT_TRUE(producer->getReadyQueue().exists(vbid));
+
+    // Step should cause stream closed message, previously this would
+    // keep the "ENGINE_SUCCESS" response due to the itemsReady flag,
+    // which is not expected with that message already being in the readyQ.
+    EXPECT_EQ(ENGINE_WANT_MORE, producer->step(producers.get()));
+    EXPECT_EQ(PROTOCOL_BINARY_CMD_DCP_STREAM_END, dcp_last_op);
+
+    // Stepping forward should now show that stream end message has been
+    // completed and no more messages are needed to send.
+    EXPECT_EQ(ENGINE_SUCCESS, producer->step(producers.get()));
+
+    // Similarly, the readyQ should be empty again
+    EXPECT_TRUE(producer->getReadyQueue().empty());
+
+    // backfill:finished() - just to cleanup.
+    runNextTask(lpAuxioQ);
+
+    // vb0Stream should be closed
+    EXPECT_FALSE(vb0Stream->isActive());
+
+    // Stop Producer checkpoint processor task
+    producer->cancelCheckpointCreatorTask();
+}
+
 INSTANTIATE_TEST_CASE_P(XattrSystemUserTest,
                         XattrSystemUserTest,
                         ::testing::Bool(), );