refactor(NODE-5063): make DestroyOptions required on connection.destroy (#3568)

Author: baileympearson

src/cmap/connect.ts

@@ -96,7 +96,7 @@ function performInitialHandshake(
 ) {
   const callback: Callback<Document> = function (err, ret) {
     if (err && conn) {
-      conn.destroy();
+      conn.destroy({ force: false });
     }
     _callback(err, ret);
   };

src/cmap/connection.ts

@@ -133,7 +133,7 @@ export interface ConnectionOptions
 /** @internal */
 export interface DestroyOptions {
   /** Force the destruction. */
-  force?: boolean;
+  force: boolean;
 }
 
 /** @public */
@@ -443,16 +443,10 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
     callback(undefined, message.documents[0]);
   }
 
-  destroy(options?: DestroyOptions, callback?: Callback): void {
-    if (typeof options === 'function') {
-      callback = options;
-      options = { force: false };
-    }
-
+  destroy(options: DestroyOptions, callback?: Callback): void {
     this.removeAllListeners(Connection.PINNED);
     this.removeAllListeners(Connection.UNPINNED);
 
-    options = Object.assign({ force: false }, options);
     if (this[kStream] == null || this.destroyed) {
       this.destroyed = true;
       if (typeof callback === 'function') {

src/cmap/connection_pool.ts

@@ -510,7 +510,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
           ConnectionPool.CONNECTION_CLOSED,
           new ConnectionClosedEvent(this, conn, 'poolClosed')
         );
-        conn.destroy(options, cb);
+        conn.destroy({ force: !!options.force }, cb);
       },
       err => {
         this[kConnections].clear();
@@ -586,7 +586,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
       new ConnectionClosedEvent(this, connection, reason)
     );
     // destroy the connection
-    process.nextTick(() => connection.destroy());
+    process.nextTick(() => connection.destroy({ force: false }));
   }
 
   private connectionIsStale(connection: Connection) {

src/sdam/server.ts

@@ -241,7 +241,10 @@ export class Server extends TypedEventEmitter<ServerEvents> {
 
   /** Destroy the server connection */
   destroy(options?: DestroyOptions, callback?: Callback): void {
-    if (typeof options === 'function') (callback = options), (options = {});
+    if (typeof options === 'function') {
+      callback = options;
+      options = { force: false };
+    }
     options = Object.assign({}, { force: false }, options);
 
     if (this.s.state === STATE_CLOSED) {

src/sdam/topology.ts

@@ -466,26 +466,17 @@ export class Topology extends TypedEventEmitter<TopologyEvents> {
   }
 
   /** Close this topology */
-  close(callback: Callback): void;
   close(options: CloseOptions): void;
   close(options: CloseOptions, callback: Callback): void;
-  close(options?: CloseOptions | Callback, callback?: Callback): void {
-    if (typeof options === 'function') {
-      callback = options;
-      options = {};
-    }
-
-    if (typeof options === 'boolean') {
-      options = { force: options };
-    }
-    options = options ?? {};
+  close(options?: CloseOptions, callback?: Callback): void {
+    options = options ?? { force: false };
 
     if (this.s.state === STATE_CLOSED || this.s.state === STATE_CLOSING) {
       return callback?.();
     }
 
     const destroyedServers = Array.from(this.s.servers.values(), server => {
-      return promisify(destroyServer)(server, this, options as CloseOptions);
+      return promisify(destroyServer)(server, this, { force: !!options?.force });
     });
 
     Promise.all(destroyedServers)
@@ -740,7 +731,7 @@ function destroyServer(
   options?: DestroyOptions,
   callback?: Callback
 ) {
-  options = options ?? {};
+  options = options ?? { force: false };
   for (const event of LOCAL_SERVER_EVENTS) {
     server.removeAllListeners(event);
   }

test/integration/crud/misc_cursors.test.js

@@ -10,7 +10,7 @@ const sinon = require('sinon');
 const { Writable } = require('stream');
 const { once, on } = require('events');
 const { setTimeout } = require('timers');
-const { ReadPreference } = require('../../mongodb');
+const { ReadPreference, MongoExpiredSessionError } = require('../../mongodb');
 const { ServerType } = require('../../mongodb');
 const { formatSort } = require('../../mongodb');
 const { getSymbolFrom } = require('../../tools/utils');
@@ -1852,61 +1852,31 @@ describe('Cursor', function () {
     }
   });
 
-  it('should close dead tailable cursors', {
-    metadata: {
-      os: '!win32' // NODE-2943: timeout on windows
-    },
-
-    test: function (done) {
-      // http://www.mongodb.org/display/DOCS/Tailable+Cursors
-
-      const configuration = this.configuration;
-      client.connect((err, client) => {
-        expect(err).to.not.exist;
-        this.defer(() => client.close());
-
-        const db = client.db(configuration.db);
-        const options = { capped: true, size: 10000000 };
-        db.createCollection(
-          'test_if_dead_tailable_cursors_close',
-          options,
-          function (err, collection) {
-            expect(err).to.not.exist;
+  it('closes cursors when client is closed even if it has not been exhausted', async function () {
+    await client
+      .db()
+      .dropCollection('test_cleanup_tailable')
+      .catch(() => null);
 
-            let closeCount = 0;
-            const docs = Array.from({ length: 100 }).map(() => ({ a: 1 }));
-            collection.insertMany(docs, { w: 'majority', wtimeoutMS: 5000 }, err => {
-              expect(err).to.not.exist;
-
-              const cursor = collection.find({}, { tailable: true, awaitData: true });
-              const stream = cursor.stream();
+    const collection = await client
+      .db()
+      .createCollection('test_cleanup_tailable', { capped: true, size: 1000, max: 3 });
 
-              stream.resume();
-
-              var validator = () => {
-                closeCount++;
-                if (closeCount === 2) {
-                  done();
-                }
-              };
+    // insert only 2 docs in capped coll of 3
+    await collection.insertMany([{ a: 1 }, { a: 1 }]);
 
-              // we validate that the stream "ends" either cleanly or with an error
-              stream.on('end', validator);
-              stream.on('error', validator);
+    const cursor = collection.find({}, { tailable: true, awaitData: true, maxAwaitTimeMS: 2000 });
 
-              cursor.on('close', validator);
+    await cursor.next();
+    await cursor.next();
+    // will block for maxAwaitTimeMS (except we are closing the client)
+    const rejectedEarlyBecauseClientClosed = cursor.next().catch(error => error);
 
-              const docs = Array.from({ length: 100 }).map(() => ({ a: 1 }));
-              collection.insertMany(docs, err => {
-                expect(err).to.not.exist;
+    await client.close();
+    expect(cursor).to.have.property('killed', true);
 
-                setTimeout(() => client.close());
-              });
-            });
-          }
-        );
-      });
-    }
+    const error = await rejectedEarlyBecauseClientClosed;
+    expect(error).to.be.instanceOf(MongoExpiredSessionError);
   });
 
   it('shouldAwaitData', {

test/integration/node-specific/topology.test.js

@@ -10,12 +10,20 @@ describe('Topology', function () {
       const states = [];
       topology.on('stateChanged', (_, newState) => states.push(newState));
       topology.connect(err => {
-        expect(err).to.not.exist;
-        topology.close(err => {
+        try {
           expect(err).to.not.exist;
-          expect(topology.isDestroyed()).to.be.true;
-          expect(states).to.eql(['connecting', 'connected', 'closing', 'closed']);
-          done();
+        } catch (error) {
+          done(error);
+        }
+        topology.close({}, err => {
+          try {
+            expect(err).to.not.exist;
+            expect(topology.isDestroyed()).to.be.true;
+            expect(states).to.eql(['connecting', 'connected', 'closing', 'closed']);
+            done();
+          } catch (error) {
+            done(error);
+          }
         });
       });
     }

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

@@ -103,7 +103,7 @@ describe('Polling Srv Records for Mongos Discovery', () => {
 
     afterEach(function (done) {
       if (context.topology) {
-        context.topology.close(done);
+        context.topology.close({}, done);
       } else {
         done();
       }

test/unit/assorted/server_selection_spec_helper.js

@@ -109,7 +109,7 @@ function executeServerSelectionTest(testDefinition, testDone) {
     });
 
   function done(err) {
-    topology.close(e => testDone(e || err));
+    topology.close({}, e => testDone(e || err));
   }
 
   topology.connect(err => {

test/unit/error.test.ts

@@ -381,7 +381,7 @@ describe('MongoErrors', () => {
 
       makeAndConnectReplSet((err, topology) => {
         // cleanup the server before calling done
-        const cleanup = err => topology.close(err2 => done(err || err2));
+        const cleanup = err => topology.close({}, err2 => done(err || err2));
 
         if (err) {
           return cleanup(err);