latest changes

Author: baileympearson

src/operations/execute_operation.ts

@@ -32,13 +32,12 @@ import { TimeoutContext } from '../timeout';
 import { RETRY_COST, TOKEN_REFRESH_RATE } from '../token_bucket';
 import {
   abortable,
-  exponentialBackoffDelayProvider,
+  ExponentialBackoffProvider,
   maxWireVersion,
   supportsRetryableWrites
 } from '../utils';
 import { AggregateOperation } from './aggregate';
 import { AbstractOperation, Aspect } from './operation';
-import { RunCommandOperation } from './run_command';
 
 const MMAPv1_RETRY_WRITES_ERROR_CODE = MONGODB_ERROR_CODES.IllegalOperation;
 const MMAPv1_RETRY_WRITES_ERROR_MESSAGE =
@@ -246,46 +245,48 @@ async function executeOperationWithRetries<
     session.incrementTransactionNumber();
   }
 
-  // The maximum number of retry attempts using regular retryable reads/writes logic (not including
-  // SystemOverLoad error retries).
-  const maxNonOverloadRetryAttempts = willRetry ? (timeoutContext.csotEnabled() ? Infinity : 2) : 1;
   let previousOperationError: MongoError | undefined;
   const deprioritizedServers = new DeprioritizedServers();
-  let nonOverloadRetryAttempt = 0;
 
-  let systemOverloadRetryAttempt = 0;
-  const maxSystemOverloadRetryAttempts = 5;
-  const backoffDelayProvider = exponentialBackoffDelayProvider(
+  const backoffDelayProvider = new ExponentialBackoffProvider(
     10_000, // MAX_BACKOFF
     100, // base backoff
     2 // backoff rate
   );
 
-  while (true) {
+  for (
+    let attempt = 0, maxAttempts = willRetry ? (timeoutContext.csotEnabled() ? Infinity : 2) : 1;
+    attempt <= maxAttempts;
+    attempt++,
+      maxAttempts = previousOperationError?.hasErrorLabel(MongoErrorLabel.SystemOverloadedError)
+        ? 5
+        : maxAttempts
+  ) {
     if (previousOperationError) {
-      if (previousOperationError.hasErrorLabel(MongoErrorLabel.SystemOverloadedError)) {
-        systemOverloadRetryAttempt += 1;
-
-        // if retryable writes or reads are not configured, throw.
-        const isOperationConfiguredForRetry =
-          (hasReadAspect && topology.s.options.retryReads) ||
-          (hasWriteAspect && topology.s.options.retryWrites);
-        const isRunCommand = operation instanceof RunCommandOperation;
-
-        if (
-          // if the SystemOverloadError is not retryable, throw.
-          !previousOperationError.hasErrorLabel(MongoErrorLabel.RetryableError) ||
-          !(isOperationConfiguredForRetry || isRunCommand)
-        ) {
-          throw previousOperationError;
-        }
+      if (hasWriteAspect && previousOperationError.code === MMAPv1_RETRY_WRITES_ERROR_CODE) {
+        throw new MongoServerError({
+          message: MMAPv1_RETRY_WRITES_ERROR_MESSAGE,
+          errmsg: MMAPv1_RETRY_WRITES_ERROR_MESSAGE,
+          originalError: previousOperationError
+        });
+      }
 
-        // if we have exhausted overload retry attempts, throw.
-        if (systemOverloadRetryAttempt > maxSystemOverloadRetryAttempts) {
-          throw previousOperationError;
-        }
+      const isRetryable =
+        // bulk write commands are retryable if all operations in the batch are retryable
+        (operation.hasAspect(Aspect.COMMAND_BATCHING) && operation.canRetryWrite) ||
+        // if we have a retryable read or write operation, we can retry
+        (hasWriteAspect && isRetryableWriteError(previousOperationError)) ||
+        (hasReadAspect && isRetryableReadError(previousOperationError)) ||
+        // if we have a retryable, system overloaded error, we can retry
+        (previousOperationError.hasErrorLabel(MongoErrorLabel.SystemOverloadedError) &&
+          previousOperationError.hasErrorLabel(MongoErrorLabel.RetryableError));
+
+      if (!isRetryable) {
+        throw previousOperationError;
+      }
 
-        const { value: delayMS } = backoffDelayProvider.next();
+      if (previousOperationError.hasErrorLabel(MongoErrorLabel.SystemOverloadedError)) {
+        const delayMS = backoffDelayProvider.getNextBackoffDuration();
 
         // if the delay would exhaust the CSOT timeout, short-circuit.
         if (timeoutContext.csotEnabled() && delayMS > timeoutContext.remainingTimeMS) {
@@ -298,80 +299,49 @@ async function executeOperationWithRetries<
           );
         }
 
-        await setTimeout(delayMS);
-
         if (!topology.tokenBucket.consume(RETRY_COST)) {
           throw previousOperationError;
         }
 
-        server = await topology.selectServer(selector, {
-          session,
-          operationName: operation.commandName,
-          deprioritizedServers,
-          signal: operation.options.signal
-        });
-      } else {
-        nonOverloadRetryAttempt++;
-        // we have no more retry attempts, throw.
-        if (nonOverloadRetryAttempt >= maxNonOverloadRetryAttempts) {
-          throw previousOperationError;
-        }
-
-        if (hasWriteAspect && previousOperationError.code === MMAPv1_RETRY_WRITES_ERROR_CODE) {
-          throw new MongoServerError({
-            message: MMAPv1_RETRY_WRITES_ERROR_MESSAGE,
-            errmsg: MMAPv1_RETRY_WRITES_ERROR_MESSAGE,
-            originalError: previousOperationError
-          });
-        }
+        await setTimeout(delayMS);
+      }
 
-        if (
-          (operation.hasAspect(Aspect.COMMAND_BATCHING) && !operation.canRetryWrite) ||
-          (hasWriteAspect && !isRetryableWriteError(previousOperationError)) ||
-          (hasReadAspect && !isRetryableReadError(previousOperationError))
-        ) {
-          throw previousOperationError;
-        }
+      if (
+        previousOperationError instanceof MongoNetworkError &&
+        operation.hasAspect(Aspect.CURSOR_CREATING) &&
+        session != null &&
+        session.isPinned &&
+        !session.inTransaction()
+      ) {
+        session.unpin({ force: true, forceClear: true });
+      }
 
-        if (
-          previousOperationError instanceof MongoNetworkError &&
-          operation.hasAspect(Aspect.CURSOR_CREATING) &&
-          session != null &&
-          session.isPinned &&
-          !session.inTransaction()
-        ) {
-          session.unpin({ force: true, forceClear: true });
-        }
+      server = await topology.selectServer(selector, {
+        session,
+        operationName: operation.commandName,
+        deprioritizedServers,
+        signal: operation.options.signal
+      });
 
-        server = await topology.selectServer(selector, {
-          session,
-          operationName: operation.commandName,
-          deprioritizedServers,
-          signal: operation.options.signal
-        });
-
-        if (hasWriteAspect && !supportsRetryableWrites(server)) {
-          throw new MongoUnexpectedServerResponseError(
-            'Selected server does not support retryable writes'
-          );
-        }
+      if (hasWriteAspect && !supportsRetryableWrites(server)) {
+        throw new MongoUnexpectedServerResponseError(
+          'Selected server does not support retryable writes'
+        );
       }
     }
 
     operation.server = server;
 
     try {
+      const isRetry = attempt > 0;
+
       // If attempt > 0 and we are command batching we need to reset the batch.
-      if (
-        (nonOverloadRetryAttempt > 0 || systemOverloadRetryAttempt > 0) &&
-        operation.hasAspect(Aspect.COMMAND_BATCHING)
-      ) {
+      if (isRetry && operation.hasAspect(Aspect.COMMAND_BATCHING)) {
         operation.resetBatch();
       }
 
       try {
         const result = await server.command(operation, timeoutContext);
-        const isRetry = nonOverloadRetryAttempt > 0 || systemOverloadRetryAttempt > 0;
         topology.tokenBucket.deposit(
           isRetry
             ? // on successful retry, deposit the retry cost + the refresh rate.
@@ -404,4 +374,6 @@ async function executeOperationWithRetries<
       timeoutContext.clear();
     }
   }
+
+  throw previousOperationError ?? new MongoRuntimeError('ahh');
 }

src/utils.ts

@@ -1432,12 +1432,18 @@ export async function abortable<T>(
   }
 }
 
-export function* exponentialBackoffDelayProvider(
-  maxBackoff: number,
-  baseBackoff: number,
-  backoffIncreaseRate: number
-): Generator<number> {
-  for (let i = 0; ; i++) {
-    yield Math.random() * Math.min(maxBackoff, baseBackoff * backoffIncreaseRate ** i);
+export class ExponentialBackoffProvider {
+  constructor(
+    public readonly maxBackoff: number,
+    public readonly baseBackoff: number,
+    public readonly backoffIncreaseRate: number,
+    public iteration = 0
+  ) {}
+
+  getNextBackoffDuration(): number {
+    return (
+      Math.random() *
+      Math.min(this.maxBackoff, this.baseBackoff * this.backoffIncreaseRate ** this.iteration)
+    );
   }
 }