diff --git a/scripts/alert-expiring-allowances.ts b/scripts/alert-expiring-allowances.ts index 43c07dd..b131e07 100644 --- a/scripts/alert-expiring-allowances.ts +++ b/scripts/alert-expiring-allowances.ts @@ -1,28 +1,31 @@ #!/usr/bin/env tsx /** - * alert-expiring-allowances.ts — Proactive alerting for soon-to-expire token allowances + * alert-expiring-allowances.ts — Proactive alerting for soon-to-expire token + * allowances. * - * Checks each subscriber's token allowance expiry ledger. Subscribers whose + * Checks each subscriber's token allowance expiry ledger. Subscribers whose * allowance will expire within ALERT_WINDOW_LEDGERS (default 17280 ≈ 24 h at - * ~5 s/ledger) are included in the report. + * ~5 s/ledger) are included in the report. Checks run concurrently under a + * configurable cap; transient RPC errors are retried with exponential backoff. * * Usage: - * tsx alert-expiring-allowances.ts [--file subscribers.txt] [--dry-run] [address1 ...] + * tsx alert-expiring-allowances.ts [--file subscribers.txt] [--dry-run] [addr ...] * * Environment variables: * CONTRACT_ID Required. Deployed FlowPay contract ID. - * RPC_URL Optional. Soroban RPC endpoint (default: testnet). - * NETWORK_PASSPHRASE Optional. Network passphrase (default: testnet). - * ALERT_WINDOW_LEDGERS Optional. Ledgers ahead to consider "expiring soon" - * (default: 17280 ≈ 24 h). - * WEBHOOK_URL Optional. POST the JSON report here if set. + * RPC_URL Soroban RPC endpoint (default: testnet). + * NETWORK_PASSPHRASE Network passphrase (default: testnet). + * ALERT_WINDOW_LEDGERS Ledgers ahead to consider "expiring soon" (default: 17280). + * WEBHOOK_URL POST the JSON report here if set. + * CONCURRENCY Max simultaneous RPC calls (default: 5). + * MAX_RETRIES Retry attempts per transient error (default: 3). + * RETRY_BASE_MS Base backoff delay in ms (default: 300). * * Exit codes: * 0 — no expiring allowances found (or dry-run with none found) - * 1 — one or more allowances are expiring within the alert window + * 1 — one or more allowances expiring within the alert window */ -import { Server } from "@stellar/stellar-sdk/rpc"; import { Contract, Networks, @@ -33,11 +36,18 @@ import { xdr, Account, } from "@stellar/stellar-sdk"; +import { MultiEndpointServer } from "./rpc-client.js"; +import { + withRetry, + runConcurrent, + type ExpiryEntry, + type ExpiryAlertReport, +} from "./allowance-utils.js"; // ── Configuration ───────────────────────────────────────────────────────────── -const RPC_URL = process.env.RPC_URL ?? "https://soroban-testnet.stellar.org"; -const CONTRACT_ID = process.env.CONTRACT_ID ?? ""; +const CONTRACT_ID = + process.env.CONTRACT_ID ?? process.env.VITE_CONTRACT_ID ?? ""; const NETWORK_PASSPHRASE = (process.env.NETWORK_PASSPHRASE ?? Networks.TESTNET) as string; const ALERT_WINDOW_LEDGERS = parseInt( @@ -45,6 +55,12 @@ const ALERT_WINDOW_LEDGERS = parseInt( 10, ); const WEBHOOK_URL = process.env.WEBHOOK_URL; +const CONCURRENCY = Math.max(1, parseInt(process.env.CONCURRENCY ?? "5", 10)); +const MAX_RETRIES = Math.max(0, parseInt(process.env.MAX_RETRIES ?? "3", 10)); +const RETRY_BASE_MS = Math.max( + 0, + parseInt(process.env.RETRY_BASE_MS ?? "300", 10), +); if (!CONTRACT_ID) { console.error("Error: CONTRACT_ID environment variable is required."); @@ -54,10 +70,10 @@ if (!CONTRACT_ID) { process.exit(1); } -/** A stable dummy source used for read-only simulations (no funds needed). */ +/** Stable dummy source account for read-only simulations. */ const SIM_SOURCE = "GAAZI4TCR3TY5OJHCTJC2A4QSY6CJWJH5IAJTGKIN2ER7LBNVKOCCWN"; -const server = new Server(RPC_URL); +const server = new MultiEndpointServer(); const FlowPayAddress = Address.fromString(CONTRACT_ID); // ── Types ───────────────────────────────────────────────────────────────────── @@ -71,29 +87,6 @@ interface Subscription { paused: boolean; } -/** - * Per-subscriber expiry report entry. - * Emitted in the JSON report and sent to the webhook (if configured). - */ -export interface ExpiryEntry { - address: string; - merchant: string; - allowance_amount: string; - /** Ledger sequence number at which the allowance expires (0 = no expiry set). */ - expires_at_ledger: number; - ledgers_remaining: number; -} - -/** Top-level report posted to the webhook and/or printed to stdout. */ -interface AlertReport { - generated_at: string; - contract: string; - current_ledger: number; - alert_window_ledgers: number; - expiring_count: number; - expiring: ExpiryEntry[]; -} - // ── Helpers ─────────────────────────────────────────────────────────────────── function addressVal(addr: string): xdr.ScVal { @@ -129,10 +122,13 @@ Environment variables: NETWORK_PASSPHRASE Network passphrase (default: Test SDF Network ; September 2015) ALERT_WINDOW_LEDGERS Ledgers ahead to flag as expiring (default: 17280 ≈ 24 h) WEBHOOK_URL If set, POST the JSON report to this URL + CONCURRENCY Max simultaneous RPC calls (default: 5) + MAX_RETRIES Retry attempts per transient error (default: 3) + RETRY_BASE_MS Base backoff delay in ms (default: 300) Exit codes: 0 No allowances expiring within the alert window - 1 One or more allowances expiring soon (or webhook returned non-2xx) + 1 One or more allowances expiring soon Examples: CONTRACT_ID=CD123... tsx alert-expiring-allowances.ts GXYZ... GABC... @@ -148,120 +144,147 @@ Examples: /** * Fetches the subscription record for `user` from the FlowPay contract. - * Returns null if the user has no subscription or on any RPC error. + * Retries on transient RPC failures. Returns null on no subscription or after + * exhausting retries. */ -async function getSubscription(user: string): Promise { - try { - const contract = new Contract(CONTRACT_ID); - const tx = new TransactionBuilder(new Account(SIM_SOURCE, "0"), { - fee: BASE_FEE, - networkPassphrase: NETWORK_PASSPHRASE, - }) - .addOperation(contract.call("get_subscription", addressVal(user))) - .setTimeout(30) - .build(); - - const result = await server.simulateTransaction(tx); - if ("error" in result) return null; - - const retval = (result as { result?: { retval?: xdr.ScVal } }).result - ?.retval; - if (!retval || retval.switch().name === "scvVoid") return null; - - const fields: Record = {}; - for (const entry of retval.map() ?? []) { - const key = entry.key().sym().toString(); - const val = entry.val(); - switch (key) { - case "amount": - fields[key] = BigInt(val.i128().toString()); - break; - case "token": - fields[key] = Address.fromScVal(val).toString(); - break; - case "merchant": - fields[key] = Address.fromScVal(val).toString(); - break; - case "active": - fields[key] = val.b(); - break; - case "paused": - fields[key] = val.b(); - break; +export async function getSubscription( + user: string, + opts?: { maxRetries?: number; baseDelayMs?: number }, +): Promise { + return withRetry( + async () => { + const contract = new Contract(CONTRACT_ID); + const tx = new TransactionBuilder(new Account(SIM_SOURCE, "0"), { + fee: BASE_FEE, + networkPassphrase: NETWORK_PASSPHRASE, + }) + .addOperation(contract.call("get_subscription", addressVal(user))) + .setTimeout(30) + .build(); + + const result = await server.simulateTransaction(tx); + if ("error" in result) return null; + + const retval = (result as { result?: { retval?: xdr.ScVal } }).result + ?.retval; + if (!retval || retval.switch().name === "scvVoid") return null; + + const fields: Record = {}; + for (const entry of retval.map() ?? []) { + const key = entry.key().sym().toString(); + const val = entry.val(); + switch (key) { + case "amount": + fields[key] = BigInt(val.i128().toString()); + break; + case "token": + fields[key] = Address.fromScVal(val).toString(); + break; + case "merchant": + fields[key] = Address.fromScVal(val).toString(); + break; + case "active": + fields[key] = val.b(); + break; + case "paused": + fields[key] = val.b(); + break; + } } - } - if ( - fields.amount === undefined || - fields.token === undefined || - fields.merchant === undefined - ) { - return null; - } + if ( + fields.amount === undefined || + fields.token === undefined || + fields.merchant === undefined + ) { + return null; + } - return { - merchant: fields.merchant as string, - amount: fields.amount as bigint, - token: fields.token as string, - active: (fields.active as boolean | undefined) ?? false, - paused: (fields.paused as boolean | undefined) ?? false, - }; - } catch { + return { + merchant: fields.merchant as string, + amount: fields.amount as bigint, + token: fields.token as string, + active: (fields.active as boolean | undefined) ?? false, + paused: (fields.paused as boolean | undefined) ?? false, + }; + }, + { + maxRetries: opts?.maxRetries ?? MAX_RETRIES, + baseDelayMs: opts?.baseDelayMs ?? RETRY_BASE_MS, + onRetry: (attempt, err) => + console.error( + `[alert-expiring] getSubscription retry ${attempt} for ${user}: ${err instanceof Error ? err.message : String(err)}`, + ), + }, + ).catch((err) => { + console.error( + `[alert-expiring] getSubscription failed after retries for ${user}: ${err instanceof Error ? err.message : String(err)}`, + ); return null; - } + }); } /** * Returns the current allowance amount that the FlowPay contract is approved * to spend on behalf of `owner` for token `tokenId`. + * Retries on transient RPC failures. */ -async function getAllowanceAmount( +export async function getAllowanceAmount( owner: string, tokenId: string, + opts?: { maxRetries?: number; baseDelayMs?: number }, ): Promise { - try { - const tokenContract = new Contract(tokenId); - const tx = new TransactionBuilder(new Account(SIM_SOURCE, "0"), { - fee: BASE_FEE, - networkPassphrase: NETWORK_PASSPHRASE, - }) - .addOperation( - tokenContract.call( - "allowance", - addressVal(owner), - nativeToScVal(FlowPayAddress, { type: "address" }), + return withRetry( + async () => { + const tokenContract = new Contract(tokenId); + const tx = new TransactionBuilder(new Account(SIM_SOURCE, "0"), { + fee: BASE_FEE, + networkPassphrase: NETWORK_PASSPHRASE, + }) + .addOperation( + tokenContract.call( + "allowance", + addressVal(owner), + nativeToScVal(FlowPayAddress, { type: "address" }), + ), + ) + .setTimeout(30) + .build(); + + const result = await server.simulateTransaction(tx); + if ("error" in result) return 0n; + + const retval = (result as { result?: { retval?: xdr.ScVal } }).result + ?.retval; + if (!retval || retval.switch().name === "scvVoid") return 0n; + + return BigInt(retval.i128().toString()); + }, + { + maxRetries: opts?.maxRetries ?? MAX_RETRIES, + baseDelayMs: opts?.baseDelayMs ?? RETRY_BASE_MS, + onRetry: (attempt, err) => + console.error( + `[alert-expiring] getAllowanceAmount retry ${attempt} for ${owner}: ${err instanceof Error ? err.message : String(err)}`, ), - ) - .setTimeout(30) - .build(); - - const result = await server.simulateTransaction(tx); - if ("error" in result) return 0n; - - const retval = (result as { result?: { retval?: xdr.ScVal } }).result - ?.retval; - if (!retval || retval.switch().name === "scvVoid") return 0n; - - return BigInt(retval.i128().toString()); - } catch { + }, + ).catch((err) => { + console.error( + `[alert-expiring] getAllowanceAmount failed after retries for ${owner}: ${err instanceof Error ? err.message : String(err)}`, + ); return 0n; - } + }); } /** * Builds the ledger key for the allowance entry stored in a SEP-41 / SAC token - * contract. Allowances are stored as Temporary ContractData with a Map key of - * the form `{ "from": Address, "spender": Address }`. - * - * The `liveUntilLedgerSeq` of this entry is the expiry ledger. + * contract. The `liveUntilLedgerSeq` of this entry is the expiry ledger. */ function buildAllowanceLedgerKey( tokenId: string, owner: string, spender: string, ): xdr.LedgerKey { - // The SEP-41 / Stellar Asset Contract stores allowances as a Temporary - // ContractData entry keyed by a ScMap: { "from": owner, "spender": spender } const mapKey = xdr.ScVal.scvMap([ new xdr.ScMapEntry({ key: xdr.ScVal.scvSymbol("from"), @@ -283,29 +306,41 @@ function buildAllowanceLedgerKey( } /** - * Returns the `liveUntilLedgerSeq` for the allowance ledger entry, or 0 if the - * entry doesn't exist (meaning no expiry / zero allowance). + * Returns the `liveUntilLedgerSeq` for the allowance ledger entry, or 0 if + * the entry doesn't exist. Retries on transient RPC failures. */ -async function getAllowanceExpiryLedger( +export async function getAllowanceExpiryLedger( tokenId: string, owner: string, spender: string, + opts?: { maxRetries?: number; baseDelayMs?: number }, ): Promise { - try { - const ledgerKey = buildAllowanceLedgerKey(tokenId, owner, spender); - const response = await server.getLedgerEntries(ledgerKey); - - if (!response.entries || response.entries.length === 0) return 0; - - const entry = response.entries[0]; - // liveUntilLedgerSeq is the last ledger the entry is live on. - // The field is named liveUntilLedgerSeq in the SDK response object. - const liveUntil = (entry as { liveUntilLedgerSeq?: number }) - .liveUntilLedgerSeq; - return typeof liveUntil === "number" ? liveUntil : 0; - } catch { + return withRetry( + async () => { + const ledgerKey = buildAllowanceLedgerKey(tokenId, owner, spender); + const response = await server.getLedgerEntries(ledgerKey); + + if (!response.entries || response.entries.length === 0) return 0; + + const entry = response.entries[0]; + const liveUntil = (entry as { liveUntilLedgerSeq?: number }) + .liveUntilLedgerSeq; + return typeof liveUntil === "number" ? liveUntil : 0; + }, + { + maxRetries: opts?.maxRetries ?? MAX_RETRIES, + baseDelayMs: opts?.baseDelayMs ?? RETRY_BASE_MS, + onRetry: (attempt, err) => + console.error( + `[alert-expiring] getAllowanceExpiryLedger retry ${attempt} for ${owner}: ${err instanceof Error ? err.message : String(err)}`, + ), + }, + ).catch((err) => { + console.error( + `[alert-expiring] getAllowanceExpiryLedger failed after retries for ${owner}: ${err instanceof Error ? err.message : String(err)}`, + ); return 0; - } + }); } // ── Core Logic ──────────────────────────────────────────────────────────────── @@ -314,17 +349,13 @@ async function getAllowanceExpiryLedger( * Checks a single subscriber and returns an ExpiryEntry if their allowance is * expiring within the alert window, or null otherwise. * - * Skips: - * - Invalid addresses - * - Subscribers with no active subscription - * - Allowances with no expiry set (liveUntil = 0) - * - Allowances that are still healthy (ledgers_remaining > ALERT_WINDOW_LEDGERS) + * Skips: invalid addresses, no active subscription, no expiry set, + * allowances with more than ALERT_WINDOW_LEDGERS remaining. */ -async function checkSubscriber( +export async function checkSubscriber( address: string, currentLedger: number, ): Promise { - // Validate the G-address format try { Address.fromString(address); } catch { @@ -333,9 +364,7 @@ async function checkSubscriber( } const sub = await getSubscription(address); - if (!sub) return null; // No subscription — nothing to alert on - - // Paused or inactive subscribers won't be charged, so skip them. + if (!sub) return null; if (!sub.active || sub.paused) return null; const [allowanceAmount, expiresAtLedger] = await Promise.all([ @@ -343,11 +372,9 @@ async function checkSubscriber( getAllowanceExpiryLedger(sub.token, address, CONTRACT_ID), ]); - // No expiry set — this allowance never expires, skip it. if (expiresAtLedger === 0) return null; const ledgersRemaining = Math.max(0, expiresAtLedger - currentLedger); - if (ledgersRemaining > ALERT_WINDOW_LEDGERS) return null; return { @@ -363,13 +390,12 @@ async function checkSubscriber( /** * POSTs the report as JSON to WEBHOOK_URL. - * - * Logs the HTTP status on success, logs the error on failure. - * Does NOT throw — callers rely on the exit code, not exceptions here. - * - * @returns true if the webhook was delivered with a 2xx status, false otherwise. + * Returns true on HTTP 2xx, false otherwise. Never throws. */ -async function sendWebhook(url: string, report: AlertReport): Promise { +async function sendWebhook( + url: string, + report: ExpiryAlertReport, +): Promise { try { const response = await fetch(url, { method: "POST", @@ -377,16 +403,13 @@ async function sendWebhook(url: string, report: AlertReport): Promise { body: JSON.stringify(report), }); if (response.ok) { - console.error( - `Webhook delivered successfully (HTTP ${response.status}).`, - ); + console.error(`Webhook delivered successfully (HTTP ${response.status}).`); return true; - } else { - console.error( - `Webhook returned non-2xx response: HTTP ${response.status} ${response.statusText}`, - ); - return false; } + console.error( + `Webhook returned non-2xx response: HTTP ${response.status} ${response.statusText}`, + ); + return false; } catch (err) { console.error(`Webhook delivery failed: ${err}`); return false; @@ -418,15 +441,11 @@ async function main(): Promise { } } - if (cliAddresses.length === 0 && !filePath) { - showHelp(); - } + if (cliAddresses.length === 0 && !filePath) showHelp(); - // Collect all addresses const allAddresses = [...cliAddresses]; if (filePath) { - const fileAddresses = await readAddressesFromFile(filePath); - allAddresses.push(...fileAddresses); + allAddresses.push(...(await readAddressesFromFile(filePath))); } if (allAddresses.length === 0) { @@ -444,19 +463,35 @@ async function main(): Promise { process.exit(1); } - // Check each subscriber sequentially to avoid hammering the RPC endpoint. + console.error( + `Checking ${allAddresses.length} subscriber(s) — concurrency=${CONCURRENCY}, maxRetries=${MAX_RETRIES}, window=${ALERT_WINDOW_LEDGERS} ledgers`, + ); + + // Run all checks concurrently under the configured cap. + const rawResults = await runConcurrent( + allAddresses.map( + (addr) => () => checkSubscriber(addr, currentLedger), + ), + CONCURRENCY, + ); + + // Collect non-null entries; log batch-level errors but don't abort. const expiring: ExpiryEntry[] = []; - for (const addr of allAddresses) { - const entry = await checkSubscriber(addr, currentLedger); - if (entry !== null) { - expiring.push(entry); + for (let i = 0; i < rawResults.length; i++) { + const r = rawResults[i]; + if (r instanceof Error) { + console.error( + `Unexpected error for ${allAddresses[i]}: ${r.message}`, + ); + } else if (r !== null) { + expiring.push(r); } } - // Sort by ledgers_remaining ascending (most urgent first). + // Sort by urgency (fewest ledgers remaining first). expiring.sort((a, b) => a.ledgers_remaining - b.ledgers_remaining); - const report: AlertReport = { + const report: ExpiryAlertReport = { generated_at: new Date().toISOString(), contract: CONTRACT_ID, current_ledger: currentLedger, @@ -465,8 +500,8 @@ async function main(): Promise { expiring, }; - // Always print the report to stdout. - console.log(JSON.stringify(report, null, 2)); + // Always print the JSON report to stdout. + process.stdout.write(JSON.stringify(report, null, 2) + "\n"); if (expiring.length === 0) { console.error( @@ -479,13 +514,8 @@ async function main(): Promise { `${expiring.length} allowance(s) expiring within ${ALERT_WINDOW_LEDGERS} ledgers.`, ); - // Send webhook unless --dry-run is set. if (WEBHOOK_URL && !dryRun) { - const delivered = await sendWebhook(WEBHOOK_URL, report); - // Exit 1 regardless of webhook success — the expiring allowances are the signal. - if (!delivered) { - // Already logged the error inside sendWebhook; still exit 1 below. - } + await sendWebhook(WEBHOOK_URL, report); } else if (dryRun) { console.error("Dry-run mode: webhook not sent."); } else if (!WEBHOOK_URL) { diff --git a/scripts/allowance-utils.ts b/scripts/allowance-utils.ts new file mode 100644 index 0000000..c4d4cdd --- /dev/null +++ b/scripts/allowance-utils.ts @@ -0,0 +1,306 @@ +/** + * allowance-utils.ts — Shared concurrency, retry, and type utilities for + * allowance audit scripts (check-allowances.ts, alert-expiring-allowances.ts). + * + * Exports + * ─────── + * withRetry() — bounded exponential-backoff retry for transient RPC errors + * pLimit() — token-bucket concurrency limiter + * isTransientError() — classifier: network/timeout = transient; business = definitive + * runConcurrent() — fan-out a list of tasks under a concurrency cap + * + * Shared JSON schema types are also exported so both scripts produce compatible + * shapes consumable by downstream alerting pipelines. + * + * Environment knobs (read by scripts, documented here for reference): + * CONCURRENCY Max simultaneous RPC calls (default: 5) + * MAX_RETRIES Max retry attempts per call (default: 3) + * RETRY_BASE_MS Base delay in ms before first retry (default: 300) + */ + +// ── Transient-error classifier ──────────────────────────────────────────────── + +/** + * Signals that an error is definitively a business-logic result, not a + * transport fault — callers should NOT retry on this. + * + * Throw or wrap errors with this to mark them as non-retryable. + */ +export class DefinitiveError extends Error { + constructor(message: string, public readonly cause?: unknown) { + super(message); + this.name = "DefinitiveError"; + } +} + +/** + * Returns `true` for errors that are transient transport or RPC infrastructure + * failures (network timeout, connection refused, rate-limit 429, server-side + * 5xx) that are safe to retry. + * + * Returns `false` for: + * - `DefinitiveError` — business logic results (e.g. insufficient allowance) + * - 4xx client errors other than 429 (bad request, auth, not found) + * - Any error that does not look like a network/transport fault + */ +export function isTransientError(err: unknown): boolean { + if (err instanceof DefinitiveError) return false; + + if (err instanceof Error) { + const msg = err.message.toLowerCase(); + + // Explicit non-retriable patterns + if ( + msg.includes("invalid") || + msg.includes("unauthorized") || + msg.includes("forbidden") || + msg.includes("not found") || + msg.includes("bad request") || + msg.includes("400") || + msg.includes("401") || + msg.includes("403") || + msg.includes("404") + ) { + return false; + } + + // Transient patterns + if ( + msg.includes("econnrefused") || + msg.includes("econnreset") || + msg.includes("etimedout") || + msg.includes("enotfound") || + msg.includes("socket") || + msg.includes("network") || + msg.includes("timeout") || + msg.includes("fetch failed") || + msg.includes("429") || + msg.includes("too many requests") || + msg.includes("502") || + msg.includes("503") || + msg.includes("504") || + msg.includes("internal server error") || + msg.includes("service unavailable") + ) { + return true; + } + } + + // Unknown errors — conservatively treat as transient so we don't silently + // drop valid subscribers on ephemeral infrastructure blips. + return true; +} + +// ── Retry ───────────────────────────────────────────────────────────────────── + +export interface RetryOptions { + /** Maximum number of retry attempts (not counting the first call). Default 3. */ + maxRetries?: number; + /** Base delay in ms before the first retry. Doubles each attempt. Default 300. */ + baseDelayMs?: number; + /** Optional jitter fraction [0, 1] applied to each delay. Default 0.2. */ + jitter?: number; + /** Called before each retry attempt with the attempt index (1-based) and error. */ + onRetry?: (attempt: number, err: unknown) => void; +} + +/** + * Executes `fn` and retries it on transient errors with truncated exponential + * backoff plus jitter. + * + * - DefinitiveError and non-transient errors propagate immediately without retry. + * - After `maxRetries` exhausted retries, the last error is rethrown. + * + * @example + * const result = await withRetry(() => server.simulateTransaction(tx), { maxRetries: 3 }); + */ +export async function withRetry( + fn: () => Promise, + options: RetryOptions = {}, +): Promise { + const maxRetries = options.maxRetries ?? 3; + const baseDelayMs = options.baseDelayMs ?? 300; + const jitter = options.jitter ?? 0.2; + + let lastErr: unknown; + + for (let attempt = 0; attempt <= maxRetries; attempt++) { + try { + return await fn(); + } catch (err: unknown) { + lastErr = err; + + // Do not retry definitive or non-transient errors + if (!isTransientError(err)) { + throw err; + } + + // Last attempt — rethrow + if (attempt === maxRetries) { + break; + } + + const delay = + baseDelayMs * 2 ** attempt * (1 + jitter * (Math.random() * 2 - 1)); + options.onRetry?.(attempt + 1, err); + await sleep(delay); + } + } + + throw lastErr; +} + +// ── Concurrency limiter ─────────────────────────────────────────────────────── + +/** + * Returns a function that wraps async tasks so at most `limit` run in parallel. + * Excess tasks are queued and started as slots free up. + * + * @example + * const run = pLimit(5); + * const results = await Promise.all(addresses.map(addr => run(() => check(addr)))); + */ +export function pLimit( + limit: number, +): (fn: () => Promise) => Promise { + if (limit < 1) throw new RangeError("pLimit: limit must be >= 1"); + + let active = 0; + const queue: Array<() => void> = []; + + function next(): void { + if (queue.length > 0 && active < limit) { + active++; + const resume = queue.shift()!; + resume(); + } + } + + return function run(fn: () => Promise): Promise { + return new Promise((resolve, reject) => { + const execute = (): void => { + fn().then( + (v) => { + active--; + next(); + resolve(v); + }, + (e) => { + active--; + next(); + reject(e); + }, + ); + }; + + if (active < limit) { + active++; + execute(); + } else { + queue.push(execute); + } + }); + }; +} + +// ── Fan-out runner ──────────────────────────────────────────────────────────── + +/** + * Runs `tasks` with at most `concurrency` tasks in flight simultaneously. + * Returns results in the same order as `tasks`, analogous to Promise.all but + * bounded. + * + * Individual task failures are caught and returned as `Error` instances in the + * result array so one failure never aborts the entire batch. + */ +export async function runConcurrent( + tasks: Array<() => Promise>, + concurrency: number, +): Promise> { + const run = pLimit(concurrency); + return Promise.all( + tasks.map((task) => + run(() => task()).catch((err: unknown) => + err instanceof Error ? err : new Error(String(err)), + ), + ), + ); +} + +// ── Sleep ───────────────────────────────────────────────────────────────────── + +/** Resolves after `ms` milliseconds. */ +export function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +// ── Shared JSON schema types ────────────────────────────────────────────────── + +/** + * Per-subscriber result emitted by check-allowances.ts. + * Stable schema: downstream alerting pipelines may parse this. + */ +export interface AuditResult { + /** Stellar G-address of the subscriber. */ + address: string; + /** Subscription charge amount in stroops. */ + subscription_amount: string; + /** Current approved allowance in stroops. */ + allowance: string; + /** Shortfall (subscription_amount - allowance) in stroops, "0" when sufficient. */ + gap: string; + /** SAC token contract address. */ + token: string; + /** Whether the subscription is active. */ + active: boolean; + /** Whether the subscription is paused. */ + paused: boolean; + /** true when the subscription is active, not paused, and gap > 0. */ + at_risk: boolean; + /** + * Set when the check could not be completed normally. + * Values: "invalid_address" | "no_subscription" | "rpc_error" | "unknown_error" + */ + error?: string; +} + +/** + * Top-level report produced by check-allowances.ts. + */ +export interface AllowanceCheckReport { + generated_at: string; + contract: string; + total_checked: number; + healthy_count: number; + at_risk_count: number; + error_count: number; + results: AuditResult[]; +} + +/** + * Per-subscriber expiry entry produced by alert-expiring-allowances.ts. + */ +export interface ExpiryEntry { + /** Stellar G-address of the subscriber. */ + address: string; + /** Merchant address. */ + merchant: string; + /** Current allowance amount in stroops. */ + allowance_amount: string; + /** Ledger sequence at which the allowance expires (0 = no expiry / not found). */ + expires_at_ledger: number; + /** Ledgers until expiry. 0 if already expired. */ + ledgers_remaining: number; +} + +/** + * Top-level report produced by alert-expiring-allowances.ts. + */ +export interface ExpiryAlertReport { + generated_at: string; + contract: string; + current_ledger: number; + alert_window_ledgers: number; + expiring_count: number; + expiring: ExpiryEntry[]; +} diff --git a/scripts/check-allowances.ts b/scripts/check-allowances.ts index a79b454..44d5d93 100644 --- a/scripts/check-allowances.ts +++ b/scripts/check-allowances.ts @@ -1,44 +1,90 @@ #!/usr/bin/env tsx /** - * check-allowances.ts — Audit subscriber token allowances against FlowPay subscription amounts + * check-allowances.ts — Audit subscriber token allowances against FlowPay + * subscription amounts. * - * Accepts a list of G-addresses (from args or --file) and checks whether each subscriber's - * allowance for their subscription token covers the next charge amount. + * Accepts a list of G-addresses (from args or --file) and checks whether each + * subscriber's allowance for their subscription token covers the next charge + * amount. Checks run concurrently under a configurable cap; transient RPC + * errors are retried with exponential backoff. + * + * Usage: + * tsx check-allowances.ts [options] [addresses...] + * + * Options: + * --file Read subscriber addresses from a file (one per line) + * --json Output machine-readable JSON to stdout; human summary to stderr + * --help, -h Show this help + * + * Environment variables: + * CONTRACT_ID Required. Deployed FlowPay contract ID. + * RPC_URL Soroban RPC endpoint (default: https://soroban-testnet.stellar.org). + * NETWORK_PASSPHRASE Network passphrase (default: Test SDF Network ; September 2015). + * CONCURRENCY Max simultaneous RPC calls (default: 5). + * MAX_RETRIES Retry attempts per transient error (default: 3). + * RETRY_BASE_MS Base backoff delay in ms (default: 300). + * + * Exit codes: + * 0 — all subscribers healthy (or errors / no subscription only) + * 1 — one or more subscribers are at risk of a failed charge */ -import { MultiEndpointServer } from "./rpc-client.js"; -import { logger } from "./logger"; import { + Contract, Networks, TransactionBuilder, BASE_FEE, nativeToScVal, Address, xdr, + Account, } from "@stellar/stellar-sdk"; - -// ── Configuration ──────────────────────────────────────────────────────────────── - -const RPC_URL = process.env.RPC_URL || "https://soroban-testnet.stellar.org"; -const CONTRACT_ID = process.env.CONTRACT_ID || ""; +import { MultiEndpointServer } from "./rpc-client.js"; +import { logger } from "./logger.js"; +import { + withRetry, + runConcurrent, + isTransientError, + DefinitiveError, + type AuditResult, + type AllowanceCheckReport, +} from "./allowance-utils.js"; + +// ── Configuration ───────────────────────────────────────────────────────────── + +const CONTRACT_ID = + process.env.CONTRACT_ID ?? process.env.VITE_CONTRACT_ID ?? ""; const NETWORK_PASSPHRASE = (process.env.NETWORK_PASSPHRASE ?? Networks.TESTNET) as string; +const CONCURRENCY = Math.max(1, parseInt(process.env.CONCURRENCY ?? "5", 10)); +const MAX_RETRIES = Math.max(0, parseInt(process.env.MAX_RETRIES ?? "3", 10)); +const RETRY_BASE_MS = Math.max( + 0, + parseInt(process.env.RETRY_BASE_MS ?? "300", 10), +); if (!CONTRACT_ID) { - logger.error("Error: CONTRACT_ID environment variable is required"); logger.error( - "Usage: CONTRACT_ID=your_contract_id tsx check-allowances.ts [--file subscribers.txt] [--json] [address1 address2 ...]" + "Error: CONTRACT_ID environment variable is required.\n" + + "Usage: CONTRACT_ID= tsx check-allowances.ts [--json] [--file subs.txt] [addr ...]", ); process.exit(1); } -const server = new MultiEndpointServer(RPC_URL); +/** Stable dummy source account for read-only simulations. */ +const SIM_SOURCE = "GAAZI4TCR3TY5OJHCTJC2A4QSY6CJWJH5IAJTGKIN2ER7LBNVKOCCWN"; + +const server = new MultiEndpointServer(); +const FlowPayAddress = Address.fromString(CONTRACT_ID); + +// ── Helpers ─────────────────────────────────────────────────────────────────── -// ── Helpers ─────────────────────────────────────────────────────────────────────── +function stroopsToXlm(stroops: string): string { + return (Number(stroops) / 10_000_000).toFixed(7); +} -function stroopsToXlm(stroops: string | bigint): string { - const value = typeof stroops === "bigint" ? Number(stroops) : Number(stroops); - return (value / 10_000_000).toFixed(7); +function addressVal(addr: string): xdr.ScVal { + return nativeToScVal(Address.fromString(addr), { type: "address" }); } async function parseAddressListFromFile(path: string): Promise { @@ -55,19 +101,26 @@ async function parseAddressListFromFile(path: string): Promise { } } -function showHelp(): void { +function showHelp(): never { logger.info(` Usage: tsx check-allowances.ts [options] [addresses...] Options: --file Read subscriber addresses from a file (one per line, # comments allowed) - --json Output machine-readable JSON instead of human-readable table + --json Output machine-readable JSON to stdout (human summary always on stderr) --help, -h Show this help message Environment: CONTRACT_ID Required. Deployed FlowPay contract ID - RPC_URL Optional. Soroban RPC endpoint (default: https://soroban-testnet.stellar.org) - NETWORK_PASSPHRASE Optional. Network passphrase (default: Test SDF Network ; September 2015) + RPC_URL Soroban RPC endpoint (default: https://soroban-testnet.stellar.org) + NETWORK_PASSPHRASE Network passphrase (default: Test SDF Network ; September 2015) + CONCURRENCY Max simultaneous RPC calls (default: 5) + MAX_RETRIES Retry attempts per transient error (default: 3) + RETRY_BASE_MS Base backoff delay in ms (default: 300) + +Exit codes: + 0 All subscribers healthy (or only errors/no-subscription results) + 1 One or more subscribers at risk of a failed charge Examples: CONTRACT_ID=CD123... tsx check-allowances.ts GXYZ... GABC... @@ -77,152 +130,199 @@ Examples: process.exit(0); } -// ── Contract Reads ─────────────────────────────────────────────────────────────── - -const FlowPayAddress = Address.fromString(CONTRACT_ID); - -function addressVal(addr: string): xdr.ScVal { - return nativeToScVal(Address.fromString(addr), { type: "address" }); -} +// ── Contract Reads ──────────────────────────────────────────────────────────── -async function getSubscription( - user: string, -): Promise<{ +interface Subscription { amount: bigint; token: string; active: boolean; paused: boolean; -} | null> { - try { - const contract = new Contract(CONTRACT_ID); - const account = await server.getAccount(user); - - const tx = new TransactionBuilder(account, { - fee: BASE_FEE, - networkPassphrase: NETWORK_PASSPHRASE, - }) - .addOperation(contract.call("get_subscription", addressVal(user))) - .setTimeout(30) - .build(); - - const result = await server.simulateTransaction(tx); - if ("error" in result) return null; - - const retval = (result as { result?: { retval?: xdr.ScVal } }).result - ?.retval; - if (!retval || retval.switch().name === "scvVoid") return null; - - const fields: Record = {}; - for (const entry of retval.map() ?? []) { - const key = entry.key().sym().toString(); - const val = entry.val(); - switch (key) { - case "amount": - fields[key] = BigInt(val.i128().toString()); - break; - case "token": - fields[key] = Address.fromScVal(val).toString(); - break; - case "active": - fields[key] = val.b(); - break; - case "paused": - fields[key] = val.b(); - break; +} + +/** + * Fetches the FlowPay subscription record for `user`. + * Retries on transient RPC failures; returns null when the user has no + * subscription or on any error after exhausting retries. + */ +export async function getSubscription( + user: string, + opts?: { maxRetries?: number; baseDelayMs?: number }, +): Promise { + return withRetry( + async () => { + const contract = new Contract(CONTRACT_ID); + const tx = new TransactionBuilder(new Account(SIM_SOURCE, "0"), { + fee: BASE_FEE, + networkPassphrase: NETWORK_PASSPHRASE, + }) + .addOperation(contract.call("get_subscription", addressVal(user))) + .setTimeout(30) + .build(); + + const result = await server.simulateTransaction(tx); + if ("error" in result) return null; + + const retval = (result as { result?: { retval?: xdr.ScVal } }).result + ?.retval; + if (!retval || retval.switch().name === "scvVoid") return null; + + const fields: Record = {}; + for (const entry of retval.map() ?? []) { + const key = entry.key().sym().toString(); + const val = entry.val(); + switch (key) { + case "amount": + fields[key] = BigInt(val.i128().toString()); + break; + case "token": + fields[key] = Address.fromScVal(val).toString(); + break; + case "active": + fields[key] = val.b(); + break; + case "paused": + fields[key] = val.b(); + break; + } } - } - return { - amount: fields.amount as bigint, - token: fields.token as string, - active: fields.active as boolean, - paused: fields.paused as boolean, - }; - } catch { + if (fields.amount === undefined || fields.token === undefined) { + return null; + } + + return { + amount: fields.amount as bigint, + token: fields.token as string, + active: (fields.active as boolean | undefined) ?? false, + paused: (fields.paused as boolean | undefined) ?? false, + }; + }, + { + maxRetries: opts?.maxRetries ?? MAX_RETRIES, + baseDelayMs: opts?.baseDelayMs ?? RETRY_BASE_MS, + onRetry: (attempt, err) => + logger.warn(`getSubscription retry ${attempt}`, { + address: user, + error: err instanceof Error ? err.message : String(err), + }), + }, + ).catch((err) => { + logger.error("getSubscription failed after retries", { + address: user, + error: err instanceof Error ? err.message : String(err), + }); return null; - } + }); } -async function getAllowance(owner: string, tokenId: string): Promise { - try { - const tokenContract = new Contract(tokenId); - - const account = await server.getAccount(owner).catch(() => null); - if (!account) return 0n; - - const tx = new TransactionBuilder(account, { - fee: BASE_FEE, - networkPassphrase: NETWORK_PASSPHRASE, - }) - .addOperation( - tokenContract.call( - "allowance", - addressVal(owner), - nativeToScVal(FlowPayAddress, { type: "address" }), - ), - ) - .setTimeout(30) - .build(); - - const result = await server.simulateTransaction(tx); - if ("error" in result) return 0n; - - const retval = (result as { result?: { retval?: xdr.ScVal } }).result - ?.retval; - if (!retval || retval.switch().name === "scvVoid") return 0n; - - return BigInt(retval.i128().toString()); - } catch { +/** + * Returns the current approved allowance (in stroops) that the FlowPay + * contract may spend on behalf of `owner` for token `tokenId`. + * Retries on transient RPC failures; returns 0n on error after exhausting retries. + */ +export async function getAllowance( + owner: string, + tokenId: string, + opts?: { maxRetries?: number; baseDelayMs?: number }, +): Promise { + return withRetry( + async () => { + const tokenContract = new Contract(tokenId); + const tx = new TransactionBuilder(new Account(SIM_SOURCE, "0"), { + fee: BASE_FEE, + networkPassphrase: NETWORK_PASSPHRASE, + }) + .addOperation( + tokenContract.call( + "allowance", + addressVal(owner), + nativeToScVal(FlowPayAddress, { type: "address" }), + ), + ) + .setTimeout(30) + .build(); + + const result = await server.simulateTransaction(tx); + if ("error" in result) return 0n; + + const retval = (result as { result?: { retval?: xdr.ScVal } }).result + ?.retval; + if (!retval || retval.switch().name === "scvVoid") return 0n; + + return BigInt(retval.i128().toString()); + }, + { + maxRetries: opts?.maxRetries ?? MAX_RETRIES, + baseDelayMs: opts?.baseDelayMs ?? RETRY_BASE_MS, + onRetry: (attempt, err) => + logger.warn(`getAllowance retry ${attempt}`, { + owner, + error: err instanceof Error ? err.message : String(err), + }), + }, + ).catch((err) => { + logger.error("getAllowance failed after retries", { + owner, + error: err instanceof Error ? err.message : String(err), + }); return 0n; - } + }); } -// ── Audit Logic ────────────────────────────────────────────────────────────────── - -interface AuditResult { - address: string; - subscriptionAmount: string; - allowance: string; - gap: string; - token: string; - active: boolean; - paused: boolean; - atRisk: boolean; - error?: string; -} +// ── Audit Logic ─────────────────────────────────────────────────────────────── -async function auditSubscriber(address: string): Promise { - let isValid = true; +/** + * Audits a single subscriber address. + * + * Definitive outcomes (invalid address, no subscription) are NOT retried here — + * only the underlying RPC calls inside getSubscription/getAllowance retry on + * transient errors. + */ +export async function auditSubscriber(address: string): Promise { + // Validate G-address format — definitive result, never retry try { Address.fromString(address); } catch { - isValid = false; - } - if (!isValid) { return { address, - subscriptionAmount: "0", + subscription_amount: "0", allowance: "0", gap: "0", token: "", active: false, paused: false, - atRisk: false, + at_risk: false, error: "invalid_address", }; } - const sub = await getSubscription(address); + let sub: Subscription | null; + try { + sub = await getSubscription(address); + } catch (err: unknown) { + return { + address, + subscription_amount: "0", + allowance: "0", + gap: "0", + token: "", + active: false, + paused: false, + at_risk: false, + error: err instanceof DefinitiveError ? "rpc_error" : "unknown_error", + }; + } + if (!sub) { return { address, - subscriptionAmount: "0", + subscription_amount: "0", allowance: "0", gap: "0", token: "", active: false, paused: false, - atRisk: false, + at_risk: false, error: "no_subscription", }; } @@ -233,78 +333,92 @@ async function auditSubscriber(address: string): Promise { return { address, - subscriptionAmount: sub.amount.toString(), + subscription_amount: sub.amount.toString(), allowance: allowance.toString(), gap: gap.toString(), token: sub.token, active: sub.active, paused: sub.paused, - atRisk, + at_risk: atRisk, }; } -// ── Output ─────────────────────────────────────────────────────────────────────── +// ── Output ──────────────────────────────────────────────────────────────────── + +export function buildReport(results: AuditResult[]): AllowanceCheckReport { + const healthy = results.filter((r) => !r.at_risk && !r.error && r.active); + const atRisk = results.filter((r) => r.at_risk); + const errors = results.filter((r) => !!r.error); -function printHumanReadable(results: AuditResult[]): void { - const atRisk = results.filter((r) => r.atRisk); + return { + generated_at: new Date().toISOString(), + contract: CONTRACT_ID, + total_checked: results.length, + healthy_count: healthy.length, + at_risk_count: atRisk.length, + error_count: errors.length, + results, + }; +} + +function printHumanSummary(report: AllowanceCheckReport): void { + const { results } = report; + const atRisk = results.filter((r) => r.at_risk); const noSub = results.filter((r) => r.error === "no_subscription"); - const healthy = results.filter((r) => !r.atRisk && !r.error && r.active); + const healthy = results.filter((r) => !r.at_risk && !r.error && r.active); + const errors = results.filter( + (r) => r.error && r.error !== "no_subscription", + ); logger.info(`\nAudited ${results.length} subscriber(s)\n`); if (noSub.length > 0) { logger.info(`${noSub.length} with no subscription:`); - for (const r of noSub) { - logger.info(` ${r.address}`); - } - logger.info(); + for (const r of noSub) logger.info(` ${r.address}`); + logger.info(""); + } + + if (errors.length > 0) { + logger.info(`${errors.length} with errors:`); + for (const r of errors) + logger.info(` ${r.address} [${r.error}]`); + logger.info(""); } if (atRisk.length > 0) { logger.info(`${atRisk.length} at risk of failed charge:`); const header = " ADDRESS".padEnd(56) + - "AMOUNT".padStart(10) + - "ALLOWANCE".padStart(12) + - "GAP".padStart(10) + - "TOKEN".padStart(56); + "AMOUNT".padStart(12) + + "ALLOWANCE".padStart(14) + + "GAP".padStart(12) + + " TOKEN"; logger.info(header); for (const r of atRisk) { - const line = - r.address.padEnd(56) + - stroopsToXlm(r.subscriptionAmount).padStart(10) + - stroopsToXlm(r.allowance).padStart(12) + - stroopsToXlm(r.gap).padStart(10) + - r.token.padStart(56); - logger.info(` ${line}`); + logger.info( + ` ${r.address.padEnd(56)}${stroopsToXlm(r.subscription_amount).padStart(12)}${stroopsToXlm(r.allowance).padStart(14)}${stroopsToXlm(r.gap).padStart(12)} ${r.token}`, + ); } - logger.info(); + logger.info(""); } if (healthy.length > 0) { logger.info(`${healthy.length} healthy:`); for (const r of healthy) { - console.log( - ` ${r.address.padEnd(56)} ${stroopsToXlm(r.subscriptionAmount).padStart(10)} ${stroopsToXlm(r.allowance).padStart(10)}`, logger.info( - ` ${r.address.padEnd(56)} ${stroopsToXlm(r.subscriptionAmount).padStart(10)} ${stroopsToXlm(r.allowance).padStart(10)}` + ` ${r.address.padEnd(56)} ${stroopsToXlm(r.subscription_amount).padStart(12)} ${stroopsToXlm(r.allowance).padStart(12)}`, ); } - logger.info(); + logger.info(""); } - console.log( - `Summary: healthy=${healthy.length}, atRisk=${atRisk.length}, noSubscription=${noSub.length}`, logger.info( - `Summary: healthy=${healthy.length}, atRisk=${atRisk.length}, noSubscription=${noSub.length}` + `Summary: healthy=${report.healthy_count}, atRisk=${report.at_risk_count}, ` + + `errors=${report.error_count}, total=${report.total_checked}`, ); } -function printJson(results: AuditResult[]): void { - logger.info(JSON.stringify(results, null, 2)); -} - -// ── Main ───────────────────────────────────────────────────────────────────────── +// ── Main ────────────────────────────────────────────────────────────────────── async function main(): Promise { const argv = process.argv.slice(2); @@ -329,14 +443,11 @@ async function main(): Promise { } } - if (addresses.length === 0 && !filePath) { - showHelp(); - } + if (addresses.length === 0 && !filePath) showHelp(); const allAddresses = [...addresses]; if (filePath) { - const fileAddresses = await parseAddressListFromFile(filePath); - allAddresses.push(...fileAddresses); + allAddresses.push(...(await parseAddressListFromFile(filePath))); } if (allAddresses.length === 0) { @@ -344,20 +455,52 @@ async function main(): Promise { process.exit(1); } - const results: AuditResult[] = []; - for (const addr of allAddresses) { - const result = await auditSubscriber(addr); - results.push(result); - } + logger.info( + `Auditing ${allAddresses.length} subscriber(s) — concurrency=${CONCURRENCY}, maxRetries=${MAX_RETRIES}`, + ); + + // Run all audits concurrently under the configured cap. + const rawResults = await runConcurrent( + allAddresses.map((addr) => () => auditSubscriber(addr)), + CONCURRENCY, + ); + + // Wrap any unexpected batch-level errors as error entries. + const results: AuditResult[] = rawResults.map((r, i) => { + if (r instanceof Error) { + logger.error(`Unexpected error for ${allAddresses[i]}`, { + error: r.message, + }); + return { + address: allAddresses[i]!, + subscription_amount: "0", + allowance: "0", + gap: "0", + token: "", + active: false, + paused: false, + at_risk: false, + error: "unknown_error", + }; + } + return r; + }); + + const report = buildReport(results); if (jsonOutput) { - printJson(results); - } else { - printHumanReadable(results); + // JSON report to stdout; human summary to stderr so they don't interleave + process.stdout.write(JSON.stringify(report, null, 2) + "\n"); } + // Human summary always emitted (logger writes to stderr for warn/error, stdout for info) + printHumanSummary(report); + + process.exit(report.at_risk_count > 0 ? 1 : 0); } -main().catch((error) => { - logger.error(`Fatal error: ${error}`); +main().catch((err: unknown) => { + logger.error( + `Fatal error: ${err instanceof Error ? err.message : String(err)}`, + ); process.exit(1); }); diff --git a/scripts/package.json b/scripts/package.json index a932381..d7fb8b0 100644 --- a/scripts/package.json +++ b/scripts/package.json @@ -41,6 +41,8 @@ "test:event-dedup": "tsx test-event-dedup-integration.ts", "test:renewal-forecast": "tsx test-renewal-forecast.ts", "test:rpc-failover": "tsx test-rpc-failover.ts", + "test:audit-trail-reconcile": "tsx test-audit-trail-reconcile.ts", + "test:allowance-utils": "tsx test-allowance-utils.ts" "test:audit-trail-reconcile": "tsx test-audit-trail-reconcile.ts" "test": "tsx test-churn-analysis.ts" "test:rpc-failover": "tsx test-rpc-failover.ts" diff --git a/scripts/test-allowance-utils.ts b/scripts/test-allowance-utils.ts new file mode 100644 index 0000000..94fc7c1 --- /dev/null +++ b/scripts/test-allowance-utils.ts @@ -0,0 +1,456 @@ +#!/usr/bin/env tsx +/** + * test-allowance-utils.ts — Unit tests for concurrency, retry, and error- + * classification logic in allowance-utils.ts. + * + * Uses in-process mock RPC functions — no network, no CONTRACT_ID required. + * + * Run: + * npx tsx scripts/test-allowance-utils.ts + * + * Exit codes: + * 0 — all tests passed + * 1 — one or more tests failed + */ + +import { + withRetry, + pLimit, + runConcurrent, + isTransientError, + DefinitiveError, + sleep, +} from "./allowance-utils.js"; + +// ── Test harness ────────────────────────────────────────────────────────────── + +let passed = 0; +let failed = 0; + +function assert(condition: boolean, label: string): void { + if (condition) { + console.log(` [PASS] ${label}`); + passed++; + } else { + console.error(` [FAIL] ${label}`); + failed++; + } +} + +function assertEqual(actual: T, expected: T, label: string): void { + const ok = + typeof actual === "object" + ? JSON.stringify(actual) === JSON.stringify(expected) + : actual === expected; + if (ok) { + console.log(` [PASS] ${label}`); + passed++; + } else { + console.error(` [FAIL] ${label}`); + console.error(` expected: ${JSON.stringify(expected)}`); + console.error(` actual: ${JSON.stringify(actual)}`); + failed++; + } +} + +async function assertRejects( + fn: () => Promise, + label: string, + msgContains?: string, +): Promise { + try { + await fn(); + console.error(` [FAIL] ${label} (expected rejection, got resolution)`); + failed++; + } catch (err) { + if ( + msgContains && + !(err instanceof Error && err.message.includes(msgContains)) + ) { + console.error( + ` [FAIL] ${label} (error message "${err instanceof Error ? err.message : String(err)}" does not contain "${msgContains}")`, + ); + failed++; + } else { + console.log(` [PASS] ${label}`); + passed++; + } + } +} + +// ── Suite 1: isTransientError ───────────────────────────────────────────────── +console.log("\n=== Suite 1: isTransientError ==="); + +{ + assert( + !isTransientError(new DefinitiveError("insufficient allowance")), + "DefinitiveError is NOT transient", + ); + + const transientMessages = [ + "ECONNREFUSED", + "ECONNRESET", + "ETIMEDOUT", + "socket hang up", + "fetch failed", + "429 too many requests", + "503 service unavailable", + "504 gateway timeout", + "internal server error", + "network error", + ]; + + for (const msg of transientMessages) { + assert( + isTransientError(new Error(msg)), + `"${msg}" is transient`, + ); + } + + const definitiveMessages = [ + "invalid address", + "400 bad request", + "401 unauthorized", + "403 forbidden", + "404 not found", + ]; + + for (const msg of definitiveMessages) { + assert( + !isTransientError(new Error(msg)), + `"${msg}" is NOT transient`, + ); + } + + // Unknown error type — conservatively transient + assert(isTransientError("some string error"), "unknown error is transient"); + assert(isTransientError(42), "numeric error is transient"); +} + +// ── Suite 2: withRetry — success on first attempt ───────────────────────────── +console.log("\n=== Suite 2: withRetry — success on first attempt ==="); + +{ + let calls = 0; + const result = await withRetry(async () => { + calls++; + return "ok"; + }); + assertEqual(result, "ok", "returns value on success"); + assertEqual(calls, 1, "fn called exactly once"); +} + +// ── Suite 3: withRetry — retries on transient, succeeds eventually ───────────── +console.log("\n=== Suite 3: withRetry — retries on transient errors ==="); + +{ + let calls = 0; + const retryEvents: number[] = []; + + // Fail twice with a transient error, then succeed on the third attempt + const result = await withRetry( + async () => { + calls++; + if (calls < 3) throw new Error("ECONNRESET simulated"); + return "recovered"; + }, + { + maxRetries: 3, + baseDelayMs: 0, // no delay in tests + onRetry: (attempt) => retryEvents.push(attempt), + }, + ); + + assertEqual(result, "recovered", "returns value after retries"); + assertEqual(calls, 3, "fn called 3 times total (2 failures + 1 success)"); + assertEqual(retryEvents, [1, 2], "onRetry called for attempts 1 and 2"); +} + +// ── Suite 4: withRetry — exhausts retries and rethrows ──────────────────────── +console.log("\n=== Suite 4: withRetry — exhausts retries ==="); + +{ + let calls = 0; + await assertRejects( + () => + withRetry( + async () => { + calls++; + throw new Error("ECONNREFUSED persistent"); + }, + { maxRetries: 2, baseDelayMs: 0 }, + ), + "rethrows after exhausting retries", + "ECONNREFUSED", + ); + assertEqual(calls, 3, "fn called 3 times (1 original + 2 retries)"); +} + +// ── Suite 5: withRetry — does NOT retry on DefinitiveError ──────────────────── +console.log("\n=== Suite 5: withRetry — no retry on DefinitiveError ==="); + +{ + let calls = 0; + await assertRejects( + () => + withRetry( + async () => { + calls++; + throw new DefinitiveError("subscription not found"); + }, + { maxRetries: 5, baseDelayMs: 0 }, + ), + "DefinitiveError propagates immediately without retry", + "subscription not found", + ); + assertEqual(calls, 1, "fn called exactly once (no retries)"); +} + +// ── Suite 6: withRetry — does NOT retry on non-transient HTTP errors ────────── +console.log( + "\n=== Suite 6: withRetry — no retry on non-transient errors ===", +); + +{ + let calls = 0; + await assertRejects( + () => + withRetry( + async () => { + calls++; + throw new Error("404 not found"); + }, + { maxRetries: 5, baseDelayMs: 0 }, + ), + "404 error propagates immediately", + "404", + ); + assertEqual(calls, 1, "fn called exactly once for 404"); +} + +// ── Suite 7: withRetry — maxRetries=0 means no retries ─────────────────────── +console.log("\n=== Suite 7: withRetry — maxRetries=0 ==="); + +{ + let calls = 0; + await assertRejects( + () => + withRetry( + async () => { + calls++; + throw new Error("ETIMEDOUT"); + }, + { maxRetries: 0, baseDelayMs: 0 }, + ), + "maxRetries=0 propagates on first failure", + ); + assertEqual(calls, 1, "fn called exactly once when maxRetries=0"); +} + +// ── Suite 8: pLimit — concurrency cap ──────────────────────────────────────── +console.log("\n=== Suite 8: pLimit — concurrency cap ==="); + +{ + const LIMIT = 3; + const run = pLimit(LIMIT); + + let activeNow = 0; + let peakActive = 0; + const results: number[] = []; + + const tasks = Array.from({ length: 10 }, (_, i) => + run(async () => { + activeNow++; + if (activeNow > peakActive) peakActive = activeNow; + await sleep(5); // tiny async gap so concurrent tasks overlap + activeNow--; + results.push(i); + return i; + }), + ); + + const resolved = await Promise.all(tasks); + + assert(peakActive <= LIMIT, `peak concurrency (${peakActive}) ≤ limit (${LIMIT})`); + assertEqual(resolved.length, 10, "all 10 tasks resolved"); + assert( + resolved.every((v, i) => v === i), + "results are in input order", + ); +} + +// ── Suite 9: pLimit — limit=1 serialises tasks ──────────────────────────────── +console.log("\n=== Suite 9: pLimit — limit=1 serialises ==="); + +{ + const run = pLimit(1); + const order: number[] = []; + let activeNow = 0; + let concurrent = false; + + await Promise.all( + [0, 1, 2, 3, 4].map((i) => + run(async () => { + activeNow++; + if (activeNow > 1) concurrent = true; + await sleep(2); + order.push(i); + activeNow--; + }), + ), + ); + + assert(!concurrent, "limit=1: no two tasks ran simultaneously"); + assertEqual(order, [0, 1, 2, 3, 4], "limit=1: tasks executed in submission order"); +} + +// ── Suite 10: pLimit — rejects on invalid limit ─────────────────────────────── +console.log("\n=== Suite 10: pLimit — rejects invalid limit ==="); + +{ + let threw = false; + try { + pLimit(0); + } catch { + threw = true; + } + assert(threw, "pLimit(0) throws RangeError"); +} + +// ── Suite 11: runConcurrent — all tasks succeed ─────────────────────────────── +console.log("\n=== Suite 11: runConcurrent — all succeed ==="); + +{ + const tasks = [1, 2, 3, 4, 5].map((n) => async () => n * 2); + const results = await runConcurrent(tasks, 3); + assertEqual(results, [2, 4, 6, 8, 10], "runConcurrent returns results in order"); +} + +// ── Suite 12: runConcurrent — partial failures captured as Error objects ─────── +console.log("\n=== Suite 12: runConcurrent — partial failures ==="); + +{ + const tasks = [ + async () => "a", + async (): Promise => { + throw new Error("task-2-failed"); + }, + async () => "c", + async (): Promise => { + throw new Error("task-4-failed"); + }, + async () => "e", + ]; + + const results = await runConcurrent(tasks, 2); + + assertEqual(results.length, 5, "runConcurrent returns an entry for every task"); + assertEqual(results[0], "a", "task 0 succeeded"); + assert(results[1] instanceof Error, "task 1 failure is an Error object"); + assert( + (results[1] as Error).message.includes("task-2-failed"), + "task 1 error message preserved", + ); + assertEqual(results[2], "c", "task 2 succeeded"); + assert(results[3] instanceof Error, "task 3 failure is an Error object"); + assertEqual(results[4], "e", "task 4 succeeded"); +} + +// ── Suite 13: runConcurrent — concurrency cap respected ─────────────────────── +console.log("\n=== Suite 13: runConcurrent — concurrency cap ==="); + +{ + const LIMIT = 4; + let active = 0; + let peak = 0; + + const tasks = Array.from({ length: 20 }, () => async () => { + active++; + if (active > peak) peak = active; + await sleep(3); + active--; + return true; + }); + + await runConcurrent(tasks, LIMIT); + + assert(peak <= LIMIT, `runConcurrent peak (${peak}) ≤ limit (${LIMIT})`); +} + +// ── Suite 14: withRetry — mock flaky RPC (3-of-5 failures then success) ──────── +console.log( + "\n=== Suite 14: withRetry — mock flaky RPC (intermittent 503) ===", +); + +{ + const callLog: Array<{ attempt: number; outcome: string }> = []; + let attempt = 0; + const retryAttempts: number[] = []; + + // Simulates an RPC that fails 4 times then succeeds + const mockRpc = async (): Promise => { + attempt++; + if (attempt <= 4) { + callLog.push({ attempt, outcome: "503 service unavailable" }); + throw new Error("503 service unavailable"); + } + callLog.push({ attempt, outcome: "success" }); + return "data"; + }; + + const result = await withRetry(mockRpc, { + maxRetries: 5, + baseDelayMs: 0, + onRetry: (a) => retryAttempts.push(a), + }); + + assertEqual(result, "data", "mock flaky RPC eventually returns data"); + assertEqual(callLog.length, 5, "mock RPC called 5 times"); + assertEqual(retryAttempts, [1, 2, 3, 4], "onRetry fired for retries 1-4"); + assertEqual( + callLog.filter((e) => e.outcome === "503 service unavailable").length, + 4, + "4 transient failures before success", + ); +} + +// ── Suite 15: withRetry — retry count visible in onRetry callback ────────────── +console.log("\n=== Suite 15: withRetry — onRetry receives correct attempt numbers ==="); + +{ + const attempts: number[] = []; + await withRetry( + async () => { + if (attempts.length < 3) throw new Error("ECONNRESET"); + return "done"; + }, + { + maxRetries: 5, + baseDelayMs: 0, + onRetry: (n) => attempts.push(n), + }, + ); + // attempts should be [1, 2, 3] (three retries before success) + assertEqual(attempts, [1, 2, 3], "onRetry receives 1-based attempt numbers"); +} + +// ── Suite 16: runConcurrent — empty task list ───────────────────────────────── +console.log("\n=== Suite 16: runConcurrent — empty task list ==="); + +{ + const results = await runConcurrent([], 5); + assertEqual(results, [], "empty task list returns empty array"); +} + +// ── Results ─────────────────────────────────────────────────────────────────── + +console.log(`\n${"=".repeat(60)}`); +console.log(`Results: ${passed} passed, ${failed} failed`); +console.log(`${"=".repeat(60)}`); + +if (failed > 0) { + process.exit(1); +} else { + console.log("All allowance-utils tests passed ✓"); + process.exit(0); +}