diff --git a/README.md b/README.md index 4ff9cd3..87d43e0 100644 --- a/README.md +++ b/README.md @@ -239,6 +239,27 @@ BASE_URL=http://localhost:3001/api/v1 k6 run loadtest/k6/pool-stress.js --- +## Stellar Horizon timeouts & retries + +Horizon requests made by `StellarService` (currently the account-balance fallback in `getAccountBalance`) go through a resilience wrapper defined in `src/stellar/horizon-resilience.ts`, so a transient network issue no longer surfaces as an unhandled exception or a 500. + +**Retry with exponential backoff.** Each attempt has a 10s timeout (callers may pass a shorter one). Up to 3 attempts are made, with backoff of 500ms → 1s (capped at 5s); rate-limited responses (429) back off longer, 5s → 10s (capped at 30s). + +| Failure | Retried? | +|---|---| +| Timeout (per-attempt deadline, `ETIMEDOUT`, `ECONNABORTED`) | Yes | +| Network errors (`ECONNRESET`, `ECONNREFUSED`, `EAI_AGAIN`, …) | Yes | +| `429` / `5xx` responses | Yes | +| Other `4xx` (e.g. `404` account not found) | No — returned to the caller immediately | + +**Typed errors.** When every attempt times out, the service throws `HorizonTimeoutError` (HTTP 504, `errorCode: HORIZON_TIMEOUT`). Non-timeout failures that exhaust their retries rethrow the original error. + +**Timeout circuit breaker.** After 3 consecutive calls exhaust their retries on timeouts, the circuit opens and Horizon calls fail fast with `HorizonUnavailableError` (HTTP 503, `errorCode: HORIZON_UNAVAILABLE`) for 30s without touching the network. After that window, a single probe request is allowed through: any response within the deadline closes the circuit, another timeout reopens it. + +**Health checks** make a single Horizon attempt (no retries) with the 3s health-check timeout, so a slow Horizon can't multiply probe latency and trigger pod restarts. + +--- + ## Error handling (RFC 7807 Problem Details) All error responses use `Content-Type: application/problem+json` conforming to [RFC 7807](https://datatracker.ietf.org/doc/html/rfc7807). Legacy fields remain present in the response body for backward compatibility, but are deprecated: @@ -298,6 +319,8 @@ All error responses use `Content-Type: application/problem+json` conforming to [ | `GONE` | 410 | Resource existed but is no longer accessible (e.g. an expired policy). | | `TOO_MANY_REQUESTS` | 429 | Rate limit exceeded (60 requests per minute per IP). Back off and retry after the `Retry-After` header value. | | `SERVICE_UNAVAILABLE` | 503 | Downstream dependency unavailable (database, Redis, Stellar RPC). | +| `HORIZON_TIMEOUT` | 504 | Stellar Horizon did not respond in time after all retry attempts. Transient — safe to retry. See [Stellar Horizon timeouts](#stellar-horizon-timeouts--retries). | +| `HORIZON_UNAVAILABLE` | 503 | The Horizon circuit breaker is open after repeated timeouts, so the request was not sent. Retry after ~30s. | | `BAD_REQUEST` | 400 | Generic bad request not covered by validation (malformed path param, unsupported value, etc.). | | `INTERNAL_ERROR` | 500 | Unexpected server-side failure. The error is logged server-side; the response body intentionally omits internal details. | diff --git a/src/common/errors/error-codes.ts b/src/common/errors/error-codes.ts index 2bc6b63..c87eec6 100644 --- a/src/common/errors/error-codes.ts +++ b/src/common/errors/error-codes.ts @@ -38,6 +38,12 @@ export enum ErrorCode { // ── Service availability ────────────────────────────────────────────────── SERVICE_UNAVAILABLE = 'SERVICE_UNAVAILABLE', + + // ── Stellar Horizon ─────────────────────────────────────────────────────── + // Horizon did not respond in time after retries; transient, safe to retry. + HORIZON_TIMEOUT = 'HORIZON_TIMEOUT', + // Horizon circuit is open after repeated timeouts; retry after a short wait. + HORIZON_UNAVAILABLE = 'HORIZON_UNAVAILABLE', } /** diff --git a/src/health/health.controller.ts b/src/health/health.controller.ts index b21602a..e686fda 100644 --- a/src/health/health.controller.ts +++ b/src/health/health.controller.ts @@ -481,6 +481,8 @@ export class HealthController { keeperBalanceXlm = await this.stellar.getAccountBalance( this.stellar.keeperPublicKey, HEALTH_CHECK_RPC_TIMEOUT_MS, + // One Horizon attempt: retries would multiply probe latency (#338). + 1, ); // #191 — RPC reachability alone isn't enough: a keeper account diff --git a/src/stellar/horizon-resilience.ts b/src/stellar/horizon-resilience.ts new file mode 100644 index 0000000..5c11a2d --- /dev/null +++ b/src/stellar/horizon-resilience.ts @@ -0,0 +1,132 @@ +import { HttpException, HttpStatus, Logger } from "@nestjs/common"; +import { ErrorCode } from "../common/errors/error-codes"; + +/** + * Horizon timeout handling — retry with exponential backoff, a typed error on + * final failure, and a circuit breaker that fails fast once Horizon has timed + * out repeatedly. Previously a slow Horizon surfaced as an untyped timeout + * exception on the first attempt, turning a transient network blip into a 500. + */ + +/** Thrown when every attempt of a Horizon call timed out. Maps to 504. */ +export class HorizonTimeoutError extends HttpException { + constructor( + readonly operation: string, + readonly attempts: number, + readonly timeoutMs: number, + ) { + super( + { + message: `Stellar Horizon request "${operation}" timed out after ${attempts} attempt(s) (${timeoutMs}ms each)`, + errorCode: ErrorCode.HORIZON_TIMEOUT, + }, + HttpStatus.GATEWAY_TIMEOUT, + ); + this.name = "HorizonTimeoutError"; + } +} + +/** Thrown without calling Horizon while the timeout circuit is open. Maps to 503. */ +export class HorizonUnavailableError extends HttpException { + constructor(readonly operation: string, readonly retryAfterMs: number) { + super( + { + message: `Stellar Horizon is temporarily unavailable (circuit open after repeated timeouts) — "${operation}" not attempted`, + errorCode: ErrorCode.HORIZON_UNAVAILABLE, + }, + HttpStatus.SERVICE_UNAVAILABLE, + ); + this.name = "HorizonUnavailableError"; + } +} + +const NETWORK_ERROR_CODES = new Set([ + "ECONNABORTED", + "ECONNRESET", + "ECONNREFUSED", + "ETIMEDOUT", + "ESOCKETTIMEDOUT", + "EAI_AGAIN", + "EPIPE", +]); + +function statusOf(err: unknown): number | undefined { + return (err as { response?: { status?: number } })?.response?.status + ?? (err as { status?: number })?.status; +} + +/** True for our own withTimeout rejection and for socket/axios-level timeouts. */ +export function isTimeoutError(err: unknown): boolean { + if (err instanceof HorizonTimeoutError) return true; + if (err instanceof HttpException && err.getStatus() === HttpStatus.GATEWAY_TIMEOUT) return true; + const code = (err as { code?: string })?.code; + if (code === "ECONNABORTED" || code === "ETIMEDOUT" || code === "ESOCKETTIMEDOUT") return true; + const message = err instanceof Error ? err.message : String(err); + return /timed? ?out/i.test(message); +} + +/** + * Timeouts, network failures, 429s and 5xx responses are worth retrying. + * Other 4xx responses (e.g. 404 account not found) are deterministic and are + * rethrown immediately so callers can handle them. + */ +export function isRetryableHorizonError(err: unknown): boolean { + if (isTimeoutError(err)) return true; + const status = statusOf(err); + if (typeof status === "number") return status === 429 || status >= 500; + const code = (err as { code?: string })?.code; + return typeof code === "string" && NETWORK_ERROR_CODES.has(code); +} + +/** + * Counts consecutive Horizon calls that exhausted their retries on timeouts. + * At `threshold` the circuit opens and calls fail fast for `resetTimeoutMs`; + * after that a single probe is let through — its outcome closes or reopens + * the circuit. + */ +export class HorizonCircuitBreaker { + private consecutiveTimeouts = 0; + private openedAt: number | null = null; + private probeInFlight = false; + private readonly logger = new Logger("CircuitBreaker:horizon"); + + constructor( + private readonly threshold: number = 3, + private readonly resetTimeoutMs: number = 30_000, + ) {} + + get isOpen(): boolean { + return this.openedAt !== null; + } + + /** Throws HorizonUnavailableError if the call must not reach Horizon. */ + acquire(operation: string): void { + if (this.openedAt === null) return; + const elapsed = Date.now() - this.openedAt; + if (elapsed < this.resetTimeoutMs || this.probeInFlight) { + throw new HorizonUnavailableError(operation, Math.max(this.resetTimeoutMs - elapsed, 0)); + } + this.probeInFlight = true; + this.logger.warn(`Circuit half-open — allowing probe "${operation}"`); + } + + /** Horizon answered within the deadline (success or a non-timeout error). */ + recordResponse(): void { + if (this.openedAt !== null) this.logger.log("Probe succeeded — circuit closed"); + this.consecutiveTimeouts = 0; + this.openedAt = null; + this.probeInFlight = false; + } + + /** A call exhausted its retries on timeouts. */ + recordTimeout(): void { + this.probeInFlight = false; + this.consecutiveTimeouts++; + if (this.openedAt !== null || this.consecutiveTimeouts >= this.threshold) { + this.openedAt = Date.now(); + this.logger.error( + `Circuit opened after ${this.consecutiveTimeouts} consecutive Horizon timeouts — failing fast for ${this.resetTimeoutMs}ms`, + ); + } + } +} diff --git a/src/stellar/stellar.service.spec.ts b/src/stellar/stellar.service.spec.ts index 6c74c06..1b16ef5 100644 --- a/src/stellar/stellar.service.spec.ts +++ b/src/stellar/stellar.service.spec.ts @@ -1,6 +1,8 @@ import { Test, TestingModule } from "@nestjs/testing"; import { ConfigService } from "@nestjs/config"; import { StellarService } from "./stellar.service"; +import { HorizonTimeoutError, HorizonUnavailableError } from "./horizon-resilience"; +import { ErrorCode } from "../common/errors/error-codes"; jest.mock("@stellar/stellar-sdk", () => ({ Networks: { @@ -753,6 +755,99 @@ describe("StellarService", () => { mockHorizon.loadAccount.mockRejectedValue(new Error("account not found")); await expect(service.getAccountBalance("GSOMEACCOUNT")).rejects.toThrow("account not found"); + // Deterministic errors are not retried. + expect(mockHorizon.loadAccount).toHaveBeenCalledTimes(1); + }); + + describe("Horizon timeouts", () => { + let sleepSpy: jest.SpyInstance; + const hang = () => new Promise(() => {}); + + beforeEach(() => { + sleepSpy = jest + .spyOn(service as unknown as { sleep: (ms: number) => Promise }, "sleep") + .mockResolvedValue(undefined); + jest.spyOn((service as any).logger, "warn").mockImplementation(); + jest.spyOn((service as any).logger, "error").mockImplementation(); + }); + + it("retries a timed-out request with exponential backoff and returns on recovery", async () => { + mockHorizon.loadAccount + .mockImplementationOnce(hang) + .mockImplementationOnce(hang) + .mockResolvedValueOnce({ balances: [{ asset_type: "native", balance: "42.0000000" }] }); + + const balance = await service.getAccountBalance("GSOMEACCOUNT", 5); + + expect(balance).toBe("42.0000000"); + expect(mockHorizon.loadAccount).toHaveBeenCalledTimes(3); + expect(sleepSpy.mock.calls.map(([ms]) => ms)).toEqual([500, 1000]); + }); + + it("throws a typed HorizonTimeoutError (504) once every attempt times out", async () => { + mockHorizon.loadAccount.mockImplementation(hang); + + const err = await service.getAccountBalance("GSOMEACCOUNT", 5).catch((e) => e); + + expect(err).toBeInstanceOf(HorizonTimeoutError); + expect(err.getStatus()).toBe(504); + expect(err.getResponse()).toMatchObject({ errorCode: ErrorCode.HORIZON_TIMEOUT }); + expect(err.attempts).toBe(3); + expect(mockHorizon.loadAccount).toHaveBeenCalledTimes(3); + }); + + it("retries transient 5xx responses but not 404s", async () => { + mockHorizon.loadAccount + .mockRejectedValueOnce({ response: { status: 503 } }) + .mockRejectedValueOnce({ response: { status: 404 } }); + + await expect(service.getAccountBalance("GSOMEACCOUNT", 5)).resolves.toBe("0"); + expect(mockHorizon.loadAccount).toHaveBeenCalledTimes(2); + }); + + it("honours maxAttempts (health checks make a single attempt)", async () => { + mockHorizon.loadAccount.mockImplementation(hang); + + await expect(service.getAccountBalance("GSOMEACCOUNT", 5, 1)).rejects.toBeInstanceOf(HorizonTimeoutError); + expect(mockHorizon.loadAccount).toHaveBeenCalledTimes(1); + expect(sleepSpy).not.toHaveBeenCalled(); + }); + + it("opens the circuit after repeated timeouts and fails fast with HorizonUnavailableError (503)", async () => { + mockHorizon.loadAccount.mockImplementation(hang); + + for (let i = 0; i < 3; i++) { + await expect(service.getAccountBalance("GSOMEACCOUNT", 5, 1)).rejects.toBeInstanceOf(HorizonTimeoutError); + } + mockHorizon.loadAccount.mockClear(); + + const err = await service.getAccountBalance("GSOMEACCOUNT", 5).catch((e) => e); + expect(err).toBeInstanceOf(HorizonUnavailableError); + expect(err.getStatus()).toBe(503); + expect(err.getResponse()).toMatchObject({ errorCode: ErrorCode.HORIZON_UNAVAILABLE }); + expect(mockHorizon.loadAccount).not.toHaveBeenCalled(); + }); + + it("lets a probe through after the reset window and closes the circuit on success", async () => { + const nowSpy = jest.spyOn(Date, "now"); + let now = 1_000_000; + nowSpy.mockImplementation(() => now); + try { + mockHorizon.loadAccount.mockImplementation(hang); + for (let i = 0; i < 3; i++) { + await service.getAccountBalance("GSOMEACCOUNT", 5, 1).catch(() => undefined); + } + await expect(service.getAccountBalance("GSOMEACCOUNT", 5)).rejects.toBeInstanceOf(HorizonUnavailableError); + + now += 30_000; + mockHorizon.loadAccount.mockResolvedValue({ balances: [{ asset_type: "native", balance: "1.0000000" }] }); + + await expect(service.getAccountBalance("GSOMEACCOUNT", 5)).resolves.toBe("1.0000000"); + await expect(service.getAccountBalance("GSOMEACCOUNT", 5)).resolves.toBe("1.0000000"); + } finally { + nowSpy.mockRestore(); + } + }); }); }); diff --git a/src/stellar/stellar.service.ts b/src/stellar/stellar.service.ts index bcaa300..5b4c24f 100644 --- a/src/stellar/stellar.service.ts +++ b/src/stellar/stellar.service.ts @@ -14,6 +14,18 @@ import { Horizon, xdr, } from "@stellar/stellar-sdk"; +import { + HorizonCircuitBreaker, + HorizonTimeoutError, + isRetryableHorizonError, + isTimeoutError, +} from "./horizon-resilience"; + +/** Horizon retry tuning: attempts per call and the exponential backoff schedule. */ +const HORIZON_MAX_ATTEMPTS = 3; +const HORIZON_BACKOFF_BASE_MS = 500; // 500ms, 1s, 2s... +const HORIZON_BACKOFF_MAX_MS = 5000; +const HORIZON_DEFAULT_TIMEOUT_MS = 10000; /** * #187 — a reverting contract call is a deterministic simulation failure: @@ -93,6 +105,10 @@ export class StellarService { // TRANSACTION_BAD_SEQ. A simple promise-chain mutex serializes them. private readonly accountLocks = new Map>(); + // Fails Horizon calls fast after repeated timeouts instead of letting every + // request wait out the full retry schedule against an unresponsive node. + private readonly horizonBreaker = new HorizonCircuitBreaker(); + private async withAccountLock(publicKey: string, fn: () => Promise): Promise { const prev = this.accountLocks.get(publicKey) ?? Promise.resolve(); let release: () => void; @@ -666,6 +682,47 @@ export class StellarService { } } + /** + * Run a Horizon request with a per-attempt timeout, retrying timeouts and + * transient failures (network errors, 429, 5xx) with exponential backoff. + * Deterministic errors (e.g. 404) are rethrown immediately. If every attempt + * timed out, throws a typed HorizonTimeoutError and counts it toward the + * Horizon circuit breaker, which fails calls fast while open. + */ + private async callHorizon( + operation: string, + fn: () => Promise, + timeoutMs: number = HORIZON_DEFAULT_TIMEOUT_MS, + maxAttempts: number = HORIZON_MAX_ATTEMPTS, + ): Promise { + this.horizonBreaker.acquire(operation); + const attempts = Math.max(1, maxAttempts); + for (let attempt = 1; ; attempt++) { + try { + const result = await this.withTimeout(fn(), `Horizon ${operation}`, timeoutMs); + this.horizonBreaker.recordResponse(); + return result; + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + if (!isRetryableHorizonError(err) || attempt >= attempts) { + if (isTimeoutError(err)) { + this.horizonBreaker.recordTimeout(); + throw new HorizonTimeoutError(operation, attempt, timeoutMs); + } + this.horizonBreaker.recordResponse(); + throw err; + } + const backoffMs = isRateLimitError(err) + ? Math.min(5000 * Math.pow(2, attempt - 1), 30000) + : Math.min(HORIZON_BACKOFF_BASE_MS * Math.pow(2, attempt - 1), HORIZON_BACKOFF_MAX_MS); + this.logger.warn( + `Horizon ${operation} attempt ${attempt}/${attempts} failed: ${message} — retrying in ${backoffMs}ms`, + ); + await this.sleep(backoffMs); + } + } + } + /** * Reads an account's native XLM balance from its ledger entry over the RPC. * Returns a 7-decimal string like Horizon does, or "0" when the account @@ -703,8 +760,16 @@ export class StellarService { * Health checks (#338) pass a much shorter timeout so a * slow Horizon response doesn't block a load balancer's * health probe long enough to trigger a pod restart. + * @param maxAttempts Horizon attempts before giving up (default 3). Health + * checks pass 1 so retries don't multiply probe latency. + * @throws HorizonTimeoutError when every Horizon attempt timed out. + * @throws HorizonUnavailableError when the Horizon circuit is open. */ - async getAccountBalance(publicKey: string, timeoutMs?: number): Promise { + async getAccountBalance( + publicKey: string, + timeoutMs?: number, + maxAttempts?: number, + ): Promise { // #530 — the account's ledger entry carries its native balance, so ask the // RPC first and only fall back to Horizon if the RPC cannot answer. try { @@ -718,10 +783,11 @@ export class StellarService { } let account: Awaited>; try { - account = await this.withTimeout( - this.horizon.loadAccount(publicKey), + account = await this.callHorizon( "loadAccount", + () => this.horizon.loadAccount(publicKey), timeoutMs, + maxAttempts, ); } catch (err) { // #515 — horizon.loadAccount throws when the account does not exist on