POC

Author: baileympearson

src/db.ts

@@ -6,7 +6,7 @@ import * as CONSTANTS from './constants';
 import { AggregationCursor } from './cursor/aggregation_cursor';
 import { ListCollectionsCursor } from './cursor/list_collections_cursor';
 import { RunCommandCursor, type RunCursorCommandOptions } from './cursor/run_command_cursor';
-import { MongoInvalidArgumentError } from './error';
+import { MONGODB_ERROR_CODES, MongoInvalidArgumentError, MongoServerError } from './error';
 import type { MongoClient, PkFactory } from './mongo_client';
 import type { Abortable, TODO_NODE_3286 } from './mongo_types';
 import type { AggregateOptions } from './operations/aggregate';

src/operations/execute_operation.ts

@@ -26,7 +26,7 @@ import type { Topology } from '../sdam/topology';
 import type { ClientSession } from '../sessions';
 import { TimeoutContext } from '../timeout';
 import { abortable, supportsRetryableWrites } from '../utils';
-import { AbstractOperation, Aspect } from './operation';
+import { AbstractOperation, Aspect, ModernOperation } from './operation';
 
 const MMAPv1_RETRY_WRITES_ERROR_CODE = MONGODB_ERROR_CODES.IllegalOperation;
 const MMAPv1_RETRY_WRITES_ERROR_MESSAGE =
@@ -85,6 +85,8 @@ export async function executeOperation<
     throw new MongoInvalidArgumentError('ClientSession must be from the same MongoClient');
   }
 
+  operation.session ??= session;
+
   const readPreference = operation.readPreference ?? ReadPreference.primary;
   const inTransaction = !!session?.inTransaction();
 
@@ -231,6 +233,8 @@ async function tryOperation<
   let previousOperationError: MongoError | undefined;
   let previousServer: ServerDescription | undefined;
 
+  const isModernOperation = operation instanceof ModernOperation;
+
   for (let tries = 0; tries < maxTries; tries++) {
     if (previousOperationError) {
       if (hasWriteAspect && previousOperationError.code === MMAPv1_RETRY_WRITES_ERROR_CODE) {
@@ -280,7 +284,17 @@ async function tryOperation<
       if (tries > 0 && operation.hasAspect(Aspect.COMMAND_BATCHING)) {
         operation.resetBatch();
       }
-      return await operation.execute(server, session, timeoutContext);
+
+      if (!isModernOperation) {
+        return await operation.execute(server, session, timeoutContext);
+      }
+
+      try {
+        const result = await server.modernCommand(operation, timeoutContext);
+        return operation.handleOk(result) as TResult;
+      } catch (error) {
+        operation.handleError(error);
+      }
     } catch (operationError) {
       if (!(operationError instanceof MongoError)) throw operationError;
       if (

src/operations/operation.ts

@@ -1,7 +1,8 @@
+import { type Connection, type MongoError } from '..';
 import { type BSONSerializeOptions, type Document, resolveBSONOptions } from '../bson';
 import { type Abortable } from '../mongo_types';
 import { ReadPreference, type ReadPreferenceLike } from '../read_preference';
-import type { Server } from '../sdam/server';
+import type { Server, ServerCommandOptions } from '../sdam/server';
 import type { ClientSession } from '../sessions';
 import { type TimeoutContext } from '../timeout';
 import type { MongoDBNamespace } from '../utils';
@@ -108,6 +109,10 @@ export abstract class AbstractOperation<TResult = any> {
     return this._session;
   }
 
+  set session(session: ClientSession) {
+    this._session = session;
+  }
+
   clearSession() {
     this._session = undefined;
   }
@@ -125,6 +130,37 @@ export abstract class AbstractOperation<TResult = any> {
   }
 }
 
+export abstract class ModernOperation<T> extends AbstractOperation<T> {
+  /** this will never be used - but we must implement it to satisfy AbstractOperation's interface */
+  override execute(
+    _server: Server,
+    _session: ClientSession | undefined,
+    _timeoutContext: TimeoutContext
+  ): Promise<T> {
+    throw new Error('cannot execute!!');
+  }
+
+  abstract buildCommand(connection: Connection, session?: ClientSession): Document;
+
+  abstract buildOptions(timeoutContext: TimeoutContext): ServerCommandOptions;
+
+  /**
+   * Optional - if the operation performs error handling, such as wrapping or renaming the error,
+   * this method can be overridden.
+   */
+  handleOk(response: Document) {
+    return response;
+  }
+
+  /**
+   * Optional - if the operation performs post-processing
+   * on the result document, this method can be overridden.
+   */
+  handleError(error: MongoError): void {
+    throw error;
+  }
+}
+
 export function defineAspects(
   operation: { aspects?: Set<symbol> },
   aspects: symbol | symbol[] | Set<symbol>

src/operations/search_indexes/drop.ts

@@ -1,31 +1,29 @@
+import { type Connection, type MongoError } from '../..';
 import type { Document } from '../../bson';
 import type { Collection } from '../../collection';
 import { MONGODB_ERROR_CODES, MongoServerError } from '../../error';
-import type { Server } from '../../sdam/server';
+import type { Server, ServerCommandOptions } from '../../sdam/server';
 import type { ClientSession } from '../../sessions';
 import { type TimeoutContext } from '../../timeout';
-import { AbstractOperation } from '../operation';
+import { AbstractOperation, ModernOperation } from '../operation';
 
 /** @internal */
-export class DropSearchIndexOperation extends AbstractOperation<void> {
+export class DropSearchIndexOperation extends ModernOperation<void> {
   private readonly collection: Collection;
   private readonly name: string;
 
   constructor(collection: Collection, name: string) {
     super();
     this.collection = collection;
     this.name = name;
+    this.ns = collection.fullNamespace;
   }
 
   override get commandName() {
     return 'dropSearchIndex' as const;
   }
 
-  override async execute(
-    server: Server,
-    session: ClientSession | undefined,
-    timeoutContext: TimeoutContext
-  ): Promise<void> {
+  override buildCommand(_connection: Connection, _session?: ClientSession): Document {
     const namespace = this.collection.fullNamespace;
 
     const command: Document = {
@@ -35,15 +33,18 @@ export class DropSearchIndexOperation extends AbstractOperation<void> {
     if (typeof this.name === 'string') {
       command.name = this.name;
     }
+    return command;
+  }
+
+  override buildOptions(timeoutContext: TimeoutContext): ServerCommandOptions {
+    return { session: this.session, timeoutContext };
+  }
 
-    try {
-      await server.command(namespace, command, { session, timeoutContext });
-    } catch (error) {
-      const isNamespaceNotFoundError =
-        error instanceof MongoServerError && error.code === MONGODB_ERROR_CODES.NamespaceNotFound;
-      if (!isNamespaceNotFoundError) {
-        throw error;
-      }
+  override handleError(error: MongoError): void {
+    const isNamespaceNotFoundError =
+      error instanceof MongoServerError && error.code === MONGODB_ERROR_CODES.NamespaceNotFound;
+    if (!isNamespaceNotFoundError) {
+      throw error;
     }
   }
 }

src/sdam/server.ts

@@ -38,8 +38,9 @@ import {
 import type { ServerApi } from '../mongo_client';
 import { type Abortable, TypedEventEmitter } from '../mongo_types';
 import type { GetMoreOptions } from '../operations/get_more';
+import { type ModernOperation } from '../operations/operation';
 import type { ClientSession } from '../sessions';
-import { type TimeoutContext } from '../timeout';
+import { Timeout, type TimeoutContext } from '../timeout';
 import { isTransactionCommand } from '../transactions';
 import {
   abortable,
@@ -277,6 +278,100 @@ export class Server extends TypedEventEmitter<ServerEvents> {
     }
   }
 
+  public async modernCommand(
+    operation: ModernOperation<any>,
+    timeoutContext: TimeoutContext
+  ): Promise<Document> {
+    if (this.s.state === STATE_CLOSING || this.s.state === STATE_CLOSED) {
+      throw new MongoServerClosedError();
+    }
+    const session = operation.session;
+
+    let conn = session?.pinnedConnection;
+
+    this.incrementOperationCount();
+    if (conn == null) {
+      try {
+        conn = await this.pool.checkOut({ timeoutContext });
+      } catch (checkoutError) {
+        this.decrementOperationCount();
+        if (!(checkoutError instanceof PoolClearedError)) this.handleError(checkoutError);
+        throw checkoutError;
+      }
+    }
+
+    const cmd = operation.buildCommand(conn, session);
+    const options = operation.buildOptions(timeoutContext);
+    const ns = operation.ns;
+
+    if (this.loadBalanced && isPinnableCommand(cmd, session)) {
+      session?.pin(conn);
+    }
+
+    options.directConnection = this.topology.s.options.directConnection;
+
+    // There are cases where we need to flag the read preference not to get sent in
+    // the command, such as pre-5.0 servers attempting to perform an aggregate write
+    // with a non-primary read preference. In this case the effective read preference
+    // (primary) is not the same as the provided and must be removed completely.
+    if (options.omitReadPreference) {
+      delete options.readPreference;
+    }
+
+    if (this.description.iscryptd) {
+      options.omitMaxTimeMS = true;
+    }
+
+    let reauthPromise: Promise<void> | null = null;
+
+    try {
+      try {
+        const res = await conn.command(ns, cmd, options);
+        throwIfWriteConcernError(res);
+        return res;
+      } catch (commandError) {
+        throw this.decorateCommandError(conn, cmd, options, commandError);
+      }
+    } catch (operationError) {
+      if (
+        operationError instanceof MongoError &&
+        operationError.code === MONGODB_ERROR_CODES.Reauthenticate
+      ) {
+        reauthPromise = this.pool.reauthenticate(conn);
+        reauthPromise.then(undefined, error => {
+          reauthPromise = null;
+          squashError(error);
+        });
+
+        await abortable(reauthPromise, options);
+        reauthPromise = null; // only reachable if reauth succeeds
+
+        try {
+          const res = await conn.command(ns, cmd, options);
+          throwIfWriteConcernError(res);
+          return res;
+        } catch (commandError) {
+          throw this.decorateCommandError(conn, cmd, options, commandError);
+        }
+      } else {
+        throw operationError;
+      }
+    } finally {
+      this.decrementOperationCount();
+      if (session?.pinnedConnection !== conn) {
+        if (reauthPromise != null) {
+          // The reauth promise only exists if it hasn't thrown.
+          const checkBackIn = () => {
+            this.pool.checkIn(conn);
+          };
+          void reauthPromise.then(checkBackIn, checkBackIn);
+        } else {
+          this.pool.checkIn(conn);
+        }
+      }
+    }
+  }
+
   public async command<T extends MongoDBResponseConstructor>(
     ns: MongoDBNamespace,
     command: Document,