MB-34996: Do not abort prepares with infinite timeout

If we warmup or promote a replica to active then we create
SyncWrite objects with an infinite timeout as these prepares
MUST be committed as they could have been committed pre-warmup
or by the old active. Currently, the code will attempt to abort
these prepares if a subsequent topology change makes durability
impossible. This breaks durability though as we MUST commit these
prepares.

Update the code to never attempt to abort a SyncWrite with an
infinite timeout.

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

Author: BenHuddleston

engines/ep/src/durability/active_durability_monitor.cc

@@ -1178,16 +1178,28 @@ void ActiveDurabilityMonitor::State::setReplicationTopology(
     }
 
     // If durability is not possible for the new chains, then we should abort
-    // the in-flight SyncWrites so that the client can decide what to do. We
-    // have already reset the topology of the in flight SyncWrites so that they
-    // do not contain any invalid pointers post topology change.
+    // any in-flight SyncWrites that do not have an infinite timeout so that the
+    // client can decide what to do. We do not abort and infinite timeout
+    // SyncWrites as we MUST complete them as they exist due to a warmup or
+    // Passive->Active transition. We have already reset the topology of the in
+    // flight SyncWrites so that they do not contain any invalid pointers post
+    // topology change.
     if (!(newFirstChain && newFirstChain->isDurabilityPossible() &&
           (!newSecondChain || newSecondChain->isDurabilityPossible()))) {
         // We can't use a for loop with iterators here because they will be
         // modified to point to invalid memory as we use std::list.splice in
         // removeSyncWrite.
-        while (!trackedWrites.empty()) {
-            toAbort.enqueue(*this, removeSyncWrite(trackedWrites.begin()));
+        auto itr = trackedWrites.begin();
+        while (itr != trackedWrites.end()) {
+            if (!itr->getDurabilityReqs().getTimeout().isInfinite()) {
+                // 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()));
+                itr = next;
+            } else {
+                itr++;
+            }
         }
     }
 

engines/ep/tests/module_tests/durability_monitor_test.cc

@@ -2719,7 +2719,7 @@ TEST_P(ActiveDurabilityMonitorTest, HPSResetOnTopologyChange) {
  *
  * We have 1 replica that we failover. Topology is changed from
  * {{active, replica1}} to {{active, undefined}}. In this case we should abort
- * any in-flight SyncWrites and not throw assertions.
+ * any in-flight SyncWrites with a non-infinite and not throw assertions.
  */
 TEST_P(ActiveDurabilityMonitorTest,
        DurabilityImpossibleTopologyChangeAbortsInFlightSyncWrites) {
@@ -2742,6 +2742,37 @@ TEST_P(ActiveDurabilityMonitorTest,
     EXPECT_EQ(1, getActiveDM().getNumAborted());
 }
 
+/**
+ * Failover scenario:
+ *
+ * We have 1 replica that we failover. Topology is changed from
+ * {{active, replica1}} to {{active, undefined}}. In this case we should not
+ * abort any in-flight SyncWrites with an infinite timeout as this breaks
+ * durability. We create SyncWrites with an infinite timeout at warmup and at
+ * promotion from replica to active as we MUST commit these SyncWrites.
+ */
+TEST_P(ActiveDurabilityMonitorTest,
+       DurabilityImpossibleTopologyChangeDoesNotAbortsInfiniteTimeoutSyncWrites) {
+    // To start, we have 1 chain with active and replica1
+    using namespace cb::durability;
+    addSyncWrite(1, Requirements{Level::Majority, Timeout::Infinity()});
+    {
+        SCOPED_TRACE("");
+        assertNumTrackedAndHPSAndHCS(1, 1, 0);
+    }
+
+    // Failover
+    EXPECT_NO_THROW(getActiveDM().setReplicationTopology(
+            nlohmann::json::array({{active, nullptr}})));
+
+    {
+        SCOPED_TRACE("");
+        assertNumTrackedAndHPSAndHCS(1, 1, 0);
+    }
+    EXPECT_EQ(0, getActiveDM().getNumCommitted());
+    EXPECT_EQ(0, getActiveDM().getNumAborted());
+}
+
 INSTANTIATE_TEST_CASE_P(AllBucketTypes,
                         ActiveDurabilityMonitorTest,
                         STParameterizedBucketTest::allConfigValues(),