feat(CSOT) - feature branch

src/cmap/connection_pool.ts

@@ -103,8 +103,8 @@ export interface ConnectionPoolOptions extends Omit<ConnectionOptions, 'id' | 'g
 export interface WaitQueueMember {
   resolve: (conn: Connection) => void;
   reject: (err: AnyError) => void;
-  timeout: Timeout;
-  clearTimeout: boolean;
+  timeout: Timeout | null;
+  clearTimeout: boolean | null;
   [kCancelled]?: boolean;
 }
 
@@ -367,19 +367,21 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
 
     const { promise, resolve, reject } = promiseWithResolvers<Connection>();
 
-    let timeout: Timeout;
-    let clearTimeout: boolean;
+    let timeout: Timeout | null = null;
+    let clearTimeout: boolean | null = null;
     if (options?.timeout) {
       // CSOT enabled
       // Determine if we're using the timeout passed in or a new timeout
-      if (
-        csotMin(options.timeout.duration, serverSelectionTimeoutMS) === serverSelectionTimeoutMS
-      ) {
-        timeout = Timeout.expires(serverSelectionTimeoutMS - options.timeout.timeElapsed);
-        clearTimeout = true;
-      } else {
-        timeout = options.timeout;
-        clearTimeout = false;
+      if (options.timeout.duration > 0 || serverSelectionTimeoutMS > 0) {
+        if (
+          csotMin(options.timeout.duration, serverSelectionTimeoutMS) === serverSelectionTimeoutMS
+        ) {
+          timeout = Timeout.expires(serverSelectionTimeoutMS - options.timeout.timeElapsed);
+          clearTimeout = true;
+        } else {
+          timeout = options.timeout;
+          clearTimeout = false;
+        }
       }
     } else {
       timeout = Timeout.expires(waitQueueTimeoutMS);
@@ -397,8 +399,8 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
     process.nextTick(() => this.processWaitQueue());
 
     try {
-      waitQueueMember.timeout.throwIfExpired();
-      return await Promise.race([promise, waitQueueMember.timeout]);
+      timeout?.throwIfExpired();
+      return await (timeout ? Promise.race([promise, timeout]) : promise);
     } catch (error) {
       if (TimeoutError.is(error)) {
         waitQueueMember[kCancelled] = true;
@@ -422,7 +424,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
       }
       throw error;
     } finally {
-      if (clearTimeout) timeout.clear();
+      if (clearTimeout) timeout?.clear();
     }
   }
 
@@ -787,7 +789,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
           ConnectionPool.CONNECTION_CHECK_OUT_FAILED,
           new ConnectionCheckOutFailedEvent(this, reason, error)
         );
-        if (waitQueueMember.clearTimeout) waitQueueMember.timeout.clear();
+        if (waitQueueMember.clearTimeout) waitQueueMember.timeout?.clear();
         this[kWaitQueue].shift();
         waitQueueMember.reject(error);
         continue;
@@ -808,7 +810,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
           ConnectionPool.CONNECTION_CHECKED_OUT,
           new ConnectionCheckedOutEvent(this, connection)
         );
-        if (waitQueueMember.clearTimeout) waitQueueMember.timeout.clear();
+        if (waitQueueMember.clearTimeout) waitQueueMember.timeout?.clear();
 
         this[kWaitQueue].shift();
         waitQueueMember.resolve(connection);
@@ -847,7 +849,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
             waitQueueMember.resolve(connection);
           }
 
-          if (waitQueueMember.clearTimeout) waitQueueMember.timeout.clear();
+          if (waitQueueMember.clearTimeout) waitQueueMember.timeout?.clear();
         }
         process.nextTick(() => this.processWaitQueue());
       });

src/sdam/topology.ts

@@ -108,8 +108,8 @@ export interface ServerSelectionRequest {
   resolve: (server: Server) => void;
   reject: (error: MongoError) => void;
   [kCancelled]?: boolean;
-  clearTimeout: boolean;
-  timeout: Timeout;
+  clearTimeout: boolean | null;
+  timeout: Timeout | null;
   operationName: string;
   waitingLogged: boolean;
   previousServer?: ServerDescription;
@@ -565,19 +565,21 @@ export class Topology extends TypedEventEmitter<TopologyEvents> {
       );
     }
     const serverSelectionTimeoutMS = options.serverSelectionTimeoutMS ?? 0;
-    let timeout: Timeout;
-    let clearTimeout: boolean;
+    let timeout: Timeout | null = null;
+    let clearTimeout: boolean | null = null;
     if (options.timeout) {
       // CSOT Enabled
-      if (
-        Math.min(options.timeout.remainingTime, serverSelectionTimeoutMS) ===
-        serverSelectionTimeoutMS
-      ) {
-        timeout = Timeout.expires(serverSelectionTimeoutMS);
-        clearTimeout = true;
-      } else {
-        timeout = options.timeout;
-        clearTimeout = false;
+      if (options.timeout.duration !== 0 || serverSelectionTimeoutMS !== 0) {
+        if (
+          Math.min(options.timeout.remainingTime, serverSelectionTimeoutMS) ===
+          serverSelectionTimeoutMS
+        ) {
+          timeout = Timeout.expires(serverSelectionTimeoutMS);
+          clearTimeout = true;
+        } else {
+          timeout = options.timeout;
+          clearTimeout = false;
+        }
       }
     } else {
       timeout = Timeout.expires(serverSelectionTimeoutMS);
@@ -629,8 +631,8 @@ export class Topology extends TypedEventEmitter<TopologyEvents> {
     processWaitQueue(this);
 
     try {
-      timeout.throwIfExpired();
-      return await Promise.race([serverPromise, waitQueueMember.timeout]);
+      timeout?.throwIfExpired();
+      return await (timeout ? Promise.race([serverPromise, timeout]) : serverPromise);
     } catch (error) {
       if (TimeoutError.is(error)) {
         // Timeout
@@ -666,7 +668,7 @@ export class Topology extends TypedEventEmitter<TopologyEvents> {
       // Other server selection error
       throw error;
     } finally {
-      if (clearTimeout) timeout.clear();
+      if (clearTimeout) timeout?.clear();
     }
   }
   /**
@@ -924,7 +926,7 @@ function drainWaitQueue(queue: List<ServerSelectionRequest>, drainError: MongoDr
       continue;
     }
 
-    if (waitQueueMember.clearTimeout) waitQueueMember.timeout.clear();
+    if (waitQueueMember.clearTimeout) waitQueueMember.timeout?.clear();
 
     if (!waitQueueMember[kCancelled]) {
       if (
@@ -979,7 +981,7 @@ function processWaitQueue(topology: Topology) {
           )
         : serverDescriptions;
     } catch (selectorError) {
-      if (waitQueueMember.clearTimeout) waitQueueMember.timeout.clear();
+      if (waitQueueMember.clearTimeout) waitQueueMember.timeout?.clear();
       if (
         topology.client.mongoLogger?.willLog(
           MongoLoggableComponent.SERVER_SELECTION,
@@ -1067,7 +1069,7 @@ function processWaitQueue(topology: Topology) {
       transaction.pinServer(selectedServer);
     }
 
-    if (waitQueueMember.clearTimeout) waitQueueMember.timeout.clear();
+    if (waitQueueMember.clearTimeout) waitQueueMember.timeout?.clear();
 
     if (
       topology.client.mongoLogger?.willLog(