SERVER-53520 Assert recipients own all their docs in ReshardingTest.

2 files changed, 143 insertions, 21 deletions

diff --git a/jstests/sharding/libs/resharding_test_fixture.js b/jstests/sharding/libs/resharding_test_fixture.js
index d445b42fdf2..213ce387152 100644
--- a/jstests/sharding/libs/resharding_test_fixture.js
+++ b/jstests/sharding/libs/resharding_test_fixture.js
@@ -62,6 +62,8 @@ var ReshardingTest = class {
         /** @private */
         this._pauseCoordinatorInSteadyStateFailpoint = undefined;
         /** @private */
+        this._pauseCoordinatorBeforeDecisionPersistedFailpoint = undefined;
+        /** @private */
         this._reshardingThread = undefined;
         /** @private */
         this._isReshardingActive = false;
@@ -182,8 +184,11 @@ var ReshardingTest = class {
 
         this._newShardKey = Object.assign({}, newShardKeyPattern);
 
-        this._pauseCoordinatorInSteadyStateFailpoint = configureFailPoint(
-            this._st.configRS.getPrimary(), "reshardingPauseCoordinatorInSteadyState");
+        const configPrimary = this._st.configRS.getPrimary();
+        this._pauseCoordinatorInSteadyStateFailpoint =
+            configureFailPoint(configPrimary, "reshardingPauseCoordinatorInSteadyState");
+        this._pauseCoordinatorBeforeDecisionPersistedFailpoint =
+            configureFailPoint(configPrimary, "reshardingPauseCoordinatorBeforeDecisionPersisted");
 
         const commandDoneSignal = new CountDownLatch(1);
 
@@ -276,14 +281,18 @@ var ReshardingTest = class {
         try {
             fn();
         } catch (duringReshardingError) {
-            try {
-                this._pauseCoordinatorInSteadyStateFailpoint.off();
-            } catch (disableFailpointError) {
-                print(`Ignoring error from disabling the resharding coordinator failpoint: ${
-                    tojson(disableFailpointError)}`);
-
-                print("The config server primary and the mongo shell along with it are expected" +
-                      " to hang due to the resharding coordinator being left uninterrupted");
+            for (const fp of [this._pauseCoordinatorInSteadyStateFailpoint,
+                              this._pauseCoordinatorBeforeDecisionPersistedFailpoint]) {
+                try {
+                    fp.off();
+                } catch (disableFailpointError) {
+                    print(`Ignoring error from disabling the resharding coordinator failpoint: ${
+                        tojson(disableFailpointError)}`);
+
+                    print(
+                        "The config server primary and the mongo shell along with it are expected" +
+                        " to hang due to the resharding coordinator being left uninterrupted");
+                }
             }
 
             try {
@@ -294,6 +303,8 @@ var ReshardingTest = class {
                 } catch (joinError) {
                     print(`Ignoring error from the resharding thread: ${tojson(joinError)}`);
                 }
+
+                this._isReshardingActive = false;
             } catch (killOpError) {
                 print(`Ignoring error from sending killOp to the reshardCollection command: ${
                     tojson(killOpError)}`);
@@ -323,20 +334,18 @@ var ReshardingTest = class {
                 // wait for all of the recipient shards to have applied through all of the oplog
                 // entries from all of the donor shards.
                 this._pauseCoordinatorInSteadyStateFailpoint.wait();
-                const pauseCoordinatorBeforeDecisionPersistedFailpoint =
-                    configureFailPoint(this._pauseCoordinatorInSteadyStateFailpoint.conn,
-                                       "reshardingPauseCoordinatorBeforeDecisionPersisted");
-
                 this._pauseCoordinatorInSteadyStateFailpoint.off();
-                pauseCoordinatorBeforeDecisionPersistedFailpoint.wait();
+                this._pauseCoordinatorBeforeDecisionPersistedFailpoint.wait();
 
                 this._checkConsistency();
+                this._checkDocumentOwnership();
 
-                pauseCoordinatorBeforeDecisionPersistedFailpoint.off();
+                this._pauseCoordinatorBeforeDecisionPersistedFailpoint.off();
             });
         } else {
             this._callFunctionSafely(() => {
                 this._pauseCoordinatorInSteadyStateFailpoint.off();
+                this._pauseCoordinatorBeforeDecisionPersistedFailpoint.off();
             });
         }
 
@@ -364,11 +373,48 @@ var ReshardingTest = class {
             };
         })(DataConsistencyChecker.getDiff(nsCursor, tempNsCursor));
 
-        assert.eq(diff, {
-            docsWithDifferentContents: [],
-            docsExtraAfterResharding: [],
-            docsMissingAfterResharding: [],
-        });
+        assert.eq(diff,
+                  {
+                      docsWithDifferentContents: [],
+                      docsExtraAfterResharding: [],
+                      docsMissingAfterResharding: [],
+                  },
+                  "existing sharded collection and temporary resharding collection had different" +
+                      " contents");
+    }
+
+    /** @private */
+    _checkDocumentOwnership() {
+        // The "available" read concern level won't perform any ownership filtering. Any documents
+        // which were copied by a recipient shard that are actually owned by a different recipient
+        // shard would appear as extra documents.
+        const tempColl = this._st.s.getCollection(this._tempNs);
+        const localReadCursor = tempColl.find().sort({_id: 1});
+        // tempColl.find().readConcern("available") would be an error when the mongo shell is
+        // started with --readMode=legacy. We call runCommand() directly to avoid needing to tag
+        // every test which uses ReshardingTest with "requires_find_command".
+        const availableReadCursor =
+            new DBCommandCursor(tempColl.getDB(), assert.commandWorked(tempColl.runCommand("find", {
+                sort: {_id: 1},
+                readConcern: {level: "available"},
+            })));
+
+        const diff = ((diff) => {
+            return {
+                docsWithDifferentContents: diff.docsWithDifferentContents.map(
+                    ({first, second}) => ({local: first, available: second})),
+                docsFoundUnownedWithReadAvailable: diff.docsMissingOnFirst,
+                docsNotFoundWithReadAvailable: diff.docsMissingOnSecond,
+            };
+        })(DataConsistencyChecker.getDiff(localReadCursor, availableReadCursor));
+
+        assert.eq(diff,
+                  {
+                      docsWithDifferentContents: [],
+                      docsFoundUnownedWithReadAvailable: [],
+                      docsNotFoundWithReadAvailable: [],
+                  },
+                  "temporary resharding collection had unowned documents");
     }
 
     /** @private */
diff --git a/jstests/sharding/test_resharding_test_fixture_detects_unowned_docs.js b/jstests/sharding/test_resharding_test_fixture_detects_unowned_docs.js
new file mode 100644
index 00000000000..e2eb940aebb
--- /dev/null
+++ b/jstests/sharding/test_resharding_test_fixture_detects_unowned_docs.js
@@ -0,0 +1,76 @@
+/**
+ * Test for the ReshardingTest fixture itself.
+ *
+ * Verifies that an exception is thrown if a recipient shard has a document it doesn't actually own.
+ *
+ * @tags: [
+ *   requires_fcv_49,
+ *   uses_atclustertime,
+ * ]
+ */
+(function() {
+"use strict";
+
+load('jstests/libs/discover_topology.js');
+load("jstests/sharding/libs/resharding_test_fixture.js");
+
+const reshardingTest = new ReshardingTest({numDonors: 1, numRecipients: 2});
+reshardingTest.setup();
+
+const ns = "reshardingDb.coll";
+const donorShardNames = reshardingTest.donorShardNames;
+const sourceCollection = reshardingTest.createShardedCollection({
+    ns,
+    shardKeyPattern: {oldKey: 1},
+    chunks: [{min: {oldKey: MinKey}, max: {oldKey: MaxKey}, shard: donorShardNames[0]}],
+});
+
+// Perform some inserts before resharding starts so there's data to clone.
+assert.commandWorked(sourceCollection.insert([
+    {_id: "moves to recipient0", oldKey: -10, newKey: -10},
+    {_id: "moves to recipient1", oldKey: 10, newKey: 10},
+]));
+
+const mongos = sourceCollection.getMongo();
+const topology = DiscoverTopology.findConnectedNodes(mongos);
+
+const recipientShardNames = reshardingTest.recipientShardNames;
+const recipient0 = new Mongo(topology.shards[recipientShardNames[0]].primary);
+
+const err = assert.throws(() => {
+    reshardingTest.withReshardingInBackground(
+        {
+            newShardKeyPattern: {newKey: 1},
+            newChunks: [
+                {min: {newKey: MinKey}, max: {newKey: 0}, shard: recipientShardNames[0]},
+                {min: {newKey: 0}, max: {newKey: MaxKey}, shard: recipientShardNames[1]},
+            ],
+        },
+        (tempNs) => {
+            // Wait for the recipients to have finished cloning so the temporary resharding
+            // collection is known to exist.
+            assert.soon(() => {
+                const coordinatorDoc =
+                    mongos.getCollection("config.reshardingOperations").findOne();
+                return coordinatorDoc !== null && coordinatorDoc.state === "applying";
+            });
+
+            // Insert a document directly into recipient0 that is truly owned by recipient1.
+            const tempColl = recipient0.getCollection(tempNs);
+            assert.commandWorked(
+                tempColl.insert({_id: "unowned by recipient0", oldKey: 10, newKey: 10}));
+        });
+});
+
+assert(/temporary resharding collection had unowned documents/.test(err.message), err);
+
+// The ReshardingTest fixture will have interrupted the reshardCollection command on mongos so the
+// JavaScript thread running the command can be joined. The resharding operation is still active on
+// config server so we must manually wait for it to complete.
+assert.soon(() => {
+    const coordinatorDoc = mongos.getCollection("config.reshardingOperations").findOne();
+    return coordinatorDoc === null;
+});
+
+reshardingTest.teardown();
+})();