Copy chain Positions between first and second chain

If we do not do this then the nodeWriteSeqno of a replica that
existed and acked in the secondChain then was moved to the first
will be incorrect until it acks again. This bug is benign as we use
the ack count to determine if we should commit, not the
nodeWriteSeqno so we do not fail to commit anything. No monotonic
invariant exceptions are thrown as a new Position is created in the
second chain which effectively resets the monotonic nodeWriteSeqno
value.

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

Author: BenHuddleston

engines/ep/src/durability/active_durability_monitor.cc

@@ -77,6 +77,12 @@ struct ActiveDurabilityMonitor::State {
     void setReplicationTopology(const nlohmann::json& topology,
                                 CompletedQueue& toComplete);
 
+    /**
+     * Copy the Positions from the old chain to the new
+     */
+    void copyChainPositions(ReplicationChain& oldChain,
+                            ReplicationChain& newChain);
+
     void addSyncWrite(const void* cookie, queued_item item);
 
     /**
@@ -1323,20 +1329,20 @@ void ActiveDurabilityMonitor::State::setReplicationTopology(
         // Copy over the trackedWrites position for all nodes which still exist
         // in the new chain. This ensures that if we manually set the HPS on the
         // firstChain then the secondChain will also be correctly set.
-        for (const auto& node : firstChain->positions) {
-            auto newNode = newFirstChain->positions.find(node.first);
-            if (newNode != newFirstChain->positions.end()) {
-                newNode->second = node.second;
-            }
+        copyChainPositions(*firstChain, *newFirstChain);
+        if (newSecondChain) {
+            // This stage should never matter because we will find the node in
+            // the firstChain and return early from processSeqnoAck. Added for
+            // the sake of completeness.
+            // @TODO make iterators optional and remove this
+            copyChainPositions(*firstChain, *newSecondChain);
         }
     }
 
-    if (secondChain && newSecondChain) {
-        for (const auto& node : secondChain->positions) {
-            auto newNode = newSecondChain->positions.find(node.first);
-            if (newNode != newSecondChain->positions.end()) {
-                newNode->second = node.second;
-            }
+    if (secondChain) {
+        copyChainPositions(*secondChain, *newFirstChain);
+        if (newSecondChain) {
+            copyChainPositions(*secondChain, *newSecondChain);
         }
     }
 
@@ -1383,6 +1389,16 @@ void ActiveDurabilityMonitor::State::setReplicationTopology(
     cleanUpTrackedWritesPostTopologyChange(toComplete);
 }
 
+void ActiveDurabilityMonitor::State::copyChainPositions(
+        ReplicationChain& oldChain, ReplicationChain& newChain) {
+    for (const auto& node : oldChain.positions) {
+        auto newNode = newChain.positions.find(node.first);
+        if (newNode != newChain.positions.end()) {
+            newNode->second = node.second;
+        }
+    }
+}
+
 void ActiveDurabilityMonitor::State::performQueuedAckForChain(
         const DurabilityMonitor::ReplicationChain& chain,
         CompletedQueue& toCommit) {

engines/ep/tests/module_tests/vbucket_durability_test.cc

@@ -1488,8 +1488,9 @@ void VBucketDurabilityTest::testConvertPDMToADMWithNullTopology(
     testConvertPDMToADMWithNullTopologySetup(initialState, writes);
 
     // ns_server then sets the topology
-    vbucket->setState(vbucket_state_active,
-                      {{"topology", nlohmann::json::array({{active}})}});
+    vbucket->setState(
+            vbucket_state_active,
+            {{"topology", nlohmann::json::array({{active}, {active}})}});
 
     auto& adm = VBucketTestIntrospector::public_getActiveDM(*vbucket);
 
@@ -2130,6 +2131,71 @@ TEST_P(VBucketDurabilityTest, ActiveDM_DoubleSetVBState) {
     }
 }
 
+TEST_P(EPVBucketDurabilityTest,
+       ActiveDM_SecondChainNodeWriteSeqnoMaintainedOntopologyChange) {
+    // Set a topology with 2 chains - i.e. approaching the end of a replica swap
+    // rebalance
+    nlohmann::json topology({{"topology",
+                              nlohmann::json::array({{active, replica1},
+                                                     {active, replica2}})}});
+    vbucket->setState(vbucket_state_active, topology);
+
+    // Setup: queue two Prepares into the ADM
+    auto& adm = VBucketTestIntrospector::public_getActiveDM(*vbucket);
+    ASSERT_EQ(0, adm.getNumTracked());
+    using namespace cb::durability;
+    const std::vector<SyncWriteSpec> seqnos{
+            {1, false, Level::PersistToMajority}, 2};
+    testAddPrepare(seqnos);
+    ASSERT_EQ(seqnos.size(), adm.getNumTracked());
+    // checkForCommit will be called after every normal vBucket op and will
+    // set the HPS for us
+    adm.checkForCommit();
+
+    // Prepare at seqno 1 requires persistence so we have not moved HPS
+    ASSERT_EQ(0, adm.getHighPreparedSeqno());
+    ASSERT_EQ(2, adm.getNumTracked());
+
+    // We have acked replica2
+    adm.seqnoAckReceived(replica2, 2);
+    ASSERT_EQ(2, adm.getNodeWriteSeqno(replica2));
+    ASSERT_EQ(2, adm.getNumTracked());
+
+    // Complete the "rebalance" by settings the topology to a single chain with
+    // the new replica
+    topology = nlohmann::json(
+            {{"topology", nlohmann::json::array({{active, replica2}})}});
+    vbucket->setState(vbucket_state_active, topology);
+
+    // Nothing has yet been committed because the active must persist
+    EXPECT_EQ(2, adm.getNumTracked());
+    EXPECT_EQ(0, adm.getHighPreparedSeqno());
+
+    // The node write seqno is transferred in the topology change
+    EXPECT_EQ(2, adm.getNodeWriteSeqno(replica2));
+
+    // Notify persistence and commit our prepares
+    vbucket->setPersistenceSeqno(2);
+    adm.notifyLocalPersistence();
+    EXPECT_EQ(0, adm.getNumTracked());
+
+    // Client should be notified.
+    EXPECT_EQ(SWCompleteTrace(2 /*count*/, cookie, ENGINE_SUCCESS),
+              swCompleteTrace);
+
+    // After commit() check that the keys are now accessible and appear as
+    // committed.
+    for (const auto& spec : seqnos) {
+        auto key = makeStoredDocKey("key"s + std::to_string(spec.seqno));
+        auto result = vbucket->fetchValidValue(WantsDeleted::No,
+                                               TrackReference::No,
+                                               QueueExpired::No,
+                                               vbucket->lockCollections(key));
+        ASSERT_TRUE(result.storedValue);
+        EXPECT_TRUE(result.storedValue->isCommitted());
+    }
+}
+
 TEST_P(VBucketDurabilityTest, IgnoreAckAtTakeoverDead) {
     // Queue some Prepares into the PDM
     const auto& adm = VBucketTestIntrospector::public_getActiveDM(*vbucket);