diff --git a/packages/mongo/asynchronous_cursor.js b/packages/mongo/asynchronous_cursor.js index 0c2668560a0..53f1c643f1e 100644 --- a/packages/mongo/asynchronous_cursor.js +++ b/packages/mongo/asynchronous_cursor.js @@ -44,11 +44,14 @@ export class AsynchronousCursor { } try { this._pendingNext = this._dbCursor.next(); - const result = await this._pendingNext; - this._pendingNext = null; - return result; + return await this._pendingNext; } catch (e) { - console.error(e); + // close() waits for a pending next(). One that fails while the cursor is closing ends the + // cursor, as close() intends; any other failure goes to the caller. + if (this._closing) { + return null; + } + throw e; } finally { this._pendingNext = null; } @@ -96,10 +99,10 @@ export class AsynchronousCursor { resolve(this.close()); }, timeoutMS); - // If the `_nextObjectPromise` returned first, cancel the timeout. - nextObjectPromise.finally(() => { - clearTimeout(timeoutId); - }); + // If the `_nextObjectPromise` settled first, cancel the timeout. Its rejection reaches the + // caller through the race below. + const cancel = () => clearTimeout(timeoutId); + nextObjectPromise.then(cancel, cancel); }); return Promise.race([nextObjectPromise, timeoutPromise]); diff --git a/packages/mongo/package.js b/packages/mongo/package.js index 3a1f2a68470..6f09c75b772 100644 --- a/packages/mongo/package.js +++ b/packages/mongo/package.js @@ -143,5 +143,6 @@ Package.onTest(function (api) { api.addFiles("tests/changestream_observe_driver_tests.js", "server"); api.addFiles("tests/oplog_v2_converter_tests.js", "server"); api.addFiles("tests/doc_fetcher_tests.js", "server"); + api.addFiles("tests/asynchronous_cursor_tests.js", "server"); api.addFiles("tests/collation_tests.js", ["client", "server"]); }); diff --git a/packages/mongo/tests/asynchronous_cursor_tests.js b/packages/mongo/tests/asynchronous_cursor_tests.js new file mode 100644 index 00000000000..b4d924e40bf --- /dev/null +++ b/packages/mongo/tests/asynchronous_cursor_tests.js @@ -0,0 +1,71 @@ +// A cursor read that fails rejects with the driver's error, so a caller can tell it from an empty +// result (https://github.com/meteor/meteor/issues/14794). + +Tinytest.addAsync('asynchronous cursor - fetchAsync rejects when the read fails', async test => { + const collection = new Mongo.Collection(`asyncCursorFetch${test.id}`); + await collection.insertAsync({ a: 1 }); + + await test.throwsAsync( + () => collection.find({ $foo: 1 }).fetchAsync(), + /unknown top level operator/ + ); +}); + +Tinytest.addAsync('asynchronous cursor - findOneAsync rejects when the read fails', async test => { + const collection = new Mongo.Collection(`asyncCursorFindOne${test.id}`); + await collection.insertAsync({ a: 1 }); + + await test.throwsAsync( + () => collection.findOneAsync({ $foo: 1 }), + /unknown top level operator/ + ); +}); + +Tinytest.addAsync('asynchronous cursor - forEachAsync rejects when the read fails', async test => { + const collection = new Mongo.Collection(`asyncCursorForEach${test.id}`); + await collection.insertAsync({ a: 1 }); + + await test.throwsAsync( + () => collection.find({ $foo: 1 }).forEachAsync(() => {}), + /unknown top level operator/ + ); +}); + +Tinytest.addAsync('asynchronous cursor - a read that fails while the cursor closes ends it', async test => { + let failNext; + const cursor = asynchronousCursorWith( + test, + () => new Promise((resolve, reject) => { failNext = reject; }) + ); + + const read = cursor._rawNextObjectPromise(); + cursor._closing = true; + failNext(new Error('cursor killed')); + + test.isTrue((await read) == null); +}); + +Tinytest.addAsync('asynchronous cursor - a timed read that fails rejects without an unhandled rejection', async test => { + const cursor = asynchronousCursorWith(test, () => Promise.reject(new Error('connection closed'))); + const unhandled = []; + const onUnhandled = reason => unhandled.push(reason); + process.on('unhandledRejection', onUnhandled); + + try { + await test.throwsAsync(() => cursor._nextObjectPromiseWithTimeout(10000), /connection closed/); + await new Promise(resolve => setTimeout(resolve, 50)); + + test.equal(unhandled.length, 0); + } finally { + process.off('unhandledRejection', onUnhandled); + } +}); + +// An AsynchronousCursor whose driver cursor answers next() with the given function. +function asynchronousCursorWith(test, next) { + const collection = new Mongo.Collection(`asyncCursorStub${test.id}`); + const meteorCursor = collection.find(); + const cursor = meteorCursor._mongo._createAsynchronousCursor(meteorCursor._cursorDescription); + cursor._dbCursor = { next, close() {}, rewind() {} }; + return cursor; +}