MB-37468: Force notification of stream completing backfill

A stream completing a backfill may race with a stepping producer.
The stream may have just completed the backfill (scanned all items
but not yet completed execution of the completeBackfill function)
whilst the producer steps and finds nothing in the readyQueue (already
processed all items). If the completing backfill then does not notify
the stream as the itemsReady flag has not yet been reset (by the
stepping producer) then the producer will remove the stream from its
VBReadyQueue and not process it again. Nothing else would notify
the stream causing us to fail to call endStream or transitionState.

Change-Id: Ib430ef78261b910616576572bf76cd97c1cf47be
Reviewed-on: http://review.couchbase.org/120292
Reviewed-by: James Harrison <[email protected]>
Tested-by: Build Bot <[email protected]>
Reviewed-by: Jim Walker <[email protected]>
Well-Formed: Build Bot <[email protected]>

Author: BenHuddleston

engines/ep/src/dcp/stream.cc

@@ -374,6 +374,10 @@ std::unique_ptr<DcpResponse> ActiveStream::next(
             break;
     }
 
+    if (nextHook) {
+        nextHook();
+    }
+
     itemsReady.store(response ? true : false);
     return response;
 }
@@ -558,9 +562,21 @@ void ActiveStream::completeBackfill() {
         }
     }
 
+    if (completeBackfillHook) {
+        completeBackfillHook();
+    }
+
     bool inverse = true;
     isBackfillTaskRunning.compare_exchange_strong(inverse, false);
-    notifyStreamReady();
+
+    // MB-37468: Items may not be ready, but we need to notify the stream
+    // regardless as a racing stepping producer that had just finished
+    // processing all items and found an empty ready queue could clear the flag
+    // immediately after we call notifyStreamReady (which does not notify as
+    // itemsReady is true). This would then result in us not notifying the
+    // stream and not putting it back in the producer's readyQueue. A similar
+    // case exists for transitioning state to TakeoverSend or InMemory.
+    notifyStreamReady(true);
 }
 
 void ActiveStream::snapshotMarkerAckReceived() {

engines/ep/src/dcp/stream.h

@@ -410,6 +410,10 @@ class ActiveStream : public Stream,
 
     std::unique_ptr<DcpResponse> backfillPhase(std::lock_guard<std::mutex>& lh);
 
+    // MB-37468: Test only hooks set via Mock class
+    std::function<void()> completeBackfillHook;
+    std::function<void()> nextHook;
+
 private:
     std::unique_ptr<DcpResponse> next(std::lock_guard<std::mutex>& lh);
 

engines/ep/tests/mock/mock_stream.h

@@ -160,6 +160,14 @@ class MockActiveStream : public ActiveStream {
     std::function<void()> preGetOutstandingItemsCallback = [] { return; };
 
     bool isDead() { return ActiveStream::getState() == StreamState::Dead; };
+
+    void setCompleteBackfillHook(std::function<void()> hook) {
+        completeBackfillHook = hook;
+    }
+
+    void setNextHook(std::function<void()> hook) {
+        nextHook = hook;
+    }
 };
 
 /**

engines/ep/tests/module_tests/dcp_test.cc

@@ -4268,6 +4268,118 @@ TEST_P(SingleThreadedActiveStreamTest,
     EXPECT_EQ(ActiveStream::StreamState::InMemory, stream->getState());
 }
 
+// MB-37468: A stepping producer that has found no items (backfill fully
+// processed can race with a completing backfill in such a way that we fail to
+// notify the producer that the stream needs further processing. This causes us
+// to fail to send a StreamEnd message. A similar case exists for transitioning
+// state to TakeoverSend or InMemory.
+TEST_P(SingleThreadedActiveStreamTest, CompleteBackfillRaceNoStreamEnd) {
+    auto vb = engine->getVBucket(vbid);
+    auto& ckptMgr = *vb->checkpointManager;
+
+    // Delete initial stream (so we can re-create after items are available
+    // from backing store).
+    stream.reset();
+
+    // Add items, flush it to disk, then clear checkpoint to force backfill.
+    store_item(vbid, makeStoredDocKey("key1"), "value");
+    ckptMgr.createNewCheckpoint();
+
+    flushVBucketToDiskIfPersistent(vbid, 1);
+    bool newCKptCreated;
+    ASSERT_EQ(1, ckptMgr.removeClosedUnrefCheckpoints(*vb, newCKptCreated));
+
+    // Re-create producer now we have items only on disk. We want to stream up
+    // to seqno 1 (our only item) to test that we get the StreamEnd message.
+    stream = producer->mockActiveStreamRequest(0 /*flags*/,
+                                               0 /*opaque*/,
+                                               *vb,
+                                               0 /*st_seqno*/,
+                                               1 /*en_seqno*/,
+                                               0x0 /*vb_uuid*/,
+                                               0 /*snap_start_seqno*/,
+                                               ~0 /*snap_end_seqno*/);
+    ASSERT_TRUE(stream->isBackfilling());
+
+    auto producers = get_dcp_producers(
+            reinterpret_cast<ENGINE_HANDLE*>(engine.get()),
+            reinterpret_cast<ENGINE_HANDLE_V1*>(engine.get()));
+
+    // Step to schedule our backfill
+    EXPECT_EQ(ENGINE_SUCCESS, producer->step(producers.get()));
+    EXPECT_EQ(0, stream->public_readyQ().size());
+
+    auto& bfm = producer->getBFM();
+
+    // Ephemeral has a single stage backfill and we only compare about the
+    // complete stage so skip over create and scan for persistent buckets
+    if (persistent()) {
+        bfm.backfill();
+        bfm.backfill();
+    }
+
+    ThreadGate tg1 (2);
+    ThreadGate tg2 (2);
+    std::thread t1;
+    stream->setCompleteBackfillHook([this, &t1, &tg1, &tg2, &producers]() {
+        // Step past our normal items to expose the race with backfill complete
+        // and an empty readyQueue.
+
+        EXPECT_EQ(1, *stream->getNumBackfillItemsRemaining());
+        EXPECT_EQ(2, stream->public_readyQ().size());
+
+        // Step snapshot marker
+        EXPECT_EQ(ENGINE_WANT_MORE, producer->step(producers.get()));
+        EXPECT_EQ((uint8_t)cb::mcbp::ClientOpcode::DcpSnapshotMarker,
+                dcp_last_op);
+
+        // Step mutation
+        EXPECT_EQ(ENGINE_WANT_MORE, producer->step(producers.get()));
+        EXPECT_EQ((uint8_t)cb::mcbp::ClientOpcode::DcpMutation, dcp_last_op);
+
+        stream->setNextHook([&tg1, &tg2]() {
+            if (!tg1.isComplete()) {
+                tg1.threadUp();
+
+                // Wait for the completeBackfill thread to have attempted to
+                // notify that the stream is ready before exiting the hook and
+                // setting itemsReady.
+                tg2.threadUp();
+            }
+        });
+
+
+        // Run the step in a different thread
+        t1 = std::thread{[this, &producers]() {
+            // This step should produce the stream end
+            EXPECT_EQ(ENGINE_WANT_MORE, producer->step(producers.get()));
+            EXPECT_EQ((uint8_t)cb::mcbp::ClientOpcode::DcpStreamEnd, dcp_last_op);
+        }};
+
+        // Wait for the stepping thread to have reached the point at which it is
+        // about to set itemsReady before we attempt to set itemsReady after we
+        // exit this hook.
+        tg1.threadUp();
+    });
+
+    // Complete the backfill to expose the race condition
+    bfm.backfill();
+
+    // Unblock the stepping thread to now find the stream end
+    tg2.threadUp();
+
+    t1.join();
+
+    // Should have sent StreamEnd but stream still in queue
+    EXPECT_FALSE(producer->findStream(vbid)->isActive());
+    EXPECT_FALSE(producer->getReadyQueue().empty());
+
+    // Step to remove stream from queue
+    EXPECT_EQ(ENGINE_SUCCESS, producer->step(producers.get()));
+    EXPECT_FALSE(producer->findStream(vbid)->isActive());
+    EXPECT_TRUE(producer->getReadyQueue().empty());
+}
+
 INSTANTIATE_TEST_CASE_P(AllBucketTypes,
                         SingleThreadedActiveStreamTest,
-                        allConfigValues, );
+                        allConfigValues, );