fix(NODE-5106): prevent multiple mongo client connect()s from leaking topology

src/mongo_client.ts

@@ -318,6 +318,8 @@ export class MongoClient extends TypedEventEmitter<MongoClientEvents> {
   topology?: Topology;
   /** @internal */
   readonly mongoLogger: MongoLogger;
+  /** @internal */
+  private connectionLock?: Promise<this>;
 
   /**
    * The consolidate, parsed, transformed and merged options.
@@ -405,6 +407,30 @@ export class MongoClient extends TypedEventEmitter<MongoClientEvents> {
    * @see docs.mongodb.org/manual/reference/connection-string/
    */
   async connect(): Promise<this> {
+    if (this.connectionLock) {
+      return this.connectionLock;
+    }
+
+    try {
+      this.connectionLock = this._connect();
+      await this.connectionLock;
+    } finally {
+      // release
+      this.connectionLock = undefined;
+    }
+
+    return this;
+  }
+
+  /**
+   * Create a topology to open the connection, must be locked to avoid topology leaks in concurrency scenario.
+   * Locking is enforced by the connect method.
+   *
+   * When decorators available put implementation back to original connect methods
+   * and enforce locking via a dedicated decorator.
+   * @see https://github.com/mongodb/node-mongodb-native/pull/3596#discussion_r1134211500
+   */
+  private async _connect(): Promise<this> {
     if (this.topology && this.topology.isConnected()) {
       return this;
     }

test/integration/mongo_client.test.ts

@@ -0,0 +1,68 @@
+import { expect } from 'chai';
+import * as sinon from 'sinon';
+
+import { MongoClient } from '../../src';
+
+describe('MongoClient', () => {
+  let client: MongoClient;
+  let topologyOpenEvents;
+
+  /** Keep track number of call to client connect to close as many as connect (otherwise leak_checker hook will failed) */
+  let clientConnectCounter: number;
+
+  /**
+   * Wrap the connect method of the client to keep track
+   * of number of times connect is called
+   */
+  function clientConnect() {
+    if (!client) {
+      return;
+    }
+    clientConnectCounter++;
+    return client.connect();
+  }
+
+  beforeEach(async function () {
+    client = this.configuration.newClient();
+    topologyOpenEvents = [];
+    clientConnectCounter = 0;
+    client.on('open', event => topologyOpenEvents.push(event));
+  });
+
+  afterEach(async function () {
+    /** Close as many times as connect calls in the runned test (tracked by clientConnectCounter) */
+    const clientClosePromises = [...new Array(clientConnectCounter)].map(() => client.close());
+    await Promise.all(clientClosePromises);
+  });
+
+  it('Concurrents client connect correctly locked (only one topology created)', async function () {
+    await Promise.all([clientConnect(), clientConnect(), clientConnect()]);
+
+    expect(topologyOpenEvents).to.have.lengthOf(1);
+    expect(client.topology?.isConnected()).to.be.true;
+  });
+
+  it('Failed client connect must properly release lock', async function () {
+    const internalConnectStub = sinon.stub(client, '_connect' as keyof MongoClient);
+    internalConnectStub.onFirstCall().rejects();
+
+    // first call rejected to simulate a connection failure
+    try {
+      await clientConnect();
+    } catch (err) {
+      expect(err).to.exist;
+    }
+
+    internalConnectStub.restore();
+
+    // second call should connect
+    try {
+      await clientConnect();
+    } catch (err) {
+      expect.fail(`client connect throwed unexpected error`);
+    }
+
+    expect(topologyOpenEvents).to.have.lengthOf(1);
+    expect(client.topology?.isConnected()).to.be.true;
+  });
+});