diff --git a/src/account/streamAccount.ts b/src/account/streamAccount.ts index 2b88c94..9b63044 100644 --- a/src/account/streamAccount.ts +++ b/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 | 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, diff --git a/src/shared/utils.ts b/src/shared/utils.ts index 52fc3aa..c9b70e9 100644 --- a/src/shared/utils.ts +++ b/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 = { + 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, + }; +} diff --git a/src/tests/account.test.ts b/src/tests/account.test.ts index c091f8c..0a84b86 100644 --- a/src/tests/account.test.ts +++ b/src/tests/account.test.ts @@ -33,8 +33,17 @@ vi.mock("../account/getAccount", () => ({ }), })); -vi.mock("@stellar/stellar-sdk", async (importOriginal) => { - const actual = await importOriginal(); +// Add error state for retry tests +const accountErrorMockState = vi.hoisted(() => ({ + sleepCalls: [] as number[], + errors: [] as Error[], + index: 0, + shouldFail: false, + failCount: 0, +})); + +vi.mock("@stellar/stellar-sdk", async (importOriginal: any) => { + const actual = await importOriginal("@stellar/stellar-sdk"); return { ...actual, Horizon: { @@ -1480,7 +1489,7 @@ describe("streamAccount — onBalanceChange callback (#11)", () => { const a2 = createAccount("2"); const callCounts: Record = {}; - vi.mocked(getAccount).mockImplementation(async (_url, key) => { + vi.mocked(getAccount).mockImplementation(async (_url: string, key: string) => { callCounts[key] = (callCounts[key] ?? 0) + 1; return key === "key1" ? ok(a1) : ok(a2); }); @@ -1714,7 +1723,7 @@ describe("getMultipleAssetBalances — bulk account queries (#42)", () => { const DELAY = 30; vi.mocked(getAccount).mockImplementation( - (_, publicKey) => + (_: string, publicKey: string) => new Promise((resolve) => setTimeout(() => resolve(ok(makeAccount(publicKey, "1"))), DELAY), ), @@ -2553,4 +2562,160 @@ describe("streamAccount — emitOnStart, maxPolls, AbortSignal, mid-stream error expect(results[1]).toEqual(expect.objectContaining({ status: "error" })); expect(results[2]).toEqual(expect.objectContaining({ status: "ok" })); }, 10_000); + + describe("automatic retry with exponential backoff", () => { + beforeEach(() => { + accountMockState.sleepCalls.length = 0; + }); + + it("retries transient errors with exponential backoff", async () => { + const { getAccount } = await import("../account/getAccount"); + const { streamAccount } = await import("../account/streamAccount"); + + const account1 = createAccount("1"); + const transientError = new Error("ECONNREFUSED"); + + vi.mocked(getAccount) + .mockRejectedValueOnce(transientError) + .mockRejectedValueOnce(transientError) + .mockResolvedValueOnce(ok(account1)); + + const results: unknown[] = []; + for await (const r of streamAccount("https://horizon.test", "G...", { + maxPolls: 3, + emitOnStart: true, + intervalMs: 1, + enableAutoRetry: true, + })) { + results.push(r); + } + + // Should have retry delays (1s, 2s for exponential backoff) + expect(accountMockState.sleepCalls.length).toBeGreaterThan(0); + // First retry should be around 1s, second around 2s + expect(accountMockState.sleepCalls[0]).toBeGreaterThanOrEqual(1000); + expect(accountMockState.sleepCalls[0]).toBeLessThanOrEqual(1100); // with jitter + if (accountMockState.sleepCalls.length > 1) { + expect(accountMockState.sleepCalls[1]).toBeGreaterThanOrEqual(2000); + expect(accountMockState.sleepCalls[1]).toBeLessThanOrEqual(2200); // with jitter + } + + // Should eventually succeed + expect(results.some((r: any) => r?.status === "ok")).toBe(true); + }, 10_000); + + it("emits error after max consecutive failures and enters cooldown", async () => { + const { getAccount } = await import("../account/getAccount"); + const { streamAccount } = await import("../account/streamAccount"); + + const transientError = new Error("ETIMEDOUT"); + + vi.mocked(getAccount).mockRejectedValue(transientError); + + const results: unknown[] = []; + for await (const r of streamAccount("https://horizon.test", "G...", { + maxPolls: 7, // Allow enough polls to hit max consecutive failures (5) + emitOnStart: true, + intervalMs: 1, + enableAutoRetry: true, + })) { + results.push(r); + if (results.length >= 2) break; // Stop after we get the error emission + } + + // Should have emitted an error after 5 consecutive failures + expect(results.some((r: any) => r?.status === "error")).toBe(true); + + // Should have entered cooldown (60s delay) + const cooldownIndex = accountMockState.sleepCalls.findIndex( + (ms: number) => ms >= 60000 + ); + expect(cooldownIndex).toBeGreaterThanOrEqual(0); + }, 10_000); + + it("resets failure counter on successful poll", async () => { + const { getAccount } = await import("../account/getAccount"); + const { streamAccount } = await import("../account/streamAccount"); + + const account1 = createAccount("1"); + const account2 = createAccount("2"); + const transientError = new Error("ECONNRESET"); + + vi.mocked(getAccount) + .mockRejectedValueOnce(transientError) + .mockRejectedValueOnce(transientError) + .mockResolvedValueOnce(ok(account1)) + .mockRejectedValueOnce(transientError) + .mockRejectedValueOnce(transientError) + .mockRejectedValueOnce(transientError) + .mockResolvedValueOnce(ok(account2)); + + const results: unknown[] = []; + for await (const r of streamAccount("https://horizon.test", "G...", { + maxPolls: 7, + emitOnStart: true, + intervalMs: 1, + enableAutoRetry: true, + })) { + results.push(r); + } + + // Should succeed overall despite multiple failure sequences + expect(results.filter((r: any) => r?.status === "ok").length).toBe(2); + }, 10_000); + + it("does not retry when enableAutoRetry is false", async () => { + const { getAccount } = await import("../account/getAccount"); + const { streamAccount } = await import("../account/streamAccount"); + + const account1 = createAccount("1"); + const transientError = new Error("ETIMEDOUT"); + + vi.mocked(getAccount) + .mockRejectedValueOnce(transientError) + .mockResolvedValueOnce(ok(account1)); + + const results: unknown[] = []; + for await (const r of streamAccount("https://horizon.test", "G...", { + maxPolls: 2, + emitOnStart: true, + intervalMs: 1, + enableAutoRetry: false, + })) { + results.push(r); + } + + // Should emit error immediately without retry backoff + expect(results.some((r: any) => r?.status === "error")).toBe(true); + // Should not have retry delays (only normal interval) + expect(accountMockState.sleepCalls.every((ms: number) => ms < 1000)).toBe(true); + }, 10_000); + + it("does not retry non-transient errors", async () => { + const { getAccount } = await import("../account/getAccount"); + const { streamAccount } = await import("../account/streamAccount"); + + const account1 = createAccount("1"); + const permanentError = new Error("404 Not Found"); + + vi.mocked(getAccount) + .mockRejectedValueOnce(permanentError) + .mockResolvedValueOnce(ok(account1)); + + const results: unknown[] = []; + for await (const r of streamAccount("https://horizon.test", "G...", { + maxPolls: 2, + emitOnStart: true, + intervalMs: 1, + enableAutoRetry: true, + })) { + results.push(r); + } + + // Should emit error immediately without retry + expect(results.some((r: any) => r?.status === "error")).toBe(true); + // Should not have retry delays + expect(accountMockState.sleepCalls.every((ms: number) => ms < 1000)).toBe(true); + }, 10_000); + }); }); diff --git a/src/tests/transaction.test.ts b/src/tests/transaction.test.ts index 04e2d62..dfab26a 100644 --- a/src/tests/transaction.test.ts +++ b/src/tests/transaction.test.ts @@ -115,6 +115,10 @@ const { // ─── Hoisted mocks (must be defined before vi.mock is hoisted) ──────────────── +const transactionSleepMockState = vi.hoisted(() => ({ + sleepCalls: [] as number[], +})); + const transactionBuilderInstances: Array<{ memo?: unknown }> = []; const mocks = vi.hoisted(() => ({ @@ -231,6 +235,17 @@ vi.mock("../transaction/buildTransaction", async (importOriginal) => { }; }); +vi.mock("../shared", async (importOriginal) => { + const actual = await importOriginal("../shared"); + return { + ...actual, + sleep: vi.fn((ms: number) => { + transactionSleepMockState.sleepCalls.push(ms); + return Promise.resolve(); + }), + }; +}); + import { calculateMedian, isFeeSurge, @@ -2270,7 +2285,6 @@ describe("TokenBucketRateLimiter — rate limiting on submit", () => { ); }); }); -import { Operation } from "@stellar/stellar-sdk"; import { buildReverseTransaction, buildPathPayment, @@ -4502,3 +4516,171 @@ describe("createTransactionBuilder — undo/redo (#139)", () => { expect(b2.size()).toBe(2); // b2 unaffected }); }); + +describe("transaction streaming retry with exponential backoff", () => { + beforeEach(() => { + transactionSleepMockState.sleepCalls.length = 0; + vi.clearAllMocks(); + }); + + it("retries transient errors with exponential backoff", async () => { + const { streamTransactions } = await import("../transaction/streamTransactions"); + + const transientError = new Error("ECONNREFUSED"); + const mockPage = { + records: TRANSACTION_FIXTURES.map((tx, index) => + makeHorizonRecord(tx, `cursor_${index + 1}`), + ), + }; + + mockTransactionsCall + .mockRejectedValueOnce(transientError) + .mockRejectedValueOnce(transientError) + .mockResolvedValueOnce(mockPage); + + const results: unknown[] = []; + for await (const r of streamTransactions("https://horizon.test", "G...", { + maxPolls: 3, + emitOnStart: true, + intervalMs: 1, + enableAutoRetry: true, + })) { + results.push(r); + } + + // Should have retry delays (1s, 2s for exponential backoff) + expect(transactionSleepMockState.sleepCalls.length).toBeGreaterThan(0); + // First retry should be around 1s, second around 2s + expect(transactionSleepMockState.sleepCalls[0]).toBeGreaterThanOrEqual(1000); + expect(transactionSleepMockState.sleepCalls[0]).toBeLessThanOrEqual(1100); // with jitter + if (transactionSleepMockState.sleepCalls.length > 1) { + expect(transactionSleepMockState.sleepCalls[1]).toBeGreaterThanOrEqual(2000); + expect(transactionSleepMockState.sleepCalls[1]).toBeLessThanOrEqual(2200); // with jitter + } + + // Should eventually succeed + expect(results.some((r: any) => r?.status === "ok")).toBe(true); + }, 10_000); + + it("emits error after max consecutive failures and enters cooldown", async () => { + const { streamTransactions } = await import("../transaction/streamTransactions"); + + const transientError = new Error("ETIMEDOUT"); + + mockTransactionsCall.mockRejectedValue(transientError); + + const results: unknown[] = []; + for await (const r of streamTransactions("https://horizon.test", "G...", { + maxPolls: 7, // Allow enough polls to hit max consecutive failures (5) + emitOnStart: true, + intervalMs: 1, + enableAutoRetry: true, + })) { + results.push(r); + if (results.length >= 2) break; // Stop after we get the error emission + } + + // Should have emitted an error after 5 consecutive failures + expect(results.some((r: any) => r?.status === "error")).toBe(true); + + // Should have entered cooldown (60s delay) + const cooldownIndex = transactionSleepMockState.sleepCalls.findIndex( + (ms) => ms >= 60000 + ); + expect(cooldownIndex).toBeGreaterThanOrEqual(0); + }, 10_000); + + it("resets failure counter on successful poll", async () => { + const { streamTransactions } = await import("../transaction/streamTransactions"); + + const transientError = new Error("ECONNRESET"); + const mockPage = { + records: TRANSACTION_FIXTURES.slice(0, 2).map((tx, index) => + makeHorizonRecord(tx, `cursor_${index + 1}`), + ), + }; + + mockTransactionsCall + .mockRejectedValueOnce(transientError) + .mockRejectedValueOnce(transientError) + .mockResolvedValueOnce(mockPage) + .mockRejectedValueOnce(transientError) + .mockRejectedValueOnce(transientError) + .mockRejectedValueOnce(transientError) + .mockResolvedValueOnce(mockPage); + + const results: unknown[] = []; + for await (const r of streamTransactions("https://horizon.test", "G...", { + maxPolls: 7, + emitOnStart: true, + intervalMs: 1, + enableAutoRetry: true, + })) { + results.push(r); + } + + // Should succeed overall despite multiple failure sequences + expect(results.filter((r: any) => r?.status === "ok").length).toBe(2); + }, 10_000); + + it("does not retry when enableAutoRetry is false", async () => { + const { streamTransactions } = await import("../transaction/streamTransactions"); + + const transientError = new Error("ETIMEDOUT"); + const mockPage = { + records: TRANSACTION_FIXTURES.slice(0, 1).map((tx, index) => + makeHorizonRecord(tx, `cursor_${index + 1}`), + ), + }; + + mockTransactionsCall + .mockRejectedValueOnce(transientError) + .mockResolvedValueOnce(mockPage); + + const results: unknown[] = []; + for await (const r of streamTransactions("https://horizon.test", "G...", { + maxPolls: 2, + emitOnStart: true, + intervalMs: 1, + enableAutoRetry: false, + })) { + results.push(r); + } + + // Should emit error immediately without retry backoff + expect(results.some((r: any) => r?.status === "error")).toBe(true); + // Should not have retry delays (only normal interval) + expect(transactionSleepMockState.sleepCalls.every((ms) => ms < 1000)).toBe(true); + }, 10_000); + + it("does not retry non-transient errors (404)", async () => { + const { streamTransactions } = await import("../transaction/streamTransactions"); + + const notFoundError = new Error("404 Not Found"); + (notFoundError as any).response = { status: 404 }; + const mockPage = { + records: TRANSACTION_FIXTURES.slice(0, 1).map((tx, index) => + makeHorizonRecord(tx, `cursor_${index + 1}`), + ), + }; + + mockTransactionsCall + .mockRejectedValueOnce(notFoundError) + .mockResolvedValueOnce(mockPage); + + const results: unknown[] = []; + for await (const r of streamTransactions("https://horizon.test", "G...", { + maxPolls: 2, + emitOnStart: true, + intervalMs: 1, + enableAutoRetry: true, + })) { + results.push(r); + } + + // Should emit error immediately without retry + expect(results.some((r: any) => r?.status === "error")).toBe(true); + // Should not have retry delays + expect(transactionSleepMockState.sleepCalls.every((ms) => ms < 1000)).toBe(true); + }, 10_000); +}); diff --git a/src/transaction/streamTransactions.ts b/src/transaction/streamTransactions.ts index 6c73fa3..32e84e4 100644 --- a/src/transaction/streamTransactions.ts +++ b/src/transaction/streamTransactions.ts @@ -5,6 +5,12 @@ import { sleep, toMessage, isNotFoundError } from "../shared"; import type { SorokitLogger } from "../shared/logger"; import type { TransactionResult, TransactionStatus } from "./types"; import { createHorizonServer, createSorobanServer } from "../shared/serverFactory"; +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; @@ -91,6 +97,14 @@ export interface TransactionStreamConfig { * If true, emit the current page immediately on start. Default: true. */ emitOnStart?: boolean; + /** + * 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; } /** @@ -246,6 +260,15 @@ export async function* streamTransactions( const order = config?.order ?? "desc"; const emitOnStart = config?.emitOnStart ?? true; + // Retry configuration + const retryConfig: StreamingRetryConfig = { + enabled: config?.enableAutoRetry ?? true, + }; + let retryState: StreamingRetryState = { + consecutiveFailures: 0, + inCooldown: false, + }; + let cursor = config?.cursor; let polls = 0; let currentIntervalMs = Math.min( @@ -312,6 +335,9 @@ export async function* streamTransactions( if (signal?.aborted) return; const pollStartedAt = Date.now(); + let pollSuccess = false; + let errorResult: SorokitResult | null = null; + try { logger?.debug("transaction.stream.poll", { operation: "transaction.stream.poll", @@ -335,7 +361,7 @@ export async function* streamTransactions( const page = await builder.call(); - const transactions: TransactionResult[] = page.records.map((tx) => ({ + const transactions: TransactionResult[] = page.records.map((tx: any) => ({ hash: tx.hash, status: tx.successful ? ("success" as const) : ("failed" as const), ledger: tx.ledger_attr, @@ -369,6 +395,8 @@ export async function* streamTransactions( lastEmitted = transactionPage; yield ok(transactionPage); } + + pollSuccess = true; } catch (cause) { const code = isNotFoundError(cause) ? SorokitErrorCode.ACCOUNT_NOT_FOUND @@ -387,8 +415,48 @@ export async function* streamTransactions( }); adjustInterval(false); - yield err(code, message, cause); - } finally { + errorResult = err(code, message, cause); + } + + // Handle retry logic after poll attempt + const isNotFound = errorResult && errorResult.error && errorResult.error.code === SorokitErrorCode.ACCOUNT_NOT_FOUND; + const isTransient = errorResult && errorResult.error && !isNotFound ? 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 found, not a transient error, or we're in cooldown, or auto-retry is disabled + const shouldYieldError = errorResult && + (isNotFound || !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("transaction.stream.retry", { + operation: "transaction.stream.retry", + status: "retrying", + publicKey, + consecutiveFailures: retryState.consecutiveFailures, + delayMs: nextDelayMs, + }); + } else if (retryState.inCooldown) { + nextDelayMs = retryDecision.delayMs; + logger?.debug("transaction.stream.retry", { + operation: "transaction.stream.retry", + status: "cooldown", + publicKey, + delayMs: nextDelayMs, + }); + } else { nextDelayMs = getLatencyCompensatedDelay( currentIntervalMs, Date.now() - pollStartedAt,