From fbceb18cf7e26a99f0ab170abaf12cffe81c4029 Mon Sep 17 00:00:00 2001 From: bibi-fay Date: Tue, 20 Oct 2026 17:36:59 +0100 Subject: [PATCH] feat:alerting for stuck entries and a fee-bump ceiling --- package-lock.json | 1 + src/outbox/dispatcher.ts | 95 ++++- src/outbox/executors.ts | 820 ++++++++++++++++++++++++++----------- src/outbox/service.ts | 92 ++++- src/outbox/stateMachine.ts | 37 ++ 5 files changed, 797 insertions(+), 248 deletions(-) diff --git a/package-lock.json b/package-lock.json index f003b087..f1eecb3a 100644 --- a/package-lock.json +++ b/package-lock.json @@ -7340,6 +7340,7 @@ "version": "2.3.3", "resolved": "https://registry.npmjs.org/fsevents/-/fsevents-2.3.3.tgz", "integrity": "sha512-5xoDfX+fL7faATnagmWPpbFtwh/R77WmMMqqHGS65C3vvB0YHrgF+B1YmZ3441tMj5n63k0212XNoJwzlhffQw==", + "dev": true, "hasInstallScript": true, "license": "MIT", "optional": true, diff --git a/src/outbox/dispatcher.ts b/src/outbox/dispatcher.ts index 21c39b32..d557a2f6 100644 --- a/src/outbox/dispatcher.ts +++ b/src/outbox/dispatcher.ts @@ -63,6 +63,12 @@ import { isStale as isFeeSnapshotStale, } from '../stellar/feeOracle' +/** + * Accessor for the shared per-signer concurrency limiter. Resolved on each + * call (rather than cached at module load) so it always reflects the current + * config. Limits apply both globally and per signer account, which keeps + * submissions from one signer serialized (avoids sequence-number clashes). + */ function signerLock() { return getSignerLock( config.outbox.globalMaxInFlight, @@ -76,6 +82,8 @@ function signerLock() { * documented cap in docs/OUTBOX.md. */ function computeFeeMultiplier(attempts: number): number { + // attempts is 1 on the first try (claimOp increments before submit), so + // the first attempt gets 0 bumps (multiplier 1) and each retry adds one. const bumps = Math.min( Math.max(attempts - 1, 0), config.outbox.feeBumpMaxAttempts @@ -83,6 +91,8 @@ function computeFeeMultiplier(attempts: number): number { return config.outbox.feeBumpMultiplier ** bumps } +// Numeric ranking of the fee oracle's congestion levels so they can be +// compared with >= / <. Unknown levels are treated as 0 (lowest) by callers. const CONGESTION_ORDER: Record = { low: 0, elevated: 1, @@ -90,15 +100,27 @@ const CONGESTION_ORDER: Record = { severe: 3, } +/** + * Should this op be held back for now? Only LOW-priority ops are ever + * deferred, and only while congestion is "high" or worse. The deferral is + * bounded: once an op has waited longer than lowMaxDeferMs it is dispatched + * regardless, so LOW work can't starve forever. + */ function shouldDeferLowOp(op: OutboxOpRecord, congestion: string): boolean { if (op.priority !== 'LOW') return false if ((CONGESTION_ORDER[congestion] ?? 0) < CONGESTION_ORDER['high']) return false const ageMs = Date.now() - new Date(op.createdAt).getTime() + // Waited too long already — stop deferring and let it through. if (ageMs > config.outbox.lowMaxDeferMs) return false return true } +/** + * Pick the base fee (before attempt-based bumps) from the current fee + * snapshot. CRITICAL ops pay the "aggressive" fee once congestion reaches + * "elevated"; everything else uses the normal recommended fee. + */ function getBaseFeeForOp(op: OutboxOpRecord): { baseFee: number isAggressive: boolean @@ -114,6 +136,12 @@ function getBaseFeeForOp(op: OutboxOpRecord): { return { baseFee: snapshot.recommendedBaseFee, isAggressive: false } } +/** + * Side effects when an op reaches permanent FAILED: log it, raise a critical + * alert, and notify the user via an event. The alert and event are + * best-effort — their own failures are swallowed/logged so they can never + * mask the original failure or crash the dispatcher. + */ async function onTerminalFailure( op: OutboxOpRecord, errorMessage: string @@ -163,6 +191,9 @@ async function onTerminalFailure( * caller's existing error handling is unaffected by the outbox underneath it. */ async function submitClaimedOp(op: OutboxOpRecord): Promise { + // --- Fee calculation ----------------------------------------------------- + // Base fee from congestion/priority, scaled by the per-attempt bump + // multiplier, then clamped to the absolute ceiling (maxAbsFee). const { baseFee, isAggressive } = getBaseFeeForOp(op) if (isAggressive) recordOutboxAggressiveFeeUsed() @@ -172,6 +203,9 @@ async function submitClaimedOp(op: OutboxOpRecord): Promise { const hitCap = rawFee > config.outbox.maxAbsFee if (hitCap) { recordOutboxMaxFeeHit(op.priority) + // A CRITICAL op being capped means it may be underpriced for the current + // network, so page someone. The second argument is a dedupe key so the + // alert isn't re-sent on every attempt. if (op.priority === 'CRITICAL') { await alertingService .emit( @@ -192,6 +226,11 @@ async function submitClaimedOp(op: OutboxOpRecord): Promise { const lock = signerLock() try { + // Run the submission under the per-signer lock so concurrent ops from + // the same signer account don't collide. The callback persists the + // txHash as soon as it exists, so a crash mid-confirmation can still be + // reconciled later. + // (signerPublicKey is always set here: claimOp stored it at claim time.) const result = await lock.withLock(op.signerPublicKey!, () => executeOutboxPayload(op.payload, feeMultiplier, (txHash) => recordSubmittedTxHash(op.id, txHash) @@ -199,6 +238,8 @@ async function submitClaimedOp(op: OutboxOpRecord): Promise { ) if (result.status === 'pending') { + // Submitted but not yet confirmed. Leave the op SUBMITTED and keep the + // hash; the stuck-op reconciler or the events oracle will resolve it. await recordSubmittedTxHash(op.id, result.hash) await mirrorPendingTransaction(op.payload, result.hash) logger.warn('[Outbox] Confirmation pending; preserving submitted hash', { @@ -206,12 +247,14 @@ async function submitClaimedOp(op: OutboxOpRecord): Promise { txHash: result.hash, }) } else if (!result.status || result.status === 'success') { + // Success path (a missing status is treated as success). await markConfirmed(op.id, result.hash) await mirrorLinkedTransaction(op.payload, { txHash: result.hash, status: 'CONFIRMED', }) recordOutboxOp(op.kind, op.priority, 'confirmed') + // End-to-end latency from enqueue to confirmation, in seconds. recordOutboxLatency(op.kind, (Date.now() - op.createdAt.getTime()) / 1000) logger.info('[Outbox] Op confirmed', { opId: op.id, @@ -220,6 +263,8 @@ async function submitClaimedOp(op: OutboxOpRecord): Promise { attempts: op.attempts, }) } else { + // The network resolved but rejected the op (status === 'failed'). + // Terminal and not retried; returned (not thrown) to the caller. await markFailedTerminal( op.id, result.hash, @@ -241,6 +286,9 @@ async function submitClaimedOp(op: OutboxOpRecord): Promise { return result } catch (err) { const message = err instanceof Error ? err.message : String(err) + // A confirmation timeout is not a failure: the tx was submitted and may + // still land. Keep the hash, leave the op SUBMITTED, and report pending + // rather than retrying (which could double-submit). if (err instanceof TransactionConfirmationTimeoutError) { await recordSubmittedTxHash(op.id, err.txHash) await mirrorPendingTransaction(op.payload, err.txHash) @@ -254,10 +302,13 @@ async function submitClaimedOp(op: OutboxOpRecord): Promise { return { hash: err.txHash, status: 'pending' } } + // Any other exception: either schedule a retry with backoff or, if + // attempts are exhausted, fail permanently. const { terminal } = await markFailedOrRetry(op, message) recordOutboxOp(op.kind, op.priority, terminal ? 'failed' : 'retry') if (terminal) { + // No txHash is available on this path, hence only the status mirrors. await mirrorLinkedTransaction(op.payload, { status: 'FAILED' }) await onTerminalFailure(op, message) } else { @@ -269,6 +320,8 @@ async function submitClaimedOp(op: OutboxOpRecord): Promise { }) } + // Re-throw so synchronous callers see the same error they would have + // seen before the outbox existed. throw err } } @@ -287,6 +340,8 @@ export async function dispatchOne(opId: string): Promise { const op = await getOp(opId) if (!op) throw new Error(`Outbox op ${opId} not found`) + // Idempotent short-circuit: if it already confirmed (e.g. via the sweep), + // return the existing result instead of erroring. if (op.status === 'CONFIRMED' && op.txHash) { return { hash: op.txHash, status: 'success' } } @@ -294,6 +349,7 @@ export async function dispatchOne(opId: string): Promise { throw new Error(`Outbox op ${opId} is not PENDING (status=${op.status})`) } + // Compliance/freeze guard: never dispatch for a halted user. if (await isUserHalted(op.userId)) { throw new Error( `Outbox op ${opId} not dispatched: user ${op.userId} is frozen` @@ -304,6 +360,8 @@ export async function dispatchOne(opId: string): Promise { try { const snap = getFeeSnapshot() if (shouldDeferLowOp(op, snap.congestionLevel)) { + // Push nextAttemptAt out so the sweep also leaves it alone until then, + // then throw so the inline caller knows it wasn't dispatched. const deferUntil = new Date(Date.now() + config.outbox.lowDeferMs) await db.outboxOp.update({ where: { id: op.id }, @@ -316,6 +374,7 @@ export async function dispatchOne(opId: string): Promise { } } catch (err) { // If the error is our deferral throw, rethrow; otherwise log and proceed + // (a broken fee oracle must not block dispatching — fail open here). if ( err instanceof Error && err.message.includes('deferred due to high congestion') @@ -327,9 +386,12 @@ export async function dispatchOne(opId: string): Promise { }) } + // Decide which signer account will submit, then atomically claim the op. const signerPublicKey = await resolveSignerPublicKey(op.payload, op.userId) const claimed = await claimOp(opId, signerPublicKey) if (!claimed) { + // Lost the claim race. Re-read: if the winner already confirmed it, + // return that result; otherwise surface the conflict as an error. const latest = await getOp(opId) if (latest?.status === 'CONFIRMED' && latest.txHash) { return { hash: latest.txHash, status: 'success' } @@ -350,6 +412,7 @@ export async function dispatchOne(opId: string): Promise { * attempt doesn't win the claim race first. */ export function dispatchInBackground(opId: string): void { + // Errors are only logged: the op stays durable, so the sweep can retry. dispatchOne(opId).catch((err) => { logger.warn('[Outbox] Background dispatch attempt failed (retriable)', { opId, @@ -361,14 +424,17 @@ export function dispatchInBackground(opId: string): void { /** Move ops SUBMITTED-but-unconfirmed past the timeout back into the retry pipeline. */ async function reconcileStuckSubmitted(): Promise { const stuck = await findStuckSubmittedOps(config.outbox.submittedTimeoutMs) - updateOutboxStuckSubmitted(stuck.length) + updateOutboxStuckSubmitted(stuck.length) // gauge for monitoring/alerting for (const op of stuck) { + // Step 1: if we have a txHash, ask the chain what actually happened + // before deciding to resubmit (avoids double-submitting a tx that landed). if (op.txHash) { let chainStatus: 'success' | 'failed' | 'not_found' try { chainStatus = await getTransactionStatus(op.txHash) } catch (err) { + // RPC unavailable: leave the op alone and try again next sweep. logger.warn('[Outbox] Could not reconcile submitted transaction hash', { opId: op.id, txHash: op.txHash, @@ -377,6 +443,7 @@ async function reconcileStuckSubmitted(): Promise { continue } + // It landed after all: close the op out as confirmed. if (chainStatus === 'success') { await markConfirmed(op.id, op.txHash) await mirrorLinkedTransaction(op.payload, { @@ -387,6 +454,7 @@ async function reconcileStuckSubmitted(): Promise { continue } + // It landed but failed on-chain: terminal, no retry. if (chainStatus === 'failed') { await markFailedTerminal( op.id, @@ -401,8 +469,12 @@ async function reconcileStuckSubmitted(): Promise { await onTerminalFailure(op, 'Submitted Stellar transaction failed') continue } + // 'not_found' falls through to the escalation logic below. } + // Step 2: out of fee-bump attempts. Passing attempts=maxAttempts here + // forces markFailedOrRetry down its terminal branch, so this always + // ends in FAILED (the `terminal` flag is effectively always true). if (op.attempts >= config.outbox.feeBumpMaxAttempts) { const { terminal } = await markFailedOrRetry( { id: op.id, attempts: config.outbox.maxAttempts }, @@ -413,6 +485,8 @@ async function reconcileStuckSubmitted(): Promise { continue } + // Step 3: still within the fee-bump budget — requeue as PENDING. The + // next claim bumps `attempts`, which raises the fee on resubmission. logger.warn( '[Outbox] Op unconfirmed past timeout — escalating to fee-bump retry', { @@ -433,16 +507,20 @@ async function reconcileStuckSubmitted(): Promise { * CRITICAL withdrawals never wait behind a NORMAL/LOW wave). */ export async function runDispatchSweep(): Promise { + // First clean up anything stuck, so requeued ops can be picked up below. await reconcileStuckSubmitted() + // Refresh the queue-depth gauges. const stats = await getQueueStats() updateOutboxQueueDepth(stats) + // Fetch up to batchSize eligible ops, then order by priority. const claimable = sortForDispatch( await findClaimableOps(config.outbox.batchSize) ) for (const op of claimable) { + // Skip (but leave PENDING) ops belonging to frozen users. if (await isUserHalted(op.userId)) { logger.info('[Outbox] Skipping op for frozen user', { opId: op.id, @@ -452,6 +530,7 @@ export async function runDispatchSweep(): Promise { } // LOW deferral during high/severe congestion (bounded) + // Same rule as dispatchOne, but here we `continue` instead of throwing. try { const snap = getFeeSnapshot() if (shouldDeferLowOp(op, snap.congestionLevel)) { @@ -469,12 +548,15 @@ export async function runDispatchSweep(): Promise { continue } } catch (err) { + // Fail open: a fee-oracle problem shouldn't stall the whole sweep. logger.warn('[Outbox] Defer check failed, proceeding to dispatch', { opId: op.id, error: err instanceof Error ? err.message : String(err), }) } + // Work out which signer would submit; skip this op if that fails so one + // bad op can't abort the rest of the batch. let signerPublicKey: string try { signerPublicKey = await resolveSignerPublicKey(op.payload, op.userId) @@ -486,6 +568,7 @@ export async function runDispatchSweep(): Promise { continue } + // Respect global and per-signer in-flight limits before claiming. if (!signerLock().hasCapacity(signerPublicKey)) { continue // over the concurrency cap — left PENDING for the next sweep } @@ -502,8 +585,18 @@ export async function runDispatchSweep(): Promise { } } +// Handle to the running interval, kept at module scope (e.g. for shutdown). let dispatcherHandle: NodeJS.Timeout | null = null +/** + * Start the background dispatcher: run one sweep immediately, then repeat + * every dispatchIntervalMs. Sweep errors are logged and never propagate, so + * a single failed sweep can't kill the timer. Returns the interval handle so + * callers can clear it on shutdown. + * + * Note: sweeps are not guarded against overlap — if one runs longer than the + * interval, the next starts anyway. This is safe because claimOp is atomic. + */ export function scheduleOutboxDispatcher(): NodeJS.Timeout { runDispatchSweep().catch((err) => logger.error('[Outbox] Initial dispatch sweep failed', { diff --git a/src/outbox/executors.ts b/src/outbox/executors.ts index 59438966..d557a2f6 100644 --- a/src/outbox/executors.ts +++ b/src/outbox/executors.ts @@ -1,289 +1,619 @@ /** - * The single point where a durable OutboxOp becomes a real on-chain - * submission (#325). + * The prioritized dispatcher (#325): claims durable outbox intents and turns + * them into real Stellar submissions, with retry/backoff, fee-bump-on- + * congestion, per-signer serialization, and a compliance halt guard. * - * Besides src/stellar/contract.ts itself, THIS is the only module allowed to - * import the raw write functions (depositForUser, withdrawForUser, - * triggerRebalance, payReferralReward). Every money-moving caller - * (src/routes/deposit.ts, withdraw.ts, src/agent/router.ts, - * src/referral/service.ts) goes through src/outbox/service.ts instead — - * enforced by tests/unit/outbox/structural.test.ts, which fails CI the - * moment a new caller bypasses the outbox. + * Two entry points: + * - dispatchOne(opId) — claim-and-submit ONE specific PENDING op inline, + * awaited by synchronous callers (deposit/withdraw + * routes, referral payout) that need the result now. + * - runDispatchSweep() — the background pass: claims whatever PENDING + * backlog exists (crash recovery, retries whose + * backoff has elapsed, and any genuinely + * fire-and-forget op such as an agent rebalance), + * priority-ordered, and reconciles SUBMITTED ops + * that have gone quiet too long. + * + * Both paths converge on submitClaimedOp — there is exactly one code path + * that ever calls src/outbox/executors.ts. */ +import { logger } from '../utils/logger' +import { config } from '../config/env' +import { alertingService } from '../services/alerting' +import db from '../db' +import { publishUserEvent } from '../events/publisher' +import { EVENT_TYPE_TOPIC } from '../events/types' import { - depositForUser, - withdrawForUser, - triggerRebalance as submitRebalance, - payReferralReward, -} from '../stellar/contract' -import { getWalletByUserId } from '../stellar/wallet' -import { getAgentKeypair } from '../stellar/client' -import { TransactionResult } from '../stellar/types' -import { OutboxPayload } from './types' + TransactionConfirmationTimeoutError, + TransactionResult, +} from '../stellar/types' +import { getTransactionStatus } from '../stellar/client' +import { getSignerLock } from './signerLock' +import { resolveSignerPublicKey, executeOutboxPayload } from './executors' +import { sortForDispatch } from './stateMachine' +import { OutboxOpRecord } from './types' import { - Keypair, - Asset, - rpc, - TransactionBuilder, - Operation, - BASE_FEE, -} from '@stellar/stellar-sdk' + claimOp, + findClaimableOps, + findStuckSubmittedOps, + getOp, + getQueueStats, + isUserHalted, + markConfirmed, + markFailedOrRetry, + markFailedTerminal, + mirrorLinkedTransaction, + mirrorPendingTransaction, + recordSubmittedTxHash, + returnStuckOpToPending, +} from './service' import { - buildSponsoredCreateAccount, - buildSponsoredTrustline, - buildRevokeSponsorship, - getSponsorKeypairs, -} from '../stellar/sponsorship' + recordOutboxOp, + recordOutboxFeeBump, + recordOutboxLatency, + updateOutboxQueueDepth, + updateOutboxStuckSubmitted, + recordOutboxLowDeferred, + recordOutboxAggressiveFeeUsed, + recordOutboxMaxFeeHit, +} from '../utils/metrics' import { - submitTransaction, - waitForConfirmation, - prepareTransaction, - simulateTransaction, - getAccount, - getNetworkPassphrase, -} from '../stellar/client' -import db from '../db' + getFeeSnapshot, + isStale as isFeeSnapshotStale, +} from '../stellar/feeOracle' /** - * Which Stellar public key an op's submission will sign with — resolved - * without decrypting anything, so the dispatcher can serialize per-signer - * (src/outbox/signerLock.ts) before it commits to actually claiming the op. + * Accessor for the shared per-signer concurrency limiter. Resolved on each + * call (rather than cached at module load) so it always reflects the current + * config. Limits apply both globally and per signer account, which keeps + * submissions from one signer serialized (avoids sequence-number clashes). */ -export async function resolveSignerPublicKey( - payload: OutboxPayload, - userId: string -): Promise { - switch (payload.method) { - case 'deposit': - case 'withdraw': { - const wallet = await getWalletByUserId(userId) - if (!wallet) { - throw new Error(`No custodial wallet found for user ${userId}`) - } - return wallet.publicKey - } - case 'rebalance': - case 'referral_reward': - return getAgentKeypair().publicKey() - case 'sponsor_create_account': - case 'sponsor_trustline': - case 'revoke_sponsorship': - return (payload as any).sponsorAccount as string - case 'treasury_sweep': { - const account = await db.treasuryAccount.findFirst({ - where: { tier: payload.fromTier as any, isActive: true }, - select: { publicKey: true }, - }) - if (!account) { - throw new Error( - `No active treasury account found for tier ${payload.fromTier}` - ) - } - return account.publicKey - } - default: - throw new Error(`Unsupported outbox method: ${(payload as any).method}`) - } +function signerLock() { + return getSignerLock( + config.outbox.globalMaxInFlight, + config.outbox.perAccountMaxInFlight + ) } -function getSponsorKeypair(publicKey: string): Keypair { - const kps = getSponsorKeypairs() - const kp = kps.find((k) => k.publicKey() === publicKey) - if (!kp) throw new Error(`Sponsor key not found for ${publicKey}`) - return kp +/** + * Fee multiplier for a submission, derived from how many attempts have + * already been made. Compounds up to feeBumpMaxAttempts, matching the + * documented cap in docs/OUTBOX.md. + */ +function computeFeeMultiplier(attempts: number): number { + // attempts is 1 on the first try (claimOp increments before submit), so + // the first attempt gets 0 bumps (multiplier 1) and each retry adds one. + const bumps = Math.min( + Math.max(attempts - 1, 0), + config.outbox.feeBumpMaxAttempts + ) + return config.outbox.feeBumpMultiplier ** bumps } -async function submitSponsoredTransaction( - tx: ReturnType, - signer: Keypair -): Promise { - const simulation = await simulateTransaction(tx as any) - if (rpc.Api.isSimulationError(simulation as any)) { - throw new Error( - `Sponsorship simulation failed: ${(simulation as any).error}` - ) - } - const prepared = await prepareTransaction(tx as any) - prepared.sign(signer) - // Sponsored create also needs sponsored account signature? For createAccount with 0 balance sponsored, only sponsor signs. Add sponsored sig if needed? No. - const hash = await submitTransaction(prepared) - const result = await waitForConfirmation(hash) +// Numeric ranking of the fee oracle's congestion levels so they can be +// compared with >= / <. Unknown levels are treated as 0 (lowest) by callers. +const CONGESTION_ORDER: Record = { + low: 0, + elevated: 1, + high: 2, + severe: 3, +} + +/** + * Should this op be held back for now? Only LOW-priority ops are ever + * deferred, and only while congestion is "high" or worse. The deferral is + * bounded: once an op has waited longer than lowMaxDeferMs it is dispatched + * regardless, so LOW work can't starve forever. + */ +function shouldDeferLowOp(op: OutboxOpRecord, congestion: string): boolean { + if (op.priority !== 'LOW') return false + if ((CONGESTION_ORDER[congestion] ?? 0) < CONGESTION_ORDER['high']) + return false + const ageMs = Date.now() - new Date(op.createdAt).getTime() + // Waited too long already — stop deferring and let it through. + if (ageMs > config.outbox.lowMaxDeferMs) return false + return true +} + +/** + * Pick the base fee (before attempt-based bumps) from the current fee + * snapshot. CRITICAL ops pay the "aggressive" fee once congestion reaches + * "elevated"; everything else uses the normal recommended fee. + */ +function getBaseFeeForOp(op: OutboxOpRecord): { + baseFee: number + isAggressive: boolean +} { + const snapshot = getFeeSnapshot() + const level = snapshot.congestionLevel if ( - (result as any).status !== 'success' && - (result as any).status !== undefined + op.priority === 'CRITICAL' && + (CONGESTION_ORDER[level] ?? 0) >= CONGESTION_ORDER['elevated'] ) { - // waitForConfirmation returns {hash, status:'success'|...} - if ((result as any).status === 'failed') - throw new Error('Sponsored transaction failed on-chain') + return { baseFee: snapshot.aggressiveBaseFee, isAggressive: true } } - return result + return { baseFee: snapshot.recommendedBaseFee, isAggressive: false } } /** - * Perform the actual on-chain submission for a claimed op. `feeMultiplier` - * implements the dispatcher's fee-bump-on-congestion retry strategy (docs/OUTBOX.md). + * Side effects when an op reaches permanent FAILED: log it, raise a critical + * alert, and notify the user via an event. The alert and event are + * best-effort — their own failures are swallowed/logged so they can never + * mask the original failure or crash the dispatcher. */ -export async function executeOutboxPayload( - payload: OutboxPayload, - feeMultiplier: number = 1, - onSubmitted?: (txHash: string) => Promise -): Promise { - switch (payload.method) { - case 'deposit': - return depositForUser( - payload.userId, - payload.userAddress, - payload.amount, - payload.assetSymbol, - feeMultiplier, - onSubmitted - ) - case 'withdraw': - return withdrawForUser( - payload.userId, - payload.userAddress, - payload.amount, - payload.assetSymbol, - feeMultiplier, - onSubmitted - ) - case 'rebalance': - return submitRebalance( - payload.toProtocol, - payload.expectedApyBasisPoints, - feeMultiplier, - onSubmitted - ) - case 'referral_reward': - return payReferralReward( - payload.recipientAddress, - payload.amount, - payload.assetSymbol, - feeMultiplier, - onSubmitted - ) - case 'sponsor_create_account': { - const sponsor = getSponsorKeypair(payload.sponsorAccount) - const tx = buildSponsoredCreateAccount({ - newAccountId: payload.newAccountId, - sponsorKeypair: sponsor, - startingBalance: '0', - }) - const result = await submitSponsoredTransaction(tx, sponsor) - // Record reserve ledger on success (best-effort, backfillable via reconciliation) - await db.reserveSponsorship - .create({ - data: { - sponsoredId: payload.sponsoredId, - sponsorAccount: payload.sponsorAccount, - entryType: 'ACCOUNT', - ledgerKey: payload.ledgerKey, - xlmReserved: payload.xlmReserved, - status: 'ACTIVE', +async function onTerminalFailure( + op: OutboxOpRecord, + errorMessage: string +): Promise { + logger.error('[Outbox] Op permanently FAILED', { + opId: op.id, + kind: op.kind, + userId: op.userId, + attempts: op.attempts, + error: errorMessage, + }) + + await alertingService + .emit({ + title: `Outbox op permanently failed: ${op.kind}`, + description: `Outbox op ${op.id} (${op.kind}, priority ${op.priority}) for user ${op.userId} failed after ${op.attempts} attempts: ${errorMessage}`, + severity: 'critical', + component: 'outbox', + metadata: { opId: op.id, kind: op.kind, userId: op.userId }, + }) + .catch((err: unknown) => + logger.error('[Outbox] Failed to emit terminal-failure alert', { err }) + ) + + await publishUserEvent( + op.userId, + EVENT_TYPE_TOPIC['outbox.op_failed'], + 'outbox.op_failed', + { + opId: op.id, + kind: op.kind, + userId: op.userId, + attempts: op.attempts, + error: errorMessage, + } + ).catch(() => {}) +} + +/** + * Submit an already-claimed (status=SUBMITTED in the DB) op. Resolves with + * the on-chain result whether the network accepted or rejected it — a + * resolved `status: 'failed'` (a vault-contract precondition failure, not a + * transient error) is recorded as terminally FAILED but NOT retried and NOT + * thrown, matching the single-attempt behavior the deposit/withdraw routes + * had before #325. Only an actual exception (network/simulation error) goes + * through the retry/backoff pipeline and is re-thrown, so a synchronous + * caller's existing error handling is unaffected by the outbox underneath it. + */ +async function submitClaimedOp(op: OutboxOpRecord): Promise { + // --- Fee calculation ----------------------------------------------------- + // Base fee from congestion/priority, scaled by the per-attempt bump + // multiplier, then clamped to the absolute ceiling (maxAbsFee). + const { baseFee, isAggressive } = getBaseFeeForOp(op) + if (isAggressive) recordOutboxAggressiveFeeUsed() + + const attemptMultiplier = computeFeeMultiplier(op.attempts) + const rawFee = baseFee * attemptMultiplier + const effectiveFee = Math.min(rawFee, config.outbox.maxAbsFee) + const hitCap = rawFee > config.outbox.maxAbsFee + if (hitCap) { + recordOutboxMaxFeeHit(op.priority) + // A CRITICAL op being capped means it may be underpriced for the current + // network, so page someone. The second argument is a dedupe key so the + // alert isn't re-sent on every attempt. + if (op.priority === 'CRITICAL') { + await alertingService + .emit( + { + title: 'Outbox fee cap hit on CRITICAL op', + description: `CRITICAL op ${op.id} hit OUTBOX_MAX_ABS_FEE ${config.outbox.maxAbsFee} stroops (raw ${rawFee}). Submitting at cap.`, + severity: 'critical', + component: 'outbox', + metadata: { opId: op.id, rawFee, cap: config.outbox.maxAbsFee }, }, - }) + 'outbox:max-fee-hit' + ) .catch(() => {}) - return result } - case 'sponsor_trustline': { - const sponsor = getSponsorKeypair(payload.sponsorAccount) - const asset = new Asset(payload.assetCode, payload.assetIssuer) - const tx = buildSponsoredTrustline({ - accountId: payload.accountId, - asset, - sponsorKeypair: sponsor, + } + // Contract expects multiplier relative to BASE_FEE (100 stroops) + const feeMultiplier = Math.max(1, Math.round(effectiveFee / 100)) + const lock = signerLock() + + try { + // Run the submission under the per-signer lock so concurrent ops from + // the same signer account don't collide. The callback persists the + // txHash as soon as it exists, so a crash mid-confirmation can still be + // reconciled later. + // (signerPublicKey is always set here: claimOp stored it at claim time.) + const result = await lock.withLock(op.signerPublicKey!, () => + executeOutboxPayload(op.payload, feeMultiplier, (txHash) => + recordSubmittedTxHash(op.id, txHash) + ) + ) + + if (result.status === 'pending') { + // Submitted but not yet confirmed. Leave the op SUBMITTED and keep the + // hash; the stuck-op reconciler or the events oracle will resolve it. + await recordSubmittedTxHash(op.id, result.hash) + await mirrorPendingTransaction(op.payload, result.hash) + logger.warn('[Outbox] Confirmation pending; preserving submitted hash', { + opId: op.id, + txHash: result.hash, }) - const result = await submitSponsoredTransaction(tx, sponsor) - await db.reserveSponsorship - .create({ - data: { - sponsoredId: payload.sponsoredId, - sponsorAccount: payload.sponsorAccount, - entryType: 'TRUSTLINE', - ledgerKey: payload.ledgerKey, - xlmReserved: payload.xlmReserved, - status: 'ACTIVE', - }, - }) - .catch(() => {}) - return result + } else if (!result.status || result.status === 'success') { + // Success path (a missing status is treated as success). + await markConfirmed(op.id, result.hash) + await mirrorLinkedTransaction(op.payload, { + txHash: result.hash, + status: 'CONFIRMED', + }) + recordOutboxOp(op.kind, op.priority, 'confirmed') + // End-to-end latency from enqueue to confirmation, in seconds. + recordOutboxLatency(op.kind, (Date.now() - op.createdAt.getTime()) / 1000) + logger.info('[Outbox] Op confirmed', { + opId: op.id, + kind: op.kind, + txHash: result.hash, + attempts: op.attempts, + }) + } else { + // The network resolved but rejected the op (status === 'failed'). + // Terminal and not retried; returned (not thrown) to the caller. + await markFailedTerminal( + op.id, + result.hash, + 'On-chain submission returned status=failed' + ) + await mirrorLinkedTransaction(op.payload, { + txHash: result.hash, + status: 'FAILED', + }) + recordOutboxOp(op.kind, op.priority, 'failed') + logger.warn('[Outbox] Op rejected on-chain (terminal, not retried)', { + opId: op.id, + kind: op.kind, + txHash: result.hash, + }) + await onTerminalFailure(op, 'On-chain submission returned status=failed') + } + + return result + } catch (err) { + const message = err instanceof Error ? err.message : String(err) + // A confirmation timeout is not a failure: the tx was submitted and may + // still land. Keep the hash, leave the op SUBMITTED, and report pending + // rather than retrying (which could double-submit). + if (err instanceof TransactionConfirmationTimeoutError) { + await recordSubmittedTxHash(op.id, err.txHash) + await mirrorPendingTransaction(op.payload, err.txHash) + logger.warn( + '[Outbox] Confirmation timed out; preserving submitted hash', + { + opId: op.id, + txHash: err.txHash, + } + ) + return { hash: err.txHash, status: 'pending' } } - case 'revoke_sponsorship': { - const sponsor = getSponsorKeypair(payload.sponsorAccount) - const tx = buildRevokeSponsorship({ - sponsorKeypair: sponsor, - accountId: payload.sponsoredId, - ledgerKey: payload.ledgerKey, + + // Any other exception: either schedule a retry with backoff or, if + // attempts are exhausted, fail permanently. + const { terminal } = await markFailedOrRetry(op, message) + recordOutboxOp(op.kind, op.priority, terminal ? 'failed' : 'retry') + + if (terminal) { + // No txHash is available on this path, hence only the status mirrors. + await mirrorLinkedTransaction(op.payload, { status: 'FAILED' }) + await onTerminalFailure(op, message) + } else { + logger.warn('[Outbox] Op submit failed — retriable', { + opId: op.id, + kind: op.kind, + attempts: op.attempts, + error: message, }) - const result = await submitSponsoredTransaction(tx, sponsor) - await db.reserveSponsorship - .updateMany({ - where: { ledgerKey: payload.ledgerKey, status: 'ACTIVE' }, - data: { status: 'RECLAIMED', revokedAt: new Date() }, - }) - .catch(() => {}) - return result } - case 'treasury_sweep': - return submitTreasurySweepPayment(payload) - default: - throw new Error(`Unsupported outbox method: ${(payload as any).method}`) + + // Re-throw so synchronous callers see the same error they would have + // seen before the outbox existed. + throw err } } /** - * Moves funds from one treasury tier's account to another via a plain - * Stellar payment. Signing here uses the fromTier account's own keypair - * resolved through the sponsor-key env convention (STELLAR_SPONSOR_KEYS) — - * the same lookup-by-public-key pattern getSponsorKeypair already uses. - * Full multisig-threshold co-signing for treasury accounts is handled - * upstream by src/stellar/multisig.ts / src/jobs/treasurySweep.ts's approval - * gate before an op ever reaches the outbox; this executor's job is only to - * submit an already-authorized sweep. + * Claim-and-submit one specific PENDING op inline. Used by callers that want + * the on-chain result synchronously (deposit/withdraw, referral payout). + * + * Throws if the op cannot be claimed (already claimed by a concurrent + * dispatcher pass, the user is halted, or submission fails) — callers treat + * that exactly as they treat a direct executeWriteContractCall failure today, + * except the durable OutboxOp record now survives the failure for the + * background sweep or an admin force-retry. */ -async function submitTreasurySweepPayment(payload: { - fromTier: string - toTier: string - asset: string - amount: number - sweepId: string -}): Promise { - const [fromAccount, toAccount] = await Promise.all([ - db.treasuryAccount.findFirst({ - where: { tier: payload.fromTier as any, isActive: true }, - }), - db.treasuryAccount.findFirst({ - where: { tier: payload.toTier as any, isActive: true }, - }), - ]) - if (!fromAccount || !toAccount) { +export async function dispatchOne(opId: string): Promise { + const op = await getOp(opId) + if (!op) throw new Error(`Outbox op ${opId} not found`) + + // Idempotent short-circuit: if it already confirmed (e.g. via the sweep), + // return the existing result instead of erroring. + if (op.status === 'CONFIRMED' && op.txHash) { + return { hash: op.txHash, status: 'success' } + } + if (op.status !== 'PENDING') { + throw new Error(`Outbox op ${opId} is not PENDING (status=${op.status})`) + } + + // Compliance/freeze guard: never dispatch for a halted user. + if (await isUserHalted(op.userId)) { throw new Error( - `Missing active treasury account for tier ${!fromAccount ? payload.fromTier : payload.toTier}` + `Outbox op ${opId} not dispatched: user ${op.userId} is frozen` ) } - const signer = getSponsorKeypair(fromAccount.publicKey) - const source = await getAccount(fromAccount.publicKey) - const asset = - payload.asset === 'XLM' ? Asset.native() : new Asset(payload.asset, '') + // LOW deferral during high/severe congestion (bounded) + try { + const snap = getFeeSnapshot() + if (shouldDeferLowOp(op, snap.congestionLevel)) { + // Push nextAttemptAt out so the sweep also leaves it alone until then, + // then throw so the inline caller knows it wasn't dispatched. + const deferUntil = new Date(Date.now() + config.outbox.lowDeferMs) + await db.outboxOp.update({ + where: { id: op.id }, + data: { nextAttemptAt: deferUntil }, + }) + recordOutboxLowDeferred() + throw new Error( + `LOW op ${opId} deferred due to high congestion (${snap.congestionLevel})` + ) + } + } catch (err) { + // If the error is our deferral throw, rethrow; otherwise log and proceed + // (a broken fee oracle must not block dispatching — fail open here). + if ( + err instanceof Error && + err.message.includes('deferred due to high congestion') + ) + throw err + logger.warn('[Outbox] Defer check failed, proceeding to dispatch', { + opId: op.id, + error: err instanceof Error ? err.message : String(err), + }) + } - const tx = new TransactionBuilder(source, { - fee: BASE_FEE, - networkPassphrase: getNetworkPassphrase(), + // Decide which signer account will submit, then atomically claim the op. + const signerPublicKey = await resolveSignerPublicKey(op.payload, op.userId) + const claimed = await claimOp(opId, signerPublicKey) + if (!claimed) { + // Lost the claim race. Re-read: if the winner already confirmed it, + // return that result; otherwise surface the conflict as an error. + const latest = await getOp(opId) + if (latest?.status === 'CONFIRMED' && latest.txHash) { + return { hash: latest.txHash, status: 'success' } + } + throw new Error( + `Outbox op ${opId} could not be claimed (status=${latest?.status ?? 'unknown'})` + ) + } + + return submitClaimedOp(claimed) +} + +/** + * Non-blocking fire-and-forget dispatch: enqueue is already durable, so the + * caller (the agent loop, for a LOW-priority rebalance) does not need to + * await the on-chain round trip at all — the background sweep will pick the + * op up on its own cadence, respecting priority ordering, if this opportunistic + * attempt doesn't win the claim race first. + */ +export function dispatchInBackground(opId: string): void { + // Errors are only logged: the op stays durable, so the sweep can retry. + dispatchOne(opId).catch((err) => { + logger.warn('[Outbox] Background dispatch attempt failed (retriable)', { + opId, + error: err instanceof Error ? err.message : String(err), + }) }) - .addOperation( - Operation.payment({ - destination: toAccount.publicKey, - asset, - amount: payload.amount.toFixed(7), +} + +/** Move ops SUBMITTED-but-unconfirmed past the timeout back into the retry pipeline. */ +async function reconcileStuckSubmitted(): Promise { + const stuck = await findStuckSubmittedOps(config.outbox.submittedTimeoutMs) + updateOutboxStuckSubmitted(stuck.length) // gauge for monitoring/alerting + + for (const op of stuck) { + // Step 1: if we have a txHash, ask the chain what actually happened + // before deciding to resubmit (avoids double-submitting a tx that landed). + if (op.txHash) { + let chainStatus: 'success' | 'failed' | 'not_found' + try { + chainStatus = await getTransactionStatus(op.txHash) + } catch (err) { + // RPC unavailable: leave the op alone and try again next sweep. + logger.warn('[Outbox] Could not reconcile submitted transaction hash', { + opId: op.id, + txHash: op.txHash, + error: err instanceof Error ? err.message : String(err), + }) + continue + } + + // It landed after all: close the op out as confirmed. + if (chainStatus === 'success') { + await markConfirmed(op.id, op.txHash) + await mirrorLinkedTransaction(op.payload, { + txHash: op.txHash, + status: 'CONFIRMED', + }) + recordOutboxOp(op.kind, op.priority, 'confirmed') + continue + } + + // It landed but failed on-chain: terminal, no retry. + if (chainStatus === 'failed') { + await markFailedTerminal( + op.id, + op.txHash, + 'Stellar RPC confirmed the submitted transaction failed' + ) + await mirrorLinkedTransaction(op.payload, { + txHash: op.txHash, + status: 'FAILED', + }) + recordOutboxOp(op.kind, op.priority, 'failed') + await onTerminalFailure(op, 'Submitted Stellar transaction failed') + continue + } + // 'not_found' falls through to the escalation logic below. + } + + // Step 2: out of fee-bump attempts. Passing attempts=maxAttempts here + // forces markFailedOrRetry down its terminal branch, so this always + // ends in FAILED (the `terminal` flag is effectively always true). + if (op.attempts >= config.outbox.feeBumpMaxAttempts) { + const { terminal } = await markFailedOrRetry( + { id: op.id, attempts: config.outbox.maxAttempts }, + `Unconfirmed after ${config.outbox.feeBumpMaxAttempts} fee-bump attempts (submittedAt=${op.submittedAt?.toISOString()})` + ) + recordOutboxOp(op.kind, op.priority, terminal ? 'failed' : 'retry') + if (terminal) await onTerminalFailure(op, 'Fee-bump cap exceeded') + continue + } + + // Step 3: still within the fee-bump budget — requeue as PENDING. The + // next claim bumps `attempts`, which raises the fee on resubmission. + logger.warn( + '[Outbox] Op unconfirmed past timeout — escalating to fee-bump retry', + { + opId: op.id, + kind: op.kind, + attempts: op.attempts, + submittedAt: op.submittedAt, + } + ) + recordOutboxFeeBump(op.kind) + await returnStuckOpToPending(op.id) + } +} + +/** + * One background sweep: reconcile stuck SUBMITTED ops, then claim and submit + * PENDING backlog in priority order (see src/outbox/stateMachine.ts — + * CRITICAL withdrawals never wait behind a NORMAL/LOW wave). + */ +export async function runDispatchSweep(): Promise { + // First clean up anything stuck, so requeued ops can be picked up below. + await reconcileStuckSubmitted() + + // Refresh the queue-depth gauges. + const stats = await getQueueStats() + updateOutboxQueueDepth(stats) + + // Fetch up to batchSize eligible ops, then order by priority. + const claimable = sortForDispatch( + await findClaimableOps(config.outbox.batchSize) + ) + + for (const op of claimable) { + // Skip (but leave PENDING) ops belonging to frozen users. + if (await isUserHalted(op.userId)) { + logger.info('[Outbox] Skipping op for frozen user', { + opId: op.id, + userId: op.userId, + }) + continue + } + + // LOW deferral during high/severe congestion (bounded) + // Same rule as dispatchOne, but here we `continue` instead of throwing. + try { + const snap = getFeeSnapshot() + if (shouldDeferLowOp(op, snap.congestionLevel)) { + const deferUntil = new Date(Date.now() + config.outbox.lowDeferMs) + await db.outboxOp.update({ + where: { id: op.id }, + data: { nextAttemptAt: deferUntil }, + }) + recordOutboxLowDeferred() + logger.info('[Outbox] LOW op deferred due to high congestion', { + opId: op.id, + congestion: snap.congestionLevel, + deferUntil: deferUntil.toISOString(), + }) + continue + } + } catch (err) { + // Fail open: a fee-oracle problem shouldn't stall the whole sweep. + logger.warn('[Outbox] Defer check failed, proceeding to dispatch', { + opId: op.id, + error: err instanceof Error ? err.message : String(err), + }) + } + + // Work out which signer would submit; skip this op if that fails so one + // bad op can't abort the rest of the batch. + let signerPublicKey: string + try { + signerPublicKey = await resolveSignerPublicKey(op.payload, op.userId) + } catch (err) { + logger.error('[Outbox] Could not resolve signer for op', { + opId: op.id, + error: err instanceof Error ? err.message : String(err), + }) + continue + } + + // Respect global and per-signer in-flight limits before claiming. + if (!signerLock().hasCapacity(signerPublicKey)) { + continue // over the concurrency cap — left PENDING for the next sweep + } + + const claimed = await claimOp(op.id, signerPublicKey) + if (!claimed) continue // lost the claim race to another dispatch path + + // Bounded by the concurrency caps checked above — not literally awaited + // per-op so one slow submission cannot stall the whole priority-ordered + // batch behind it. + submitClaimedOp(claimed).catch(() => { + // Already logged/alerted inside submitClaimedOp. + }) + } +} + +// Handle to the running interval, kept at module scope (e.g. for shutdown). +let dispatcherHandle: NodeJS.Timeout | null = null + +/** + * Start the background dispatcher: run one sweep immediately, then repeat + * every dispatchIntervalMs. Sweep errors are logged and never propagate, so + * a single failed sweep can't kill the timer. Returns the interval handle so + * callers can clear it on shutdown. + * + * Note: sweeps are not guarded against overlap — if one runs longer than the + * interval, the next starts anyway. This is safe because claimOp is atomic. + */ +export function scheduleOutboxDispatcher(): NodeJS.Timeout { + runDispatchSweep().catch((err) => + logger.error('[Outbox] Initial dispatch sweep failed', { + error: err instanceof Error ? err.message : String(err), + }) + ) + + dispatcherHandle = setInterval(() => { + runDispatchSweep().catch((err) => + logger.error('[Outbox] Dispatch sweep failed', { + error: err instanceof Error ? err.message : String(err), }) ) - .setTimeout(30) - .build() + }, config.outbox.dispatchIntervalMs) - tx.sign(signer) - const hash = await submitTransaction(tx) - const result = await waitForConfirmation(hash) - return result + logger.info('[Outbox] Dispatcher scheduled', { + intervalMs: config.outbox.dispatchIntervalMs, + }) + return dispatcherHandle } diff --git a/src/outbox/service.ts b/src/outbox/service.ts index 895f94a0..dbfca53e 100644 --- a/src/outbox/service.ts +++ b/src/outbox/service.ts @@ -4,6 +4,18 @@ * * Every write here goes through the pure rules in src/outbox/stateMachine.ts * first — this module is deliberately thin glue between that and Prisma. + * + * Lifecycle overview (see stateMachine.ts for the authoritative rules): + * + * PENDING ──claimOp──▶ SUBMITTED ──markConfirmed──────────▶ CONFIRMED + * ▲ │ + * │ ├─markFailedOrRetry (attempts left)─▶ PENDING (with backoff) + * │ ├─markFailedOrRetry (exhausted)─────▶ FAILED + * │ ├─markFailedTerminal (network rejected)▶ FAILED + * │ └─returnStuckOpToPending (timeout)──▶ PENDING + * │ + * ├─forceRetry (admin, from FAILED) + * └─cancelOp (admin) ─▶ CANCELLED */ import { Prisma, OutboxOpStatus as PrismaOutboxOpStatus } from '@prisma/client' @@ -21,8 +33,18 @@ import { PRIORITY_BY_KIND, } from './types' +// Either the global Prisma client or an interactive-transaction handle, so +// callers can run the same helper inside or outside a `db.$transaction`. type DbClient = typeof db | Prisma.TransactionClient +/** + * Map a raw Prisma OutboxOp row to the domain-level OutboxOpRecord. + * + * Prisma returns `kind`, `actor`, `priority`, `status` as its own generated + * enum types (or plain strings) and `payload` as generic JSON; this narrows + * them to the app's own union types. The casts are trusted because every + * write path in this file only stores valid values. + */ function toRecord(row: { id: string idempotencyKey: string @@ -48,6 +70,8 @@ function toRecord(row: { userId: row.userId, kind: row.kind as OutboxOpKind, actor: row.actor as OutboxOpActor, + // JsonValue -> OutboxPayload goes through `unknown` because TS can't + // verify the JSON shape; the payload is validated at enqueue time. payload: row.payload as unknown as OutboxPayload, priority: row.priority as OutboxPriority, status: row.status as OutboxOpStatus, @@ -85,6 +109,9 @@ export async function enqueueOutboxOp( priority?: OutboxPriority } ): Promise { + // Look up first so a repeat call returns the existing op untouched. + // (A concurrent insert with the same key would still be caught by the + // unique constraint on idempotencyKey at the DB level.) const existing = await tx.outboxOp.findUnique({ where: { idempotencyKey: params.idempotencyKey }, }) @@ -99,7 +126,9 @@ export async function enqueueOutboxOp( kind: params.kind, actor: params.actor, payload: params.payload as unknown as Prisma.InputJsonValue, + // Fall back to the kind's default priority unless the caller overrides. priority: params.priority ?? PRIORITY_BY_KIND[params.kind], + // New ops always start life as PENDING, awaiting the dispatcher. status: 'PENDING', }, }) @@ -117,26 +146,38 @@ export async function claimOp( opId: string, signerPublicKey: string ): Promise { + // Fail fast if the state machine ever stops allowing this transition. assertTransition('PENDING', 'SUBMITTED') + // The `status: 'PENDING'` predicate is what makes this a compare-and-set: + // the row only updates if nobody else has claimed it in the meantime. const result = await db.outboxOp.updateMany({ where: { id: opId, status: 'PENDING' as PrismaOutboxOpStatus }, data: { status: 'SUBMITTED' as PrismaOutboxOpStatus, + // Each claim counts as an attempt; this also drives retry exhaustion + // and the fee-bump multiplier on resubmission. attempts: { increment: 1 }, submittedAt: new Date(), + // Record which signer account this attempt is using. signerPublicKey, }, }) + // Lost the race (or the op is no longer PENDING). if (result.count === 0) { return null } + // updateMany doesn't return rows, so re-read the freshly claimed op. const row = await db.outboxOp.findUnique({ where: { id: opId } }) return row ? toRecord(row) : null } +/** + * The network confirmed the op: SUBMITTED -> CONFIRMED. Stores the final + * txHash and timestamp and clears any error left over from earlier retries. + */ export async function markConfirmed( opId: string, txHash: string @@ -153,6 +194,12 @@ export async function markConfirmed( }) } +/** + * Persist the txHash as soon as it is known (before confirmation), so the + * confirmation fallback in reconcileOutboxOpByTxHash can find the op if the + * dispatcher crashes mid-flight. Conditional on SUBMITTED so it can't + * overwrite an op that has already moved on. + */ export async function recordSubmittedTxHash( opId: string, txHash: string @@ -173,20 +220,25 @@ export async function markFailedOrRetry( op: Pick, errorMessage: string ): Promise<{ terminal: boolean }> { + // `attempts` was already incremented by claimOp, so it equals the number + // of tries made so far, including the one that just failed. const terminal = op.attempts >= config.outbox.maxAttempts if (terminal) { + // Out of retries: give up permanently. assertTransition('SUBMITTED', 'FAILED') await db.outboxOp.update({ where: { id: op.id }, data: { status: 'FAILED' as PrismaOutboxOpStatus, + // Truncate so a huge error string can't bloat the row. error: errorMessage.slice(0, 2000), }, }) return { terminal: true } } + // Retries remain: put the op back in the queue, delayed by the backoff. assertTransition('SUBMITTED', 'PENDING') const backoffMs = computeBackoffMs( op.attempts, @@ -198,6 +250,7 @@ export async function markFailedOrRetry( data: { status: 'PENDING' as PrismaOutboxOpStatus, error: errorMessage.slice(0, 2000), + // findClaimableOps skips the op until this time has passed. nextAttemptAt: new Date(Date.now() + backoffMs), }, }) @@ -223,7 +276,7 @@ export async function markFailedTerminal( where: { id: opId }, data: { status: 'FAILED' as PrismaOutboxOpStatus, - txHash, + txHash, // kept for audit even though the op failed error: errorMessage.slice(0, 2000), }, }) @@ -233,6 +286,8 @@ export async function markFailedTerminal( export async function forceRetry(opId: string): Promise { const op = await db.outboxOp.findUnique({ where: { id: opId } }) if (!op) throw new Error(`Outbox op ${opId} not found`) + // Let the state machine decide whether this op's current status may go + // back to PENDING (e.g. reject CONFIRMED or CANCELLED ops). assertTransition(op.status as OutboxOpStatus, 'PENDING') const updated = await db.outboxOp.update({ @@ -240,6 +295,7 @@ export async function forceRetry(opId: string): Promise { data: { status: 'PENDING' as PrismaOutboxOpStatus, error: null, + // null = immediately eligible, no backoff delay. nextAttemptAt: null, }, }) @@ -249,11 +305,14 @@ export async function forceRetry(opId: string): Promise { /** Cancel an unsent op. Only PENDING ops can be cancelled — once SUBMITTED, it's on-chain. */ export async function cancelOp(opId: string): Promise { + // Conditional update again: cancelling only succeeds if the op is still + // PENDING, so it can't race with a dispatcher that just claimed it. const result = await db.outboxOp.updateMany({ where: { id: opId, status: 'PENDING' as PrismaOutboxOpStatus }, data: { status: 'CANCELLED' as PrismaOutboxOpStatus }, }) if (result.count === 0) { + // Nothing updated: work out why so the error message is useful. const existing = await db.outboxOp.findUnique({ where: { id: opId } }) if (!existing) throw new Error(`Outbox op ${opId} not found`) throw new Error( @@ -262,14 +321,16 @@ export async function cancelOp(opId: string): Promise { } const row = await db.outboxOp.findUnique({ where: { id: opId } }) logger.info('[Outbox] Admin cancel', { opId }) - return toRecord(row!) + return toRecord(row!) // non-null: the update above just matched this row } +/** Fetch a single op by id, or null if it doesn't exist. */ export async function getOp(opId: string): Promise { const row = await db.outboxOp.findUnique({ where: { id: opId } }) return row ? toRecord(row) : null } +/** Look up the op that produced a given on-chain transaction hash (txHash is unique). */ export async function findOpByTxHash( txHash: string ): Promise { @@ -277,6 +338,7 @@ export async function findOpByTxHash( return row ? toRecord(row) : null } +/** Optional filters and pagination for listOps. All fields are AND-ed together. */ export interface ListOpsFilter { status?: OutboxOpStatus kind?: OutboxOpKind @@ -286,18 +348,25 @@ export interface ListOpsFilter { offset?: number } +/** + * Paginated op listing for the admin view, newest first. Returns the page + * of ops plus the total count matching the filter (ignoring pagination). + */ export async function listOps( filter: ListOpsFilter = {} ): Promise<{ ops: OutboxOpRecord[]; total: number }> { + // Prisma ignores `undefined` fields, so unset filters simply don't apply. const where: Prisma.OutboxOpWhereInput = { status: filter.status as PrismaOutboxOpStatus | undefined, kind: filter.kind, priority: filter.priority, userId: filter.userId, } + // Clamp pagination: default page size 50, between 1 and 500; offset >= 0. const limit = Math.min(Math.max(filter.limit ?? 50, 1), 500) const offset = Math.max(filter.offset ?? 0, 0) + // Run the page query and the count in parallel. const [rows, total] = await Promise.all([ db.outboxOp.findMany({ where, @@ -315,6 +384,7 @@ export async function listOps( export async function getQueueStats(): Promise< Array<{ status: OutboxOpStatus; priority: OutboxPriority; count: number }> > { + // One GROUP BY query instead of fetching rows and counting in JS. const grouped = await db.outboxOp.groupBy({ by: ['status', 'priority'], _count: { _all: true }, @@ -331,6 +401,9 @@ export async function getQueueStats(): Promise< * window. Priority ordering is applied in JS via * src/outbox/stateMachine.ts#sortForDispatch rather than a DB ORDER BY on a * non-numeric enum, so the ordering rule stays in one pure, tested place. + * + * Note: `take: limit` is applied by the DB with no ORDER BY, so priority + * sorting only reorders the batch returned here, not the whole backlog. */ export async function findClaimableOps( limit: number @@ -339,6 +412,7 @@ export async function findClaimableOps( const rows = await db.outboxOp.findMany({ where: { status: 'PENDING' as PrismaOutboxOpStatus, + // Eligible if there's no backoff set, or the backoff has elapsed. OR: [{ nextAttemptAt: null }, { nextAttemptAt: { lte: now } }], }, take: limit, @@ -354,11 +428,15 @@ export async function findClaimableOps( */ export async function returnStuckOpToPending(opId: string): Promise { assertTransition('SUBMITTED', 'PENDING') + // Conditional on SUBMITTED so we don't clobber an op that confirmed while + // the sweep was deciding to requeue it. await db.outboxOp.updateMany({ where: { id: opId, status: 'SUBMITTED' as PrismaOutboxOpStatus }, data: { status: 'PENDING' as PrismaOutboxOpStatus, + // Clear the old hash: the resubmission will produce a new one. txHash: null, + // "Now" (rather than null) makes it claimable on the very next sweep. nextAttemptAt: new Date(), }, }) @@ -372,6 +450,7 @@ export async function findStuckSubmittedOps( const rows = await db.outboxOp.findMany({ where: { status: 'SUBMITTED' as PrismaOutboxOpStatus, + // Submitted at or before the cutoff = waiting longer than the timeout. submittedAt: { lte: cutoff }, }, }) @@ -393,7 +472,9 @@ export async function mirrorLinkedTransaction( payload: OutboxPayload, outcome: { txHash?: string; status: 'CONFIRMED' | 'FAILED' } ): Promise { + // Some payload kinds have no linked Transaction row; nothing to mirror. if (!('transactionId' in payload)) return + // updateMany (not update) so a missing row is a silent no-op, not a throw. await db.transaction.updateMany({ where: { id: payload.transactionId }, data: { @@ -404,6 +485,11 @@ export async function mirrorLinkedTransaction( }) } +/** + * Mirror the "submitted, awaiting confirmation" state onto the linked + * Transaction row: store the txHash and mark it PENDING. Like + * mirrorLinkedTransaction, a no-op for payloads without a `transactionId`. + */ export async function mirrorPendingTransaction( payload: OutboxPayload, txHash: string @@ -429,6 +515,7 @@ export async function reconcileOutboxOpByTxHash( txHash: string ): Promise { await tx.outboxOp.updateMany({ + // Only SUBMITTED ops are touched, so already-terminal ops stay as-is. where: { txHash, status: 'SUBMITTED' as PrismaOutboxOpStatus }, data: { status: 'CONFIRMED' as PrismaOutboxOpStatus, @@ -443,5 +530,6 @@ export async function isUserHalted(userId: string): Promise { where: { id: userId }, select: { isActive: true }, }) + // Fail closed: a missing user or a deactivated account both count as halted. return !user || user.isActive === false } diff --git a/src/outbox/stateMachine.ts b/src/outbox/stateMachine.ts index 1a1248f3..62b84112 100644 --- a/src/outbox/stateMachine.ts +++ b/src/outbox/stateMachine.ts @@ -14,15 +14,30 @@ import { OutboxOpStatus, OutboxOpRecord, OutboxPriority } from './types' * submit failure returns an op to PENDING with a backoff nextAttemptAt, so it * is re-claimed on a later dispatcher tick rather than living in a separate * "RETRYING" status). + * + * Reading the table: each key is the CURRENT status, and its array lists the + * statuses the op is allowed to move to next. An empty array means the status + * is terminal. */ const VALID_TRANSITIONS: Record = { + // Waiting to be dispatched. Can be claimed (SUBMITTED), cancelled by an + // admin, or failed directly. PENDING: ['SUBMITTED', 'CANCELLED', 'FAILED'], + // Claimed and (possibly) on-chain. Can confirm, be requeued for retry or + // after a stuck-op timeout (PENDING), or fail permanently. SUBMITTED: ['CONFIRMED', 'PENDING', 'FAILED'], + // Terminal: the op succeeded; nothing may follow. CONFIRMED: [], FAILED: ['PENDING'], // admin force-retry only (src/outbox/service.ts) + // Terminal: cancelled before submission; nothing may follow. CANCELLED: [], } +/** + * Non-throwing check: is `from` -> `to` an allowed transition? Returns false + * for any pair not listed in VALID_TRANSITIONS (including unknown statuses, + * via the `?.` / `?? false` guards). + */ export function canTransition( from: OutboxOpStatus, to: OutboxOpStatus @@ -30,6 +45,10 @@ export function canTransition( return VALID_TRANSITIONS[from]?.includes(to) ?? false } +/** + * Throwing variant of canTransition. DB-facing code calls this before every + * status write so an illegal move fails loudly instead of corrupting state. + */ export function assertTransition( from: OutboxOpStatus, to: OutboxOpStatus @@ -61,11 +80,18 @@ export function compareForDispatch( a: Pick, b: Pick ): number { + // Primary key: priority weight (negative => `a` sorts first). const weightDiff = PRIORITY_WEIGHT[a.priority] - PRIORITY_WEIGHT[b.priority] if (weightDiff !== 0) return weightDiff + // Tie-breaker: older op first (smaller timestamp => negative => `a` first). return a.createdAt.getTime() - b.createdAt.getTime() } +/** + * Return a new array sorted into dispatch order. Copies first (`[...ops]`) + * so the caller's array is never mutated. The generic `T` lets callers pass + * full OutboxOpRecords and get the same type back. + */ export function sortForDispatch< T extends Pick, >(ops: T[]): T[] { @@ -76,6 +102,13 @@ export function sortForDispatch< * Full jitter exponential backoff (AWS-style): a random delay in * [0, base * 2^attempt], capped. `attempt` is the 1-indexed attempt number * that just failed. + * + * Note: the code uses `2 ** (attempt - 1)`, so the ceiling is actually + * base * 2^(attempt - 1): the first failure (attempt 1) draws from + * [0, base], the second from [0, 2 * base], and so on. + * + * `random` is injectable (defaults to Math.random) so tests can make the + * result deterministic. */ export function computeBackoffMs( attempt: number, @@ -83,6 +116,10 @@ export function computeBackoffMs( maxMs: number, random: () => number = Math.random ): number { + // Exponential ceiling for this attempt, clamped to maxMs. Math.max(0, ...) + // keeps a bad `attempt` (0 or negative) from producing a fractional power. const capped = Math.min(maxMs, baseMs * 2 ** Math.max(0, attempt - 1)) + // "Full jitter": pick uniformly in [0, capped) rather than waiting the + // whole ceiling, which spreads retries out and avoids thundering herds. return Math.floor(random() * capped) }