Merge branch 'main' into NODE-6626/server-side

Author: aditi-khare-mongoDB

test/integration/node-specific/client_close.test.ts

@@ -121,22 +121,16 @@ describe('MongoClient.close() Integration', () => {
                 'monitor interval timer is cleaned up by client.close()',
                 metadata,
                 async function () {
-                  const run = async function ({
-                    MongoClient,
-                    uri,
-                    expect,
-                    getTimerCount,
-                    promiseWithResolvers
-                  }) {
+                  const run = async function ({ MongoClient, uri, expect, getTimerCount, once }) {
                     const heartbeatFrequencyMS = 2000;
                     const client = new MongoClient(uri, { heartbeatFrequencyMS });
-                    const { promise, resolve } = promiseWithResolvers();
-                    client.once('serverHeartbeatSucceeded', () => resolve());
+                    const willBeHeartbeatSucceeded = once(client, 'serverHeartbeatSucceeded');
                     await client.connect();
-                    await promise;
+                    await willBeHeartbeatSucceeded;
 
                     function monitorTimersExist(servers) {
                       for (const [, server] of servers) {
+                        // the current expected behavior is that timerId is set to undefined once it expires or is interrupted
                         if (server?.monitor.monitorId.timerId === undefined) {
                           return false;
                         }
@@ -160,19 +154,12 @@ describe('MongoClient.close() Integration', () => {
                 'the new monitor interval timer is cleaned up by client.close()',
                 metadata,
                 async () => {
-                  const run = async function ({
-                    MongoClient,
-                    expect,
-                    getTimerCount,
-                    promiseWithResolvers
-                  }) {
+                  const run = async function ({ MongoClient, expect, getTimerCount, once }) {
                     const heartbeatFrequencyMS = 2000;
                     const client = new MongoClient('mongodb://fakeUri', { heartbeatFrequencyMS });
-                    const { promise, resolve } = promiseWithResolvers();
-                    client.once('serverHeartbeatFailed', () => resolve());
+                    const willBeHeartbeatFailed = once(client, 'serverHeartbeatFailed');
                     client.connect();
-                    await promise;
-
+                    await willBeHeartbeatFailed;
                     function getMonitorTimer(servers) {
                       for (const [, server] of servers) {
                         return server?.monitor.monitorId.timerId;
@@ -181,6 +168,7 @@ describe('MongoClient.close() Integration', () => {
                     const servers = client.topology.s.servers;
                     expect(getMonitorTimer(servers)).to.exist;
                     await client.close();
+                    // the current expected behavior is that timerId is set to undefined once it expires or is interrupted
                     expect(getMonitorTimer(servers)).to.not.exist;
 
                     expect(getTimerCount()).to.equal(0);
@@ -226,22 +214,14 @@ describe('MongoClient.close() Integration', () => {
                 'the rtt pinger timer is cleaned up by client.close()',
                 metadata,
                 async function () {
-                  const run = async function ({
-                    MongoClient,
-                    uri,
-                    expect,
-                    getTimerCount,
-                    promiseWithResolvers
-                  }) {
+                  const run = async function ({ MongoClient, uri, expect, getTimerCount, once }) {
                     const heartbeatFrequencyMS = 2000;
                     const client = new MongoClient(uri, {
                       serverMonitoringMode: 'stream',
                       heartbeatFrequencyMS
                     });
                     await client.connect();
-                    const { promise, resolve } = promiseWithResolvers();
-                    client.once('serverHeartbeatSucceeded', () => resolve());
-                    await promise;
+                    await once(client, 'serverHeartbeatSucceeded');
 
                     function getRttTimer(servers) {
                       for (const [, server] of servers) {
@@ -267,13 +247,7 @@ describe('MongoClient.close() Integration', () => {
             describe('Node.js resource: Socket', () => {
               describe('when rtt monitoring is turned on', () => {
                 it.skip('no sockets remain after client.close()', metadata, async () => {
-                  const run = async ({
-                    MongoClient,
-                    uri,
-                    expect,
-                    getSockets,
-                    promiseWithResolvers
-                  }) => {
+                  const run = async ({ MongoClient, uri, expect, getSockets, once }) => {
                     const heartbeatFrequencyMS = 500;
                     const client = new MongoClient(uri, {
                       serverMonitoringMode: 'stream',
@@ -289,12 +263,8 @@ describe('MongoClient.close() Integration', () => {
                     const servers = client.topology.s.servers;
 
                     while (heartbeatOccurredSet.size < servers.size) {
-                      const { promise, resolve } = promiseWithResolvers();
-                      client.once('serverHeartbeatSucceeded', ev => {
-                        heartbeatOccurredSet.add(ev.connectionId);
-                        resolve();
-                      });
-                      await promise;
+                      const ev = await once(client, 'serverHeartbeatSucceeded');
+                      heartbeatOccurredSet.add(ev[0].connectionId);
                     }
 
                     const activeSocketsAfterHeartbeat = () =>
@@ -309,7 +279,6 @@ describe('MongoClient.close() Integration', () => {
 
                     // close the client
                     await client.close();
-
                     // upon close, assert rttPinger sockets are cleaned up
                     const activeSocketsAfterClose = activeSocketsAfterHeartbeat();
                     expect(activeSocketsAfterClose).to.have.lengthOf(0);
@@ -385,36 +354,34 @@ describe('MongoClient.close() Integration', () => {
             });
 
             it.skip('the wait queue timer is cleaned up by client.close()', async function () {
-              const run = async function ({ MongoClient, uri, expect, sleep, getTimerCount }) {
+              const run = async function ({ MongoClient, uri, expect, getTimerCount, once }) {
                 const waitQueueTimeoutMS = 1515;
 
                 const client = new MongoClient(uri, {
                   maxPoolSize: 1,
                   waitQueueTimeoutMS,
-                  appName: 'waitQueueTestClient'
+                  appName: 'waitQueueTestClient',
+                  monitorCommands: true
                 });
-                const insertPromise = client
+                client
                   .db('db')
                   .collection('collection')
                   .insertOne({ x: 1 })
                   .catch(e => e);
-                await client.once('commandStarted', v => v);
+                await once(client, 'connectionCheckedOut');
 
-                client
+                const blockedInsert = client
                   .db('db')
                   .collection('collection')
                   .insertOne({ x: 1 })
                   .catch(e => e);
-                await client.once('commandStarted', v => v);
-
-                // don't allow entire checkout timer to elapse to ensure close is called mid-timeout
-                await sleep(waitQueueTimeoutMS / 2);
+                await once(client, 'connectionCheckOutStarted');
 
                 expect(getTimerCount()).to.not.equal(0);
                 await client.close();
                 expect(getTimerCount()).to.equal(0);
 
-                const err = await insertPromise;
+                const err = await blockedInsert;
                 expect(err).to.be.instanceOf(Error);
                 expect(err.message).to.contain(
                   'Timed out while checking out a connection from connection pool'
@@ -457,24 +424,19 @@ describe('MongoClient.close() Integration', () => {
     describe('SrvPoller', () => {
       describe('Node.js resource: Timer', () => {
         // requires an srv environment that can transition to sharded
-        const metadata: MongoDBMetadataUI = {
-          requires: {
-            predicate: () =>
-              process.env.ATLAS_SRV_REPL ? true : 'Skipped: this test requires an SRV environment'
-          }
-        };
+        const metadata: MongoDBMetadataUI = { requires: { topology: 'sharded' } };
 
         describe('after SRVPoller is created', () => {
           it.skip('timers are cleaned up by client.close()', metadata, async () => {
-            const run = async function ({ MongoClient, uri, expect, getTimerCount }) {
-              const client = new MongoClient(uri);
+            const run = async function ({ MongoClient, expect, getTimerCount }) {
+              const SRV_CONNECTION_STRING = `mongodb+srv://test1.test.build.10gen.cc`;
+              // 27018 localhost.test.build.10gen.cc.
+              // 27017 localhost.test.build.10gen.cc.
+
+              const client = new MongoClient(SRV_CONNECTION_STRING);
               await client.connect();
-              const description = client.topology.s.description;
-              // simulate transition to sharded
-              client.topology.emit.skip('topologyDescriptionChanged', description, {
-                ...description,
-                type: 'Sharded'
-              });
+              // the current expected behavior is that _timeout is set to undefined until SRV polling starts
+              // then _timeout is set to undefined again when SRV polling stops
               expect(client.topology.s.srvPoller._timeout).to.exist;
               await client.close();
               expect(getTimerCount()).to.equal(0);

test/integration/node-specific/resource_tracking_script_builder.ts

@@ -1,5 +1,6 @@
 import { fork, spawn } from 'node:child_process';
 import { on, once } from 'node:events';
+import { openSync } from 'node:fs';
 import { readFile, unlink, writeFile } from 'node:fs/promises';
 import * as path from 'node:path';
 
@@ -31,7 +32,7 @@ export type ProcessResourceTestFunction = (options: {
   timers?: typeof timers;
   getSocketReport?: () => { host: string; port: string };
   getSocketEndpointReport?: () => any;
-  promiseWithResolvers?: () => any;
+  once?: () => typeof once;
 }) => Promise<void>;
 
 const HEAP_RESOURCE_SCRIPT_PATH = path.resolve(
@@ -176,7 +177,10 @@ export async function runScriptAndGetProcessInfo(
   await writeFile(scriptName, scriptContent, { encoding: 'utf8' });
   const logFile = name + '.logs.txt';
 
-  const script = spawn(process.execPath, [scriptName], { stdio: ['ignore', 'ignore', 'inherit'] });
+  const stdErrFile = 'err.out';
+  const script = spawn(process.execPath, [scriptName], {
+    stdio: ['ignore', 'ignore', openSync(stdErrFile, 'w')]
+  });
 
   const willClose = once(script, 'close');
 
@@ -190,9 +194,12 @@ export async function runScriptAndGetProcessInfo(
     .map(line => JSON.parse(line))
     .reduce((acc, curr) => ({ ...acc, ...curr }), {});
 
+  const stdErrSize = await readFile(stdErrFile, { encoding: 'utf8' });
+
   // delete temporary files
   await unlink(scriptName);
   await unlink(logFile);
+  await unlink(stdErrFile);
 
   // assertions about exit status
   if (exitCode) {
@@ -207,4 +214,7 @@ export async function runScriptAndGetProcessInfo(
   expect(messages.beforeExitHappened).to.be.true;
   expect(messages.newResources.libuvResources).to.be.empty;
   expect(messages.newResources.activeResources).to.be.empty;
+
+  // assertion about error output
+  expect(stdErrSize).to.be.empty;
 }

test/tools/fixtures/process_resource_script.in.js

@@ -15,6 +15,7 @@ const fs = require('node:fs');
 const { expect } = require('chai');
 const timers = require('node:timers');
 const { setTimeout } = timers;
+const { once } = require('node:events');
 
 let originalReport;
 const logFile = scriptName + '.logs.txt';
@@ -71,6 +72,7 @@ function getNewLibuvResourceArray() {
  * - `process.getActiveResourcesInfo()` does not contain enough server information we need for our assertions
  *
  */
+
 function getNewResources() {
   return {
     libuvResources: getNewLibuvResourceArray(),
@@ -98,16 +100,6 @@ const getSocketEndpoints = () =>
     .libuv.filter(r => r.type === 'tcp')
     .map(r => r.remoteEndpoint);
 
-function promiseWithResolvers() {
-  let resolve;
-  let reject;
-  const promise = new Promise((promiseResolve, promiseReject) => {
-    resolve = promiseResolve;
-    reject = promiseReject;
-  });
-  return { promise, resolve, reject };
-}
-
 // A log function for debugging
 function log(message) {
   // remove outer parentheses for easier parsing
@@ -130,7 +122,7 @@ async function main() {
     getTimerCount,
     getSockets,
     getSocketEndpoints,
-    promiseWithResolvers
+    once
   });
   log({ newResources: getNewResources() });
 }