Lost wakeup in ActiveStream can result in DCP clients not receiving seqnos until another mutation occurs

Summary
There exists a race in ActiveStream between a notifying that a new mutation available, and processing the previous mutation(s). If the notification occurs just before ActiveStream sets a flag indicating that there's currently no more items to process, then the notification is lost and the DCP stream is not notified of this new mutation when it should be.

This results in the DCP stream being behind by that seqno, until the the next mutation on the vBucket occurs with will re-notify the stream and cause it to process both.

Impact

When it occurs, this bug introduces additional latency between a mutation occurring on the KV node, and it getting passed to the affected DCP client. For DCP connections which require a "current" set of seqnos to proceed - e.g. GSI when using request_plus queries, this can manifest as those operations hanging.

Note the latency added is bounded by the mutation rate on the vBucket - the next mutation is guaranteed to wake up the stream (as it is now idle and there's no possible race). For workloads with moderate mutation rates (e.g. 1000 mutations per second per Bucket), the additional latency on average should be 1s. However for workloads with very low mutation rates (<100 per second) then on average the latency would be 10+ seconds.

For workloads which are intermittent and idle for extended periods, the added latency could be large - as long as it takes for that vBucket to be modified again.

This issue affects all currently supported releases (6.6.x, 7.0.x, 7.1.x).

Steps to Reproduce

1. Two node cluster, first node KV, second node query+GSI.
2. 5 buckets created - value eviction, couchstore, 1 replica: bucket-1, bucket-2, bucket-3, bucket-4, bucket-5
3. Create primary index for all buckets so projector / GSI establishes DCP streams.
4. Run the attached workload generator / monitor script ( cb_monitor_dcp_items_remaining_2.py ) which generates a burst of mutations, then monitors DCP stats waiting for items_remaining to drop to zero before generating the next batch of mutations. This is run on 5 buckets to increase likelihood / speed of hitting the issue:

   python3.4 ./cb_monitor_dcp_items_remaining_2.py localhost bucket-1

   python3.4 ./cb_monitor_dcp_items_remaining_2.py localhost bucket-2

   ...

5. Plus, Running 5 instances of dcpdrain against the first bucket to add additional load to the system:

   ./dcpdrain --host <HOST> --user Administrator --password asdasd -b bucket-1 --stream-request-flags 0 -v --name dcpdrain.5 --buffer-size 20M

The workload program will print log message when items_remaining for GSi is non-zero. If this is transient then that's not a problem, however if it gets stuck at a non-zero value then we have hit the issue:

2022-11-16T21:01:15+0000 Non-zero items outstanding: ec2-63-35-220-143.eu-west-1.compute.amazonaws.com:11210 -> 1

2022-11-16T21:01:15+0000  - conn:'eq_dcpq:secidx:proj-MAINT_STREAM_TOPIC_6787fcc9207b2f307e985b5bb108a0eb-1994742923520278715/2' vb:vb_18 num_items_for_cursor: 1

2022-11-16T21:01:17+0000 Non-zero items outstanding: ec2-63-35-220-143.eu-west-1.compute.amazonaws.com:11210 -> 1

2022-11-16T21:01:17+0000  - conn:'eq_dcpq:secidx:proj-MAINT_STREAM_TOPIC_6787fcc9207b2f307e985b5bb108a0eb-1994742923520278715/2' vb:vb_18 num_items_for_cursor: 1

...

The time taken to hit the issue can vary; but was generally seen within 10minutes.

Details

The race is between the itemsReady flag being tested in notifyStreamReady, and clearing itemsReady when there are no more items available for the stream in ActiveStream::next(). Flag is tested here:

void ActiveStream::notifyStreamReady(bool force, DcpProducer* producer, uint64_t seqno) {

    bool inverse = false;

    if (force || itemsReady.compare_exchange_strong(inverse, true)) {

        if (seqno) {

            itemsReadySeqnosForNotify.lock()->push_back(seqno);

        }

...

Adding a sleep when itemsReady is about to be cleared (when no response is available) makes window for a potential race larger, and I see the issue occur pretty much immediately, whereas before it took a varying amount of time from 1-10 mins for the issue to occur with the test workload:

std::unique_ptr<DcpResponse> ActiveStream::next(DcpProducer& producer) {

    std::unique_ptr<DcpResponse> response;

    switch (state_.load()) {

    ...

    case StreamState::InMemory:

        response = inMemoryPhase(producer);

        break;

    ...

    }

    // DEBUG

    if (!response) {

        std::this_thread::sleep_for(std::chrono::milliseconds{1});

    }

    // END DEBUG

    itemsReady.store(response ? true : false);

    return response;