SERVER-29825 Do not allow rename from unreplicated to replicated DB or vice versa

(cherry picked from commit f68d6091bad5a2747cc09c743b8c2c7b4e42a65a)

3 files changed, 75 insertions, 3 deletions

diff --git a/jstests/replsets/rename_collection_between_unrepl_and_repl.js b/jstests/replsets/rename_collection_between_unrepl_and_repl.js
new file mode 100644
index 00000000000..bba8ca8299b
--- /dev/null
+++ b/jstests/replsets/rename_collection_between_unrepl_and_repl.js
@@ -0,0 +1,44 @@
+/**
+ * Tests that renameCollection disallows renaming between an unreplicated and a replicated
+ * namespace in both directions. Unreplicated collections are unique to the nodes that own
+ * them, so allowing such renames would introduce consistency risks. This test uses the
+ * 'local' database for unreplicated namespaces.
+ */
+
+(function() {
+    "use strict";
+
+    const name = "rename_collection_between_unrepl_and_repl";
+    const rst = new ReplSetTest({"name": name, "nodes": 1});
+    rst.startSet();
+    rst.initiate();
+    const primary = rst.getPrimary();
+
+    /**
+     * Part 1: Attempt to rename from a replicated to an unreplicated namespace.
+     */
+    let sourceNs = "somedb.replicated";
+    let targetNs = "local.unreplicated";
+
+    // Ensure that the source collection exists.
+    assert.writeOK(primary.getCollection(sourceNs).insert({"fromRepl": "toUnrepl"}));
+
+    assert.commandFailedWithCode(
+        primary.adminCommand({"renameCollection": sourceNs, "to": targetNs}),
+        ErrorCodes.IllegalOperation);
+
+    /**
+     * Part 2: Attempt to rename from an unreplicated to a replicated namespace.
+     */
+    sourceNs = "local.alsoUnreplicated";
+    targetNs = "somedb.alsoReplicated";
+
+    // Ensure that the source collection exists.
+    assert.writeOK(primary.getCollection(sourceNs).insert({"fromUnrepl": "toRepl"}));
+
+    assert.commandFailedWithCode(
+        primary.adminCommand({"renameCollection": sourceNs, "to": targetNs}),
+        ErrorCodes.IllegalOperation);
+
+    rst.stopSet();
+})();
diff --git a/src/mongo/db/catalog/rename_collection.cpp b/src/mongo/db/catalog/rename_collection.cpp
index 13ddf962377..c4550272b20 100644
--- a/src/mongo/db/catalog/rename_collection.cpp
+++ b/src/mongo/db/catalog/rename_collection.cpp
@@ -143,8 +143,9 @@ Status renameCollectionCommon(OperationContext* opCtx,
     boost::optional<OldClientContext> ctx;
     ctx.emplace(opCtx, source.ns());
 
-    bool userInitiatedWritesAndNotPrimary = opCtx->writesAreReplicated() &&
-        !repl::getGlobalReplicationCoordinator()->canAcceptWritesFor(opCtx, source);
+    auto replCoord = repl::ReplicationCoordinator::get(opCtx);
+    bool userInitiatedWritesAndNotPrimary =
+        opCtx->writesAreReplicated() && !replCoord->canAcceptWritesFor(opCtx, source);
 
     if (userInitiatedWritesAndNotPrimary) {
         return Status(ErrorCodes::NotMaster,
@@ -167,6 +168,14 @@ Status renameCollectionCommon(OperationContext* opCtx,
         return {ErrorCodes::IllegalOperation, "source namespace cannot be sharded"};
     }
 
+    // Disallow renaming from a replicated to an unreplicated collection or vice versa.
+    auto sourceIsUnreplicated = replCoord->isOplogDisabledFor(opCtx, source);
+    auto targetIsUnreplicated = replCoord->isOplogDisabledFor(opCtx, target);
+    if (sourceIsUnreplicated != targetIsUnreplicated) {
+        return {ErrorCodes::IllegalOperation,
+                "Cannot rename collections between a replicated and an unreplicated database"};
+    }
+
     // Ensure that collection name does not exceed maximum length.
     // Ensure that index names do not push the length over the max.
     std::string::size_type longestIndexNameLength =
@@ -260,7 +269,6 @@ Status renameCollectionCommon(OperationContext* opCtx,
             // If this rename collection is replicated, check for long index names in the target
             // collection that may exceed the MMAPv1 namespace limit when the target collection
             // is renamed with a drop-pending namespace.
-            auto replCoord = repl::ReplicationCoordinator::get(opCtx);
             auto isOplogDisabledForNamespace = replCoord->isOplogDisabledFor(opCtx, target);
             auto isMasterSlave =
                 repl::ReplicationCoordinator::modeMasterSlave == replCoord->getReplicationMode();
diff --git a/src/mongo/db/catalog/rename_collection_test.cpp b/src/mongo/db/catalog/rename_collection_test.cpp
index 7af3899671c..060f377b0a8 100644
--- a/src/mongo/db/catalog/rename_collection_test.cpp
+++ b/src/mongo/db/catalog/rename_collection_test.cpp
@@ -1097,4 +1097,24 @@ TEST_F(RenameCollectionTest,
     ASSERT_TRUE(_opObserver->onInsertsIsGlobalWriteLockExclusive);
 }
 
+TEST_F(RenameCollectionTest, FailRenameCollectionFromReplicatedToUnreplicatedDB) {
+    NamespaceString sourceNss("foo.isReplicated");
+    NamespaceString targetNss("local.isUnreplicated");
+
+    _createCollection(_opCtx.get(), sourceNss);
+
+    ASSERT_EQUALS(ErrorCodes::IllegalOperation,
+                  renameCollection(_opCtx.get(), sourceNss, targetNss, {}));
+}
+
+TEST_F(RenameCollectionTest, FailRenameCollectionFromUnreplicatedToReplicatedDB) {
+    NamespaceString sourceNss("foo.system.profile");
+    NamespaceString targetNss("foo.bar");
+
+    _createCollection(_opCtx.get(), sourceNss);
+
+    ASSERT_EQUALS(ErrorCodes::IllegalOperation,
+                  renameCollection(_opCtx.get(), sourceNss, targetNss, {}));
+}
+
 }  // namespace