SERVER-44939 allow replication across config versions

5 files changed, 117 insertions, 60 deletions

diff --git a/src/mongo/db/repl/replication_coordinator_impl.cpp b/src/mongo/db/repl/replication_coordinator_impl.cpp
index 6be06869253..b14c37585ca 100644
--- a/src/mongo/db/repl/replication_coordinator_impl.cpp
+++ b/src/mongo/db/repl/replication_coordinator_impl.cpp
@@ -2610,8 +2610,12 @@ void ReplicationCoordinatorImpl::cancelAndRescheduleElectionTimeout() {
 
 EventHandle ReplicationCoordinatorImpl::_processReplSetMetadata_inlock(
     const rpc::ReplSetMetadata& replMetadata) {
-    if (replMetadata.getConfigVersion() != _rsConfig.getConfigVersion()) {
-        return EventHandle();
+    // If we're in FCV 4.4, allow metadata updates between config versions.
+    if (!serverGlobalParams.featureCompatibility.isVersion(
+            ServerGlobalParams::FeatureCompatibility::Version::kFullyUpgradedTo44)) {
+        if (replMetadata.getConfigVersion() != _rsConfig.getConfigVersion()) {
+            return EventHandle();
+        }
     }
     return _updateTerm_inlock(replMetadata.getTerm());
 }
diff --git a/src/mongo/db/repl/replication_coordinator_impl_test.cpp b/src/mongo/db/repl/replication_coordinator_impl_test.cpp
index 71c709f0727..15a6b44532d 100644
--- a/src/mongo/db/repl/replication_coordinator_impl_test.cpp
+++ b/src/mongo/db/repl/replication_coordinator_impl_test.cpp
@@ -3592,7 +3592,7 @@ TEST_F(ReplCoordTest, DoNotProcessSelfWhenUpdatePositionContainsInfoAboutSelf) {
                   getReplCoord()->awaitReplication(opCtx.get(), time2, writeConcern).status);
 }
 
-TEST_F(ReplCoordTest, DoNotProcessUpdatePositionWhenItsConfigVersionIsIncorrect) {
+TEST_F(ReplCoordTest, ProcessUpdatePositionWhenItsConfigVersionIsDifferent) {
     assertStartSuccess(BSON("_id"
                             << "mySet"
                             << "version" << 2 << "members"
@@ -3601,10 +3601,7 @@ TEST_F(ReplCoordTest, DoNotProcessUpdatePositionWhenItsConfigVersionIsIncorrect)
                                                << "_id" << 0)
                                           << BSON("host"
                                                   << "node2:12345"
-                                                  << "_id" << 1)
-                                          << BSON("host"
-                                                  << "node3:12345"
-                                                  << "_id" << 2))),
+                                                  << "_id" << 1))),
                        HostAndPort("node1", 12345));
     ASSERT_OK(getReplCoord()->setFollowerMode(MemberState::RS_SECONDARY));
     replCoordSetMyLastAppliedOpTime(OpTimeWithTermOne(100, 1), Date_t() + Seconds(100));
@@ -3620,28 +3617,27 @@ TEST_F(ReplCoordTest, DoNotProcessUpdatePositionWhenItsConfigVersionIsIncorrect)
     writeConcern.wTimeout = WriteConcernOptions::kNoWaiting;
     writeConcern.wNumNodes = 1;
 
-    // receive updatePosition with incorrect config version
+    // receive updatePosition with a different config version, 3
+    replCoordSetMyLastAppliedOpTime(time2, Date_t() + Seconds(100));
+    replCoordSetMyLastDurableOpTime(time2, Date_t() + Seconds(100));
+    auto updatePositionConfigVersion = 3;
     UpdatePositionArgs args;
-    ASSERT_OK(args.initialize(
-        BSON(UpdatePositionArgs::kCommandFieldName
-             << 1 << UpdatePositionArgs::kUpdateArrayFieldName
-             << BSON_ARRAY(BSON(UpdatePositionArgs::kConfigVersionFieldName
-                                << 3 << UpdatePositionArgs::kMemberIdFieldName << 1
-                                << UpdatePositionArgs::kDurableOpTimeFieldName << time2.toBSON()
-                                << UpdatePositionArgs::kDurableWallTimeFieldName
-                                << Date_t() + Seconds(time2.getSecs())
-                                << UpdatePositionArgs::kAppliedOpTimeFieldName << time2.toBSON()
-                                << UpdatePositionArgs::kAppliedWallTimeFieldName
-                                << Date_t() + Seconds(time2.getSecs()))))));
+    ASSERT_OK(args.initialize(BSON(
+        UpdatePositionArgs::kCommandFieldName
+        << 1 << UpdatePositionArgs::kUpdateArrayFieldName
+        << BSON_ARRAY(BSON(UpdatePositionArgs::kConfigVersionFieldName
+                           << updatePositionConfigVersion << UpdatePositionArgs::kMemberIdFieldName
+                           << 1 << UpdatePositionArgs::kDurableOpTimeFieldName << time2.toBSON()
+                           << UpdatePositionArgs::kDurableWallTimeFieldName
+                           << Date_t() + Seconds(time2.getSecs())
+                           << UpdatePositionArgs::kAppliedOpTimeFieldName << time2.toBSON()
+                           << UpdatePositionArgs::kAppliedWallTimeFieldName
+                           << Date_t() + Seconds(time2.getSecs()))))));
 
     auto opCtx = makeOperationContext();
 
-
-    long long cfgver;
-    ASSERT_EQUALS(ErrorCodes::InvalidReplicaSetConfig,
-                  getReplCoord()->processReplSetUpdatePosition(args, &cfgver));
-    ASSERT_EQUALS(ErrorCodes::WriteConcernFailed,
-                  getReplCoord()->awaitReplication(opCtx.get(), time2, writeConcern).status);
+    ASSERT_OK(getReplCoord()->processReplSetUpdatePosition(args, nullptr));
+    ASSERT_OK(getReplCoord()->awaitReplication(opCtx.get(), time2, writeConcern).status);
 }
 
 TEST_F(ReplCoordTest, DoNotProcessUpdatePositionOfMembersWhoseIdsAreNotInTheConfig) {
@@ -5023,7 +5019,7 @@ TEST_F(ReplCoordTest, WaitUntilOpTimeforReadRejectsUnsupportedMajorityReadConcer
     ASSERT_OK(status);
 }
 
-TEST_F(ReplCoordTest, IgnoreTheContentsOfMetadataWhenItsConfigVersionDoesNotMatchOurs) {
+TEST_F(ReplCoordTest, DoNotIgnoreTheContentsOfMetadataWhenItsConfigVersionDoesNotMatchOurs) {
     // Ensure that we do not process ReplSetMetadata when ConfigVersions do not match.
     assertStartSuccess(BSON("_id"
                             << "mySet"
@@ -5041,25 +5037,30 @@ TEST_F(ReplCoordTest, IgnoreTheContentsOfMetadataWhenItsConfigVersionDoesNotMatc
     ASSERT_EQUALS(OpTime(Timestamp(0, 0), 0), getReplCoord()->getLastCommittedOpTime());
 
     // lower configVersion
+    auto lowerConfigVersion = 1;
     StatusWith<rpc::ReplSetMetadata> metadata = rpc::ReplSetMetadata::readFromMetadata(BSON(
-        rpc::kReplSetMetadataFieldName << BSON(
-            "lastOpCommitted" << BSON("ts" << Timestamp(10, 0) << "t" << 2) << "lastCommittedWall"
-                              << Date_t() + Seconds(100) << "lastOpVisible"
-                              << BSON("ts" << Timestamp(10, 0) << "t" << 2) << "configVersion" << 1
-                              << "primaryIndex" << 2 << "term" << 2 << "syncSourceIndex" << 1)));
+        rpc::kReplSetMetadataFieldName
+        << BSON("lastOpCommitted" << BSON("ts" << Timestamp(10, 0) << "t" << 2)
+                                  << "lastCommittedWall" << Date_t() + Seconds(100)
+                                  << "lastOpVisible" << BSON("ts" << Timestamp(10, 0) << "t" << 2)
+                                  << "configVersion" << lowerConfigVersion << "primaryIndex" << 2
+                                  << "term" << 2 << "syncSourceIndex" << 1)));
     getReplCoord()->processReplSetMetadata(metadata.getValue());
-    ASSERT_EQUALS(0, getReplCoord()->getTerm());
+    // term should advance
+    ASSERT_EQUALS(2, getReplCoord()->getTerm());
 
     // higher configVersion
-    StatusWith<rpc::ReplSetMetadata> metadata2 = rpc::ReplSetMetadata::readFromMetadata(
-        BSON(rpc::kReplSetMetadataFieldName
-             << BSON("lastOpCommitted"
-                     << BSON("ts" << Timestamp(10, 0) << "t" << 2) << "lastCommittedWall"
-                     << Date_t() + Seconds(100) << "lastOpVisible"
-                     << BSON("ts" << Timestamp(10, 0) << "t" << 2) << "configVersion" << 100
-                     << "primaryIndex" << 2 << "term" << 2 << "syncSourceIndex" << 1)));
+    auto higherConfigVersion = 100;
+    StatusWith<rpc::ReplSetMetadata> metadata2 = rpc::ReplSetMetadata::readFromMetadata(BSON(
+        rpc::kReplSetMetadataFieldName
+        << BSON("lastOpCommitted" << BSON("ts" << Timestamp(10, 0) << "t" << 2)
+                                  << "lastCommittedWall" << Date_t() + Seconds(100)
+                                  << "lastOpVisible" << BSON("ts" << Timestamp(10, 0) << "t" << 2)
+                                  << "configVersion" << higherConfigVersion << "primaryIndex" << 2
+                                  << "term" << 2 << "syncSourceIndex" << 1)));
     getReplCoord()->processReplSetMetadata(metadata2.getValue());
-    ASSERT_EQUALS(0, getReplCoord()->getTerm());
+    // term should advance
+    ASSERT_EQUALS(2, getReplCoord()->getTerm());
 }
 
 TEST_F(ReplCoordTest, UpdateLastCommittedOpTimeWhenTheLastCommittedOpTimeIsNewer) {
diff --git a/src/mongo/db/repl/topology_coordinator.cpp b/src/mongo/db/repl/topology_coordinator.cpp
index e1817bca636..770440e7c1c 100644
--- a/src/mongo/db/repl/topology_coordinator.cpp
+++ b/src/mongo/db/repl/topology_coordinator.cpp
@@ -1071,16 +1071,22 @@ StatusWith<bool> TopologyCoordinator::setLastOptime(const UpdatePositionArgs::Up
            << " has reached optime: " << args.appliedOpTime
            << " and is durable through: " << args.durableOpTime;
 
-    if (args.cfgver != _rsConfig.getConfigVersion()) {
-        std::string errmsg = str::stream()
-            << "Received replSetUpdatePosition for node with memberId " << memberId
-            << " whose config version of " << args.cfgver << " doesn't match our config version of "
-            << _rsConfig.getConfigVersion();
-        LOG(1) << errmsg;
-        *configVersion = _rsConfig.getConfigVersion();
-        return Status(ErrorCodes::InvalidReplicaSetConfig, errmsg);
-    }
-
+    // If we're in FCV 4.4, allow replSetUpdatePosition commands between config versions.
+    if (!serverGlobalParams.featureCompatibility.isVersion(
+            ServerGlobalParams::FeatureCompatibility::Version::kFullyUpgradedTo44)) {
+        if (args.cfgver != _rsConfig.getConfigVersion()) {
+            std::string errmsg = str::stream()
+                << "Received replSetUpdatePosition for node with memberId " << memberId
+                << " whose config version of " << args.cfgver
+                << " doesn't match our config version of " << _rsConfig.getConfigVersion();
+            LOG(1) << errmsg;
+            *configVersion = _rsConfig.getConfigVersion();
+            return Status(ErrorCodes::InvalidReplicaSetConfig, errmsg);
+        }
+    }
+
+    // While we can accept replSetUpdatePosition commands across config versions, we still do not
+    // allow receiving them from a node that is not in our config.
     auto* memberData = _findMemberDataByMemberId(memberId.getData());
     if (!memberData) {
         invariant(!_rsConfig.findMemberByID(memberId.getData()));
@@ -2612,13 +2618,20 @@ bool TopologyCoordinator::shouldChangeSyncSource(
         return true;
     }
 
-    if (replMetadata.getConfigVersion() != _rsConfig.getConfigVersion()) {
-        log() << "Choosing new sync source because the config version supplied by " << currentSource
-              << ", " << replMetadata.getConfigVersion() << ", does not match ours, "
-              << _rsConfig.getConfigVersion();
-        return true;
+    // If we're in FCV 4.4, allow data replication between config versions. Otherwise, change
+    // our sync source.
+    if (!serverGlobalParams.featureCompatibility.isVersion(
+            ServerGlobalParams::FeatureCompatibility::Version::kFullyUpgradedTo44)) {
+        if (replMetadata.getConfigVersion() != _rsConfig.getConfigVersion()) {
+            log() << "Choosing new sync source because the config version supplied by "
+                  << currentSource << ", " << replMetadata.getConfigVersion()
+                  << ", does not match ours, " << _rsConfig.getConfigVersion();
+            return true;
+        }
     }
 
+    // While we can allow data replication across config versions, we still do not allow syncing
+    // from a node that is not in our config.
     const int currentSourceIndex = _rsConfig.findMemberIndexByHostAndPort(currentSource);
     if (currentSourceIndex == -1) {
         log() << "Choosing new sync source because " << currentSource.toString()
diff --git a/src/mongo/db/repl/topology_coordinator_v1_test.cpp b/src/mongo/db/repl/topology_coordinator_v1_test.cpp
index 54519ab8390..4720e7b01b4 100644
--- a/src/mongo/db/repl/topology_coordinator_v1_test.cpp
+++ b/src/mongo/db/repl/topology_coordinator_v1_test.cpp
@@ -208,11 +208,13 @@ protected:
     // Only set visibleOpTime, primaryIndex and syncSourceIndex
     ReplSetMetadata makeReplSetMetadata(OpTime visibleOpTime = OpTime(),
                                         int primaryIndex = -1,
-                                        int syncSourceIndex = -1) {
+                                        int syncSourceIndex = -1,
+                                        long long configVersion = -1) {
+        auto configIn = (configVersion != -1) ? configVersion : _currentConfig.getConfigVersion();
         return ReplSetMetadata(_topo->getTerm(),
                                OpTimeAndWallTime(),
                                visibleOpTime,
-                               _currentConfig.getConfigVersion(),
+                               configIn,
                                OID(),
                                primaryIndex,
                                syncSourceIndex);
@@ -5627,6 +5629,44 @@ TEST_F(HeartbeatResponseTestV1, ShouldNotChangeSyncSourceWhenFresherMemberIsNotR
         HostAndPort("host2"), makeReplSetMetadata(syncSourceOpTime), boost::none, now()));
 }
 
+TEST_F(HeartbeatResponseTestV1, ShouldNotChangeSyncSourceIfSyncSourceHasDifferentConfigVersion) {
+    // In this test, the TopologyCoordinator should not tell us to change sync sources away from
+    // "host2" because it has a different config version.
+    OpTime election = OpTime();
+    // Our last op time applied must be behind host2, or we'll hit the case where we change
+    // sync sources due to the sync source being behind, without a sync source, and not primary.
+    OpTime lastOpTimeApplied = OpTime(Timestamp(400, 0), 0);
+    OpTime syncSourceOpTime = OpTime(Timestamp(400, 1), 0);
+
+    // Update the config version to 7.
+    updateConfig(BSON("_id"
+                      << "rs0"
+                      << "version" << 7 << "term" << 2LL << "members"
+                      << BSON_ARRAY(BSON("_id" << 0 << "host"
+                                               << "hself")
+                                    << BSON("_id" << 1 << "host"
+                                                  << "host2"))),
+                 0);
+
+    topoCoordSetMyLastAppliedOpTime(lastOpTimeApplied, Date_t(), false);
+    HeartbeatResponseAction nextAction = receiveUpHeartbeat(
+        HostAndPort("host2"), "rs0", MemberState::RS_SECONDARY, election, syncSourceOpTime);
+    ASSERT_NO_ACTION(nextAction.getAction());
+
+    // We should not want to change our sync source from "host2" even though it has a different
+    // config version.
+    ASSERT_FALSE(getTopoCoord().shouldChangeSyncSource(
+        HostAndPort("host2"),
+        makeReplSetMetadata(OpTime(), -1, -1, 8 /* different config version */),
+        makeOplogQueryMetadata(syncSourceOpTime),
+        now()));
+    ASSERT_FALSE(
+        getTopoCoord().shouldChangeSyncSource(HostAndPort("host2"),
+                                              makeReplSetMetadata(syncSourceOpTime, -1, -1, 8),
+                                              boost::none,
+                                              now()));
+}
+
 class HeartbeatResponseTestOneRetryV1 : public HeartbeatResponseTestV1 {
 public:
     virtual void setUp() {
diff --git a/src/mongo/db/server_options.h b/src/mongo/db/server_options.h
index 2a7796f9307..076f8decb87 100644
--- a/src/mongo/db/server_options.h
+++ b/src/mongo/db/server_options.h
@@ -228,9 +228,8 @@ struct ServerGlobalParams {
             return _version.store(version);
         }
 
-        bool isVersionUpgradingOrUpgraded() {
-            return (getVersion() == Version::kUpgradingTo44 ||
-                    getVersion() == Version::kFullyUpgradedTo44);
+        bool isVersion(Version version) {
+            return _version.load() == version;
         }
 
     private: