refactor: remove Topology#command method (#2545)

NODE-2801

Author: mbroadst

src/cmap/wire_protocol/command.ts

@@ -26,10 +26,8 @@ export interface CommandOptions extends BSONSerializeOptions {
   documentsReturnedIn?: string;
   noResponse?: boolean;
 
-  // NOTE: these are for retryable writes and will be removed soon
+  // FIXME: NODE-2802
   willRetryWrite?: boolean;
-  retryWrites?: boolean;
-  retrying?: boolean;
 
   // FIXME: NODE-2781
   writeConcern?: WriteConcernOptions | WriteConcern | W;
@@ -137,7 +135,6 @@ function _command(
 
   // This value is not overridable
   commandOptions.slaveOk = readPreference.slaveOk();
-
   const cmdNs = `${databaseNamespace(ns)}.$cmd`;
   const message = shouldUseOpMsg
     ? new Msg(cmdNs, finalCmd, commandOptions)

src/operations/command.ts

@@ -3,7 +3,7 @@ import { ReadConcern } from '../read_concern';
 import { WriteConcern, WriteConcernOptions } from '../write_concern';
 import { maxWireVersion, MongoDBNamespace, Callback } from '../utils';
 import { ReadPreference, ReadPreferenceLike } from '../read_preference';
-import { commandSupportsReadConcern, ClientSession } from '../sessions';
+import { commandSupportsReadConcern } from '../sessions';
 import { MongoError } from '../error';
 import type { Logger } from '../logger';
 import type { Server } from '../sdam/server';
@@ -14,13 +14,12 @@ const SUPPORTS_WRITE_CONCERN_AND_COLLATION = 5;
 
 /** @public */
 export interface CommandOperationOptions extends OperationOptions, WriteConcernOptions {
+  /** Return the full server response for the command */
   fullResponse?: boolean;
   /** Specify a read concern and level for the collection. (only MongoDB 3.2 or higher supported) */
   readConcern?: ReadConcern;
   /** The preferred read preference (ReadPreference.primary, ReadPreference.primary_preferred, ReadPreference.secondary, ReadPreference.secondary_preferred, ReadPreference.nearest). */
   readPreference?: ReadPreferenceLike;
-  /** Specify ClientSession for this command */
-  session?: ClientSession;
   /** Collation */
   collation?: CollationOptions;
   maxTimeMS?: number;
@@ -32,6 +31,7 @@ export interface CommandOperationOptions extends OperationOptions, WriteConcernO
   // Admin command overrides.
   dbName?: string;
   authdb?: string;
+  noResponse?: boolean;
 }
 
 /** @internal */
@@ -56,16 +56,20 @@ export abstract class CommandOperation<
   fullResponse?: boolean;
   logger?: Logger;
 
-  constructor(parent: OperationParent, options?: T) {
+  constructor(parent?: OperationParent, options?: T) {
     super(options);
 
     // NOTE: this was explicitly added for the add/remove user operations, it's likely
     //       something we'd want to reconsider. Perhaps those commands can use `Admin`
     //       as a parent?
     const dbNameOverride = options?.dbName || options?.authdb;
-    this.ns = dbNameOverride
-      ? new MongoDBNamespace(dbNameOverride, '$cmd')
-      : parent.s.namespace.withCollection('$cmd');
+    if (dbNameOverride) {
+      this.ns = new MongoDBNamespace(dbNameOverride, '$cmd');
+    } else {
+      this.ns = parent
+        ? parent.s.namespace.withCollection('$cmd')
+        : new MongoDBNamespace('admin', '$cmd');
+    }
 
     const propertyProvider = this.hasAspect(Aspect.NO_INHERIT_OPTIONS) ? undefined : parent;
     this.readPreference = this.hasAspect(Aspect.WRITE_OPERATION)
@@ -82,7 +86,7 @@ export abstract class CommandOperation<
     this.options.readPreference = this.readPreference;
 
     // TODO(NODE-2056): make logger another "inheritable" property
-    if (parent.logger) {
+    if (parent && parent.logger) {
       this.logger = parent.logger;
     }
   }

src/operations/execute_operation.ts

@@ -123,8 +123,9 @@ function supportsRetryableReads(server: Server) {
 }
 
 function executeWithServerSelection(topology: Topology, operation: any, callback: Callback) {
+  const session = operation.session;
   const readPreference = operation.readPreference || ReadPreference.primary;
-  const inTransaction = operation.session && operation.session.inTransaction();
+  const inTransaction = session && session.inTransaction();
 
   if (inTransaction && !readPreference.equals(ReadPreference.primary)) {
     callback(
@@ -136,10 +137,7 @@ function executeWithServerSelection(topology: Topology, operation: any, callback
     return;
   }
 
-  const serverSelectionOptions = {
-    session: operation.session
-  };
-
+  const serverSelectionOptions = { session };
   function callbackWithRetry(err?: any, result?: any) {
     if (err == null) {
       return callback(undefined, result);
@@ -176,7 +174,7 @@ function executeWithServerSelection(topology: Topology, operation: any, callback
         (operation.hasAspect(Aspect.READ_OPERATION) && !supportsRetryableReads(server)) ||
         (operation.hasAspect(Aspect.WRITE_OPERATION) && !supportsRetryableWrites(server))
       ) {
-        callback(err, null);
+        callback(err);
         return;
       }
 
@@ -187,20 +185,20 @@ function executeWithServerSelection(topology: Topology, operation: any, callback
   // select a server, and execute the operation against it
   topology.selectServer(readPreference, serverSelectionOptions, (err?: any, server?: any) => {
     if (err) {
-      callback(err, null);
+      callback(err);
       return;
     }
 
     const willRetryRead =
       topology.s.options.retryReads !== false &&
-      operation.session &&
+      session &&
       !inTransaction &&
       supportsRetryableReads(server) &&
       operation.canRetryRead;
 
     const willRetryWrite =
       topology.s.options.retryWrites === true &&
-      operation.session &&
+      session &&
       !inTransaction &&
       supportsRetryableWrites(server) &&
       operation.canRetryWrite;
@@ -212,7 +210,7 @@ function executeWithServerSelection(topology: Topology, operation: any, callback
     ) {
       if (operation.hasAspect(Aspect.WRITE_OPERATION) && willRetryWrite) {
         operation.options.willRetryWrite = true;
-        operation.session.incrementTransactionNumber();
+        session.incrementTransactionNumber();
       }
 
       operation.execute(server, callbackWithRetry);

src/operations/operation.ts

@@ -21,8 +21,11 @@ export interface OperationConstructor extends Function {
 
 /** @internal */
 export interface OperationOptions extends BSONSerializeOptions {
-  explain?: boolean;
+  /** Specify ClientSession for this command */
   session?: ClientSession;
+
+  explain?: boolean;
+  willRetryWrites?: boolean;
 }
 
 /**

src/operations/run_command.ts

@@ -14,7 +14,7 @@ export class RunCommandOperation<
 > extends CommandOperation<T, TResult> {
   command: Document;
 
-  constructor(parent: OperationParent, command: Document, options?: T) {
+  constructor(parent: OperationParent | undefined, command: Document, options?: T) {
     super(parent, options);
     this.command = command;
   }
@@ -29,7 +29,7 @@ export class RunAdminCommandOperation<
   T extends RunCommandOptions = RunCommandOptions,
   TResult = Document
 > extends RunCommandOperation<T, TResult> {
-  constructor(parent: OperationParent, command: Document, options?: T) {
+  constructor(parent: OperationParent | undefined, command: Document, options?: T) {
     super(parent, command, options);
     this.ns = new MongoDBNamespace('admin');
   }

src/sdam/topology.ts

@@ -54,7 +54,6 @@ import type { Transaction } from '../transactions';
 import type { CloseOptions } from '../cmap/connection_pool';
 import type { LoggerOptions } from '../logger';
 import { DestroyOptions, Connection } from '../cmap/connection';
-import type { CommandOptions } from '../cmap/wire_protocol/command';
 import { RunCommandOperation } from '../operations/run_command';
 import type { CursorOptions } from '../cursor/cursor';
 import type { MongoClientOptions } from '../mongo_client';
@@ -356,15 +355,28 @@ export class Topology extends EventEmitter {
 
     ReadPreference.translate(options);
     const readPreference = options.readPreference || ReadPreference.primary;
-    const connectHandler = (err: Error | MongoError | undefined) => {
+    this.selectServer(readPreferenceServerSelector(readPreference), options, (err, server) => {
       if (err) {
         this.close();
 
-        if (typeof callback === 'function') {
-          callback(err);
-        } else {
-          this.emit(Topology.ERROR, err);
-        }
+        typeof callback === 'function' ? callback(err) : this.emit(Topology.ERROR, err);
+        return;
+      }
+
+      // TODO: NODE-2471
+      if (server && this.s.credentials) {
+        server.command('admin.$cmd', { ping: 1 }, err => {
+          if (err) {
+            typeof callback === 'function' ? callback(err) : this.emit(Topology.ERROR, err);
+            return;
+          }
+
+          stateTransition(this, STATE_CONNECTED);
+          this.emit(Topology.OPEN, err, this);
+          this.emit(Topology.CONNECT, this);
+
+          if (typeof callback === 'function') callback(undefined, this);
+        });
 
         return;
       }
@@ -373,16 +385,8 @@ export class Topology extends EventEmitter {
       this.emit(Topology.OPEN, err, this);
       this.emit(Topology.CONNECT, this);
 
-      if (typeof callback === 'function') callback(err, this);
-    };
-
-    // TODO: NODE-2471
-    if (this.s.credentials) {
-      this.command('admin.$cmd', { ping: 1 }, { readPreference }, connectHandler);
-      return;
-    }
-
-    this.selectServer(readPreferenceServerSelector(readPreference), options, connectHandler);
+      if (typeof callback === 'function') callback(undefined, this);
+    });
   }
 
   /** Close this topology */
@@ -567,18 +571,27 @@ export class Topology extends EventEmitter {
   }
 
   /** Send endSessions command(s) with the given session ids */
-  endSessions(sessions: ServerSessionId[], callback?: Callback): void {
+  endSessions(sessions: ServerSessionId[], callback?: Callback<Document>): void {
     if (!Array.isArray(sessions)) {
       sessions = [sessions];
     }
 
-    this.command(
-      'admin.$cmd',
-      { endSessions: sessions },
-      { readPreference: ReadPreference.primaryPreferred, noResponse: true },
-      () => {
-        // intentionally ignored, per spec
-        if (typeof callback === 'function') callback();
+    this.selectServer(
+      readPreferenceServerSelector(ReadPreference.primaryPreferred),
+      (err, server) => {
+        if (err || !server) {
+          if (typeof callback === 'function') callback(err);
+          return;
+        }
+
+        server.command(
+          'admin.$cmd',
+          { endSessions: sessions },
+          { noResponse: true },
+          (err, result) => {
+            if (typeof callback === 'function') callback(err, result);
+          }
+        );
       }
     );
   }
@@ -669,56 +682,6 @@ export class Topology extends EventEmitter {
     if (typeof callback === 'function') callback(undefined, true);
   }
 
-  /**
-   * Execute a command
-   *
-   * @param ns - The MongoDB fully qualified namespace (ex: db1.collection1)
-   * @param cmd - The command
-   */
-  command(ns: string, cmd: Document, options: CommandOptions, callback: Callback): void {
-    if (typeof options === 'function') {
-      (callback = options), (options = {}), (options = options || {});
-    }
-
-    ReadPreference.translate(options);
-    const readPreference = (options.readPreference as ReadPreference) || ReadPreference.primary;
-
-    this.selectServer(readPreferenceServerSelector(readPreference), options, (err, server) => {
-      if (err || !server) {
-        callback(err);
-        return;
-      }
-
-      const willRetryWrite =
-        !options.retrying &&
-        !!options.retryWrites &&
-        options.session &&
-        isRetryableWritesSupported(this) &&
-        !options.session.inTransaction() &&
-        isWriteCommand(cmd);
-
-      // increment and assign txnNumber
-      if (willRetryWrite) {
-        options.session?.incrementTransactionNumber();
-        options.willRetryWrite = willRetryWrite;
-      }
-
-      server.command(ns, cmd, options, (err, result) => {
-        if (!err) return callback(undefined, result);
-        if (!shouldRetryOperation(err)) {
-          return callback(err);
-        }
-
-        if (willRetryWrite) {
-          const newOptions = Object.assign({}, options, { retrying: true });
-          return this.command(ns, cmd, newOptions, callback);
-        }
-
-        return callback(err);
-      });
-    });
-  }
-
   /**
    * Create a new cursor
    *
@@ -788,11 +751,6 @@ export class Topology extends EventEmitter {
   );
 }
 
-const RETRYABLE_WRITE_OPERATIONS = ['findAndModify', 'insert', 'update', 'delete'];
-function isWriteCommand(command: Document) {
-  return RETRYABLE_WRITE_OPERATIONS.some((op: string) => command[op]);
-}
-
 /** Destroys a server, and removes all event listeners from the instance */
 function destroyServer(
   server: Server,
@@ -941,10 +899,6 @@ function updateServers(topology: Topology, incomingServerDescription?: ServerDes
   }
 }
 
-function shouldRetryOperation(err: AnyError) {
-  return err instanceof MongoError && err.hasErrorLabel('RetryableWriteError');
-}
-
 function srvPollingHandler(topology: Topology) {
   return function handleSrvPolling(ev: SrvPollingEvent) {
     const previousTopologyDescription = topology.s.description;
@@ -1058,26 +1012,6 @@ function makeCompressionInfo(options: TopologyOptions) {
   return options.compression.compressors;
 }
 
-const RETRYABLE_WIRE_VERSION = 6;
-
-/** Determines whether the provided topology supports retryable writes */
-function isRetryableWritesSupported(topology: Topology) {
-  const maxWireVersion = topology.lastIsMaster().maxWireVersion;
-  if (maxWireVersion < RETRYABLE_WIRE_VERSION) {
-    return false;
-  }
-
-  if (!topology.logicalSessionTimeoutMinutes) {
-    return false;
-  }
-
-  if (topology.description.type === TopologyType.Single) {
-    return false;
-  }
-
-  return true;
-}
-
 /** @public */
 export class ServerCapabilities {
   maxWireVersion: number;