diff --git a/deno.json b/deno.json index eb9a530..34c52c2 100644 --- a/deno.json +++ b/deno.json @@ -1,6 +1,6 @@ { "name": "@moonlight-protocol/provider-platform", - "version": "0.9.7", + "version": "0.9.8", "license": "MIT", "exports": "./src/main.ts", "tasks": { @@ -27,7 +27,7 @@ "@colibri/core": "jsr:@colibri/core@^0.22.0", "@drizzle-team/brocli": "npm:@drizzle-team/brocli@^0.11.0", "@fifo/convee": "jsr:@fifo/convee@^0.10.0", - "@moonlight/moonlight-sdk": "jsr:@moonlight/moonlight-sdk@^0.11.0", + "@moonlight/moonlight-sdk": "jsr:@moonlight/moonlight-sdk@^0.11.1", "@std/assert": "jsr:@std/assert@^1.0.0", "@std/dotenv": "jsr:@std/dotenv@^0.225.6", "@zaubrik/djwt": "jsr:@zaubrik/djwt@^3.0.2", diff --git a/src/core/service/auth/challenge/verify/verify-challenge.ts b/src/core/service/auth/challenge/verify/verify-challenge.ts index 62b6758..4e25548 100644 --- a/src/core/service/auth/challenge/verify/verify-challenge.ts +++ b/src/core/service/auth/challenge/verify/verify-challenge.ts @@ -4,6 +4,7 @@ import { getProviderAccount } from "@/core/service/auth/service/service-account. import { NETWORK_CONFIG, SERVICE_DOMAIN } from "@/config/env.ts"; import type { PostChallengeInput } from "@/core/service/auth/challenge/types.ts"; import * as E from "@/core/service/auth/challenge/verify/error.ts"; +import { PlatformError } from "@/error/index.ts"; import { assertOrThrow } from "@/utils/error/assert-or-throw.ts"; import { isDefined } from "@/utils/type-guards/is-defined.ts"; import { StrKey } from "@colibri/core"; @@ -128,6 +129,10 @@ export const P_VerifyChallenge = (deps: { log: Logger }) => : String(error), }); log.error(error, "challenge verification failed"); + // Preserve a precise auth failure (e.g. MISSING_CLIENT_SIGNATURE + // 4xx) instead of masking every cause as a generic 500 (#4). Only + // genuinely unexpected errors become CHALLENGE_VERIFICATION_FAILED. + if (error instanceof PlatformError) throw error; throw new E.CHALLENGE_VERIFICATION_FAILED(error); } }); diff --git a/src/core/service/bundle/add-bundle.process.ts b/src/core/service/bundle/add-bundle.process.ts index f202f0e..f59ae74 100644 --- a/src/core/service/bundle/add-bundle.process.ts +++ b/src/core/service/bundle/add-bundle.process.ts @@ -8,6 +8,7 @@ import type { PostEndpointInput } from "@/http/pipelines/types.ts"; import type { OperationTypes } from "@moonlight/moonlight-sdk"; import { MoonlightOperation } from "@moonlight/moonlight-sdk"; import { resolveChannelContext } from "@/core/service/executor/channel-resolver.ts"; +import { TimeoutError, withTimeout } from "@/utils/async/with-timeout.ts"; import { calculateBundleTtl, calculateBundleWeight, @@ -54,6 +55,10 @@ const operationsBundleRepository = new OperationsBundleRepository( drizzleClient, ); +// Bounded so an unreachable Soroban RPC fast-fails admission (#2) instead of +// hanging the request for minutes. +const ADMISSION_RPC_TIMEOUT_MS = 10_000; + const MEMPOOL_WEIGHT_CONFIG: WeightConfig = { expensiveOpWeight: MEMPOOL_EXPENSIVE_OP_WEIGHT, cheapOpWeight: MEMPOOL_CHEAP_OP_WEIGHT, @@ -137,16 +142,13 @@ function parseOperations( "operations.count": operationsMLXDR.length, }); log.event("parsing MLXDR operations"); + // `operationsMLXDR` is a 1:1 source for `operations` and the request + // schema already enforces `.min(1)` (returning a structured 400 before + // this runs), so an empty-operations case is unreachable here (#13). const operations = await Promise.all( operationsMLXDR.map((xdr) => MoonlightOperation.fromMLXDR(xdr)), ); - if (operations.length === 0) { - span.addEvent("no_operations"); - log.event("no operations parsed"); - throw new E.NO_OPERATIONS_PROVIDED(); - } - span.addEvent("operations_parsed", { "operations.count": operations.length, }); @@ -210,6 +212,7 @@ function persistSpendOperations( bundleId: string, accountId: string, channelClient: import("@moonlight/moonlight-sdk").PrivacyChannel, + channelContractId: string, deps: { log: Logger }, ): Promise { if (operations.length === 0) { @@ -223,11 +226,16 @@ function persistSpendOperations( }); const utxoPublicKeys = operations.map((op) => op.getUtxo()); - const balances = await fetchUtxoBalances( - utxoPublicKeys, - channelClient, - deps, - ); + const balances = await withTimeout( + fetchUtxoBalances(utxoPublicKeys, channelClient, deps), + ADMISSION_RPC_TIMEOUT_MS, + "fetchUtxoBalances", + ).catch((error) => { + if (error instanceof TimeoutError) { + throw new E.CHANNEL_RPC_UNAVAILABLE(channelContractId); + } + throw error; + }); for (let i = 0; i < operations.length; i++) { const operation = operations[i]; @@ -350,11 +358,18 @@ export const P_AddOperationsBundle = (deps: { log: Logger }) => log.debug("ppPublicKey", ppPublicKey); log.event("resolving channel context for PP"); - const channelCtx = await resolveChannelContext( - channelContractId, - ppPublicKey, - deps, - ); + const channelCtx = await withTimeout( + resolveChannelContext(channelContractId, ppPublicKey, deps), + ADMISSION_RPC_TIMEOUT_MS, + "resolveChannelContext", + ).catch((error) => { + // Fast-fail an unreachable chain during admission instead of + // hanging the request indefinitely (#2). + if (error instanceof TimeoutError) { + throw new E.CHANNEL_RPC_UNAVAILABLE(channelContractId); + } + throw error; + }); const channelClient = channelCtx.channelClient; span.addEvent("validating_session"); @@ -432,11 +447,18 @@ export const P_AddOperationsBundle = (deps: { log: Logger }) => span.addEvent("calculating_fee"); log.event("calculating fee"); - const amounts = await calculateOperationAmounts( - classified, - channelClient, - deps, - ); + // First on-chain read of admission — fast-fail if the chain is + // unreachable rather than hanging the request (#2). + const amounts = await withTimeout( + calculateOperationAmounts(classified, channelClient, deps), + ADMISSION_RPC_TIMEOUT_MS, + "calculateOperationAmounts", + ).catch((error) => { + if (error instanceof TimeoutError) { + throw new E.CHANNEL_RPC_UNAVAILABLE(channelContractId); + } + throw error; + }); const feeCalculation = calculateFee(amounts); span.addEvent("fee_calculated", { @@ -494,6 +516,7 @@ export const P_AddOperationsBundle = (deps: { log: Logger }) => bundleEntity.id, userSession.accountId, channelClient, + channelContractId, deps, ); diff --git a/src/core/service/bundle/bundle.errors.ts b/src/core/service/bundle/bundle.errors.ts index fedfb96..6d7ffaa 100644 --- a/src/core/service/bundle/bundle.errors.ts +++ b/src/core/service/bundle/bundle.errors.ts @@ -7,7 +7,8 @@ export enum BUNDLE_ERROR_CODES { INSUFFICIENT_UTXOS = "BND_004", UTXO_NOT_FOUND = "BND_005", SPEND_OPERATION_NOT_SIGNED = "BND_006", - NO_OPERATIONS_PROVIDED = "BND_007", + // BND_007 (NO_OPERATIONS_PROVIDED) removed — unreachable behind the request + // schema's operationsMLXDR.min(1) validation (#13). BUNDLE_NOT_FOUND = "BND_008", BUNDLE_ACCESS_FORBIDDEN = "BND_009", TOO_MANY_OPERATIONS = "BND_010", @@ -16,6 +17,7 @@ export enum BUNDLE_ERROR_CODES { PP_NOT_FOUND = "BND_013", PP_NOT_MEMBER_OF_CHANNEL = "BND_014", CHANNEL_DISABLED = "BND_015", + CHANNEL_RPC_UNAVAILABLE = "BND_016", } const source = "@service/bundle"; @@ -116,6 +118,32 @@ export class CHANNEL_DISABLED extends PlatformError<{ } } +/** + * Error thrown when an on-chain read needed to admit a bundle (channel + * context resolution or UTXO balance lookup) does not respond within the + * admission timeout. Fast-fails the request instead of hanging (#2). + */ +export class CHANNEL_RPC_UNAVAILABLE extends PlatformError<{ + channelContractId: string; +}> { + constructor(channelContractId: string) { + super({ + source, + code: BUNDLE_ERROR_CODES.CHANNEL_RPC_UNAVAILABLE, + message: "Channel network is temporarily unavailable", + details: + `An on-chain read for channel '${channelContractId}' did not respond in time.`, + api: { + status: 503, + message: "Channel network is temporarily unavailable", + details: + "The network could not be reached to process your bundle right now. Please try again shortly.", + }, + meta: { channelContractId }, + }); + } +} + /** * Error thrown when the submitter has no APPROVED entity record. */ @@ -229,26 +257,6 @@ export class BUNDLE_ALREADY_EXISTS extends PlatformError<{ bundleId: string }> { } } -/** - * Error thrown when no operations are provided in the bundle - */ -export class NO_OPERATIONS_PROVIDED extends PlatformError { - constructor() { - super({ - source, - code: BUNDLE_ERROR_CODES.NO_OPERATIONS_PROVIDED, - message: "No operations provided", - details: "The operations bundle must contain at least one operation.", - api: { - status: 400, - message: "No operations provided", - details: - "The request must include at least one operation in the operations bundle.", - }, - }); - } -} - /** * Error thrown when a spend operation is not signed by the UTXO owner */ diff --git a/src/core/service/event-watcher/channel-convergence.ts b/src/core/service/event-watcher/channel-convergence.ts index 50905ef..3e6a193 100644 --- a/src/core/service/event-watcher/channel-convergence.ts +++ b/src/core/service/event-watcher/channel-convergence.ts @@ -1,4 +1,5 @@ import type { ChannelRegistry } from "./channel-registry.ts"; +import type { Logger } from "@/utils/logger/index.ts"; /** * Subset of a council's public state (`GET /api/v1/public/council`) the provider @@ -17,6 +18,7 @@ export interface CouncilConfigData { export async function fetchCouncilConfig( councilUrl: string, channelAuthId: string, + deps?: { log?: Logger }, ): Promise { try { const res = await fetch( @@ -24,10 +26,23 @@ export async function fetchCouncilConfig( encodeURIComponent(channelAuthId) }`, ); - if (!res.ok) return null; + if (!res.ok) { + // Best-effort still, but no longer silent (#10): a persistently failing + // council query degrades convergence and must be visible. + deps?.log?.error( + new Error(`council config query returned ${res.status}`), + "fetchCouncilConfig non-OK response", + { councilUrl, channelAuthId, status: res.status }, + ); + return null; + } const { data } = await res.json(); return data as CouncilConfigData; - } catch { + } catch (error) { + deps?.log?.error(error, "fetchCouncilConfig failed", { + councilUrl, + channelAuthId, + }); return null; } } diff --git a/src/core/service/event-watcher/event-watcher.process.ts b/src/core/service/event-watcher/event-watcher.process.ts index 38c758e..20fbd69 100644 --- a/src/core/service/event-watcher/event-watcher.process.ts +++ b/src/core/service/event-watcher/event-watcher.process.ts @@ -2,7 +2,7 @@ import type { Logger } from "@/utils/logger/index.ts"; import type { Server } from "stellar-sdk/rpc"; import { fetchChannelAuthEvents } from "./event-watcher.service.ts"; import type { ChannelAuthEvent } from "./event-watcher.types.ts"; -import { withSpan } from "@/core/tracing.ts"; +import { currentTraceId, SpanStatusCode, withSpan } from "@/core/tracing.ts"; import { recoverFromOutOfRetention } from "./retention.ts"; import { resolveBootStartLedger } from "./start-ledger.ts"; @@ -43,6 +43,11 @@ export class EventWatcher { private rpc: Server; private startLedgerBlock: number | null; private log: Logger; + // Health signal (#10): the watcher runs in the background with no request to + // surface failures on, so its last-success / last-error is tracked here and + // exposed via getHealth() for the /health endpoint. + private lastPollOkAt: number | null = null; + private lastPollError: { at: number; message: string } | null = null; constructor( config: { contractIds: string[]; intervalMs?: number }, @@ -55,6 +60,28 @@ export class EventWatcher { this.log = deps.log.scope("EventWatcher"); } + /** + * Background-poll health for the /health endpoint. `healthy` is false once a + * poll has errored more recently than the last success (or never succeeded + * after an error). + */ + getHealth(): { + running: boolean; + lastPollOkAt: number | null; + lastPollError: { at: number; message: string } | null; + healthy: boolean; + } { + const healthy = this.lastPollError === null || + (this.lastPollOkAt !== null && + this.lastPollOkAt >= this.lastPollError.at); + return { + running: this.isRunning, + lastPollOkAt: this.lastPollOkAt, + lastPollError: this.lastPollError, + healthy, + }; + } + /** * Add a channel-auth contract to the watched set (e.g. when a PP joins a new * council). Idempotent; the next poll picks it up — no new watcher is spun up. @@ -206,6 +233,7 @@ export class EventWatcher { // Advance cursor past the latest ledger we've seen (in memory only — // there is no durable cursor; converge-by-query is the recovery path). this.lastLedger = latestLedger + 1; + this.lastPollOkAt = Date.now(); } catch (error) { span.addEvent("poll_error", { "error.message": error instanceof Error @@ -233,7 +261,23 @@ export class EventWatcher { return; } - this.log.error(error, "EventWatcher poll error"); + // Surface the failure (#10): the swallowing catch previously left the + // span un-errored and no health signal. Mark it ERROR, record the + // exception, and update the health state so /health can report it. + this.lastPollError = { + at: Date.now(), + message: error instanceof Error ? error.message : String(error), + }; + span.setStatus({ + code: SpanStatusCode.ERROR, + message: this.lastPollError.message, + }); + span.recordException( + error instanceof Error ? error : new Error(String(error)), + ); + this.log.error(error, "EventWatcher poll error", { + traceId: currentTraceId(), + }); } }); } diff --git a/src/core/service/event-watcher/index.ts b/src/core/service/event-watcher/index.ts index ca1789b..bd952b8 100644 --- a/src/core/service/event-watcher/index.ts +++ b/src/core/service/event-watcher/index.ts @@ -179,6 +179,7 @@ async function convergeChannelStatusesOnBoot(): Promise { const data = await fetchCouncilConfig( membership.councilUrl, membership.channelAuthId, + { log }, ); if (data) await reconcileChannelStatuses(channelRegistry, data); } @@ -205,6 +206,7 @@ async function convergeChannelStatusesForCouncil( const data = await fetchCouncilConfig( membership.councilUrl, channelAuthId, + { log }, ); if (data) { await reconcileChannelStatuses(channelRegistry, data); @@ -235,7 +237,13 @@ async function activateMembership( let configJson: string | null = null; let councilName = membership.councilName; - const data = await fetchCouncilConfig(membership.councilUrl, channelAuthId); + const data = await fetchCouncilConfig( + membership.councilUrl, + channelAuthId, + { + log, + }, + ); if (data) { configJson = JSON.stringify(data); councilName = data.council?.name ?? councilName; @@ -398,3 +406,15 @@ export function stopEventWatcher(): void { watcher = null; } } + +/** + * Background-poll health of the event watcher for the /health endpoint (#10). + * Returns null when no watcher is running (e.g. no active councils yet). + */ +export function getEventWatcherHealth(): + | ReturnType< + EventWatcher["getHealth"] + > + | null { + return watcher?.getHealth() ?? null; +} diff --git a/src/core/service/executor/executor-failure.helpers.ts b/src/core/service/executor/executor-failure.helpers.ts index ef0cdcf..bacfb9a 100644 --- a/src/core/service/executor/executor-failure.helpers.ts +++ b/src/core/service/executor/executor-failure.helpers.ts @@ -2,7 +2,8 @@ import type { Logger } from "@/utils/logger/index.ts"; import { BundleStatus } from "@/persistence/drizzle/entity/operations-bundle.entity.ts"; import type { OperationsBundleRepository } from "@/persistence/drizzle/repository/operations-bundle.repository.ts"; import type { SlotBundle } from "@/core/service/bundle/bundle.types.ts"; -import { withSpan } from "@/core/tracing.ts"; +import { currentTraceId, withSpan } from "@/core/tracing.ts"; +import type { StructuredFailureDetail } from "@/core/service/executor/failure-detail.ts"; export type ExecutionFailureResult = { bundleId: string; @@ -22,6 +23,10 @@ export function handleExecutionFailure( deps: { operationsBundleRepository: OperationsBundleRepository; maxRetryAttempts: number; + /** True for an on-chain revert that will fail identically on retry. */ + deterministic: boolean; + /** Structured identity persisted to `failureDetail` on terminal failure. */ + failureDetail: StructuredFailureDetail; log: Logger; }, ): Promise { @@ -35,9 +40,17 @@ export function handleExecutionFailure( span.addEvent("handling_failure", { "error.message": errorMessage, "bundles.count": bundleIds.length, + deterministic: deps.deterministic, + "failure.code": deps.failureDetail.code, + }); + log.error(error, "execution failed", { + bundleId: bundleIds, + failureCode: deps.failureDetail.code, + deterministic: deps.deterministic, + traceId: currentTraceId(), }); - log.error(error, "execution failed"); + const failureDetail = { ...deps.failureDetail }; const bundlesToRetry: ExecutionFailureResult[] = []; for (const bundleId of bundleIds) { @@ -50,18 +63,30 @@ export function handleExecutionFailure( } const nextRetryCount = (bundle.retryCount ?? 0) + 1; + // A deterministic revert is terminal immediately — retrying it only + // wastes submits and produces duplicate FAILED traces (#12). const hasReachedMaxAttempts = nextRetryCount >= deps.maxRetryAttempts; + const isTerminal = deps.deterministic || hasReachedMaxAttempts; - if (hasReachedMaxAttempts) { + if (isTerminal) { await deps.operationsBundleRepository.update(bundleId, { status: BundleStatus.FAILED, retryCount: nextRetryCount, lastFailureReason, + failureDetail, updatedAt: new Date(), }); log.debug("bundleId", bundleId); log.debug("retryCount", nextRetryCount); - log.event("bundle moved to dead-letter after max retry attempts"); + span.addEvent("bundle_failed_terminal", { + "bundle.id": bundleId, + reason: deps.deterministic ? "deterministic" : "max_retries", + }); + log.event( + deps.deterministic + ? "bundle FAILED — deterministic revert, no retry" + : "bundle moved to dead-letter after max retry attempts", + ); } else { await deps.operationsBundleRepository.update(bundleId, { status: BundleStatus.PENDING, diff --git a/src/core/service/executor/executor.process.ts b/src/core/service/executor/executor.process.ts index 9b7c868..71c1611 100644 --- a/src/core/service/executor/executor.process.ts +++ b/src/core/service/executor/executor.process.ts @@ -27,7 +27,12 @@ import { TransactionRepository, } from "@/persistence/drizzle/repository/index.ts"; import { safeJsonStringify } from "@/utils/parse/safeStringify.ts"; -import { withSpan } from "@/core/tracing.ts"; +import { currentTraceId, withSpan } from "@/core/tracing.ts"; +import { + buildFailureDetail, + isDeterministic, + type StructuredFailureDetail, +} from "@/core/service/executor/failure-detail.ts"; import { buildRetryBundles, handleExecutionFailure as _handleExecutionFailure, @@ -84,7 +89,10 @@ class ExecutionError extends Error { readonly failureContext: ExecutionFailureContext; constructor(cause: Error, failureContext: ExecutionFailureContext) { - super(cause.message); + // Preserve the original error as `cause` so its structured payload (e.g. + // the colibri contract-error-matcher's meta.data.match with the soroban + // code) survives up to the decoder and the logs, per the bubbling model. + super(cause.message, { cause }); this.name = "ExecutionError"; this.stack = cause.stack; this.failureContext = failureContext; @@ -268,11 +276,14 @@ function handleExecutionFailure( error: Error, bundleIds: string[], lastFailureReason: string, + outcome: { deterministic: boolean; failureDetail: StructuredFailureDetail }, log: Logger, ) { return _handleExecutionFailure(error, bundleIds, lastFailureReason, { operationsBundleRepository, maxRetryAttempts: EXECUTOR_CONFIG.MAX_RETRY_ATTEMPTS, + deterministic: outcome.deterministic, + failureDetail: outcome.failureDetail, log, }); } @@ -521,13 +532,24 @@ export class Executor { const lastFailureReason = safeJsonStringify(lastFailureReasonPayload) ?? truncate(errorMessage, 2000); - this.log.error( - new Error(String("Slot execution failed")), - "Slot execution failed", - ); + // Translate the failure into a stable identity ONCE, and decide + // whether it is deterministic (an on-chain revert that will fail + // identically on retry) or transient (retryable). + const failureDetail = buildFailureDetail(errorInstance, networkCtx); + const deterministic = isDeterministic(errorInstance); const failedChannelContractId = slot?.getBundles()[0] ?.channelContractId ?? null; + + // Preserve the real cause (its full chain is flattened by the logger) + // instead of discarding it behind a generic "Slot execution failed". + this.log.error(errorInstance, "Slot execution failed", { + bundleId: bundleIds, + channelContractId: failedChannelContractId, + failureCode: failureDetail.code, + deterministic, + traceId: currentTraceId(), + }); if (failedChannelContractId) { await emitForBundles(bundleIds, (scope) => ({ kind: "executor.execution_failed", @@ -548,6 +570,7 @@ export class Executor { errorInstance, bundleIds, lastFailureReason, + { deterministic, failureDetail }, this.log, ); @@ -563,10 +586,9 @@ export class Executor { } } else { this.log.error( - new Error( - String("Execution error with no slot or bundles to re-add"), - ), + errorInstance, "Execution error with no slot or bundles to re-add", + { bundleId: bundleIds, traceId: currentTraceId() }, ); } } finally { diff --git a/src/core/service/executor/failure-detail.ts b/src/core/service/executor/failure-detail.ts new file mode 100644 index 0000000..40c47ec --- /dev/null +++ b/src/core/service/executor/failure-detail.ts @@ -0,0 +1,56 @@ +import { decodeContractError } from "@moonlight/moonlight-sdk"; +import type { NetworkErrorContext } from "@/core/service/executor/error-extraction.ts"; + +/** + * Structured, safe failure identity persisted to a bundle's `failureDetail` + * and returned verbatim in the API body. This is the edge translation of an + * internal error chain into a stable machine code — it must NOT carry stack + * frames, raw internal messages, or secrets. The UI maps `code` to copy. + */ +export interface StructuredFailureDetail { + /** Stable machine code, e.g. `SOROBAN_1010` or `PROVIDER_EXECUTION_FAILED`. */ + code: string; + /** Origin layer of the root cause. */ + source: "onchain" | "provider"; + /** Safe, human-readable summary (no internal detail). */ + message: string; + /** On-chain variant name when the root cause is a decoded contract error. */ + name?: string; +} + +/** + * Whether a failure is deterministic (will fail identically on retry) — true + * for decoded on-chain contract reverts, false for transient/network errors. + * Drives the retry decision (#12): never retry a deterministic revert. + */ +export function isDeterministic(error: unknown): boolean { + return decodeContractError(error) !== null; +} + +/** + * Translate a terminal execution failure into a {@link StructuredFailureDetail}. + * + * A decoded on-chain contract revert becomes `SOROBAN_` with its catalog + * name/description. Anything else (network, RPC, build) collapses to a single + * safe provider code — the rich internal cause stays in `lastFailureReason` + * and the span, never in this edge payload. + */ +export function buildFailureDetail( + error: unknown, + _networkCtx?: NetworkErrorContext, +): StructuredFailureDetail { + const decoded = decodeContractError(error); + if (decoded) { + return { + code: `SOROBAN_${decoded.code}`, + source: "onchain", + name: decoded.name, + message: decoded.details || `On-chain rejection ${decoded.name}`, + }; + } + return { + code: "PROVIDER_EXECUTION_FAILED", + source: "provider", + message: "The bundle could not be submitted to the network.", + }; +} diff --git a/src/core/service/mempool/mempool.process.ts b/src/core/service/mempool/mempool.process.ts index c5f3bff..243299d 100644 --- a/src/core/service/mempool/mempool.process.ts +++ b/src/core/service/mempool/mempool.process.ts @@ -4,6 +4,7 @@ import { BundleStatus } from "@/persistence/drizzle/entity/operations-bundle.ent import type { OperationsBundle } from "@/persistence/drizzle/entity/operations-bundle.entity.ts"; import { AccountRepository, + BundleTransactionRepository, EntityRepository, OperationsBundleRepository, } from "@/persistence/drizzle/repository/index.ts"; @@ -48,6 +49,7 @@ const operationsBundleRepository = new OperationsBundleRepository( ); const accountRepository = new AccountRepository(drizzleClient); const entityRepository = new EntityRepository(drizzleClient); +const bundleTransactionRepository = new BundleTransactionRepository(); /** * Parses MLXDR operations from a bundle entity @@ -166,9 +168,31 @@ async function loadPendingBundlesFromDB( log.event("querying pending/processing bundles"); const bundles = await operationsBundleRepository.findPendingOrProcessing(); log.debug("count", bundles.length); + + // Reconciliation guard (#11): a PROCESSING bundle that already has a linked + // transaction was submitted AND recorded before the crash. Re-executing it + // would double-submit; instead leave it PROCESSING for the verifier to + // reconcile against the chain (the verifier confirms it, or fails it once + // past its confirmation timeout — see #3). Only truly un-submitted bundles + // (PENDING, or PROCESSING with no transaction row) are re-hydrated. + const resubmittable: OperationsBundle[] = []; + for (const bundle of bundles) { + if (bundle.status === BundleStatus.PROCESSING) { + const links = await bundleTransactionRepository.findByBundleId(bundle.id); + if (links.length > 0) { + log.debug("bundleId", bundle.id); + log.event( + "skipping resubmit — bundle already has a transaction; left for verifier reconciliation", + ); + continue; + } + } + resubmittable.push(bundle); + } + log.event("hydrating SlotBundles"); const slotBundles = await Promise.all( - bundles.map((bundle: OperationsBundle) => + resubmittable.map((bundle: OperationsBundle) => createSlotBundleFromEntity(bundle, deps) ), ); diff --git a/src/core/service/verifier/verifier-failure.helpers.ts b/src/core/service/verifier/verifier-failure.helpers.ts index a45507f..36a7030 100644 --- a/src/core/service/verifier/verifier-failure.helpers.ts +++ b/src/core/service/verifier/verifier-failure.helpers.ts @@ -12,6 +12,14 @@ export type VerifierFailureDeps = { createSlotBundleFn: (bundle: OperationsBundle) => Promise; reAddBundlesFn: (bundles: SlotBundle[]) => Promise; maxRetryAttempts: number; + /** + * Force terminal FAILED with no retry — set for deterministic outcomes + * (an on-chain result-code failure, or a confirmation timeout) that would + * fail identically on resubmit (#3, #12). Defaults to retryable. + */ + terminal?: boolean; + /** Structured identity persisted to `failureDetail` on terminal failure. */ + failureDetail?: Record; log: Logger; }; @@ -52,6 +60,7 @@ export async function handleVerificationFailure( const nextRetryCount = (bundle.retryCount ?? 0) + 1; const hasReachedMaxAttempts = nextRetryCount >= deps.maxRetryAttempts; + const isTerminal = deps.terminal || hasReachedMaxAttempts; const lastFailureReason = safeJsonStringify({ occurredAt: new Date().toISOString(), @@ -62,15 +71,16 @@ export async function handleVerificationFailure( }) ?? reason; await deps.operationsBundleRepository.update(bundleId, { - status: hasReachedMaxAttempts - ? BundleStatus.FAILED - : BundleStatus.PENDING, + status: isTerminal ? BundleStatus.FAILED : BundleStatus.PENDING, retryCount: nextRetryCount, lastFailureReason, + ...(isTerminal && deps.failureDetail + ? { failureDetail: deps.failureDetail } + : {}), updatedAt: new Date(), }); - if (!hasReachedMaxAttempts) { + if (!isTerminal) { retryableBundleIds.push(bundleId); } else { log.debug("bundleId", bundleId); diff --git a/src/core/service/verifier/verifier.process.ts b/src/core/service/verifier/verifier.process.ts index 778cfd7..6b4622d 100644 --- a/src/core/service/verifier/verifier.process.ts +++ b/src/core/service/verifier/verifier.process.ts @@ -15,7 +15,7 @@ import { import { BundleStatus } from "@/persistence/drizzle/entity/operations-bundle.entity.ts"; import { getMempool } from "@/core/mempool/index.ts"; import { createSlotBundleFromEntity } from "@/core/service/mempool/mempool.process.ts"; -import { withSpan } from "@/core/tracing.ts"; +import { currentTraceId, withSpan } from "@/core/tracing.ts"; import { handleVerificationFailure as _handleVerificationFailure, } from "@/core/service/verifier/verifier-failure.helpers.ts"; @@ -78,6 +78,7 @@ function handleVerificationFailure( reason: string, bundleIds: string[], log: Logger, + opts?: { terminal?: boolean; failureDetail?: Record }, ): Promise { return _handleVerificationFailure(txId, reason, bundleIds, { operationsBundleRepository, @@ -86,6 +87,8 @@ function handleVerificationFailure( createSlotBundleFn: (bundle) => createSlotBundleFromEntity(bundle, { log }), reAddBundlesFn: (bundles) => getMempool().reAddBundles(bundles), maxRetryAttempts: VERIFIER_MAX_RETRY_ATTEMPTS, + terminal: opts?.terminal, + failureDetail: opts?.failureDetail, log, }); } @@ -262,7 +265,7 @@ export class Verifier { ); for (const transaction of unverifiedTransactions) { - await this.verifyTransaction(transaction.id); + await this.verifyTransaction(transaction); } span.addEvent("verification_cycle_complete"); @@ -272,10 +275,9 @@ export class Verifier { ? error.message : String(error), }); - this.log.error( - new Error(String("Error during transaction verification")), - "Error during transaction verification", - ); + this.log.error(error, "Error during transaction verification", { + traceId: currentTraceId(), + }); } }); } @@ -283,7 +285,10 @@ export class Verifier { /** * Verifies a single transaction */ - private verifyTransaction(txId: string): Promise { + private verifyTransaction( + transaction: { id: string; timeout: Date }, + ): Promise { + const txId = transaction.id; return withSpan("Verifier.verifyTransaction", async (span) => { try { span.setAttribute("tx.id", txId); @@ -316,11 +321,21 @@ export class Verifier { const channelContractId = await findFirstBundleChannel(bundleIds, { log: this.log, }); + // An applied-but-failed tx is deterministic — mark terminal with a + // structured on-chain identity, no retry (#1, #12). await handleVerificationFailure( txId, result.reason, bundleIds, this.log, + { + terminal: true, + failureDetail: { + code: "ONCHAIN_TX_FAILED", + source: "onchain", + message: "The transaction failed on-chain.", + }, + }, ); if (channelContractId) { await emitForBundles(bundleIds, (scope) => ({ @@ -335,6 +350,27 @@ export class Verifier { }, }), { log: this.log }); } + } else if (Date.now() > transaction.timeout.getTime()) { + // Dropped / never-closed tx past its confirmation window: fail it + // instead of leaving the bundle PROCESSING forever (#3). + span.addEvent("transaction_confirmation_timeout", { + "tx.timeout": transaction.timeout.toISOString(), + }); + this.log.event(`Transaction ${txId} timed out awaiting confirmation`); + await handleVerificationFailure( + txId, + `Confirmation timeout: not observed on-network by ${transaction.timeout.toISOString()}`, + bundleIds, + this.log, + { + terminal: true, + failureDetail: { + code: "PROVIDER_TX_TIMEOUT", + source: "provider", + message: "The transaction was not confirmed in time.", + }, + }, + ); } else { this.log.event(`Transaction ${txId} still pending verification`); } @@ -344,10 +380,10 @@ export class Verifier { ? error.message : String(error), }); - this.log.error( - new Error(String(`Failed to verify transaction ${txId}`)), - `Failed to verify transaction ${txId}`, - ); + this.log.error(error, `Failed to verify transaction ${txId}`, { + bundleId: txId, + traceId: currentTraceId(), + }); } }); } diff --git a/src/core/tracing.ts b/src/core/tracing.ts index cb568fb..5cabac9 100644 --- a/src/core/tracing.ts +++ b/src/core/tracing.ts @@ -44,5 +44,14 @@ export function withSpan( }); } +/** + * The active span's trace id, or undefined outside a span. Used to stamp + * correlation on log lines so a log can be joined to its trace. + */ +export function currentTraceId(): string | undefined { + const id = trace.getActiveSpan()?.spanContext().traceId; + return id && id !== "00000000000000000000000000000000" ? id : undefined; +} + export { SpanStatusCode, tracer }; export type { Span }; diff --git a/src/http/default-schemas.ts b/src/http/default-schemas.ts index 3897a1f..a6b8e6a 100644 --- a/src/http/default-schemas.ts +++ b/src/http/default-schemas.ts @@ -22,6 +22,9 @@ export const errorStatusSchema = z.union([ z.literal(Status.InternalServerError), z.literal(Status.TooManyRequests), z.literal(Status.Conflict), + // 503 — a dependency (e.g. Soroban RPC) is temporarily unreachable and the + // client should retry; used by the admission fast-fail (#2). + z.literal(Status.ServiceUnavailable), ]); export const errorResponseSchema = z.object({ diff --git a/src/http/v1/health/routes.ts b/src/http/v1/health/routes.ts index 1a7bd1d..43b747e 100644 --- a/src/http/v1/health/routes.ts +++ b/src/http/v1/health/routes.ts @@ -2,6 +2,7 @@ import { Router, Status } from "@oak/oak"; import { sql } from "drizzle-orm"; import { drizzleClient } from "@/persistence/drizzle/config.ts"; import { checkDbHealth } from "@/http/v1/health/db-check.ts"; +import { getEventWatcherHealth } from "@/core/service/event-watcher/index.ts"; const denoJson = JSON.parse( await Deno.readTextFile(new URL("../../../../deno.json", import.meta.url)), @@ -22,6 +23,11 @@ healthRouter.get("/health", async (ctx) => { const db = await checkDbHealth(() => drizzleClient.execute(sql`select 1`)); const healthy = db === "ok"; + // Event-watcher health is reported but does NOT gate the deploy status — a + // transient RPC blip must not flap the Fly health-gate (same rationale as + // checkDbHealth). Ops/alerting consume the `watcher` field (#10). + const watcher = getEventWatcherHealth(); + ctx.response.status = healthy ? Status.OK : Status.ServiceUnavailable; ctx.response.body = { status: healthy ? "ok" : "error", @@ -31,6 +37,7 @@ healthRouter.get("/health", async (ctx) => { "moonlight-sdk": sdkVersion, db, }, + watcher: watcher ?? "idle", }; }); diff --git a/src/utils/async/with-timeout.ts b/src/utils/async/with-timeout.ts new file mode 100644 index 0000000..ec3e38d --- /dev/null +++ b/src/utils/async/with-timeout.ts @@ -0,0 +1,30 @@ +/** Raised when {@link withTimeout} elapses before its promise settles. */ +export class TimeoutError extends Error { + constructor(readonly opName: string, readonly ms: number) { + super(`${opName} timed out after ${ms}ms`); + this.name = "TimeoutError"; + } +} + +/** + * Race a promise against a deadline so a hung dependency (e.g. an unreachable + * Soroban RPC during bundle admission) fast-fails instead of blocking the + * request indefinitely (#2). The underlying promise is not cancelled — it is + * abandoned — so only use this for idempotent reads. + */ +export function withTimeout( + promise: Promise, + ms: number, + opName: string, +): Promise { + let timer: number | undefined; + const timeout = new Promise((_, reject) => { + timer = setTimeout( + () => reject(new TimeoutError(opName, ms)), + ms, + ) as unknown as number; + }); + return Promise.race([promise, timeout]).finally(() => { + if (timer !== undefined) clearTimeout(timer); + }); +} diff --git a/src/utils/logger/index.ts b/src/utils/logger/index.ts index 5623024..c567cb9 100644 --- a/src/utils/logger/index.ts +++ b/src/utils/logger/index.ts @@ -10,11 +10,19 @@ export enum Level { Disabled = 3, } +/** Correlation ids attached to a log line (bundle / account / trace id). */ +export type Correlation = { [key: string]: unknown }; + export interface Logger { info(msg: string): void; event(msg: string): void; debug(key: string, value: unknown): void; - error(err: unknown, msg: string): void; + /** + * Emit an error. `err`'s full `cause` chain is flattened into the message so + * nothing wrapped on the way up is discarded; `corr` carries correlation ids + * (bundle id / account / trace id) that render on every sink. + */ + error(err: unknown, msg: string, corr?: Correlation): void; scope(name: string): Logger; } @@ -37,6 +45,7 @@ interface Record { key?: string; value?: unknown; error?: string; + corr?: Correlation; } type Format = (r: Record) => string; @@ -89,6 +98,32 @@ function stringify(v: unknown): string { } } +/** + * Flatten an error's `cause` chain into a single "msg <- cause <- cause" + * string so a wrapped error preserves every layer's context in the log line. + */ +function flattenCauses(err: unknown): string { + const parts: string[] = []; + let current: unknown = err; + const seen = new Set(); + for (let depth = 0; depth < 16 && current != null; depth++) { + if (seen.has(current)) break; + seen.add(current); + parts.push(current instanceof Error ? current.message : String(current)); + current = current instanceof Error + ? (current as { cause?: unknown }).cause + : undefined; + } + return parts.join(" <- "); +} + +function formatCorr(corr: Correlation): string { + return Object.entries(corr) + .filter(([, v]) => v !== undefined && v !== null) + .map(([k, v]) => `${k}=${stringify(v)}`) + .join(" "); +} + function humanFormat(colored: boolean): Format { const grayLb = colored ? chalk.gray : (s: string) => s; const greenLb = colored ? chalk.green : (s: string) => s; @@ -107,8 +142,14 @@ function humanFormat(colored: boolean): Format { return `${ts} ${cyanLb("DBG")} ${r.key}: ${ stringify(r.value) } (${r.scope})`; - case "error": - return `${ts} ${redLb("ERR")} [${r.scope}] ${r.msg} error="${r.error}"`; + case "error": { + const corr = r.corr && Object.keys(r.corr).length + ? ` ${formatCorr(r.corr)}` + : ""; + return `${ts} ${ + redLb("ERR") + } [${r.scope}] ${r.msg} error="${r.error}"${corr}`; + } } }; } @@ -124,6 +165,7 @@ const jsonFormat: Format = (r) => { if (r.key !== undefined) out.key = r.key; if (r.value !== undefined) out.value = safeJsonValue(r.value); if (r.error !== undefined) out.error = r.error; + if (r.corr !== undefined) out.corr = safeJsonValue(r.corr) as Correlation; try { return JSON.stringify(out); } catch (err) { @@ -172,15 +214,16 @@ class LoggerImpl implements Logger { this.emit({ ts: now(), level: "debug", scope: this.scopePath, key, value }); } - error(err: unknown, msg: string): void { + error(err: unknown, msg: string, corr?: Correlation): void { // ERR always emits regardless of level (matches go-logger / zerolog). - const detail = err instanceof Error ? err.message : String(err); + // The whole cause chain is flattened so wrapped context is never lost. this.emit({ ts: now(), level: "error", scope: this.scopePath, msg, - error: detail, + error: flattenCauses(err), + corr, }); } diff --git a/tests/deno.json b/tests/deno.json index 5791c34..e081340 100644 --- a/tests/deno.json +++ b/tests/deno.json @@ -5,7 +5,7 @@ "@colibri/core": "jsr:@colibri/core@^0.22.0", "@drizzle-team/brocli": "npm:@drizzle-team/brocli@^0.11.0", "@fifo/convee": "jsr:@fifo/convee@^0.10.0", - "@moonlight/moonlight-sdk": "jsr:@moonlight/moonlight-sdk@^0.11.0", + "@moonlight/moonlight-sdk": "jsr:@moonlight/moonlight-sdk@^0.11.1", "@std/assert": "jsr:@std/assert@^1.0.0", "@std/dotenv": "jsr:@std/dotenv@^0.225.6", "@zaubrik/djwt": "jsr:@zaubrik/djwt@^3.0.2", diff --git a/tests/integration/service/executor-retry.test.ts b/tests/integration/service/executor-retry.test.ts index 66792b2..30dce29 100644 --- a/tests/integration/service/executor-retry.test.ts +++ b/tests/integration/service/executor-retry.test.ts @@ -12,6 +12,13 @@ import { newNoop } from "@/utils/logger/index.ts"; const MAX_RETRY = 3; +// Transient (retryable) failure identity used by the existing retry cases. +const PROVIDER_DETAIL = { + code: "PROVIDER_EXECUTION_FAILED", + source: "provider" as const, + message: "The bundle could not be submitted to the network.", +}; + async function setup() { await ensureInitialized(); await resetDb(); @@ -55,6 +62,8 @@ Deno.test( const retryMeta = await handleExecutionFailure(error, [id], reason, { operationsBundleRepository: repo, maxRetryAttempts: MAX_RETRY, + deterministic: false, + failureDetail: PROVIDER_DETAIL, log: newNoop(), }); @@ -72,6 +81,42 @@ Deno.test( }, ); +Deno.test( + "executor – deterministic revert: FAILED immediately with failureDetail, NOT retried", + async () => { + const repo = await setup(); + const id = testBundleId(); + // Well below max retries — a deterministic revert must not be retried. + await seedBundle({ id, retryCount: 0, status: BundleStatus.PROCESSING }); + + const error = makeError("Contract error: SignatureExpired"); + const reason = makeLastFailureReason(error, id); + const sorobanDetail = { + code: "SOROBAN_1010", + source: "onchain" as const, + name: "SignatureExpired", + message: "A signature expired before the current ledger.", + }; + + const retryMeta = await handleExecutionFailure(error, [id], reason, { + operationsBundleRepository: repo, + maxRetryAttempts: MAX_RETRY, + deterministic: true, + failureDetail: sorobanDetail, + log: newNoop(), + }); + + // Not returned for retry. + assertEquals(retryMeta.length, 0); + + const found = await repo.findById(id); + assertExists(found); + assertEquals(found.status, BundleStatus.FAILED); + assertEquals(found.failureDetail?.code, "SOROBAN_1010"); + assertEquals(found.failureDetail?.source, "onchain"); + }, +); + Deno.test( "executor – at max retries (dead-letter): status→FAILED, retryCount=MAX, NOT returned for retry", async () => { @@ -85,6 +130,8 @@ Deno.test( const retryMeta = await handleExecutionFailure(error, [id], reason, { operationsBundleRepository: repo, maxRetryAttempts: MAX_RETRY, + deterministic: false, + failureDetail: PROVIDER_DETAIL, log: newNoop(), }); @@ -134,6 +181,8 @@ Deno.test( { operationsBundleRepository: repo, maxRetryAttempts: MAX_RETRY, + deterministic: false, + failureDetail: PROVIDER_DETAIL, log: newNoop(), }, ); @@ -169,6 +218,8 @@ Deno.test( await handleExecutionFailure(error, [id], reason, { operationsBundleRepository: repo, maxRetryAttempts: MAX_RETRY, + deterministic: false, + failureDetail: PROVIDER_DETAIL, log: newNoop(), }); @@ -199,6 +250,8 @@ Deno.test( const meta = await handleExecutionFailure(error, [id], reason, { operationsBundleRepository: repo, maxRetryAttempts: MAX_RETRY, + deterministic: false, + failureDetail: PROVIDER_DETAIL, log: newNoop(), }); @@ -224,6 +277,8 @@ Deno.test( const meta = await handleExecutionFailure(error, [missingId], reason, { operationsBundleRepository: repo, maxRetryAttempts: MAX_RETRY, + deterministic: false, + failureDetail: PROVIDER_DETAIL, log: newNoop(), });