MB-34873: disk snapshots - Ack snapEnd seqno once persisted

Problem 1:
Any prepare received by a replica from a disk snapshot may have deduped
an earlier prepare of a higher level.

For example, the following ops (for the same key)

PRE(l=PersistToMajority) CMT PRE(l=Majority) CMT

May be deduped to

PRE(l=Majority) CMT

If we acked this prepare immediately, were we to (say) be promoted to
active and immediately crash, we may be left with no value on disk for
the key BUT the PersistToMajority op may have returned SUCCESS to the
client (it was committed) - we have broken the durability agreement for
that op.

Problem 2:
PRE(persistMajority), CMT, PRE(), ABORT, SET

may, after the abort has been purged be sent as:

SET

and we have no way of knowing a durability op was ever present.

Solution:
Advance the HPS to snapshotEndSeqno and seqnoAck once a full snapshot is
persisted, _just in case_ any prepares were deduped.

We could ack the latest prepare we are tracking, but we may be unaware
of some prepares due to Problem 2, so ack snapshotEndSeqno - any
prepares before this seqno are definitely prepared anyway.

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

Author: jameseh96

engines/ep/src/dcp/passive_stream.cc

@@ -948,17 +948,23 @@ void PassiveStream::handleSnapshotEnd(VBucketPtr& vb, uint64_t byseqno) {
             notifyStreamReady();
             cur_snapshot_ack = false;
         }
-        cur_snapshot_type.store(Snapshot::None);
 
         // Notify the PassiveDM that the snapshot-end mutation has been
         // received on PassiveStream, if the snapshot contains at least one
         // Prepare. That is necessary for unblocking the High Prepared Seqno
         // in PassiveDM. Note that the HPS is what the PassiveDM acks back to
         // the Active. See comments in PassiveDM for details.
-        if (cur_snapshot_prepare) {
+
+        // Disk snapshots are subject to deduplication, and may be missing
+        // purged aborts. We must notify the PDM even if we have not seen a
+        // prepare, to account for possible unseen prepares.
+        if (cur_snapshot_prepare ||
+            cur_snapshot_type.load() == Snapshot::Disk) {
             vb->notifyPassiveDMOfSnapEndReceived(byseqno);
             cur_snapshot_prepare.store(false);
         }
+
+        cur_snapshot_type.store(Snapshot::None);
     }
 }
 

engines/ep/src/durability/durability_monitor_impl.cc

@@ -323,3 +323,11 @@ bool DurabilityMonitor::ReplicationChain::hasAcked(const std::string& node,
 
     return itr->second.lastWriteSeqno >= bySeqno;
 }
+
+bool operator>(const SnapshotEndInfo& a, const SnapshotEndInfo& b) {
+    return a.seqno > b.seqno;
+}
+std::string to_string(const SnapshotEndInfo& snapshotEndInfo) {
+    return std::to_string(snapshotEndInfo.seqno) + "{" +
+           to_string(snapshotEndInfo.type) + "}";
+}

engines/ep/src/durability/durability_monitor_impl.h

@@ -261,6 +261,23 @@ struct DurabilityMonitor::ReplicationChain {
     const DurabilityMonitor::ReplicationChainName name;
 };
 
+// Used to track information necessary for the PassiveDM to correctly ACK
+// at the end of snapshots
+struct SnapshotEndInfo {
+    int64_t seqno;
+    CheckpointType type;
+};
+
+/**
+ * Compares two SnapshotEndInfos by seqno.
+ *
+ * Only provides a partial ordering; SnapshotEndInfos with the same seqno
+ * but different checkpoint type would be arbitrarily ordered if using this
+ * comparator
+ */
+bool operator>(const SnapshotEndInfo& a, const SnapshotEndInfo& b);
+std::string to_string(const SnapshotEndInfo& snapshotEndInfo);
+
 /*
  * This class embeds the state of a PDM. It has been designed for being
  * wrapped by a folly::Synchronized<T>, which manages the read/write
@@ -340,11 +357,11 @@ struct PassiveDurabilityMonitor::State {
     //         PersistToMajority
     Position highPreparedSeqno;
 
-    // Queue of snapEndSeqnos for snapshots which have been received entirely
-    // by the (owning) replica/pending VBucket.
+    // Queue of SnapshotEndInfos for snapshots which have been received entirely
+    // by the (owning) replica/pending VBucket, and the type of the checkpoint.
     // Used for implementing the correct move-logic of High Prepared Seqno.
     // Must be pushed to at snapshot-end received on PassiveStream.
-    MonotonicQueue<uint64_t> receivedSnapshotEndSeqnos;
+    MonotonicQueue<SnapshotEndInfo> receivedSnapshotEnds;
 
     // Cumulative count of accepted (tracked) SyncWrites.
     size_t totalAccepted = 0;

engines/ep/src/durability/passive_durability_monitor.cc

@@ -131,7 +131,10 @@ void PassiveDurabilityMonitor::notifySnapshotEndReceived(uint64_t snapEnd) {
     int64_t hps{0};
     {
         auto s = state.wlock();
-        s->receivedSnapshotEndSeqnos.push(snapEnd);
+        s->receivedSnapshotEnds.push({int64_t(snapEnd),
+                                      vb.isReceivingDiskSnapshot()
+                                              ? CheckpointType::Disk
+                                              : CheckpointType::Memory});
         // Maybe the new tracked Prepare is already satisfied and could be
         // ack'ed back to the Active.
         prevHps = s->highPreparedSeqno.lastWriteSeqno;
@@ -307,6 +310,7 @@ void PassiveDurabilityMonitor::State::updateHighPreparedSeqno() {
     // at PDM under the following constraints:
     //
     // (1) Nothing is ack'ed before the complete snapshot is received
+    //     (I.e., do nothing if receivedSnapshotEnds is empty)
     //
     // (2) Majority and MajorityAndPersistOnMaster Prepares (which don't need to
     //     be persisted for being locally satisfied) may be satisfied as soon as
@@ -319,6 +323,12 @@ void PassiveDurabilityMonitor::State::updateHighPreparedSeqno() {
     // (4) The durability-fence can move (ie, PersistToMajority Prepares are
     //     locally-satisfied) only when the complete snapshot is persisted.
     //
+    // (5) Once a disk snapshot is fully persisted, the HPS is advanced to the
+    //     snapshot end - even if no prepares were seen during the snapshot
+    //     or if trackedWrites is empty. This accounts for deduping; there may
+    //     have been prepares we have not seen, but they are definitely
+    //     satisfied (they are persisted) and should be acked.
+    //
     // This function implements all the logic necessary for moving the HPS by
     // enforcing the rules above. The function is called:
     //
@@ -334,16 +344,6 @@ void PassiveDurabilityMonitor::State::updateHighPreparedSeqno() {
     //     durability-fence. As already mentioned, we can move the
     //     durability-fence only if the complete snapshot is persisted.
 
-    if (trackedWrites.empty()) {
-        return;
-    }
-
-    if (receivedSnapshotEndSeqnos.empty()) {
-        // We have not received a full snapshot, we cannot advance the hps at
-        // all
-        return;
-    }
-
     const auto prevHPS = highPreparedSeqno.lastWriteSeqno;
 
     // Helper to keep conditions short and meaningful
@@ -355,49 +355,87 @@ void PassiveDurabilityMonitor::State::updateHighPreparedSeqno() {
                        snapshotEndSeqno;
     };
 
-    while (!receivedSnapshotEndSeqnos.empty() && !trackedWrites.empty()) {
-        uint64_t snapshotEndSeqno = receivedSnapshotEndSeqnos.front();
-
-        // ** If pdm.vb.getPersistenceSeqno() >= snapshotEndSeqno
-        // we have received and persisted an entire snapshot
-        // All prepares from this snapshot are satisfied and the state
-        // is consistent at snap end. The HPS can advance over Prepares of
-        // PersistToMajority or lower (i.e., everything currently)
-
-        // ** if pdm.vb.getPersistenceSeqno() < snapshotEndSeqno
-        // we have received but NOT persisted an entire snapshot
-        //  We *may* be able to advance the HPS part way
-        // into this snapshot - The HPS can be advanced over all Prepares of
-        // MajorityAndPersistOnMaster level or lower, to the last Prepare
-        // immediately preceding an *unpersisted* Prepare with Level ==
-        // PersistToMajority. We cannot move the HPS past this Prepare until
-        // it *is* persisted.
-
-        const cb::durability::Level maxLevelCanAdvanceOver =
-                (pdm.vb.getPersistenceSeqno() >= snapshotEndSeqno)
-                        ? cb::durability::Level::PersistToMajority
-                        : cb::durability::Level::MajorityAndPersistOnMaster;
-
-        for (auto next = getIteratorNext(highPreparedSeqno.it);
-             inSnapshot(snapshotEndSeqno, next) &&
-             next->getDurabilityReqs().getLevel() <= maxLevelCanAdvanceOver;
-             next = getIteratorNext(highPreparedSeqno.it)) {
-            // Note: Update last-write-seqno first to enforce monotonicity and
-            //     avoid any state-change if monotonicity checks fail
-            highPreparedSeqno.lastWriteSeqno = next->getBySeqno();
-            highPreparedSeqno.it = next;
+    while (!receivedSnapshotEnds.empty()) {
+        const auto snapshotEnd = receivedSnapshotEnds.front();
+
+        const bool snapshotFullyPersisted =
+                static_cast<int64_t>(pdm.vb.getPersistenceSeqno()) >=
+                snapshotEnd.seqno;
+
+        const bool isDiskSnapshot = snapshotEnd.type == CheckpointType::Disk;
+
+        using namespace cb::durability;
+
+        Level maxLevelCanAdvanceOver{};
+
+        if (snapshotFullyPersisted) {
+            // we have received and persisted an entire snapshot
+            // All prepares from this snapshot are satisfied and the state
+            // is consistent at snap end. The HPS can advance over Prepares of
+            // PersistToMajority or lower (i.e., everything currently)
+            maxLevelCanAdvanceOver = Level::PersistToMajority;
+        } else if (!isDiskSnapshot) {
+            // we have received but NOT persisted an entire snapshot
+            //  We *may* be able to advance the HPS part way
+            // into this snapshot - The HPS can be advanced over all Prepares of
+            // MajorityAndPersistOnMaster level or lower, to the last Prepare
+            // immediately preceding an *unpersisted* Prepare with Level ==
+            // PersistToMajority. We cannot move the HPS past this Prepare until
+            // it *is* persisted.
+            maxLevelCanAdvanceOver = Level::MajorityAndPersistOnMaster;
+        } else {
+            // we have received but NOT persisted an entire *DISK* snapshot
+            // we cannot ack anything until the entire snapshot has been
+            // persisted because PersistToMajority level Prepares may have been
+            // deduped by lower level prepares.
+            // Therefore, the HPS cannot advance over *any* prepares.
+            maxLevelCanAdvanceOver = Level::None;
         }
-            // Check if we finished an entire snapshot, and might be able to
-            // continue checking the next one.
-            if (inSnapshot(snapshotEndSeqno,
-                           getIteratorNext(highPreparedSeqno.it))) {
-                // we stopped advancing the HPS before the end of a snapshot
-                // because we reached a PersistToMajority Prepare
-                // HPS now points to the last Prepare before any
-                // PersistToMajority
-                break;
+
+        // Advance the HPS, respecting maxLevelCanAdvanceOver
+        if (!trackedWrites.empty()) {
+            for (auto next = getIteratorNext(highPreparedSeqno.it);
+                 inSnapshot(snapshotEnd.seqno, next) &&
+                 next->getDurabilityReqs().getLevel() <= maxLevelCanAdvanceOver;
+                 next = getIteratorNext(highPreparedSeqno.it)) {
+                // Note: Update last-write-seqno first to enforce monotonicity
+                // and avoid any state-change if monotonicity checks fail
+                highPreparedSeqno.lastWriteSeqno = next->getBySeqno();
+                highPreparedSeqno.it = next;
             }
-        receivedSnapshotEndSeqnos.pop();
+        }
+
+        if (isDiskSnapshot && snapshotFullyPersisted) {
+            // Special case - prepares in disk snapshots may have been
+            // deduplicated.
+            // PRE(persistMajority), CMT, PRE(), ABORT, SET
+            // may, after the abort has been purged be sent as:
+            // SET
+            // We would have no prepare for this op, but we still need to
+            // seqno ack something. To resolve this, advance the HPS seqno to
+            // the snapshotEndSeqno. There may not be an associated prepare.
+            // NB: lastWriteSeqno is NOT guaranteed to match
+            // highPreparedSeqno.it->getBySeqno()
+            // because of this case
+            highPreparedSeqno.lastWriteSeqno = snapshotEnd.seqno;
+        }
+
+        // Check if we could have acked everything within the snapshot and
+        // might be able to continue checking the next one.
+        if ((isDiskSnapshot && !snapshotFullyPersisted) ||
+            inSnapshot(snapshotEnd.seqno,
+                       getIteratorNext(highPreparedSeqno.it))) {
+            // Either we have not fully persisted a disk snapshot and
+            // the HPS is left <= the start of this snapshot
+            // OR
+            // we stopped advancing the HPS before the end of a memory
+            // snapshot because we reached a PersistToMajority Prepare
+            // HPS now points to the last Prepare before any
+            // PersistToMajority
+            break;
+        }
+
+        receivedSnapshotEnds.pop();
     }
 
     // We have now acked all the complete, persisted snapshots we received,

engines/ep/src/ep_types.cc

@@ -67,6 +67,16 @@ std::string to_string(TrackCasDrift trackCasDrift) {
             std::to_string(static_cast<TrackCasDriftUType>(trackCasDrift)));
 }
 
+std::string to_string(CheckpointType checkpointType) {
+    switch (checkpointType) {
+    case CheckpointType::Disk:
+        return "Disk";
+    case CheckpointType::Memory:
+        return "Memory";
+    }
+    folly::assume_unreachable();
+}
+
 std::string to_string(HighPriorityVBNotify hpNotifyType) {
     using HighPriorityVBNotifyUType =
             std::underlying_type<HighPriorityVBNotify>::type;

engines/ep/src/ep_types.h

@@ -63,6 +63,7 @@ std::ostream& operator<<(std::ostream&, TransferVB transfer);
 std::string to_string(GenerateBySeqno generateBySeqno);
 std::string to_string(GenerateCas generateCas);
 std::string to_string(TrackCasDrift trackCasDrift);
+std::string to_string(CheckpointType checkpointType);
 
 struct snapshot_range_t {
     snapshot_range_t(uint64_t start, uint64_t end) : start(start), end(end) {

engines/ep/tests/ep_testsuite_dcp.cc

@@ -3829,6 +3829,8 @@ static enum test_result test_dcp_consumer_takeover(EngineIface* h) {
                 "Failed to send dcp mutation");
     }
 
+    wait_for_flusher_to_settle(h);
+
     dcp->snapshot_marker(cookie, stream_opaque, Vbid(0), 6, 10, 10);
     for (int i = 6; i <= 10; i++) {
         const std::string key{"key" + std::to_string(i)};
@@ -3852,18 +3854,32 @@ static enum test_result test_dcp_consumer_takeover(EngineIface* h) {
                 "Failed to send dcp mutation");
     }
 
+    wait_for_flusher_to_settle(h);
+
     wait_for_stat_to_be(h, "eq_dcpq:unittest:stream_0_buffer_items", 0, "dcp");
 
     dcp_step(h, cookie, producers);
     cb_assert(producers.last_op == cb::mcbp::ClientOpcode::DcpSnapshotMarker);
     cb_assert(producers.last_status == cb::mcbp::Status::Success);
     cb_assert(producers.last_opaque != opaque);
 
+    dcp_step(h, cookie, producers);
+    cb_assert(producers.last_op ==
+              cb::mcbp::ClientOpcode::DcpSeqnoAcknowledged);
+    cb_assert(producers.last_status == cb::mcbp::Status::Success);
+    cb_assert(producers.last_opaque != opaque);
+
     dcp_step(h, cookie, producers);
     cb_assert(producers.last_op == cb::mcbp::ClientOpcode::DcpSnapshotMarker);
     cb_assert(producers.last_status == cb::mcbp::Status::Success);
     cb_assert(producers.last_opaque != opaque);
 
+    dcp_step(h, cookie, producers);
+    cb_assert(producers.last_op ==
+              cb::mcbp::ClientOpcode::DcpSeqnoAcknowledged);
+    cb_assert(producers.last_status == cb::mcbp::Status::Success);
+    cb_assert(producers.last_opaque != opaque);
+
     testHarness->destroy_cookie(cookie);
 
     return SUCCESS;

engines/ep/tests/module_tests/dcp_durability_stream_test.cc

@@ -868,6 +868,11 @@ void DurabilityPassiveStreamTest::
     // 4) Verify doc state
     auto vb = store->getVBucket(vbid);
     ASSERT_TRUE(vb);
+
+    // tell the DM that this snapshot was persisted
+    vb->setPersistenceSeqno(streamStartSeqno);
+    vb->notifyPersistenceToDurabilityMonitor();
+
     {
         // findForCommit will return both pending and committed perspectives
         auto res = vb->ht.findForCommit(key);