test: check that event emitters have error listeners in the same tick…

… of their creation

.mocharc.json

@@ -4,7 +4,7 @@
     "source-map-support/register",
     "ts-node/register",
     "test/tools/runner/chai_addons.ts",
-    "test/tools/runner/hooks/unhandled_checker.ts"
+    "test/tools/runner/ee_checker.ts"
   ],
   "extension": [
     "js",

src/change_stream.ts

@@ -630,6 +630,11 @@ export class ChangeStream<
     options: ChangeStreamOptions = {}
   ) {
     super();
+    const ctorCallSite = new Error('ChangeStream must add an error listener synchronously');
+    ctorCallSite.stack;
+
+    const squishErrorListenerRequirement = () => null;
+    this.on('error', squishErrorListenerRequirement);
 
     this.pipeline = pipeline;
     this.options = { ...options };
@@ -667,7 +672,11 @@ export class ChangeStream<
     // Listen for any `change` listeners being added to ChangeStream
     this.on('newListener', eventName => {
       if (eventName === 'change' && this.cursor && this.listenerCount('change') === 0) {
+        this.off('error', squishErrorListenerRequirement);
         this._streamEvents(this.cursor);
+        process.nextTick(() => {
+          if (this.listenerCount('error') === 0) throw ctorCallSite;
+        });
       }
     });
 

src/cmap/connection.ts

@@ -229,6 +229,7 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
 
   constructor(stream: Stream, options: ConnectionOptions) {
     super();
+    this.on('error', () => null);
 
     this.socket = stream;
     this.id = options.id;

src/cmap/connection_pool.ts

@@ -200,6 +200,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
 
   constructor(server: Server, options: ConnectionPoolOptions) {
     super();
+    this.on('error', () => null);
 
     this.options = Object.freeze({
       connectionType: Connection,

src/cursor/abstract_cursor.ts

@@ -267,6 +267,7 @@ export abstract class AbstractCursor<
     options: AbstractCursorOptions & Abortable = {}
   ) {
     super();
+    this.on('error', () => null);
 
     if (!client.s.isMongoClient) {
       throw new MongoRuntimeError('Cursor must be constructed with MongoClient');

src/gridfs/index.ts

@@ -87,6 +87,7 @@ export class GridFSBucket extends TypedEventEmitter<GridFSBucketEvents> {
 
   constructor(db: Db, options?: GridFSBucketOptions) {
     super();
+    this.on('error', () => null);
     this.setMaxListeners(0);
     const privateOptions = resolveOptions(db, {
       ...DEFAULT_GRIDFS_BUCKET_OPTIONS,

src/mongo_client.ts

@@ -384,6 +384,7 @@ export class MongoClient extends TypedEventEmitter<MongoClientEvents> implements
 
   constructor(url: string, options?: MongoClientOptions) {
     super();
+    this.on('error', () => null);
 
     this.options = parseOptions(url, this, options);
 

src/mongo_types.ts

@@ -472,7 +472,12 @@ export class TypedEventEmitter<Events extends EventsDescription> extends EventEm
 }
 
 /** @public */
-export class CancellationToken extends TypedEventEmitter<{ cancel(): void }> {}
+export class CancellationToken extends TypedEventEmitter<{ cancel(): void }> {
+  constructor(...args: any[]) {
+    super(...args);
+    this.on('error', () => null);
+  }
+}
 
 /** @public */
 export type Abortable = {

src/sdam/monitor.ts

@@ -102,6 +102,7 @@ export class Monitor extends TypedEventEmitter<MonitorEvents> {
 
   constructor(server: Server, options: MonitorOptions) {
     super();
+    this.on('error', () => null);
 
     this.server = server;
     this.connection = null;

src/sdam/server.ts

@@ -142,6 +142,7 @@ export class Server extends TypedEventEmitter<ServerEvents> {
    */
   constructor(topology: Topology, description: ServerDescription, options: ServerOptions) {
     super();
+    this.on('error', () => null);
 
     this.serverApi = options.serverApi;
 

src/sdam/srv_polling.ts

@@ -49,6 +49,7 @@ export class SrvPoller extends TypedEventEmitter<SrvPollerEvents> {
 
   constructor(options: SrvPollerOptions) {
     super();
+    this.on('error', () => null);
 
     if (!options || !options.srvHost) {
       throw new MongoRuntimeError('Options for SrvPoller must exist and include srvHost');

src/sdam/topology.ts

@@ -248,6 +248,7 @@ export class Topology extends TypedEventEmitter<TopologyEvents> {
     options: TopologyOptions
   ) {
     super();
+    this.on('error', () => null);
 
     this.client = client;
     // Options should only be undefined in tests, MongoClient will always have defined options

src/sessions.ts

@@ -161,6 +161,7 @@ export class ClientSession
     clientOptions: MongoOptions
   ) {
     super();
+    this.on('error', () => null);
 
     if (client == null) {
       // TODO(NODE-3483)

test/integration/change-streams/change_stream.test.ts

@@ -736,6 +736,7 @@ describe('Change Streams', function () {
           // ChangeStream detects emitter usage via 'newListener' event
           // so this covers all emitter methods
         });
+        changeStream.on('error', () => null); // one must listen for errors if they use EE mode.
 
         await once(changeStream.cursor, 'init');
         expect(changeStream).to.have.property('mode', 'emitter');
@@ -971,7 +972,7 @@ describe('Change Streams', function () {
           { requires: { topology: '!single' } },
           async function () {
             changeStream = collection.watch([]);
-            changeStream.on('change', sinon.stub());
+            changeStream.on('change', sinon.stub()).on('error', () => null);
 
             try {
               // eslint-disable-next-line @typescript-eslint/no-unused-vars

test/integration/change-streams/change_streams.prose.test.ts

@@ -858,6 +858,7 @@ describe('Change Stream prose tests', function () {
         expect(err).to.not.exist;
         coll = client.db('integration_tests').collection('setupAfterTest');
         const changeStream = coll.watch();
+        changeStream.on('error', done);
         waitForStarted(changeStream, () => {
           coll.insertOne({ x: 1 }, { writeConcern: { w: 'majority', j: true } }, err => {
             expect(err).to.not.exist;
@@ -932,6 +933,7 @@ describe('Change Stream prose tests', function () {
         let events = [];
         client.on('commandStarted', e => recordEvent(events, e));
         const changeStream = coll.watch([], { startAfter });
+        changeStream.on('error', done);
         this.defer(() => changeStream.close());
 
         changeStream.on('change', change => {

test/mocha_mongodb.json

@@ -3,9 +3,9 @@
   "require": [
     "source-map-support/register",
     "ts-node/register",
+    "test/tools/runner/ee_checker.ts",
     "test/tools/runner/chai_addons.ts",
     "test/tools/runner/hooks/configuration.ts",
-    "test/tools/runner/hooks/unhandled_checker.ts",
     "test/tools/runner/hooks/leak_checker.ts",
     "test/tools/runner/hooks/legacy_crud_shims.ts"
   ],

test/tools/reporter/mongodb_reporter.js

@@ -250,7 +250,6 @@ class MongoDBMochaReporter extends mocha.reporters.Spec {
    */
   fail(test, error) {
     if (REPORT_TO_STDIO) console.log(chalk.red(`⨯ ${test.fullTitle()} -- ${error.message}`));
-    test.error = error;
   }
 
   /**

test/tools/runner/ee_checker.ts

@@ -0,0 +1,17 @@
+// eslint-disable-next-line @typescript-eslint/no-require-imports
+const events = require('events');
+
+const EventEmitter = events.EventEmitter;
+
+events.EventEmitter = class RequireErrorListenerEventEmitter extends EventEmitter {
+  constructor(...args) {
+    super(...args);
+    const ctorCallSite = new Error('EventEmitter must add an error listener synchronously');
+    ctorCallSite.stack;
+    process.nextTick(() => {
+      if (this.listenerCount('error') === 0) {
+        throw ctorCallSite;
+      }
+    });
+  }
+};

test/tools/unified-spec-runner/entities.ts

@@ -231,6 +231,8 @@ export class UnifiedMongoClient extends MongoClient {
       // TODO(NODE-5785): We need to increase the truncation length because signature.hash is a Buffer making hellos too long
       mongodbLogMaxDocumentLength: 1250
     } as any);
+
+    this.observedEventEmitter.on('error', () => null);
     this.logCollector = logCollector;
     this.observeSensitiveCommands = description.observeSensitiveCommands ?? false;
 

test/unit/sdam/srv_polling.test.ts

@@ -188,6 +188,10 @@ describe('Mongos SRV Polling', function () {
 
   describe('topology', function () {
     class FakeSrvPoller extends EventEmitter {
+      constructor() {
+        super();
+        this.on('error', () => null);
+      }
       start() {
         // ignore
       }