fix(shell-api): .watch() signature and behavior fixes MONGOSH-976 (#1107)

Modify the `.watch()` signature so that it “officially” allows
omitting the first argument.

As a drive-by fix, stop applying `.tryNext()` for cases in which
the change stream cursor starts from a specific point in time.
(Unfortunately, `.hasNext()` is not an option here, unlike
discussed with the Node.js driver team, because it is blocking
like `.next()` and unlike `.tryNext()`.)

Author: addaleax

packages/shell-api/src/collection.ts

@@ -1715,7 +1715,11 @@ export default class Collection extends ShellApiWithMongoClass {
   @topologies([Topologies.ReplSet, Topologies.Sharded])
   @apiVersions([1])
   @returnsPromise
-  async watch(pipeline: Document[] = [], options: ChangeStreamOptions = {}): Promise<ChangeStreamCursor> {
+  async watch(pipeline: Document[] | ChangeStreamOptions = [], options: ChangeStreamOptions = {}): Promise<ChangeStreamCursor> {
+    if (!Array.isArray(pipeline)) {
+      options = pipeline;
+      pipeline = [];
+    }
     this._emitCollectionApiCall('watch', { pipeline, options });
     const cursor = new ChangeStreamCursor(
       this._mongo._serviceProvider.watch(pipeline, {
@@ -1739,7 +1743,11 @@ export default class Collection extends ShellApiWithMongoClass {
     //   .watch() call and before the .tryNext() call could also have been
     //   observed before the .watch() call, i.e. there is a race condition
     //   here either way and we can use that to our advantage.
-    await cursor.tryNext();
+    // We only do this for change streams that do not specify from a specified
+    // point in time, i.e. start from the current time.
+    if (!options.resumeAfter && !options.startAfter && !options.startAtOperationTime) {
+      await cursor.tryNext();
+    }
     this._mongo._instanceState.currentCursor = cursor;
     return cursor;
   }

packages/shell-api/src/database.ts

@@ -1436,7 +1436,11 @@ export default class Database extends ShellApiWithMongoClass {
   @topologies([Topologies.ReplSet, Topologies.Sharded])
   @apiVersions([1])
   @returnsPromise
-  async watch(pipeline: Document[] = [], options: ChangeStreamOptions = {}): Promise<ChangeStreamCursor> {
+  async watch(pipeline: Document[] | ChangeStreamOptions = [], options: ChangeStreamOptions = {}): Promise<ChangeStreamCursor> {
+    if (!Array.isArray(pipeline)) {
+      options = pipeline;
+      pipeline = [];
+    }
     this._emitDatabaseApiCall('watch', { pipeline, options });
     const cursor = new ChangeStreamCursor(
       this._mongo._serviceProvider.watch(pipeline, {
@@ -1446,7 +1450,9 @@ export default class Database extends ShellApiWithMongoClass {
       this._name,
       this._mongo
     );
-    await cursor.tryNext(); // See comment in coll.watch().
+    if (!options.resumeAfter && !options.startAfter && !options.startAtOperationTime) {
+      await cursor.tryNext(); // See comment in coll.watch().
+    }
     this._mongo._instanceState.currentCursor = cursor;
     return cursor;
   }

packages/shell-api/src/mongo.ts

@@ -507,14 +507,20 @@ export default class Mongo extends ShellApiClass {
   @topologies([Topologies.ReplSet, Topologies.Sharded])
   @apiVersions([1])
   @returnsPromise
-  async watch(pipeline: Document[] = [], options: ChangeStreamOptions = {}): Promise<ChangeStreamCursor> {
+  async watch(pipeline: Document[] | ChangeStreamOptions = [], options: ChangeStreamOptions = {}): Promise<ChangeStreamCursor> {
+    if (!Array.isArray(pipeline)) {
+      options = pipeline;
+      pipeline = [];
+    }
     this._emitMongoApiCall('watch', { pipeline, options });
     const cursor = new ChangeStreamCursor(
       this._serviceProvider.watch(pipeline, options),
       redactURICredentials(this._uri),
       this
     );
-    await cursor.tryNext(); // See comment in coll.watch().
+    if (!options.resumeAfter && !options.startAfter && !options.startAtOperationTime) {
+      await cursor.tryNext(); // See comment in coll.watch().
+    }
     this._instanceState.currentCursor = cursor;
     return cursor;
   }

packages/shell-api/src/replica-set.spec.ts

@@ -857,6 +857,7 @@ describe('ReplicaSet', () => {
     let additionalServer: MongodSetup;
     let serviceProvider: CliServiceProvider;
     let instanceState: ShellInstanceState;
+    let db: Database;
     let rs: ReplicaSet;
 
     before(async function() {
@@ -873,7 +874,8 @@ describe('ReplicaSet', () => {
 
       serviceProvider = await CliServiceProvider.connect(`${await srv0.connectionString()}?directConnection=true`, {}, {}, new EventEmitter());
       instanceState = new ShellInstanceState(serviceProvider);
-      rs = new ReplicaSet(instanceState.currentDb);
+      db = instanceState.currentDb;
+      rs = new ReplicaSet(db);
 
       // check replset uninitialized
       try {
@@ -923,6 +925,33 @@ describe('ReplicaSet', () => {
         expect(Object.keys(result)).to.include('logSizeMB');
       });
     });
+
+    describe('watch', () => {
+      afterEach(async() => {
+        await db.dropDatabase();
+      });
+
+      it('allows watching changes as they happen', async() => {
+        const coll = db.getCollection('cstest');
+        const cs = await coll.watch();
+        await coll.insertOne({ i: 42 });
+        expect((await cs.next()).fullDocument.i).to.equal(42);
+      });
+
+      it('allow to resume watching changes as they happen', async() => {
+        const coll = db.getCollection('cstest');
+        const cs = await coll.watch();
+        await coll.insertOne({ i: 123 });
+        expect((await cs.next()).fullDocument.i).to.equal(123);
+        const token = cs.getResumeToken();
+        await coll.insertOne({ i: 456 });
+        expect((await cs.next()).fullDocument.i).to.equal(456);
+
+        const cs2 = await coll.watch({ resumeAfter: token });
+        expect((await cs2.next()).fullDocument.i).to.equal(456);
+      });
+    });
+
     describe('reconfig', () => {
       it('reconfig with one less secondary', async() => {
         const newcfg: Partial<ReplSetConfig> = {