remove unneeded changes

Author: W-A-James

src/bulk/common.ts

@@ -20,7 +20,6 @@ import { makeUpdateStatement, UpdateOperation, type UpdateStatement } from '../o
 import type { Server } from '../sdam/server';
 import type { Topology } from '../sdam/topology';
 import type { ClientSession } from '../sessions';
-import { type Timeout } from '../timeout';
 import { maybeAddIdToDocuments } from '../utils';
 import {
   applyRetryableWrites,
@@ -875,8 +874,6 @@ export interface BulkWriteOptions extends CommandOperationOptions {
   forceServerObjectId?: boolean;
   /** Map of parameter names and values that can be accessed using $$var (requires MongoDB 5.0). */
   let?: Document;
-  /** @internal */
-  timeout?: Timeout | null;
 }
 
 const executeCommandsAsync = promisify(executeCommands);

src/cmap/connection.ts

@@ -29,7 +29,6 @@ import { type CancellationToken, TypedEventEmitter } from '../mongo_types';
 import { ReadPreference, type ReadPreferenceLike } from '../read_preference';
 import { ServerType } from '../sdam/common';
 import { applySession, type ClientSession, updateSessionFromResponse } from '../sessions';
-import { Timeout } from '../timeout';
 import {
   BufferPool,
   calculateDurationInMs,
@@ -59,7 +58,6 @@ import {
   type WriteProtocolMessageType
 } from './commands';
 import type { Stream } from './connect';
-import { type ConnectionPool } from './connection_pool';
 import type { ClientMetadata } from './handshake/client_metadata';
 import { StreamDescription, type StreamDescriptionOptions } from './stream_description';
 import { type CompressorName, decompressResponse } from './wire_protocol/compression';
@@ -90,8 +88,6 @@ export interface CommandOptions extends BSONSerializeOptions {
   writeConcern?: WriteConcern;
 
   directConnection?: boolean;
-
-  timeout?: Timeout | null;
 }
 
 /** @public */
@@ -184,7 +180,6 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
    * Once connection is established, command logging can log events (if enabled)
    */
   public established: boolean;
-  public pool?: ConnectionPool;
   /** Indicates that the connection (including underlying TCP socket) has been closed. */
   public closed = false;
 
@@ -281,10 +276,6 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
     );
   }
 
-  private get minRoundTripTime(): number {
-    return this.pool?.server.description.minRoundTripTime ?? 0;
-  }
-
   public markAvailable(): void {
     this.lastUseTime = now();
   }
@@ -349,10 +340,6 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
 
     let clusterTime = this.clusterTime;
 
-    if (Timeout.is(options.timeout) && options.timeout.duration > 0) {
-      cmd.maxTimeMS = options.timeout.getMaxTimeMS(this.minRoundTripTime);
-    }
-
     if (this.serverApi) {
       const { version, strict, deprecationErrors } = this.serverApi;
       cmd.apiVersion = version;
@@ -442,8 +429,7 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
     try {
       await this.writeCommand(message, {
         agreedCompressor: this.description.compressor ?? 'none',
-        zlibCompressionLevel: this.description.zlibCompressionLevel,
-        timeout: options.timeout
+        zlibCompressionLevel: this.description.zlibCompressionLevel
       });
 
       if (options.noResponse) {
@@ -453,7 +439,7 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
 
       this.throwIfAborted();
 
-      for await (const response of this.readMany({ timeout: options.timeout })) {
+      for await (const response of this.readMany()) {
         this.socket.setTimeout(0);
         const bson = response.parse();
 
@@ -646,11 +632,7 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
    */
   private async writeCommand(
     command: WriteProtocolMessageType,
-    options: {
-      agreedCompressor?: CompressorName;
-      zlibCompressionLevel?: number;
-      timeout?: Timeout | null;
-    }
+    options: { agreedCompressor?: CompressorName; zlibCompressionLevel?: number }
   ): Promise<void> {
     const finalCommand =
       options.agreedCompressor === 'none' || !OpCompressedRequest.canCompress(command)
@@ -662,15 +644,8 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
 
     const buffer = Buffer.concat(await finalCommand.toBin());
 
-    if (this.socket.write(buffer)) {
-      return;
-    }
-    const drain = once(this.socket, 'drain');
-
-    if (options.timeout) {
-      await Promise.race([drain, options.timeout]);
-    }
-    await drain;
+    if (this.socket.write(buffer)) return;
+    return await once(this.socket, 'drain');
   }
 
   /**
@@ -682,11 +657,9 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
    *
    * Note that `for-await` loops call `return` automatically when the loop is exited.
    */
-  private async *readMany(options: {
-    timeout?: Timeout | null;
-  }): AsyncGenerator<OpMsgResponse | OpReply> {
+  private async *readMany(): AsyncGenerator<OpMsgResponse | OpReply> {
     try {
-      this.dataEvents = onData(this.messageStream, options);
+      this.dataEvents = onData(this.messageStream);
       for await (const message of this.dataEvents) {
         const response = await decompressResponse(message);
         yield response;

src/cmap/connection_pool.ts

@@ -168,8 +168,6 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
   [kMetrics]: ConnectionPoolMetrics;
   [kProcessingWaitQueue]: boolean;
 
-  server: Server;
-
   /**
    * Emitted when the connection pool is created.
    * @event
@@ -249,7 +247,6 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
 
     this[kPoolState] = PoolState.paused;
     this[kServer] = server;
-    this.server = server;
     this[kConnections] = new List();
     this[kPending] = 0;
     this[kCheckedOut] = new Set();

src/cmap/wire_protocol/on_data.ts

@@ -1,6 +1,5 @@
 import { type EventEmitter } from 'events';
 
-import { type Timeout } from '../../timeout';
 import { List, promiseWithResolvers } from '../../utils';
 
 /**
@@ -19,7 +18,7 @@ type PendingPromises = Omit<
  * Returns an AsyncIterator that iterates each 'data' event emitted from emitter.
  * It will reject upon an error event.
  */
-export function onData(emitter: EventEmitter, options?: { timeout?: Timeout | null }) {
+export function onData(emitter: EventEmitter) {
   // Setup pending events and pending promise lists
   /**
    * When the caller has not yet called .next(), we store the
@@ -87,8 +86,6 @@ export function onData(emitter: EventEmitter, options?: { timeout?: Timeout | nu
   // Adding event handlers
   emitter.on('data', eventHandler);
   emitter.on('error', errorHandler);
-  // eslint-disable-next-line github/no-then
-  options?.timeout?.then(() => null, errorHandler);
 
   return iterator;
 

src/operations/bulk_write.ts

@@ -36,12 +36,7 @@ export class BulkWriteOperation extends AbstractOperation<BulkWriteResult> {
   ): Promise<BulkWriteResult> {
     const coll = this.collection;
     const operations = this.operations;
-    const options = {
-      ...this.options,
-      ...this.bsonOptions,
-      readPreference: this.readPreference,
-      timeout: this.timeout
-    };
+    const options = { ...this.options, ...this.bsonOptions, readPreference: this.readPreference };
 
     // Create the bulk operation
     const bulk: BulkOperationBase =

src/operations/command.ts

@@ -118,8 +118,7 @@ export abstract class CommandOperation<T> extends AbstractOperation<T> {
       ...this.options,
       ...this.bsonOptions,
       readPreference: this.readPreference,
-      session,
-      timeout: this.timeout
+      session
     };
 
     const serverWireVersion = maxWireVersion(server);

src/operations/execute_operation.ts

@@ -187,9 +187,6 @@ export async function executeOperation<
   }
 
   try {
-    if (operation.timeout) {
-      return await Promise.race([operation.execute(server, session), operation.timeout]);
-    }
     return await operation.execute(server, session);
   } catch (operationError) {
     if (willRetry && operationError instanceof MongoError) {
@@ -263,8 +260,7 @@ async function retryOperation<
   const server = await topology.selectServer(selector, {
     session,
     operationName: operation.commandName,
-    previousServer,
-    timeout: operation.timeout
+    previousServer
   });
 
   if (isWriteOperation && !supportsRetryableWrites(server)) {
@@ -274,9 +270,6 @@ async function retryOperation<
   }
 
   try {
-    if (operation.timeout) {
-      return await Promise.race([operation.execute(server, session), operation.timeout]);
-    }
     return await operation.execute(server, session);
   } catch (retryError) {
     if (

src/operations/find.ts

@@ -115,8 +115,7 @@ export class FindOperation extends CommandOperation<Document> {
       ...this.options,
       ...this.bsonOptions,
       documentsReturnedIn: 'firstBatch',
-      session,
-      timeout: this.timeout
+      session
     });
   }
 }

src/operations/operation.ts

@@ -2,7 +2,6 @@ import { type BSONSerializeOptions, type Document, resolveBSONOptions } from '..
 import { ReadPreference, type ReadPreferenceLike } from '../read_preference';
 import type { Server } from '../sdam/server';
 import type { ClientSession } from '../sessions';
-import { Timeout } from '../timeout';
 import type { MongoDBNamespace } from '../utils';
 
 export const Aspect = {
@@ -64,10 +63,7 @@ export abstract class AbstractOperation<TResult = any> {
 
   [kSession]: ClientSession | undefined;
 
-  timeout?: Timeout | null;
-
   constructor(options: OperationOptions = {}) {
-    this.timeout = options.timeoutMS != null ? Timeout.expires(options.timeoutMS) : null;
     this.readPreference = this.hasAspect(Aspect.WRITE_OPERATION)
       ? ReadPreference.primary
       : ReadPreference.fromOptions(options) ?? ReadPreference.primary;

src/operations/run_command.ts

@@ -13,8 +13,6 @@ export type RunCommandOptions = {
   session?: ClientSession;
   /** The read preference */
   readPreference?: ReadPreferenceLike;
-  /** @internal */
-  timeoutMS?: number;
 } & BSONSerializeOptions;
 
 /** @internal */
@@ -33,8 +31,7 @@ export class RunCommandOperation<T = Document> extends AbstractOperation<T> {
     const res: TODO_NODE_3286 = await server.command(this.ns, this.command, {
       ...this.options,
       readPreference: this.readPreference,
-      session,
-      timeout: this.timeout
+      session
     });
     return res;
   }

test/integration/client-side-operations-timeout/client_side_operations_timeout.spec.test.ts

@@ -4,6 +4,6 @@ import { loadSpecTests } from '../../spec';
 import { runUnifiedSuite } from '../../tools/unified-spec-runner/runner';
 
 // TODO(NODE-5823): Implement unified runner operations and options support for CSOT
-describe('CSOT spec tests', function () {
+describe.skip('CSOT spec tests', function () {
   runUnifiedSuite(loadSpecTests(join('client-side-operations-timeout')));
 });