From b0030a0482e40c6904bd8c7ae2af5a9ec1f2a8bc Mon Sep 17 00:00:00 2001 From: Renato Alcara Date: Mon, 27 Apr 2026 00:05:28 -0300 Subject: [PATCH] =?UTF-8?q?fix(bounded-retry):=20address=20PR=20#393=20rev?= =?UTF-8?q?iew=20=E2=80=94=20strict=20TTL,=20abort=20cancel,=20no=20leaks,?= =?UTF-8?q?=20structured=20logs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses 7 issues from Codex P2 + Copilot + CodeRabbit Critical/Major: 1. **TTL enforced BEFORE each attempt** (Codex P2 + Copilot + CodeRabbit Major) Previously TTL was only checked AFTER a failure, so an attempt could start with 0ms remaining and run for the full perAttemptTimeoutMs. Now: check at top of loop. If already exceeded, throw immediately. 2. **Per-attempt timeout capped by remaining TTL** (Copilot) `attemptTimeoutMs = min(perAttemptTimeoutMs, ttlMs - elapsed)` so a single attempt cannot run past the wall-clock deadline. 3. **Operation receives AbortSignal for cancellation** (CodeRabbit Critical) `withTimeout` now aborts an `AbortController` when the timeout fires. The signal is forwarded to the operation, so callers can opt into cancelling in-flight work (e.g. `(signal) => fetch({ signal })`). Without this, a timed-out attempt's network call kept running while bounded-retry started the next attempt — wasted work + duplicate listeners. 4. **sleep() removes abort listener on normal completion** (Codex P2 + Copilot + CodeRabbit) Previously `signal.addEventListener('abort', onAbort, { once: true })` was added on every retry but never removed when the timer resolved. Long-lived signals reused across retries accumulated listeners. Now explicit cleanup on both timer-resolve and abort paths. 5. **Outer abort listener properly cleaned up in main loop** Adds + removes via try/finally to prevent leaks when the loop throws. 6. **Metrics use optional chaining** (Copilot) `metrics.socketEvents?.inc(...)` everywhere, matching the rest of the codebase pattern (handles partial mocks in tests). 7. **Structured logging** (user request) New optional `logger?: ILogger` field. Emits: - debug: each retry scheduling (`{op, attempt, delayMs, elapsedMs, error}`) - info: recovery after retries (`{op, attempts, elapsedMs}`) - warn: give-up via TTL or shouldRetry predicate 8. **socket.ts: align uploadPreKeys query timeout with bounded-retry** `query(node, PER_ATTEMPT_TIMEOUT_MS)` so a stale attempt does not keep an iq listener registered past bounded-retry's per-attempt deadline (Copilot). 9. **Doc updated**: removed claim about `maxAttempts` (option doesn't exist; bounded by ttlMs + delay-cap + per-attempt-timeout). Tests: 18 pass (was 12; added 6 covering each fix). Full suite: 35/35 suites, 809/809 tests pass. NB: The "breaking API change" comments (Types/Socket.ts, Utils/index.ts, structured-logger.ts) are intentional and not addressed — this fork has a single consumer; downstream forks merging will absorb the change. --- src/Socket/socket.ts | 9 +- src/Utils/bounded-retry.ts | 181 +++++++++++++++++----- src/__tests__/Utils/bounded-retry.test.ts | 164 ++++++++++++++++++++ 3 files changed, 317 insertions(+), 37 deletions(-) diff --git a/src/Socket/socket.ts b/src/Socket/socket.ts index c16453f4..fda5483f 100644 --- a/src/Socket/socket.ts +++ b/src/Socket/socket.ts @@ -645,8 +645,12 @@ export const makeSocket = (config: SocketConfig) => { // Upload to server. Bounded-retry handles backoff (replaces both // the previous circuit breaker AND the manual exponential backoff // retry loop with maxRetries=3). + const PER_ATTEMPT_TIMEOUT_MS = 20_000 + // Pass the matching timeoutMs to query() so a stale attempt does + // not keep an iq listener registered after bounded-retry has moved + // on to the next attempt (Copilot review on PR #393). const uploadToServer = async () => { - await query(node) + await query(node, PER_ATTEMPT_TIMEOUT_MS) logger.info({ count }, 'uploaded pre-keys successfully') lastUploadTime = Date.now() } @@ -657,7 +661,8 @@ export const makeSocket = (config: SocketConfig) => { // 1s -> 2s -> 4s -> 8s -> 10s (cap matches old MAX backoff) delays: [1000, 2000, 4000, 8000, 10000], ttlMs: 60_000, - perAttemptTimeoutMs: 20_000 + perAttemptTimeoutMs: PER_ATTEMPT_TIMEOUT_MS, + logger }) } catch (uploadError) { logger.error({ uploadError: (uploadError as Error).toString(), count }, 'Failed to upload pre-keys to server') diff --git a/src/Utils/bounded-retry.ts b/src/Utils/bounded-retry.ts index 90269dcb..aa2563ed 100644 --- a/src/Utils/bounded-retry.ts +++ b/src/Utils/bounded-retry.ts @@ -31,13 +31,24 @@ * * - Per-operation isolation: each call to `withBoundedRetry` has its own * timer. No shared state. Operation A failing has zero effect on B. - * - Bounded by `maxAttempts + ttlMs`: prevents unbounded retry accumulation - * under prolonged outages. Eventually throws BoundedRetryGiveUpError. + * - Bounded by `ttlMs` (wall-clock budget). The TTL is enforced strictly + * at three points: (a) before each attempt, (b) cap on per-attempt + * timeout, (c) cap on retry delay. * - Per-attempt timeout: each attempt has its own deadline (default 30s), - * so a single hung call cannot consume the entire TTL budget. - * - AbortSignal cancellation: external cancellation supported. - * - Memory bound: at most one outstanding retry timer per call. Once the - * call resolves/rejects/aborts, all state is freed. + * automatically capped to remaining TTL budget so total runtime never + * exceeds ttlMs. + * - AbortSignal cancellation: external cancellation supported, both for + * the sleep between retries AND (optionally) the in-flight operation + * when the operation accepts a signal parameter. + * - Memory bound: at most one outstanding retry timer per call. Listeners + * are explicitly removed when timers settle. Once the call resolves / + * rejects / aborts, all state is freed. + * + * # Logging + * + * Pass `logger` in options to get structured logs at each retry attempt, + * give-up, and post-failure recovery. The module emits Prometheus metrics + * regardless of whether a logger is provided. * * # When to use this vs. plain query() * @@ -45,8 +56,7 @@ * what to do on failure). Most call sites in InfiniteAPI use this. * - Use `withBoundedRetry(() => query(...), { name: 'X' })` when the * operation is "must-eventually-succeed" with no upstream retry — e.g. - * `uploadPreKeys`, `assertSessions(force=true)`, or other write paths - * where the alternative is data loss. + * `uploadPreKeys` or other write paths where the alternative is data loss. * * # Examples * @@ -54,7 +64,7 @@ * // Single attempt with 5-min TTL, give up after that: * await withBoundedRetry( * () => assertSessions([jid], true), - * { name: 'assertSessions', ttlMs: 5 * 60_000 } + * { name: 'assertSessions', ttlMs: 5 * 60_000, logger } * ) * * // Tight deadline, fast give-up: @@ -63,13 +73,17 @@ * { name: 'send', ttlMs: 30_000, perAttemptTimeoutMs: 5_000 } * ) * - * // Custom delay sequence (for testing): - * await withBoundedRetry(op, { delays: [10, 20, 40], jitter: 0, ttlMs: 200 }) + * // Operation that respects abort signal (best practice): + * await withBoundedRetry( + * (signal) => fetchSomething({ signal }), + * { name: 'fetch', logger } + * ) * ``` * * @module Utils/bounded-retry */ +import type { ILogger } from './logger' import { metrics } from './prometheus-metrics.js' /** @@ -109,7 +123,7 @@ export interface BoundedRetryOptions { jitter?: number /** Total wall-clock budget — gives up after this. Default 10 min. */ ttlMs?: number - /** Per-attempt timeout (ms). Default 30s. */ + /** Per-attempt timeout (ms). Default 30s. Capped by remaining TTL. */ perAttemptTimeoutMs?: number /** Predicate: should we retry on this error? Default: always */ shouldRetry?: (err: Error, attempt: number) => boolean @@ -117,6 +131,8 @@ export interface BoundedRetryOptions { onRetry?: (err: Error, attempt: number, delayMs: number) => void /** AbortSignal — if aborted, give up immediately */ signal?: AbortSignal + /** Optional logger for structured retry/give-up/recovery logs */ + logger?: ILogger } export class BoundedRetryGiveUpError extends Error { @@ -161,12 +177,19 @@ function pickDelay(attempt: number, delays: readonly number[]): number { } /** - * Wrap a promise with a per-attempt timeout. Rejects if the promise does - * not settle before timeoutMs elapses. + * Wrap a promise with a per-attempt timeout. Aborts the supplied controller + * when the timeout fires so the caller can cancel any in-flight work + * (operations that accept the signal). Always clears the timer on settlement. */ -function withTimeout(promise: Promise, timeoutMs: number, name: string): Promise { +function withTimeout( + promise: Promise, + timeoutMs: number, + name: string, + abortOnTimeout: AbortController +): Promise { return new Promise((resolve, reject) => { const timer = setTimeout(() => { + abortOnTimeout.abort() reject(new Error(`bounded-retry "${name}" attempt timed out after ${timeoutMs}ms`)) }, timeoutMs) promise @@ -176,13 +199,14 @@ function withTimeout(promise: Promise, timeoutMs: number, name: string): P }) .catch(err => { clearTimeout(timer) - reject(err) + reject(err as Error) }) }) } /** - * Sleep with abort support. + * Sleep with abort support. Always removes the abort listener on settlement + * so listeners do not accumulate on long-lived signals. */ function sleep(ms: number, signal?: AbortSignal): Promise { return new Promise((resolve, reject) => { @@ -190,12 +214,26 @@ function sleep(ms: number, signal?: AbortSignal): Promise { reject(new Error('aborted')) return } - const timer = setTimeout(resolve, ms) + + let onAbort: (() => void) | undefined + const cleanup = () => { + if (onAbort && signal) { + signal.removeEventListener('abort', onAbort) + } + } + + const timer = setTimeout(() => { + cleanup() + resolve() + }, ms) + if (signal) { - const onAbort = () => { + onAbort = () => { clearTimeout(timer) + cleanup() reject(new Error('aborted')) } + signal.addEventListener('abort', onAbort, { once: true }) } }) @@ -207,16 +245,20 @@ function sleep(ms: number, signal?: AbortSignal): Promise { * Independent per-call: no global state, no cross-operation interaction. * Memory bound: at most one outstanding retry timer per call. * + * The operation may optionally accept an `AbortSignal` parameter — when the + * per-attempt timeout fires (or the outer signal is aborted), the inner + * signal is aborted so the operation can stop in-flight work cleanly. + * * @example * ```ts * const result = await withBoundedRetry( - * () => assertSessions([jid], true), - * { name: 'assertSessions', ttlMs: 5 * 60_000 } + * (signal) => assertSessions([jid], true, { signal }), + * { name: 'assertSessions', ttlMs: 5 * 60_000, logger } * ) * ``` */ export async function withBoundedRetry( - operation: () => Promise, + operation: (signal?: AbortSignal) => Promise, options: BoundedRetryOptions = {} ): Promise { const name = options.name ?? 'bounded-retry' @@ -225,50 +267,119 @@ export async function withBoundedRetry( const ttlMs = options.ttlMs ?? DEFAULT_TTL_MS const perAttemptTimeoutMs = options.perAttemptTimeoutMs ?? DEFAULT_PER_ATTEMPT_TIMEOUT_MS const shouldRetry = options.shouldRetry ?? (() => true) + const logger = options.logger const start = Date.now() let lastError: Error = new Error('unknown') let attempt = 0 while (true) { + // Check abort + TTL BEFORE starting an attempt — strict wall-clock + // budget. Without this check the loop could begin a new attempt with + // 0ms remaining and then run for up to `perAttemptTimeoutMs`. if (options.signal?.aborted) { throw new BoundedRetryAbortedError(name) } - try { - const result = await withTimeout(operation(), perAttemptTimeoutMs, name) - if (attempt > 0) { - metrics.socketEvents.inc({ event: 'bounded_retry_recovered' }) + const elapsedBeforeAttempt = Date.now() - start + const remainingBudget = ttlMs - elapsedBeforeAttempt + if (remainingBudget <= 0) { + metrics.errors?.inc({ category: 'bounded_retry', code: 'ttl_exceeded' }) + logger?.warn?.( + { op: name, attempts: attempt, elapsedMs: elapsedBeforeAttempt, ttlMs, lastError: lastError.message }, + 'bounded-retry: TTL exceeded before next attempt — giving up' + ) + throw new BoundedRetryGiveUpError(name, attempt, elapsedBeforeAttempt, lastError) + } + + // Cap the per-attempt timeout to the remaining TTL budget so a single + // attempt cannot run past the wall-clock deadline. + const attemptTimeoutMs = Math.min(perAttemptTimeoutMs, remainingBudget) + const attemptAbort = new AbortController() + + // Forward outer abort to the per-attempt controller so the in-flight + // operation is cancelled when the user aborts. Cleaned up below. + let onOuterAbort: (() => void) | undefined + if (options.signal) { + onOuterAbort = () => attemptAbort.abort() + if (options.signal.aborted) { + attemptAbort.abort() + } else { + options.signal.addEventListener('abort', onOuterAbort, { once: true }) } + } + + try { + const result = await withTimeout(operation(attemptAbort.signal), attemptTimeoutMs, name, attemptAbort) + + if (attempt > 0) { + metrics.socketEvents?.inc({ event: 'bounded_retry_recovered' }) + logger?.info?.( + { op: name, attempts: attempt + 1, elapsedMs: Date.now() - start }, + 'bounded-retry: operation succeeded after retries' + ) + } + return result } catch (err) { lastError = err as Error attempt++ - const elapsed = Date.now() - start - if (elapsed >= ttlMs) { - metrics.errors.inc({ category: 'bounded_retry', code: 'ttl_exceeded' }) - throw new BoundedRetryGiveUpError(name, attempt, elapsed, lastError) + // Detach outer-abort listener as soon as the attempt settles so + // it cannot fire after the controller is no longer in scope. + if (options.signal && onOuterAbort) { + options.signal.removeEventListener('abort', onOuterAbort) } + // If the outer signal aborted us mid-attempt, surface that explicitly + // rather than as a generic operation failure. + if (options.signal?.aborted) { + throw new BoundedRetryAbortedError(name) + } + + const elapsed = Date.now() - start + if (!shouldRetry(lastError, attempt)) { - metrics.errors.inc({ category: 'bounded_retry', code: 'predicate_no_retry' }) + metrics.errors?.inc({ category: 'bounded_retry', code: 'predicate_no_retry' }) + logger?.warn?.( + { op: name, attempts: attempt, elapsedMs: elapsed, error: lastError.message }, + 'bounded-retry: shouldRetry returned false — giving up' + ) throw lastError } + // Re-check budget after the failure so we do not sleep past TTL. + const remainingAfterFailure = ttlMs - elapsed + if (remainingAfterFailure <= 0) { + metrics.errors?.inc({ category: 'bounded_retry', code: 'ttl_exceeded' }) + logger?.warn?.( + { op: name, attempts: attempt, elapsedMs: elapsed, ttlMs, lastError: lastError.message }, + 'bounded-retry: TTL exceeded after attempt — giving up' + ) + throw new BoundedRetryGiveUpError(name, attempt, elapsed, lastError) + } + const baseDelay = pickDelay(attempt - 1, delays) - // Cap remaining delay so we do not blow past ttlMs - const remainingBudget = Math.max(0, ttlMs - elapsed) - const delayMs = Math.min(withJitter(baseDelay, jitter), remainingBudget) + const delayMs = Math.min(withJitter(baseDelay, jitter), remainingAfterFailure) options.onRetry?.(lastError, attempt, delayMs) - metrics.socketEvents.inc({ event: 'bounded_retry_attempt' }) + metrics.socketEvents?.inc({ event: 'bounded_retry_attempt' }) + logger?.debug?.( + { op: name, attempt, delayMs, elapsedMs: elapsed, error: lastError.message }, + 'bounded-retry: scheduling next attempt' + ) try { await sleep(delayMs, options.signal) } catch { throw new BoundedRetryAbortedError(name) } + } finally { + // Belt-and-suspenders: ensure the outer-abort listener is always + // removed even on early throws. + if (options.signal && onOuterAbort) { + options.signal.removeEventListener('abort', onOuterAbort) + } } } } diff --git a/src/__tests__/Utils/bounded-retry.test.ts b/src/__tests__/Utils/bounded-retry.test.ts index 41c760b8..fcb07af4 100644 --- a/src/__tests__/Utils/bounded-retry.test.ts +++ b/src/__tests__/Utils/bounded-retry.test.ts @@ -210,4 +210,168 @@ describe('bounded-retry — WhatsApp-aligned per-operation retry', () => { expect(call[2]).toBeLessThanOrEqual(50) } }) + + // ─── Tests for fixes from PR #393 review ────────────────────────────── + + test('TTL is enforced BEFORE the next attempt (not just after)', async () => { + // Regression: previously TTL was only checked after a failure, so a + // new attempt could start with 0ms remaining and run for the full + // per-attempt timeout, overshooting wall-clock budget by that much. + const slowOp = jest.fn<() => Promise>(() => new Promise(resolve => setTimeout(() => resolve('late'), 200))) + + const start = Date.now() + await expect( + withBoundedRetry(slowOp, { + name: 'ttl-strict', + delays: [10], + jitter: 0, + ttlMs: 50, + perAttemptTimeoutMs: 1000 // huge, must be capped by TTL + }) + ).rejects.toBeInstanceOf(BoundedRetryGiveUpError) + + const elapsed = Date.now() - start + // With strict TTL enforcement, total runtime should be very close to ttlMs. + // Allow up to 100ms slack for Node timer + microtask scheduling. + expect(elapsed).toBeLessThan(150) + }) + + test('per-attempt timeout is capped by remaining TTL budget', async () => { + // If user passes perAttemptTimeoutMs > ttlMs, the cap should apply. + const slowOp = jest.fn<() => Promise>(() => new Promise(() => {})) // hangs forever + + const start = Date.now() + await expect( + withBoundedRetry(slowOp, { + name: 'attempt-cap', + delays: [1], + jitter: 0, + ttlMs: 30, + perAttemptTimeoutMs: 5000 // way bigger than ttl + }) + ).rejects.toBeInstanceOf(BoundedRetryGiveUpError) + + const elapsed = Date.now() - start + expect(elapsed).toBeLessThan(150) // not 5000ms! + }) + + test('operation receives an AbortSignal that is aborted on per-attempt timeout', async () => { + // New API: operation can opt into cancellation via the signal arg. + const aborts: boolean[] = [] + const opThatRespectsSignal = jest.fn( + (signal?: AbortSignal) => + new Promise((resolve, reject) => { + const timer = setTimeout(() => resolve('late'), 100) + signal?.addEventListener('abort', () => { + clearTimeout(timer) + aborts.push(true) + reject(new Error('aborted by signal')) + }) + }) + ) + + await withBoundedRetry(opThatRespectsSignal, { + name: 'signal-aware', + delays: [1], + jitter: 0, + ttlMs: 80, + perAttemptTimeoutMs: 20 // smaller than 100ms op duration + }).catch(() => {}) + + expect(aborts.length).toBeGreaterThan(0) // at least one attempt was aborted + }) + + test('sleep listener is removed when timeout completes normally (no leak)', async () => { + // Regression: sleep used signal.addEventListener with { once: true } but + // never removed the listener on timer-resolve. With many retries on a + // shared signal, listeners would accumulate. + const ctrl = new AbortController() + const op = jest + .fn<() => Promise>() + .mockRejectedValueOnce(new Error('1')) + .mockRejectedValueOnce(new Error('2')) + .mockRejectedValueOnce(new Error('3')) + .mockResolvedValueOnce('ok') + + // Track listener count via getMaxListeners-style check is not portable + // across runtimes, so we instead verify completion proceeds without + // MaxListenersExceededWarning being raised. The implementation must + // remove the listener on each successful sleep. + const result = await withBoundedRetry(op, { + name: 'no-leak', + delays: [5], + jitter: 0, + ttlMs: 200, + signal: ctrl.signal + }) + + expect(result).toBe('ok') + expect(op).toHaveBeenCalledTimes(4) + // If listeners leaked, we would see them by counting. AbortSignal does + // not expose listenerCount publicly; the regression manifests as a + // runtime warning. Test passes if no warning + result is correct. + }) + + test('logger receives structured logs for retry, give-up, and recovery', async () => { + const debugCalls: unknown[][] = [] + const infoCalls: unknown[][] = [] + const warnCalls: unknown[][] = [] + const fakeLogger = { + debug: (...args: unknown[]) => debugCalls.push(args), + info: (...args: unknown[]) => infoCalls.push(args), + warn: (...args: unknown[]) => warnCalls.push(args), + error: () => {}, + fatal: () => {}, + trace: () => {}, + child: () => fakeLogger, + level: 'debug' + } + + const op = jest + .fn<() => Promise>() + .mockRejectedValueOnce(new Error('boom')) + .mockResolvedValueOnce('ok') + + const result = await withBoundedRetry(op, { + name: 'logged-op', + delays: [5], + jitter: 0, + ttlMs: 200, + logger: fakeLogger as never + }) + + expect(result).toBe('ok') + // Should have logged the scheduling at debug level + expect(debugCalls.length).toBeGreaterThanOrEqual(1) + // Should have logged the recovery at info level + expect(infoCalls.length).toBeGreaterThanOrEqual(1) + }) + + test('logger reports give-up via warn when TTL exceeded', async () => { + const warnCalls: unknown[][] = [] + const fakeLogger = { + debug: () => {}, + info: () => {}, + warn: (...args: unknown[]) => warnCalls.push(args), + error: () => {}, + fatal: () => {}, + trace: () => {}, + child: () => fakeLogger, + level: 'debug' + } + + const op = jest.fn<() => Promise>().mockRejectedValue(new Error('always-fails')) + + await expect( + withBoundedRetry(op, { + name: 'logged-fail', + delays: [5], + jitter: 0, + ttlMs: 30, + logger: fakeLogger as never + }) + ).rejects.toBeInstanceOf(BoundedRetryGiveUpError) + + expect(warnCalls.length).toBeGreaterThanOrEqual(1) + }) })