SERVER-34789: Using resume token from 'invalidate' notification for 'resumeAfter' should error

1 files changed, 40 insertions, 38 deletions

diff --git a/jstests/change_streams/metadata_notifications.js b/jstests/change_streams/metadata_notifications.js
index f1d4373606a..be892c114da 100644
--- a/jstests/change_streams/metadata_notifications.js
+++ b/jstests/change_streams/metadata_notifications.js
@@ -70,7 +70,7 @@
         {operationType: "drop"},
         {operationType: "invalidate"},
     ];
-    const changes = cst.assertNextChangesEqual(
+    let changes = cst.assertNextChangesEqual(
         {cursor: cursor, expectedChanges: expectedChanges, expectInvalidate: true});
     const resumeToken = changes[0]._id;
     const resumeTokenDrop = changes[3]._id;
@@ -89,33 +89,36 @@
     coll = assertCreateCollection(db, collName);
     assert.writeOK(coll.insert({_id: "after recreate"}));
 
-    // TODO SERVER-34789: The code below should throw an error. We exercise this behavior here to
-    // be sure that it doesn't crash the server, but the ability to resume a change stream using
-    // 'resumeAfter' with a resume token from an invalidate is a bug, not a feature.
-
     // Test resuming the change stream from the collection drop using 'resumeAfter'.
-    expectedChanges = [{
-        operationType: "insert",
-        ns: {db: db.getName(), coll: coll.getName()},
-        fullDocument: {_id: "after recreate"},
-        documentKey: {_id: "after recreate"}
-    }];
-    assertResumeExpected(
-        {coll: coll.getName(), spec: {resumeAfter: resumeTokenDrop}, expected: expectedChanges});
-
-    // Test resuming the change stream from the invalidate after the drop using 'resumeAfter'.
     assertResumeExpected({
         coll: coll.getName(),
-        spec: {resumeAfter: resumeTokenInvalidate},
-        expected: expectedChanges
+        spec: {resumeAfter: resumeTokenDrop},
+        expected: [{operationType: "invalidate"}]
     });
 
+    // Test resuming the change stream from the invalidate after the drop using 'resumeAfter'.
+    assert.commandFailedWithCode(db.runCommand({
+        aggregate: coll.getName(),
+        pipeline: [{$changeStream: {resumeAfter: resumeTokenInvalidate}}],
+        cursor: {},
+        collation: {locale: "simple"},
+    }),
+                                 ErrorCodes.InvalidResumeToken);
+
     // Test resuming the change stream from the collection drop using 'startAfter'.
-    assertResumeExpected(
-        {coll: coll.getName(), spec: {startAfter: resumeTokenDrop}, expected: expectedChanges});
+    assertResumeExpected({
+        coll: coll.getName(),
+        spec: {startAfter: resumeTokenDrop},
+        expected: [{operationType: "invalidate"}]
+    });
 
-    // Test resuming the change stream from the 'invalidate' notification using 'startAfter'. This
-    // is expected to behave identical to resuming from the drop.
+    // Test resuming the change stream from the 'invalidate' notification using 'startAfter'.
+    expectedChanges = [{
+        operationType: "insert",
+        ns: {db: db.getName(), coll: coll.getName()},
+        fullDocument: {_id: "after recreate"},
+        documentKey: {_id: "after recreate"}
+    }];
     assertResumeExpected({
         coll: coll.getName(),
         spec: {startAfter: resumeTokenInvalidate},
@@ -161,36 +164,35 @@
         coll = db[collName];
         assert.writeOK(coll.insert({_id: "after rename"}));
 
-        // TODO SERVER-34789: The code below should throw an error. We exercise this behavior here
-        // to be sure that it doesn't crash the server, but the ability to resume a change stream
-        // after an invalidate using 'resumeAfter' is a bug, not a feature.
-
         // Test resuming the change stream from the collection rename using 'resumeAfter'.
-        expectedChanges = [{
-            operationType: "insert",
-            ns: {db: db.getName(), coll: coll.getName()},
-            fullDocument: {_id: "after rename"},
-            documentKey: {_id: "after rename"}
-        }];
         assertResumeExpected({
             coll: coll.getName(),
             spec: {resumeAfter: resumeTokenRename},
-            expected: expectedChanges
+            expected: [{operationType: "invalidate"}]
         });
         // Test resuming the change stream from the invalidate after the rename using 'resumeAfter'.
-        assertResumeExpected({
-            coll: coll.getName(),
-            spec: {resumeAfter: resumeTokenInvalidate},
-            expected: expectedChanges
-        });
+        assert.commandFailedWithCode(db.runCommand({
+            aggregate: coll.getName(),
+            pipeline: [{$changeStream: {resumeAfter: resumeTokenInvalidate}}],
+            cursor: {},
+            collation: {locale: "simple"},
+        }),
+                                     ErrorCodes.InvalidResumeToken);
 
         // Test resuming the change stream from the rename using 'startAfter'.
         assertResumeExpected({
             coll: coll.getName(),
             spec: {startAfter: resumeTokenRename},
-            expected: expectedChanges
+            expected: [{operationType: "invalidate"}]
         });
+
         // Test resuming the change stream from the invalidate after the rename using 'startAfter'.
+        expectedChanges = [{
+            operationType: "insert",
+            ns: {db: db.getName(), coll: coll.getName()},
+            fullDocument: {_id: "after rename"},
+            documentKey: {_id: "after rename"}
+        }];
         assertResumeExpected({
             coll: coll.getName(),
             spec: {startAfter: resumeTokenInvalidate},