MB-35367: Remove potential race in seqno acking

From code inspection, looks like we could enter either the
PDM::notifySnapshotEndReceived or PDM::notifyLocalPersistence
functions and acquire a hps value to ack back to the active.
The locking ensures that we never attempt to ack the same value
twice, but if the first thread were to reach the check if
hps != prevHps and get deschdeduled then a second thread could
make it through the lock and ack a high value before the first.
This would cause the active to throw monotonic invariant broken
exceptions.

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

Author: BenHuddleston

engines/ep/src/durability/passive_durability_monitor.cc

@@ -204,50 +204,57 @@ size_t PassiveDurabilityMonitor::getNumAborted() const {
 }
 
 void PassiveDurabilityMonitor::notifySnapshotEndReceived(uint64_t snapEnd) {
-    int64_t prevHps{0};
-    int64_t hps{0};
-    {
+    { // state locking scope
         auto s = state.wlock();
         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;
+        auto prevHps = s->highPreparedSeqno.lastWriteSeqno;
         s->updateHighPreparedSeqno();
-        hps = s->highPreparedSeqno.lastWriteSeqno;
+
+        // Store the seqno ack to send after we drop the state lock
+        storeSeqnoAck(prevHps, s->highPreparedSeqno.lastWriteSeqno);
     }
 
-    // HPS may have not changed (e.g., a locally-non-satisfied PersistToMajority
-    // Prepare has introduced a durability-fence), which would result in
-    // re-acking the same HPS multiple times. Not wrong as HPS is weakly
-    // monotonic at Active, but we want to avoid sending unnecessary messages.
-    if (hps != prevHps) {
-        Expects(hps > prevHps);
-        vb.sendSeqnoAck(hps);
+    if (notifySnapEndSeqnoAckPreProcessHook) {
+        notifySnapEndSeqnoAckPreProcessHook();
     }
+
+    sendSeqnoAck();
 }
 
 void PassiveDurabilityMonitor::notifyLocalPersistence() {
-    int64_t prevHps{0};
-    int64_t hps{0};
-    {
+    { // state locking scope
         auto s = state.wlock();
-        prevHps = s->highPreparedSeqno.lastWriteSeqno;
+        auto prevHps = s->highPreparedSeqno.lastWriteSeqno;
         s->updateHighPreparedSeqno();
-        hps = s->highPreparedSeqno.lastWriteSeqno;
+
+        // Store the seqno ack to send after we drop the state lock
+        storeSeqnoAck(prevHps, s->highPreparedSeqno.lastWriteSeqno);
     }
 
-    // HPS may have not changed (e.g., we have just persisted a Majority Prepare
-    // for which the HPS has been already increased at ADM::addSyncWrite), which
-    // would result in re-acking the same HPS multiple times. Not wrong as HPS
-    // is weakly monotonic at Active, but we want to avoid sending unnecessary
-    // messages.
-    if (hps != prevHps) {
-        Expects(hps > prevHps);
-        vb.sendSeqnoAck(hps);
+    sendSeqnoAck();
+}
+
+void PassiveDurabilityMonitor::storeSeqnoAck(int64_t prevHps, int64_t newHps) {
+    if (prevHps != newHps) {
+        auto seqno = seqnoToAck.wlock();
+        if (*seqno < newHps) {
+            *seqno = newHps;
+        }
+    }
+}
+
+void PassiveDurabilityMonitor::sendSeqnoAck() {
+    // Hold the lock throughout to ensure that we do not race with another ack
+    auto seqno = seqnoToAck.wlock();
+    if (*seqno != 0) {
+        vb.sendSeqnoAck(*seqno);
     }
+    *seqno = 0;
 }
 
 std::string PassiveDurabilityMonitor::to_string(Resolution res) {

engines/ep/src/durability/passive_durability_monitor.h

@@ -142,6 +142,10 @@ class PassiveDurabilityMonitor : public DurabilityMonitor {
      */
     void notifySnapshotEndReceived(uint64_t snapEnd);
 
+    /**
+     * Notify this PDM that some persistence has happened. Attempts to update
+     * the HPS and ack back to the active.
+     */
     void notifyLocalPersistence() override;
 
     /**
@@ -159,7 +163,27 @@ class PassiveDurabilityMonitor : public DurabilityMonitor {
      */
     int64_t getHighestTrackedSeqno() const;
 
+    /**
+     * Test only: Hook which if non-empty is called from
+     * notifySnapshotEndReceived()
+     */
+    std::function<void()> notifySnapEndSeqnoAckPreProcessHook;
+
 protected:
+    /**
+     * Store the seqno ack that we should now send to the consumer. Overwrites
+     * any outstanding ack not yet sent if the new value is greater.
+     *
+     * @param prevHps determines if we should send an ack or not
+     * @param newHps new hps to ack
+     */
+    void storeSeqnoAck(int64_t prevHps, int64_t newHps);
+
+    /**
+     * Send, if we need to, a seqno ack to the active node.
+     */
+    void sendSeqnoAck();
+
     void toOStream(std::ostream& os) const override;
     /**
      * throw exception with the following error string:
@@ -181,6 +205,9 @@ class PassiveDurabilityMonitor : public DurabilityMonitor {
     struct State;
     folly::SynchronizedPtr<std::unique_ptr<State>> state;
 
+    /// Outstanding seqno ack to send to the active. 0 if no ack outstanding
+    folly::Synchronized<int64_t> seqnoToAck{0};
+
     // Necessary for implementing ADM(PDM&&)
     friend class ActiveDurabilityMonitor;
 

engines/ep/tests/module_tests/durability_monitor_test.cc

@@ -3233,6 +3233,43 @@ TEST_P(ActiveDurabilityMonitorPersistentTest,
     }
 }
 
+TEST_P(PassiveDurabilityMonitorTest, SendSeqnoAckRace) {
+    auto& pdm = getPassiveDM();
+    ASSERT_EQ(0, pdm.getNumTracked());
+
+    Monotonic<int64_t> ackedSeqno = 0;
+
+    // Set up our function hooks
+    pdm.notifySnapEndSeqnoAckPreProcessHook = [this, &pdm]() {
+        // We are outside the state lock at this point but have not yet called
+        // vb->sendSeqnoAck. Simulate this race by notifying persistence which
+        // should attempt to ack 2.
+        vb->setPersistenceSeqno(2);
+        pdm.notifyLocalPersistence();
+    };
+
+    // Test: Assert the monotonicity of the ack that we attempt to send
+    VBucketTestIntrospector::setSeqnoAckCb(
+            *vb, [&ackedSeqno](Vbid vbid, uint64_t hps) { ackedSeqno = hps; });
+
+    // Load two SyncWrites. For this test we want a snapshot end to attempt to
+    // seqno ack with seqno 1 but be "descheduled" before it can and for a
+    // persistence callback to ack with seqno 2 before the seqno ack for seqno 1
+    // completes. This could happen with the following two prepares in a single
+    // snapshot.
+    using namespace cb::durability;
+    addSyncWrite(1 /*seqno*/,
+                 Requirements{Level::Majority, Timeout::Infinity()});
+    addSyncWrite(2 /*seqno*/,
+                 Requirements{Level::PersistToMajority, Timeout::Infinity()});
+    ASSERT_EQ(2, pdm.getNumTracked());
+
+    // Should only be able to ack seqno 1 as we require persistence to ack seqno
+    // 2.
+    pdm.notifySnapshotEndReceived(2);
+    EXPECT_EQ(2, ackedSeqno);
+}
+
 INSTANTIATE_TEST_CASE_P(AllBucketTypes,
                         ActiveDurabilityMonitorTest,
                         STParameterizedBucketTest::allConfigValues(),

engines/ep/tests/module_tests/vbucket_utils.h

@@ -30,4 +30,8 @@ class VBucketTestIntrospector {
     static PassiveDurabilityMonitor& public_getPassiveDM(VBucket& vb) {
         return vb.getPassiveDM();
     }
+
+    static void setSeqnoAckCb(VBucket& vb, SeqnoAckCallback func) {
+        vb.seqnoAckCb = func;
+    }
 };