feat(NODE-7308): replace process.nextTick with queueMicrotask

Author: PavelSafronov

src/cmap/connection_pool.ts

@@ -231,7 +231,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
     this.mongoLogger = this.server.topology.client?.mongoLogger;
     this.component = 'connection';
 
-    process.nextTick(() => {
+    queueMicrotask(() => {
       this.emitAndLog(ConnectionPool.CONNECTION_POOL_CREATED, new ConnectionPoolCreatedEvent(this));
     });
   }
@@ -342,7 +342,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
     });
 
     this.waitQueue.push(waitQueueMember);
-    process.nextTick(() => this.processWaitQueue());
+    queueMicrotask(() => this.processWaitQueue());
 
     try {
       timeout?.throwIfExpired();
@@ -405,7 +405,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
       this.destroyConnection(connection, reason);
     }
 
-    process.nextTick(() => this.processWaitQueue());
+    queueMicrotask(() => this.processWaitQueue());
   }
 
   /**
@@ -461,7 +461,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
     }
 
     if (interruptInUseConnections) {
-      process.nextTick(() => this.interruptInUseConnections(oldGeneration));
+      queueMicrotask(() => this.interruptInUseConnections(oldGeneration));
     }
 
     this.processWaitQueue();
@@ -702,7 +702,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
       this.createConnection((err, connection) => {
         if (!err && connection) {
           this.connections.push(connection);
-          process.nextTick(() => this.processWaitQueue());
+          queueMicrotask(() => this.processWaitQueue());
         }
         if (this.poolState === PoolState.ready) {
           clearTimeout(this.minPoolSizeTimer);
@@ -809,7 +809,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
             waitQueueMember.resolve(connection);
           }
         }
-        process.nextTick(() => this.processWaitQueue());
+        queueMicrotask(() => this.processWaitQueue());
       });
     }
     this.processingWaitQueue = false;

src/gridfs/upload.ts

@@ -165,7 +165,7 @@ export class GridFSBucketWriteStream extends Writable {
         }
       );
     } else {
-      return process.nextTick(callback);
+      return queueMicrotask(callback);
     }
   }
 
@@ -188,7 +188,7 @@ export class GridFSBucketWriteStream extends Writable {
   /** @internal */
   override _final(callback: (error?: Error | null) => void): void {
     if (this.state.streamEnd) {
-      return process.nextTick(callback);
+      return queueMicrotask(callback);
     }
     this.state.streamEnd = true;
     writeRemnant(this, callback);
@@ -220,11 +220,11 @@ export class GridFSBucketWriteStream extends Writable {
 
 function handleError(stream: GridFSBucketWriteStream, error: Error, callback: Callback): void {
   if (stream.state.errored) {
-    process.nextTick(callback);
+    queueMicrotask(callback);
     return;
   }
   stream.state.errored = true;
-  process.nextTick(callback, error);
+  queueMicrotask(() => callback(error));
 }
 
 function createChunkDoc(filesId: ObjectId, n: number, data: Buffer): GridFSChunk {
@@ -283,7 +283,7 @@ async function checkChunksIndex(stream: GridFSBucketWriteStream): Promise<void>
 
 function checkDone(stream: GridFSBucketWriteStream, callback: Callback): void {
   if (stream.done) {
-    return process.nextTick(callback);
+    return queueMicrotask(callback);
   }
 
   if (stream.state.streamEnd && stream.state.outstandingRequests === 0 && !stream.state.errored) {
@@ -327,7 +327,7 @@ function checkDone(stream: GridFSBucketWriteStream, callback: Callback): void {
     return;
   }
 
-  process.nextTick(callback);
+  queueMicrotask(callback);
 }
 
 async function checkIndexes(stream: GridFSBucketWriteStream): Promise<void> {
@@ -425,7 +425,7 @@ function doWrite(
   if (stream.pos + inputBuf.length < stream.chunkSizeBytes) {
     inputBuf.copy(stream.bufToStore, stream.pos);
     stream.pos += inputBuf.length;
-    process.nextTick(callback);
+    queueMicrotask(callback);
     return;
   }
 
@@ -530,7 +530,7 @@ function writeRemnant(stream: GridFSBucketWriteStream, callback: Callback): void
 
 function isAborted(stream: GridFSBucketWriteStream, callback: Callback<void>): boolean {
   if (stream.state.aborted) {
-    process.nextTick(callback, new MongoAPIError('Stream has been aborted'));
+    queueMicrotask(() => callback(new MongoAPIError('Stream has been aborted')));
     return true;
   }
   return false;

src/sdam/monitor.ts

@@ -426,7 +426,7 @@ function checkServer(monitor: Monitor, callback: Callback<Document | null>) {
 function monitorServer(monitor: Monitor) {
   return (callback: Callback) => {
     if (monitor.s.state === STATE_MONITORING) {
-      process.nextTick(callback);
+      queueMicrotask(callback);
       return;
     }
     stateTransition(monitor, STATE_MONITORING);

src/sdam/server.ts

@@ -424,7 +424,7 @@ export class Server extends TypedEventEmitter<ServerEvents> {
               error.addErrorLabel(MongoErrorLabel.ResetPool);
             }
             markServerUnknown(this, error);
-            process.nextTick(() => this.requestCheck());
+            queueMicrotask(() => this.requestCheck());
           }
         }
       }

src/sdam/topology.ts

@@ -1075,7 +1075,7 @@ function processWaitQueue(topology: Topology) {
   if (topology.waitQueue.length > 0) {
     // ensure all server monitors attempt monitoring soon
     for (const [, server] of topology.s.servers) {
-      process.nextTick(function scheduleServerCheck() {
+      queueMicrotask(function scheduleServerCheck() {
         return server.requestCheck();
       });
     }

test/tools/runner/ee_checker.ts

@@ -9,7 +9,7 @@ events.EventEmitter = class RequireErrorListenerEventEmitter extends EventEmitte
     super(...args);
     const ctorCallSite = new Error('EventEmitter must add an error listener synchronously');
     ctorCallSite.stack;
-    process.nextTick(() => {
+    queueMicrotask(() => {
       const isChangeStream = this.constructor.name
         .toLowerCase()
         .includes('ChangeStream'.toLowerCase());

test/tools/utils.ts

@@ -166,7 +166,7 @@ export const sleep = promisify(setTimeout);
  * If you are using sinon fake timers, it can end up blocking queued IO from running
  * awaiting a nextTick call will allow the event loop to process Networking/FS callbacks
  */
-export const processTick = () => new Promise(resolve => process.nextTick(resolve));
+export const processTick = () => new Promise(resolve => queueMicrotask(resolve));
 
 export function getIndicesOfAuthInUrl(connectionString: string | string[]) {
   const doubleSlashIndex = connectionString.indexOf('//');

test/unit/assorted/polling_srv_records_for_mongos_discovery.prose.test.ts

@@ -130,7 +130,7 @@ describe('Polling Srv Records for Mongos Discovery', () => {
       expect(topology.description).to.have.property('type', TopologyType.Sharded);
       const servers = Array.from(topology.description.servers.keys());
       expect(servers).to.deep.equal(srvAddresses(recordSets[0]));
-      process.nextTick(() => srvPoller.trigger(recordSets[1]));
+      queueMicrotask(() => srvPoller.trigger(recordSets[1]));
 
       await once(topology, 'topologyDescriptionChanged');
 
@@ -296,7 +296,7 @@ describe('Polling Srv Records for Mongos Discovery', () => {
         const callback = args[args.length - 1] as (err: null, address: string, family: 4) => void;
 
         if (hostname.includes('test.mock.test.build.10gen.cc')) {
-          return process.nextTick(callback, null, '127.0.0.1', 4);
+          return queueMicrotask(() => callback(null, '127.0.0.1', 4));
         }
 
         const { wrappedMethod: lookup } = lookupStub;

test/unit/cmap/connect.test.ts

@@ -157,7 +157,7 @@ describe('Connect Tests', function () {
         const cancellationToken = new CancellationToken();
         // Make sure the cancel listener is added before emitting cancel
         cancellationToken.addListener('newListener', () => {
-          process.nextTick(() => {
+          queueMicrotask(() => {
             cancellationToken.emit('cancel');
           });
         });