MB-35652: Remove race between resolvedQ and setReplicationTopology

Currently we have a race between processing the
resolvedQueue and setting the replication topology. This happens
because processing the resolvedQueue requires use of the
ReplicationChain pointers in each SyncWrite object. We use the
ReplicationChain to see if a SyncWrite is satisfied when processing
the resolvedQueue. We do not update the pointers to the new
ReplicationChains for the SyncWrites in the resolvedQueue. This means
we could attempt to use a freed pointer when processing the
resolvedQueue. This problem existed before the processing of the queue
was moved to a separate task, but was exacerbated by it due to timings.

Fix this by setting a status in each SyncWrite when we know how to
completed it and using the status instead of the chain pointers. We
can then invalidate the pointers when removing the SyncWrites from
trackedWrites.

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

Author: BenHuddleston

engines/ep/src/durability/active_durability_monitor.cc

@@ -157,9 +157,10 @@ struct ActiveDurabilityMonitor::State {
      * Remove the given SyncWrte from tracking.
      *
      * @param it The iterator to the SyncWrite to be removed
+     * @param status The SyncWriteStatus to set the SyncWrite to as we remove it
      * @return the removed SyncWrite.
      */
-    SyncWrite removeSyncWrite(Container::iterator it);
+    SyncWrite removeSyncWrite(Container::iterator it, SyncWriteStatus status);
 
     /**
      * Logically 'moves' forward the High Prepared Seqno to the last
@@ -732,11 +733,22 @@ void ActiveDurabilityMonitor::processCompletedSyncWriteQueue() {
     std::lock_guard<ResolvedQueue::ConsumerLock> lock(
             resolvedQueue->getConsumerLock());
     while (folly::Optional<SyncWrite> sw = resolvedQueue->try_dequeue(lock)) {
-        if (sw->isSatisfied()) {
+        switch (sw->getStatus()) {
+        case SyncWriteStatus::Pending:
+        case SyncWriteStatus::Completed:
+            throw std::logic_error(
+                    "ActiveDurabilityMonitor::processCompletedSyncWriteQueue "
+                    "found a SyncWrite with unexpected state: " +
+                    to_string(sw->getStatus()));
+            continue;
+        case SyncWriteStatus::ToCommit:
             commit(*sw);
-        } else {
+            continue;
+        case SyncWriteStatus::ToAbort:
             abort(*sw);
+            continue;
         }
+        folly::assume_unreachable();
     };
 }
 
@@ -1001,11 +1013,17 @@ int64_t ActiveDurabilityMonitor::State::getNodeAckSeqno(
 }
 
 DurabilityMonitor::SyncWrite ActiveDurabilityMonitor::State::removeSyncWrite(
-        Container::iterator it) {
+        Container::iterator it, SyncWriteStatus status) {
     if (it == trackedWrites.end()) {
         throwException<std::logic_error>(__func__, "Position points to end");
     }
 
+    it->setStatus(status);
+    // Reset the chains so that we don't attempt to use some possibly re-used
+    // memory if we have any bugs that still touch the chains after we remove
+    // the SyncWrite from trackedWrites.
+    it->initialiseChains(nullptr, nullptr);
+
     Container::iterator prev;
     // Note: iterators in trackedWrites are never singular, Container::end
     //     is used as placeholder element for when an iterator cannot point to
@@ -1149,7 +1167,8 @@ void ActiveDurabilityMonitor::State::processSeqnoAck(const std::string& node,
 
         // Check if Durability Requirements satisfied now, and add for commit
         if (posIt->isSatisfied()) {
-            toCommit.enqueue(*this, removeSyncWrite(posIt));
+            toCommit.enqueue(*this,
+                             removeSyncWrite(posIt, SyncWriteStatus::ToCommit));
         }
     }
 
@@ -1175,7 +1194,8 @@ size_t ActiveDurabilityMonitor::wipeTracked() {
     while (it != s->trackedWrites.end()) {
         // Note: 'it' will be invalidated, so it will need to be reset
         const auto next = std::next(it);
-        s->removeSyncWrite(it);
+        // Status does not matter, just nuking trackedWrites
+        s->removeSyncWrite(it, SyncWriteStatus::Pending);
         removed++;
         it = next;
     }
@@ -1490,7 +1510,9 @@ void ActiveDurabilityMonitor::State::abortNoLongerPossibleSyncWrites(
                 // Grab the next itr before we overwrite ours to point to a
                 // different list.
                 auto next = std::next(itr);
-                toAbort.enqueue(*this, removeSyncWrite(trackedWrites.begin()));
+                toAbort.enqueue(*this,
+                                removeSyncWrite(trackedWrites.begin(),
+                                                SyncWriteStatus::ToAbort));
                 itr = next;
             } else {
                 itr++;
@@ -1524,9 +1546,10 @@ void ActiveDurabilityMonitor::State::cleanUpTrackedWritesPostTopologyChange(
         // snapshot. We have to do this after we set the HPS otherwise we could
         // end up with an ADM with lower HPS than the previous PDM.
         if (it->isCompleted()) {
-            removeSyncWrite(it);
+            removeSyncWrite(it, SyncWriteStatus::Completed);
         } else if (it->isSatisfied()) {
-            toCommit.enqueue(*this, removeSyncWrite(it));
+            toCommit.enqueue(*this,
+                             removeSyncWrite(it, SyncWriteStatus::ToCommit));
         }
         it = next;
     }
@@ -1556,7 +1579,8 @@ void ActiveDurabilityMonitor::State::removeExpired(
             // Note: 'it' will be invalidated, so it will need to be reset
             const auto next = std::next(it);
 
-            expired.enqueue(*this, removeSyncWrite(it));
+            expired.enqueue(*this,
+                            removeSyncWrite(it, SyncWriteStatus::ToAbort));
 
             it = next;
         } else {
@@ -1621,7 +1645,8 @@ void ActiveDurabilityMonitor::State::updateHighPreparedSeqno(
         const auto& pos = firstChain->positions.at(active);
         Expects(pos.it != trackedWrites.end());
         if (pos.it->isSatisfied()) {
-            completed.enqueue(*this, removeSyncWrite(pos.it));
+            completed.enqueue(
+                    *this, removeSyncWrite(pos.it, SyncWriteStatus::ToCommit));
         }
     };
 

engines/ep/src/durability/durability_monitor_impl.cc

@@ -372,3 +372,17 @@ std::string to_string(
     ss << "}";
     return ss.str();
 }
+
+std::string to_string(SyncWriteStatus status) {
+    switch (status) {
+    case SyncWriteStatus::Pending:
+        return "Pending";
+    case SyncWriteStatus::ToCommit:
+        return "ToCommit";
+    case SyncWriteStatus::ToAbort:
+        return "ToAbort";
+    case SyncWriteStatus::Completed:
+        return "Completed";
+    }
+    folly::assume_unreachable();
+}

engines/ep/src/durability/durability_monitor_impl.h

@@ -39,6 +39,29 @@ class PassiveDurabilityMonitor;
 // topology.
 static const std::string UndefinedNode{};
 
+/**
+ * The status of an in-flight SyncWrite
+ */
+enum class SyncWriteStatus {
+    // Still waiting for enough acks to commit or to timeout.
+    Pending = 0,
+
+    // Should be committed, enough nodes have acked. Should not exist in
+    // trackedWrites in this state.
+    ToCommit,
+
+    // Should be aborted. Should not exist in trackedWrites in this state.
+    ToAbort,
+
+    // A replica receiving a disk snapshot or a snapshot with a persist level
+    // prepare may not remove the SyncWrite object from trackedWrites until
+    // it has been persisted. This SyncWrite has been Completed but may still
+    // exist in trackedWrites.
+    Completed,
+};
+
+std::string to_string(SyncWriteStatus status);
+
 /**
  * Represents a tracked SyncWrite. It is mainly a wrapper around a pending
  * Prepare item.
@@ -151,25 +174,30 @@ class DurabilityMonitor::SyncWrite {
         return item;
     }
 
-    void setCompleted() {
-        completed = true;
+    void setStatus(SyncWriteStatus status) {
+        this->status = status;
     }
 
     /**
      * @return true if this SyncWrite has been logically completed
      */
     bool isCompleted() const {
-        return completed;
+        return status == SyncWriteStatus::Completed;
+    }
+
+    SyncWriteStatus getStatus() const {
+        return status;
     }
 
-private:
     /**
      * Performs sanity checks and initialise the replication chains
      * @param firstChain Pointer (may be null) to the first chain
      * @param secondChain Pointer (may be null) to second chain
      */
     void initialiseChains(const ReplicationChain* firstChain,
                           const ReplicationChain* secondChain);
+
+private:
     /**
      * Calculate the ackCount for this SyncWrite using the given chain.
      *
@@ -226,14 +254,7 @@ class DurabilityMonitor::SyncWrite {
     /// Used for statistics (track how long SyncWrites take to complete).
     const std::chrono::steady_clock::time_point startTime;
 
-    /**
-     * A replica receiving a disk snapshot or a snapshot with a persist level
-     * prepare may not remove the SyncWrite object from trackedWrites until
-     * it has been persisted. This field exists to distinguish prepares that
-     * have been completed but still exist in trackedWrites from those that have
-     * not yet been completed.
-     */
-    bool completed = false;
+    SyncWriteStatus status = SyncWriteStatus::Pending;
 
     friend std::ostream& operator<<(std::ostream&, const SyncWrite&);
 };

engines/ep/src/durability/passive_durability_monitor.cc

@@ -342,7 +342,7 @@ void PassiveDurabilityMonitor::completeSyncWrite(
     // duplicates in trackedWrites in case it is not removed because it requires
     // persistence.
     Expects(!next->isCompleted());
-    next->setCompleted();
+    next->setStatus(SyncWriteStatus::Completed);
 
     // HCS has moved, which could make some Prepare eligible for removal.
     s->checkForAndRemovePrepares();

engines/ep/tests/module_tests/durability_monitor_test.cc

@@ -986,6 +986,21 @@ TEST_P(ActiveDurabilityMonitorTest, SeqnoAckReceivedConcurrentDataRace) {
     EXPECT_EQ(makeStoredDocKey("key2"), items[1]->getKey());
 }
 
+TEST_P(ActiveDurabilityMonitorTest,
+       CommitTopologyWithSyncWriteInCompletedQueue) {
+    auto& adm = getActiveDM();
+    adm.setReplicationTopology(nlohmann::json({{active, replica1, replica2}}));
+    DurabilityMonitorTest::addSyncWrites({1});
+
+    notifyPersistence(1, 1, 0);
+
+    setSeqnoAckReceivedPostProcessHook([this, &adm]() {
+        adm.setReplicationTopology(nlohmann::json::array({{active, replica1}}));
+    });
+
+    adm.seqnoAckReceived("replica1", 1);
+}
+
 // @todo: Refactor test suite and expand test cases
 TEST_P(ActiveDurabilityMonitorPersistentTest,
        SeqnoAckReceived_PersistToMajority) {