Automatic Exponential Backoff for Account & Transaction Stream Polling

src/account/streamAccount.ts

@@ -5,6 +5,12 @@ import type { SorokitLogger } from "../shared/logger";
 import type { AccountInfo, BalanceAlert, BalanceAlertRule } from "./types";
 import { getAccount } from "./getAccount";
 import { evaluateBalanceAlerts } from "./balanceAlerts";
+import {
+  retryStreamingPoll,
+  type StreamingRetryState,
+  type StreamingRetryConfig,
+} from "../shared/utils";
+import { isTransientError } from "../shared/errors";
 
 const MIN_POLL_INTERVAL_MS = 1_000;
 const DEFAULT_POLL_INTERVAL_MS = 5_000;
@@ -84,6 +90,14 @@ export interface AccountStreamConfig {
    * Fired after balance event callbacks for the same poll.
    */
   onAlert?: (alert: BalanceAlert) => void;
+  /**
+   * Enable automatic retry with exponential backoff for transient network errors.
+   * When enabled, transient errors (timeouts, network issues, 5xx) trigger automatic
+   * retry with exponential backoff (1s, 2s, 4s, 8s, 16s, then 30s max). After 5 consecutive
+   * failures, an error is emitted and polling pauses for 60s before resuming.
+   * Default: true.
+   */
+  enableAutoRetry?: boolean;
 }
 
 /**
@@ -151,6 +165,15 @@ export async function* streamAccount(
   const maxPolls = config?.maxPolls;
   const emitOnStart = config?.emitOnStart ?? true;
 
+  // Retry configuration
+  const retryConfig: StreamingRetryConfig = {
+    enabled: config?.enableAutoRetry ?? true,
+  };
+  let retryState: StreamingRetryState = {
+    consecutiveFailures: 0,
+    inCooldown: false,
+  };
+
   let polls = 0;
   let currentIntervalMs = Math.min(
     Math.max(baseIntervalMs, minIntervalMs),
@@ -215,6 +238,9 @@ export async function* streamAccount(
     if (signal?.aborted) return;
 
     const pollStartedAt = Date.now();
+    let pollSuccess = false;
+    let errorResult: SorokitResult<AccountInfo> | null = null;
+
     try {
       logger?.debug("account.stream.poll", {
         operation: "account.stream.poll",
@@ -294,6 +320,8 @@ export async function* streamAccount(
           for (const alert of alerts) config.onAlert(alert);
         }
 
+        pollSuccess = true;
+
       } else {
         logger?.warn("account.stream.poll", {
           operation: "account.stream.poll",
@@ -303,6 +331,8 @@ export async function* streamAccount(
           errorCode: result.error.code,
           errorMessage: result.error.message,
         });
+
+        errorResult = result;
       }
 
       if (result.status === "ok") {
@@ -316,7 +346,6 @@ export async function* streamAccount(
         }
       } else {
         adjustInterval(false);
-        yield result;
       }
     } catch (cause) {
       const message = `Account stream poll failed: ${toMessage(cause)}`;
@@ -327,8 +356,48 @@ export async function* streamAccount(
         poll: polls + 1,
         errorMessage: message,
       });
-      yield err(SorokitErrorCode.ACCOUNT_FETCH_FAILED, message, cause);
-    } finally {
+
+      errorResult = err(SorokitErrorCode.ACCOUNT_FETCH_FAILED, message, cause);
+    }
+
+    // Handle retry logic after poll attempt
+    const isTransient = errorResult && errorResult.error ? isTransientError(errorResult.error.cause || errorResult.error) : false;
+    const retryDecision = retryStreamingPoll(
+      pollSuccess,
+      retryState,
+      retryConfig,
+    );
+    retryState = retryDecision.updatedState;
+
+    // Determine if we should yield the error
+    // Yield error if: not a transient error, or we're in cooldown, or auto-retry is disabled
+    const shouldYieldError = errorResult && 
+      (!isTransient || !retryDecision.shouldRetry || retryState.inCooldown);
+
+    // Yield error if needed
+    if (shouldYieldError && errorResult) {
+      yield errorResult;
+    }
+
+    // Calculate next delay
+    if (retryDecision.shouldRetry && isTransient && !retryState.inCooldown) {
+      nextDelayMs = retryDecision.delayMs;
+      logger?.debug("account.stream.retry", {
+        operation: "account.stream.retry",
+        status: "retrying",
+        publicKey,
+        consecutiveFailures: retryState.consecutiveFailures,
+        delayMs: nextDelayMs,
+      });
+    } else if (retryState.inCooldown) {
+      nextDelayMs = retryDecision.delayMs;
+      logger?.debug("account.stream.retry", {
+        operation: "account.stream.retry",
+        status: "cooldown",
+        publicKey,
+        delayMs: nextDelayMs,
+      });
+    } else {
       nextDelayMs = getLatencyCompensatedDelay(
         currentIntervalMs,
         Date.now() - pollStartedAt,

src/shared/utils.ts

@@ -548,3 +548,148 @@ export function getInflightRequestCount(): number {
 export function clearInflightRequests(): void {
   _inflightRequests.clear();
 }
+
+/**
+ * Configuration for streaming poll retry behavior.
+ */
+export interface StreamingRetryConfig {
+  /** Enable automatic retry with exponential backoff. Default: true. */
+  enabled?: boolean;
+  /** Maximum consecutive failures before emitting error and entering cooldown. Default: 5. */
+  maxConsecutiveFailures?: number;
+  /** Cooldown period in milliseconds after max consecutive failures. Default: 60000 (60s). */
+  cooldownMs?: number;
+  /** Initial backoff delay in milliseconds. Default: 1000 (1s). */
+  initialBackoffMs?: number;
+  /** Maximum backoff delay in milliseconds. Default: 30000 (30s). */
+  maxBackoffMs?: number;
+  /** Whether to add random jitter to backoff delays. Default: true. */
+  jitter?: boolean;
+}
+
+/**
+ * Default streaming retry configuration.
+ */
+const DEFAULT_STREAMING_RETRY_CONFIG: Required<StreamingRetryConfig> = {
+  enabled: true,
+  maxConsecutiveFailures: 5,
+  cooldownMs: 60_000,
+  initialBackoffMs: 1_000,
+  maxBackoffMs: 30_000,
+  jitter: true,
+};
+
+/**
+ * State for tracking retry attempts across stream polls.
+ */
+export interface StreamingRetryState {
+  consecutiveFailures: number;
+  inCooldown: boolean;
+}
+
+/**
+ * Calculate exponential backoff delay with jitter.
+ * Progression: 1s, 2s, 4s, 8s, 16s, then capped at 30s.
+ */
+function calculateBackoff(
+  attempt: number,
+  initialBackoffMs: number,
+  maxBackoffMs: number,
+  jitter: boolean,
+): number {
+  const baseDelay = initialBackoffMs * Math.pow(2, attempt);
+  const cappedDelay = Math.min(baseDelay, maxBackoffMs);
+  const jitterMs = jitter ? Math.random() * cappedDelay * 0.1 : 0;
+  return cappedDelay + jitterMs;
+}
+
+/**
+ * Handle retry logic for streaming polls with exponential backoff.
+ *
+ * This function is designed to be called within stream generators to handle
+ * transient errors automatically. It implements:
+ * - Exponential backoff (1s, 2s, 4s, 8s, 16s, capped at 30s)
+ * - Circuit breaker after 5 consecutive failures (60s cooldown)
+ * - Automatic state reset on success
+ *
+ * @param success - Whether the poll succeeded
+ * @param state - Current retry state (consecutive failures, cooldown status)
+ * @param config - Retry configuration
+ * @returns Object indicating whether to retry, delay in ms, and updated state
+ */
+export function retryStreamingPoll(
+  success: boolean,
+  state: StreamingRetryState,
+  config: StreamingRetryConfig = {},
+): {
+  shouldRetry: boolean;
+  delayMs: number;
+  updatedState: StreamingRetryState;
+} {
+  const {
+    enabled,
+    maxConsecutiveFailures,
+    cooldownMs,
+    initialBackoffMs,
+    maxBackoffMs,
+    jitter,
+  } = { ...DEFAULT_STREAMING_RETRY_CONFIG, ...config };
+
+  // Reset state on success
+  if (success) {
+    return {
+      shouldRetry: false,
+      delayMs: 0,
+      updatedState: { consecutiveFailures: 0, inCooldown: false },
+    };
+  }
+
+  // If auto-retry is disabled, don't retry
+  if (!enabled) {
+    return {
+      shouldRetry: false,
+      delayMs: 0,
+      updatedState: state,
+    };
+  }
+
+  // If we're in cooldown, continue cooldown and don't retry
+  if (state.inCooldown) {
+    return {
+      shouldRetry: false,
+      delayMs: cooldownMs,
+      updatedState: state,
+    };
+  }
+
+  // Increment failure counter
+  const updatedState: StreamingRetryState = {
+    consecutiveFailures: state.consecutiveFailures + 1,
+    inCooldown: false,
+  };
+
+  // Check if we've hit the failure threshold
+  if (updatedState.consecutiveFailures >= maxConsecutiveFailures) {
+    // Enter cooldown mode
+    updatedState.inCooldown = true;
+    return {
+      shouldRetry: false,
+      delayMs: cooldownMs,
+      updatedState,
+    };
+  }
+
+  // Calculate backoff delay and retry
+  const delayMs = calculateBackoff(
+    updatedState.consecutiveFailures - 1,
+    initialBackoffMs,
+    maxBackoffMs,
+    jitter,
+  );
+
+  return {
+    shouldRetry: true,
+    delayMs,
+    updatedState,
+  };
+}