MB-44877: Introduce ActiveStream::lastSentSeqnoAdvance

Useful info on stream-filtering for understanding whether lastSentSeqno
is a mutation or a seqno-advance.

Change-Id: Ib134373ff482889b00b0dafc8f8895ddfddb99ba
Reviewed-on: http://review.couchbase.org/c/kv_engine/+/151111
Reviewed-by: Richard de Mellow <[email protected]>
Tested-by: Build Bot <[email protected]>

Author: paolococchi

engines/ep/src/dcp/active_stream.cc

@@ -56,6 +56,7 @@ ActiveStream::ActiveStream(EventuallyPersistentEngine* e,
       includeDeletedUserXattrs(includeDeletedUserXattrs),
       lastReadSeqnoUnSnapshotted(st_seqno),
       lastSentSeqno(st_seqno),
+      lastSentSeqnoAdvance(0),
       curChkSeqno(st_seqno),
       takeoverState(vbucket_state_pending),
       itemsFromMemoryPhase(0),
@@ -825,6 +826,14 @@ void ActiveStream::addStats(const AddStatFn& add_stat, const void* c) {
                          name_.c_str(),
                          vb_.get());
         add_casted_stat(buffer, lastSentSeqno.load(), add_stat, c);
+
+        checked_snprintf(buffer,
+                         bsize,
+                         "%s:stream_%d_last_sent_seqno_advance",
+                         name_.c_str(),
+                         vb_.get());
+        add_casted_stat(buffer, lastSentSeqnoAdvance.load(), add_stat, c);
+
         checked_snprintf(buffer,
                          bsize,
                          "%s:stream_%d_last_sent_snap_end_seqno",
@@ -2178,8 +2187,9 @@ std::unique_ptr<DcpResponse> ActiveStream::makeEndStreamResponse(
 }
 
 void ActiveStream::queueSeqnoAdvanced() {
-    pushToReadyQ(std::make_unique<SeqnoAdvanced>(
-            opaque_, vb_, sid, lastSentSnapEndSeqno.load()));
+    const auto seqno = lastSentSnapEndSeqno.load();
+    pushToReadyQ(std::make_unique<SeqnoAdvanced>(opaque_, vb_, sid, seqno));
+    lastSentSeqnoAdvance.store(seqno);
 }
 
 bool ActiveStream::isDiskOnly() const {

engines/ep/src/dcp/active_stream.h

@@ -677,6 +677,9 @@ class ActiveStream : public Stream,
     //! The last sequence number sent to the network layer
     std::atomic<uint64_t> lastSentSeqno;
 
+    //! The seqno of the last SeqnoAdvance sent
+    std::atomic<uint64_t> lastSentSeqnoAdvance;
+
     //! The last known seqno pointed to by the checkpoint cursor
     std::atomic<uint64_t> curChkSeqno;