feat(NODE-7609): add interruptInUseConnections client option

src/cmap/connection_pool.ts

@@ -76,6 +76,11 @@ export interface ConnectionPoolOptions extends Omit<ConnectionOptions, 'id' | 'g
   waitQueueTimeoutMS: number;
   /** If we are in load balancer mode. */
   loadBalanced: boolean;
+  /**
+   * Whether a monitor-driven pool clear may interrupt in-use connections.
+   * See {@link MongoClientOptions.interruptInUseConnections}. Defaults to `true`.
+   */
+  interruptInUseConnections?: boolean;
   /** @internal */
   minPoolSizeCheckFrequencyMS?: number;
 }

src/connection_string.ts

@@ -850,6 +850,10 @@ export const OPTIONS = {
   ignoreUndefined: {
     type: 'boolean'
   },
+  interruptInUseConnections: {
+    default: true,
+    type: 'boolean'
+  },
   j: {
     deprecated: 'Please use journal instead',
     target: 'writeConcern',

src/mongo_client.ts

@@ -229,6 +229,16 @@ export interface MongoClientOptions extends BSONSerializeOptions, SupportedNodeC
   retryReads?: boolean;
   /** Enable retryable writes. */
   retryWrites?: boolean;
+  /**
+   * When the server monitor detects a network timeout it clears the connection pool
+   * and, by default, interrupts ("kills") in-use connections, rejecting their
+   * in-flight operations with a `PoolClearedOnNetworkError`. Set this to `false` to
+   * leave in-use connections untouched on a monitor timeout, so a transient blip
+   * (for example the host being suspended/resumed) does not abort in-flight
+   * operations that the driver does not retry, such as tailable `getMore`s.
+   * Defaults to `true` (the Server Discovery and Monitoring spec behaviour).
+   */
+  interruptInUseConnections?: boolean;
   /**
    * The maximum number of retries during server overload. Set to 0 to disable overload retries. Defaults to 2.
    * @see https://www.mongodb.com/docs/atlas/overload-errors
@@ -1062,6 +1072,7 @@ export interface MongoOptions
         | 'forceServerObjectId'
         | 'minHeartbeatFrequencyMS'
         | 'heartbeatFrequencyMS'
+        | 'interruptInUseConnections'
         | 'localThresholdMS'
         | 'maxConnecting'
         | 'maxIdleTimeMS'

src/sdam/topology.ts

@@ -869,9 +869,15 @@ function updateServers(topology: Topology, incomingServerDescription?: ServerDes
         incomingServerDescription.error instanceof MongoError &&
         incomingServerDescription.error.hasErrorLabel(MongoErrorLabel.ResetPool)
       ) {
-        const interruptInUseConnections = incomingServerDescription.error.hasErrorLabel(
-          MongoErrorLabel.InterruptInUseConnections
-        );
+        // Honour the `interruptInUseConnections` client option (default true). When
+        // disabled, a monitor network timeout still clears the pool (ResetPool) but
+        // does NOT kill in-use connections, so a transient blip (e.g. a suspended/
+        // resumed host) doesn't abort in-flight operations the driver won't retry.
+        // See meteor/meteor#13108.
+        const interruptInUseConnections =
+          incomingServerDescription.error.hasErrorLabel(
+            MongoErrorLabel.InterruptInUseConnections
+          ) && topology.s.options.interruptInUseConnections !== false;
 
         server.pool.clear({ interruptInUseConnections });
       } else if (incomingServerDescription.error == null) {

test/unit/connection_string.test.ts

@@ -281,6 +281,25 @@ describe('Connection String', function () {
     }
   });
 
+  describe('interruptInUseConnections', function () {
+    it('defaults to true', function () {
+      const options = parseOptions('mongodb://localhost:27017');
+      expect(options).to.have.property('interruptInUseConnections', true);
+    });
+
+    it('can be disabled via the options object', function () {
+      const options = parseOptions('mongodb://localhost:27017', {
+        interruptInUseConnections: false
+      });
+      expect(options).to.have.property('interruptInUseConnections', false);
+    });
+
+    it('can be disabled via the connection string', function () {
+      const options = parseOptions('mongodb://localhost:27017/?interruptInUseConnections=false');
+      expect(options).to.have.property('interruptInUseConnections', false);
+    });
+  });
+
   it('should parse compression options', function () {
     const options = parseOptions('mongodb://localhost/?compressors=zlib&zlibCompressionLevel=4');
     expect(options).to.have.property('compressors');

test/unit/sdam/topology.test.ts

@@ -12,12 +12,15 @@ import {
   LEGACY_NOT_WRITABLE_PRIMARY_ERROR_MESSAGE,
   makeClientMetadata,
   MongoClient,
+  MongoErrorLabel,
+  MongoNetworkTimeoutError,
   MongoServerSelectionError,
   ns,
   ReadPreference,
   RunCommandOperation,
   RunCursorCommandOperation,
   Server,
+  ServerDescription,
   SrvPoller,
   SrvPollingEvent,
   TimeoutContext,
@@ -43,6 +46,59 @@ describe('Topology (unit)', function () {
     }
   });
 
+  describe('interruptInUseConnections option (meteor/meteor#13108)', function () {
+    // When the SDAM monitor sees a network timeout it labels the error with both
+    // ResetPool and InterruptInUseConnections. updateServers() then clears the pool;
+    // the new `interruptInUseConnections` client option (default true) gates whether
+    // in-use connections are interrupted.
+    let mockServer;
+
+    beforeEach(async () => {
+      mockServer = await mock.createServer();
+      mockServer.setMessageHandler(request => {
+        const doc = request.document;
+        if (isHello(doc)) {
+          request.reply(Object.assign({}, mock.HELLO, { maxWireVersion: 9 }));
+        } else {
+          request.reply({ ok: 1 });
+        }
+      });
+    });
+
+    afterEach(async () => {
+      await mock.cleanup();
+      sinon.restore();
+    });
+
+    for (const { setting, expected } of [
+      { setting: undefined, expected: true },
+      { setting: true, expected: true },
+      { setting: false, expected: false }
+    ]) {
+      it(`clears the pool with interruptInUseConnections=${expected} when the option is ${setting}`, async function () {
+        topology = topologyWithPlaceholderClient(
+          [mockServer.hostAddress()],
+          setting === undefined ? {} : { interruptInUseConnections: setting }
+        );
+        await topology.connect();
+
+        const [address] = topology.s.servers.keys();
+        const server = topology.s.servers.get(address);
+        const clearStub = sinon.stub(server.pool, 'clear');
+
+        // Simulate the SDAM monitor reporting a network timeout for this server.
+        const error = new MongoNetworkTimeoutError('connection <monitor> timed out');
+        error.addErrorLabel(MongoErrorLabel.ResetPool);
+        error.addErrorLabel(MongoErrorLabel.InterruptInUseConnections);
+        topology.serverUpdateHandler(new ServerDescription(address, undefined, { error }));
+
+        expect(clearStub).to.have.been.calledOnceWithExactly({
+          interruptInUseConnections: expected
+        });
+      });
+    }
+  });
+
   describe('client metadata', function () {
     let mockServer;