From 42cb202bd80ea7d45bd0d49db3e18f28d7290f75 Mon Sep 17 00:00:00 2001 From: xeladev4 Date: Wed, 19 Aug 2026 18:26:23 +0100 Subject: [PATCH] feat(outbox): add durable transactional outbox + prioritized dispatcher Every on-chain money movement (deposit, withdraw, agent rebalance, referral reward) was fire-and-forget: build a Stellar operation, submit it, handle failure locally, with no durable record of intent that survives a crash and no ordering/prioritization/fee-bumping under congestion. Adds src/outbox/ as the single choke point every write now passes through: a transactional outbox (intent + business row commit/roll back together), an idempotency-keyed atomic claim (no double-submission), priority-ordered dispatch (CRITICAL withdrawals never starve behind a NORMAL/LOW wave), retry/backoff/fee-bump, a compliance halt guard, per-signer serialization, admin tooling, and Prometheus metrics. A structural test fails CI the moment a money path bypasses the outbox. Closes #325 --- docs/DOCUMENTATION_INDEX.md | 1 + docs/OUTBOX.md | 215 +++++++++ docs/openapi.yaml | 259 +++++++++++ jest.config.js | 6 +- .../migration.sql | 55 +++ .../20260819120000_add_outbox_op/rollback.sql | 26 ++ prisma/schema.prisma | 84 ++++ src/agent/router.ts | 67 ++- src/config/env.ts | 42 ++ src/controllers/transaction-controller.ts | 196 +++++--- src/index.ts | 8 + src/middleware/adminAuth.ts | 4 + src/outbox/dispatcher.ts | 338 ++++++++++++++ src/outbox/executors.ts | 88 ++++ src/outbox/idempotency.ts | 23 + src/outbox/service.ts | 427 ++++++++++++++++++ src/outbox/signerLock.ts | 86 ++++ src/outbox/stateMachine.ts | 88 ++++ src/outbox/types.ts | 88 ++++ src/referral/service.ts | 100 +++- src/routes/admin.ts | 154 +++++++ src/stellar/contract.ts | 57 ++- src/stellar/events.ts | 11 + src/utils/metrics.ts | 69 +++ src/validators/webhook-validators.ts | 3 + .../agent/rebalance.integration.test.ts | 33 +- .../deposit-withdraw.integration.test.ts | 6 + .../outbox/dispatcher.integration.test.ts | 254 +++++++++++ tests/unit/outbox/idempotency.test.ts | 35 ++ tests/unit/outbox/stateMachine.test.ts | 162 +++++++ tests/unit/outbox/structural.test.ts | 134 ++++++ tests/unit/referral/service.test.ts | 48 +- 32 files changed, 3020 insertions(+), 147 deletions(-) create mode 100644 docs/OUTBOX.md create mode 100644 prisma/migrations/20260819120000_add_outbox_op/migration.sql create mode 100644 prisma/migrations/20260819120000_add_outbox_op/rollback.sql create mode 100644 src/outbox/dispatcher.ts create mode 100644 src/outbox/executors.ts create mode 100644 src/outbox/idempotency.ts create mode 100644 src/outbox/service.ts create mode 100644 src/outbox/signerLock.ts create mode 100644 src/outbox/stateMachine.ts create mode 100644 src/outbox/types.ts create mode 100644 tests/integration/outbox/dispatcher.integration.test.ts create mode 100644 tests/unit/outbox/idempotency.test.ts create mode 100644 tests/unit/outbox/stateMachine.test.ts create mode 100644 tests/unit/outbox/structural.test.ts diff --git a/docs/DOCUMENTATION_INDEX.md b/docs/DOCUMENTATION_INDEX.md index 4af6d96..13c4e17 100644 --- a/docs/DOCUMENTATION_INDEX.md +++ b/docs/DOCUMENTATION_INDEX.md @@ -12,6 +12,7 @@ - **[STRATEGY_MARKETPLACE.md](STRATEGY_MARKETPLACE.md)** - Strategy marketplace / opt-in copy-trading: metric formula, eligibility gate, privacy & custody boundaries (#285) - **[PORTFOLIO_OPTIMIZATION.md](PORTFOLIO_OPTIMIZATION.md)** - Portfolio optimization & allocation suggestions: objective, λ mapping, estimation method, advisory invariant, limitations (#322) - **[PERFORMANCE_ATTRIBUTION.md](PERFORMANCE_ATTRIBUTION.md)** - Benchmark-relative Brinson attribution: allocation/selection effects, Cariño linking, benchmark definition, `vsBenchmark` on the marketplace (#320) +- **[OUTBOX.md](OUTBOX.md)** - Durable outbox & prioritized on-chain transaction queue: state machine, idempotency, retry/fee-bump policy, priority ordering, admin API (#325) ### For DevOps/Deployment diff --git a/docs/OUTBOX.md b/docs/OUTBOX.md new file mode 100644 index 0000000..58e7f59 --- /dev/null +++ b/docs/OUTBOX.md @@ -0,0 +1,215 @@ +# Durable Outbox & Prioritized On-Chain Transaction Queue (#325) + +Every on-chain money movement — a user deposit or withdrawal, an agent +rebalance, a referral reward payout — used to be fire-and-forget from the +caller's perspective: build a Stellar operation, submit it, handle failure +locally. There was no shared, durable record of intent that survives a crash +or a mid-flight RPC timeout, no global ordering or prioritization, and no +fee-bumping strategy under network congestion. + +This adds a durable outbox + prioritized dispatcher — the single choke point +every on-chain write passes through — so a money move is atomic (the intent +and the business state that caused it commit or roll back together), +retriable, priority-ordered, and observable. + +## The state machine + +``` + ┌────────────────────────────────────────┐ + │ │ + ▼ │ + enqueue ──────► PENDING ──────claim──────► SUBMITTED ──────► │ (backoff retry) + (in a DB tx) │ │ │ + │ │ └────success────► CONFIRMED (terminal) + │ │ + cancel (admin, exhausted attempts / + unsent only) fee-bump cap reached + │ │ + ▼ ▼ + CANCELLED (terminal) FAILED (terminal) ◄── admin force-retry ──► PENDING +``` + +| Status | Meaning | +| --- | --- | +| `PENDING` | Durable intent persisted; not yet claimed, or returned here after a retriable failure (with `nextAttemptAt` backoff) | +| `SUBMITTED` | Claimed by a dispatcher and mid-flight; carries `signerPublicKey` and an incremented `attempts` | +| `CONFIRMED` | On-chain success observed (terminal) | +| `FAILED` | Exhausted retries, a non-retriable on-chain rejection, or the fee-bump cap was reached (terminal; full error + attempt count retained for audit) | +| `CANCELLED` | Admin-cancelled while still unsent (terminal) | + +The pure rules — legal transitions and priority ordering — live in +`src/outbox/stateMachine.ts` with no I/O, so they are unit-tested directly +(`tests/unit/outbox/stateMachine.test.ts`) without a database. + +## Priority & ordering + +```ts +CRITICAL // user withdrawals — capital leaving the platform +NORMAL // deposits, recurring deposits, referral rewards +LOW // agent-triggered rebalances +``` + +The dispatcher claims `PENDING` ops ordered by priority, then FIFO +(`createdAt`) within a tier. A CRITICAL withdrawal never waits behind a wave +of NORMAL or LOW ops queued ahead of it, however large — see +`src/outbox/stateMachine.ts#compareForDispatch` and the starvation tests in +`tests/unit/outbox/stateMachine.test.ts`. + +## Idempotency + +`idempotencyKey = "::"` +(`src/outbox/idempotency.ts`), where `businessRecordId` is the row this op is +the durable intent for — a `Transaction.id` for DEPOSIT/WITHDRAW/REBALANCE, a +`":"` pair for REFERRAL_REWARD. `enqueueOutboxOp` is an +upsert by this key: re-running a caller against the same business record +(a retried job tick, a crash-and-restart) resolves to the same op row instead +of creating a duplicate. + +## Atomic claim (no double-submission) + +Claiming is a single conditional update: + +```sql +UPDATE outbox_ops SET status = 'SUBMITTED', attempts = attempts + 1, ... +WHERE id = $1 AND status = 'PENDING' +``` + +Exactly one caller ever sees `count = 1`; every other concurrent claim +attempt for the same op sees `count = 0` and treats it as a no-op +(`src/outbox/service.ts#claimOp`). This is what makes it safe for a +synchronous request handler and the background sweep to race for the same op +— see `tests/integration/outbox/dispatcher.integration.test.ts`. + +## Retry, backoff, and fee-bump + +- **Transient submit failure** (a thrown exception — network error, RPC + timeout, simulation failure): the op returns to `PENDING` with a + full-jitter exponential backoff `nextAttemptAt` + (`src/outbox/stateMachine.ts#computeBackoffMs`, bounded by + `OUTBOX_BACKOFF_BASE_MS`/`OUTBOX_BACKOFF_MAX_MS`). After + `OUTBOX_MAX_ATTEMPTS` the op moves to `FAILED`. +- **Non-retriable on-chain rejection** (the submission resolves with + `status: 'failed'` rather than throwing — e.g. a vault-contract + precondition failure): recorded as `FAILED` immediately, not retried. This + matches the single-attempt behavior the deposit/withdraw routes had before + this change. +- **Unconfirmed too long** (`SUBMITTED` past `OUTBOX_SUBMITTED_TIMEOUT_MS`, + typically because the dispatcher process crashed between submitting and + observing its own confirmation, or the network is congested): escalated + back to `PENDING` for reclaim; the next submission uses a bumped fee + (`feeBumpMultiplier ^ attempts`, via `src/stellar/contract.ts`'s + `feeMultiplier` parameter on every write call). After + `OUTBOX_FEE_BUMP_MAX_ATTEMPTS` bumps, the op is escalated straight to + `FAILED` with the full attempt history preserved. + +All of the above is driven by `src/outbox/dispatcher.ts#runDispatchSweep`, +scheduled every `OUTBOX_DISPATCH_INTERVAL_MS` (default 15s). + +## Confirmation oracle + +Two paths close an op out to `CONFIRMED`: + +1. **The dispatcher's own submission.** `executeOutboxPayload` (via + `src/stellar/contract.ts`) already waits for on-chain confirmation before + resolving, so in the normal case the same call that submits also confirms. +2. **The event listener, as a fallback.** `src/stellar/events.ts` calls + `reconcileOutboxOpByTxHash` on the same DB transaction that confirms a + `Transaction` row by `txHash`. This closes out a `SUBMITTED` op left + behind by a dispatcher crash between submission and its own confirmation + — the event listener remains the durable source of truth, matching the + existing `ProcessedEvent`/`DeadLetterEvent` design. + +For a synchronous caller (deposit, withdraw, referral payout), the linked +`Transaction` row is updated directly after `dispatchOne` resolves. For the +one genuinely non-blocking path — an agent rebalance, enqueued via +`dispatchInBackground` and never awaited by the loop — nobody would otherwise +update that row, so `src/outbox/dispatcher.ts#submitClaimedOp` mirrors the +final outcome onto it directly via `mirrorLinkedTransaction` +(`transactionId` carried in the payload). This also covers the general +crash-recovery case: if the original synchronous caller already returned +(possibly with an error) before the background sweep independently resubmits +and confirms the same op, the Transaction row still gets updated. + +## Per-signer serialization & concurrency caps + +The custodial/agent wallet submits transactions using that account's Stellar +sequence number: two ops signed by the same key must never be in flight at +once. `src/outbox/signerLock.ts` is an in-process mutex keyed by +`signerPublicKey`, plus a global and a per-account cap +(`OUTBOX_GLOBAL_MAX_IN_FLIGHT`, `OUTBOX_PER_ACCOUNT_MAX_IN_FLIGHT`, default +`1` — i.e. strictly serial per signer). This is sufficient for a single +dispatcher process; the atomic claim above is what would make running more +than one dispatcher process safe, but that configuration is out of scope for +this change. + +## Compliance halt guard + +Before dispatching any op, `src/outbox/service.ts#isUserHalted` checks +`User.isActive` — the same field `src/middleware/authenticate.ts` already +checks to reject a frozen user's session. A user frozen after their op was +already queued has that op skipped on every sweep until they are reactivated; +new requests from a frozen user are already rejected at the route layer. + +## Caller migration + +| Path | Kind | Priority | Dispatch | +| --- | --- | --- | --- | +| `POST /deposit`, `POST /withdraw` (`src/controllers/transaction-controller.ts`) | DEPOSIT / WITHDRAW | NORMAL / CRITICAL | Enqueued transactionally with the `Transaction` row, then dispatched inline and awaited — the HTTP response contract (`201`, synchronous `CONFIRMED`/`FAILED`) is unchanged from before this PR | +| Recurring deposits (`src/jobs/recurringDeposits.ts`) | DEPOSIT | NORMAL | Shares `executeDeposit` with the HTTP path — gets outbox durability for free | +| Referral payouts (`src/referral/service.ts#payOneReward`) | REFERRAL_REWARD | NORMAL | Enqueued transactionally, dispatched inline and awaited (the sweep already processes conversions serially) | +| Agent rebalance (`src/agent/router.ts#triggerRebalance`) | REBALANCE | LOW | Enqueued transactionally, then `dispatchInBackground` — **not awaited**. The loop moves to the next batch immediately; the background sweep (or the opportunistic in-process attempt) submits it on its own cadence | + +No code path outside `src/stellar/contract.ts` (which defines the raw write +functions) and `src/outbox/executors.ts` (the one place that calls them) may +import `depositForUser`, `withdrawForUser`, `triggerRebalance`, or +`payReferralReward` — enforced by +`tests/unit/outbox/structural.test.ts`, which fails CI the moment a new money +path bypasses the outbox. + +## Admin API + +All under `/api/admin/outbox`, scoped keys (`outbox:read` / `outbox:write`), +fully `AdminAuditLog`-audited (`src/routes/admin.ts`): + +| Endpoint | Scope | Purpose | +| --- | --- | --- | +| `GET /api/admin/outbox` | `outbox:read` | List/query ops — filter by `status`, `kind`, `priority`, `userId` | +| `GET /api/admin/outbox/stats` | `outbox:read` | Queue depth grouped by status/priority | +| `GET /api/admin/outbox/:id` | `outbox:read` | Inspect a single op | +| `POST /api/admin/outbox/:id/retry` | `outbox:write` | Force a `FAILED` op back to `PENDING`, clearing backoff | +| `POST /api/admin/outbox/:id/cancel` | `outbox:write` | Cancel an unsent op — `PENDING` only; a `SUBMITTED` op is already on-chain and cannot be cancelled | + +## Metrics (`src/utils/metrics.ts`) + +| Metric | Type | Labels | +| --- | --- | --- | +| `outbox_ops_total` | Counter | `kind`, `priority`, `outcome` (`confirmed`\|`retry`\|`failed`) | +| `outbox_queue_depth` | Gauge | `status`, `priority` | +| `outbox_op_latency_seconds` | Histogram | `kind` — creation to confirmation | +| `outbox_fee_bump_total` | Counter | `kind` | +| `outbox_stuck_submitted` | Gauge | — ops `SUBMITTED` past the timeout; the "lost in flight" alarm | + +## Configuration (`src/config/env.ts` → `config.outbox`) + +| Env var | Default | Meaning | +| --- | --- | --- | +| `OUTBOX_DISPATCH_INTERVAL_MS` | `15000` | Background sweep cadence | +| `OUTBOX_MAX_ATTEMPTS` | `5` | Attempts before a transient failure gives up (→ `FAILED`) | +| `OUTBOX_BACKOFF_BASE_MS` / `OUTBOX_BACKOFF_MAX_MS` | `2000` / `120000` | Full-jitter backoff bounds | +| `OUTBOX_SUBMITTED_TIMEOUT_MS` | `90000` | How long `SUBMITTED` may sit unconfirmed before fee-bump escalation | +| `OUTBOX_FEE_BUMP_MULTIPLIER` | `2` | Fee multiplier per bump (compounds) | +| `OUTBOX_FEE_BUMP_MAX_ATTEMPTS` | `3` | Fee-bump cap before a stuck op is escalated to `FAILED` | +| `OUTBOX_GLOBAL_MAX_IN_FLIGHT` | `10` | Global in-flight cap | +| `OUTBOX_PER_ACCOUNT_MAX_IN_FLIGHT` | `1` | Per-signer in-flight cap (serial per account) | +| `OUTBOX_BATCH_SIZE` | `20` | Ops claimed per sweep | + +## Out of scope + +- Multi-signer/HSM signing infrastructure — the dispatcher stays on the + existing agent-signed / custodial-user-signed path. +- Cross-chain submission — Stellar-only; `src/outbox/executors.ts` is the + intended seam for adding another chain's executor later. +- Replacing the event listener as confirmation source of truth. +- Running more than one dispatcher process — the atomic claim would make it + safe, but the in-process signer mutex would not coordinate across + processes without an additional distributed lock. diff --git a/docs/openapi.yaml b/docs/openapi.yaml index 90a6ead..213c0e9 100644 --- a/docs/openapi.yaml +++ b/docs/openapi.yaml @@ -2771,6 +2771,213 @@ paths: '403': $ref: '#/components/responses/Forbidden' + # ── Durable outbox (#325) ──────────────────────────────────────────────────── + /api/v1/admin/outbox: + get: + tags: [admin] + operationId: adminListOutboxOps + summary: List/query durable outbox ops + description: | + Lists on-chain money-movement intents queued through the durable + outbox. See docs/OUTBOX.md. Requires admin scope `outbox:read`. + security: + - AdminToken: [] + parameters: + - in: query + name: status + schema: + type: string + enum: [PENDING, SUBMITTED, CONFIRMED, FAILED, CANCELLED] + - in: query + name: kind + schema: + type: string + enum: [DEPOSIT, WITHDRAW, REBALANCE, RECURRING_DEPOSIT, REFERRAL_REWARD, YIELD_CLAIM] + - in: query + name: priority + schema: + type: string + enum: [CRITICAL, NORMAL, LOW] + - in: query + name: userId + schema: + type: string + - in: query + name: limit + schema: + type: integer + default: 50 + maximum: 500 + - in: query + name: offset + schema: + type: integer + default: 0 + responses: + '200': + description: Outbox ops matching the filter + content: + application/json: + schema: + type: object + properties: + success: + type: boolean + data: + type: object + properties: + ops: + type: array + items: + $ref: '#/components/schemas/OutboxOp' + total: + type: integer + '401': + $ref: '#/components/responses/Unauthorized' + '403': + $ref: '#/components/responses/Forbidden' + + /api/v1/admin/outbox/stats: + get: + tags: [admin] + operationId: adminGetOutboxStats + summary: Outbox queue depth by status/priority + description: | + Throughput view over the outbox queue. Requires admin scope `outbox:read`. + security: + - AdminToken: [] + responses: + '200': + description: Queue depth grouped by status and priority + content: + application/json: + schema: + type: object + properties: + success: + type: boolean + data: + type: object + properties: + stats: + type: array + items: + type: object + properties: + status: + type: string + priority: + type: string + count: + type: integer + '401': + $ref: '#/components/responses/Unauthorized' + '403': + $ref: '#/components/responses/Forbidden' + + /api/v1/admin/outbox/{id}: + get: + tags: [admin] + operationId: adminGetOutboxOp + summary: Inspect a single outbox op + description: Requires admin scope `outbox:read`. + security: + - AdminToken: [] + parameters: + - in: path + name: id + required: true + schema: + type: string + responses: + '200': + description: The outbox op + content: + application/json: + schema: + type: object + properties: + success: + type: boolean + data: + $ref: '#/components/schemas/OutboxOp' + '401': + $ref: '#/components/responses/Unauthorized' + '403': + $ref: '#/components/responses/Forbidden' + '404': + description: Outbox op not found + + /api/v1/admin/outbox/{id}/retry: + post: + tags: [admin] + operationId: adminRetryOutboxOp + summary: Force a FAILED op back to PENDING + description: | + Clears backoff and re-queues a terminally FAILED op for the + dispatcher to re-attempt. Requires admin scope `outbox:write`. + security: + - AdminToken: [] + parameters: + - in: path + name: id + required: true + schema: + type: string + responses: + '200': + description: Op returned to PENDING + content: + application/json: + schema: + type: object + properties: + success: + type: boolean + data: + $ref: '#/components/schemas/OutboxOp' + '400': + description: Op is not FAILED + '401': + $ref: '#/components/responses/Unauthorized' + '403': + $ref: '#/components/responses/Forbidden' + + /api/v1/admin/outbox/{id}/cancel: + post: + tags: [admin] + operationId: adminCancelOutboxOp + summary: Cancel an unsent op + description: | + Cancels a PENDING op. A SUBMITTED op is already on-chain and cannot + be cancelled. Requires admin scope `outbox:write`. + security: + - AdminToken: [] + parameters: + - in: path + name: id + required: true + schema: + type: string + responses: + '200': + description: Op cancelled + content: + application/json: + schema: + type: object + properties: + success: + type: boolean + data: + $ref: '#/components/schemas/OutboxOp' + '400': + description: Op is not PENDING (already submitted, confirmed, failed, or cancelled) + '401': + $ref: '#/components/responses/Unauthorized' + '403': + $ref: '#/components/responses/Forbidden' + # ── Metrics ──────────────────────────────────────────────────────────────── /metrics: get: @@ -3960,6 +4167,58 @@ components: type: string format: date-time + OutboxOp: + type: object + description: A durable on-chain money-movement intent (#325). See docs/OUTBOX.md. + properties: + id: + type: string + idempotencyKey: + type: string + userId: + type: string + kind: + type: string + enum: [DEPOSIT, WITHDRAW, REBALANCE, RECURRING_DEPOSIT, REFERRAL_REWARD, YIELD_CLAIM] + actor: + type: string + enum: [USER, AGENT, SYSTEM] + priority: + type: string + enum: [CRITICAL, NORMAL, LOW] + status: + type: string + enum: [PENDING, SUBMITTED, CONFIRMED, FAILED, CANCELLED] + txHash: + type: string + nullable: true + attempts: + type: integer + nextAttemptAt: + type: string + format: date-time + nullable: true + error: + type: string + nullable: true + submittedAt: + type: string + format: date-time + nullable: true + confirmedAt: + type: string + format: date-time + nullable: true + signerPublicKey: + type: string + nullable: true + createdAt: + type: string + format: date-time + updatedAt: + type: string + format: date-time + TransactionResponse: type: object properties: diff --git a/jest.config.js b/jest.config.js index 4cd152e..b98b200 100644 --- a/jest.config.js +++ b/jest.config.js @@ -15,12 +15,16 @@ module.exports = { testMatch: ['**/*.test.ts'], // Integration tests that require a live Postgres instance are excluded from // the default `npm test` run (they fail in CI without a provisioned DB). - // Run them manually with: npx jest --testPathPattern='integration/(deposit-withdraw|tax-report)' --setupFilesAfterSetup=... + // Run them manually with: npx jest --testPathPattern='integration/(deposit-withdraw|tax-report|outbox)' --setupFilesAfterSetup=... testPathIgnorePatterns: [ '/node_modules/', 'deposit-withdraw\\.integration\\.test\\.ts$', 'tax-report\\.integration\\.test\\.ts$', 'regression\\.test\\.ts$', + // #325 — atomic-claim/kill-the-worker/priority-ordering scenarios against + // a real DB. See the header comment in + // tests/integration/outbox/dispatcher.integration.test.ts for how to run it. + 'outbox/dispatcher\\.integration\\.test\\.ts$', ], // Must run before any test module so src/config/env.ts sees the test config // at import time. See tests/setup-env.ts. diff --git a/prisma/migrations/20260819120000_add_outbox_op/migration.sql b/prisma/migrations/20260819120000_add_outbox_op/migration.sql new file mode 100644 index 0000000..843a895 --- /dev/null +++ b/prisma/migrations/20260819120000_add_outbox_op/migration.sql @@ -0,0 +1,55 @@ +-- CreateEnum +CREATE TYPE "OutboxOpKind" AS ENUM ('DEPOSIT', 'WITHDRAW', 'REBALANCE', 'RECURRING_DEPOSIT', 'REFERRAL_REWARD', 'YIELD_CLAIM'); + +-- CreateEnum +CREATE TYPE "OutboxOpActor" AS ENUM ('USER', 'AGENT', 'SYSTEM'); + +-- CreateEnum +CREATE TYPE "OutboxPriority" AS ENUM ('CRITICAL', 'NORMAL', 'LOW'); + +-- CreateEnum +CREATE TYPE "OutboxOpStatus" AS ENUM ('PENDING', 'SUBMITTED', 'CONFIRMED', 'FAILED', 'CANCELLED'); + +-- CreateTable +CREATE TABLE "outbox_ops" ( + "id" TEXT NOT NULL, + "idempotencyKey" TEXT NOT NULL, + "userId" TEXT NOT NULL, + "kind" "OutboxOpKind" NOT NULL, + "actor" "OutboxOpActor" NOT NULL, + "payload" JSONB NOT NULL, + "priority" "OutboxPriority" NOT NULL, + "status" "OutboxOpStatus" NOT NULL DEFAULT 'PENDING', + "txHash" TEXT, + "attempts" INTEGER NOT NULL DEFAULT 0, + "nextAttemptAt" TIMESTAMP(3), + "error" TEXT, + "submittedAt" TIMESTAMP(3), + "confirmedAt" TIMESTAMP(3), + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updatedAt" TIMESTAMP(3) NOT NULL, + "signerPublicKey" TEXT, + + CONSTRAINT "outbox_ops_pkey" PRIMARY KEY ("id") +); + +-- CreateIndex +CREATE UNIQUE INDEX "outbox_ops_idempotencyKey_key" ON "outbox_ops"("idempotencyKey"); + +-- CreateIndex +CREATE UNIQUE INDEX "outbox_ops_txHash_key" ON "outbox_ops"("txHash"); + +-- CreateIndex +CREATE INDEX "outbox_ops_status_priority_createdAt_idx" ON "outbox_ops"("status", "priority", "createdAt"); + +-- CreateIndex +CREATE INDEX "outbox_ops_userId_idx" ON "outbox_ops"("userId"); + +-- CreateIndex +CREATE INDEX "outbox_ops_kind_idx" ON "outbox_ops"("kind"); + +-- CreateIndex +CREATE INDEX "outbox_ops_signerPublicKey_status_idx" ON "outbox_ops"("signerPublicKey", "status"); + +-- CreateIndex +CREATE INDEX "outbox_ops_status_nextAttemptAt_idx" ON "outbox_ops"("status", "nextAttemptAt"); diff --git a/prisma/migrations/20260819120000_add_outbox_op/rollback.sql b/prisma/migrations/20260819120000_add_outbox_op/rollback.sql new file mode 100644 index 0000000..72afec6 --- /dev/null +++ b/prisma/migrations/20260819120000_add_outbox_op/rollback.sql @@ -0,0 +1,26 @@ +-- Rollback for 20260819120000_add_outbox_op +-- Drops the durable outbox table and its enums (#325). +-- +-- Safe to run before or after deploying the reverted application code: the +-- outbox is additive infrastructure — src/outbox/*, the migrated callers in +-- src/controllers/transaction-controller.ts, src/referral/service.ts, and +-- src/agent/router.ts all import from it, so the application code must be +-- rolled back to the pre-#325 revision FIRST (it will not start with these +-- imports unresolved). Once that revision is deployed, dropping this table +-- loses no confirmed on-chain history: txHash/status/confirmedAt are already +-- mirrored onto the Transaction table by every caller, which is untouched by +-- this migration. Any op still PENDING/SUBMITTED at rollback time represents +-- an in-flight submission that has not been mirrored back yet — drain the +-- queue (or capture it via `SELECT * FROM outbox_ops WHERE status IN +-- ('PENDING','SUBMITTED')`) before rolling back if that matters for your +-- deployment. + +DROP TABLE IF EXISTS "outbox_ops"; + +DROP TYPE IF EXISTS "OutboxOpStatus"; + +DROP TYPE IF EXISTS "OutboxPriority"; + +DROP TYPE IF EXISTS "OutboxOpActor"; + +DROP TYPE IF EXISTS "OutboxOpKind"; diff --git a/prisma/schema.prisma b/prisma/schema.prisma index c3d7723..2ebd8cd 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -67,6 +67,39 @@ enum DeadLetterEventStatus { RESOLVED } +// #325 — durable outbox for on-chain money movements. +enum OutboxOpKind { + DEPOSIT + WITHDRAW + REBALANCE + RECURRING_DEPOSIT + REFERRAL_REWARD + YIELD_CLAIM +} + +enum OutboxOpActor { + USER + AGENT + SYSTEM +} + +// CRITICAL: user withdrawals (capital leaving the platform). +// NORMAL: recurring deposits / referral rewards. +// LOW: agent-triggered rebalances. +enum OutboxPriority { + CRITICAL + NORMAL + LOW +} + +enum OutboxOpStatus { + PENDING + SUBMITTED + CONFIRMED + FAILED + CANCELLED +} + enum FiatDirection { ON_RAMP OFF_RAMP @@ -464,6 +497,57 @@ model DeadLetterEvent { @@map("dead_letter_events") } +// Durable outbox + prioritized dispatcher for on-chain money movements (#325). +// This is the single choke point every write to the vault contract must pass +// through — src/outbox/executors.ts is the only module besides +// src/stellar/contract.ts itself allowed to call the raw write functions +// (enforced by tests/unit/outbox/structural.test.ts). +model OutboxOp { + id String @id @default(uuid()) + + // Deterministic per-caller anchor (e.g. "DEPOSIT::"). + // Guarantees at-most-once submission across dispatcher crashes/restarts — + // see src/outbox/idempotency.ts. + idempotencyKey String @unique + + userId String + kind OutboxOpKind + actor OutboxOpActor + + // The exact, validated operation: { method, params, asset, amount, + // destination, ... } — never a key, never unvalidated user input. + payload Json + + priority OutboxPriority + status OutboxOpStatus @default(PENDING) + + // Set once the dispatcher successfully submits to the network. + txHash String? @unique + + attempts Int @default(0) + nextAttemptAt DateTime? + error String? @db.Text + + submittedAt DateTime? + confirmedAt DateTime? + + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + // Which signer (Stellar public key) this op executes under. The dispatcher + // serializes ops sharing a signer so sequence-number ordering on that + // account is never raced (see src/outbox/signerLock.ts). Populated at + // claim time, once the signer is known. + signerPublicKey String? + + @@index([status, priority, createdAt]) + @@index([userId]) + @@index([kind]) + @@index([signerPublicKey, status]) + @@index([status, nextAttemptAt]) + @@map("outbox_ops") +} + model CustodialWallet { id String @id @default(uuid()) userId String @unique diff --git a/src/agent/router.ts b/src/agent/router.ts index 5cd12b6..f61eb4b 100644 --- a/src/agent/router.ts +++ b/src/agent/router.ts @@ -12,13 +12,15 @@ import { UserStrategyPreferences, } from './types' import { scanAllProtocols, getCurrentOnChainApy } from './scanner' -import { triggerRebalance as submitRebalance } from '../stellar/contract' import { MaxYieldStrategy, TargetAllocationStrategy, GoalTrackingStrategy, } from './strategies' import db from '../db' +import { enqueueOutboxOp } from '../outbox/service' +import { dispatchInBackground } from '../outbox/dispatcher' +import { deriveIdempotencyKey } from '../outbox/idempotency' const DEFAULT_THRESHOLDS: RebalanceThresholds = { minimumImprovement: 0.5, // Must improve by at least 0.5% @@ -210,10 +212,14 @@ export async function triggerRebalance( expectedApyBasisPoints, }) - const onChainTransaction = await submitRebalance( - toProtocol, - expectedApyBasisPoints - ) + // #325: a rebalance is a LOW-priority, durable, NON-BLOCKING outbox op — + // the loop enqueues the intent (transactionally with the Transaction row) + // and moves on without waiting for the on-chain round trip. The + // background dispatcher (src/outbox/dispatcher.ts) submits it on its own + // cadence, and the event listener confirms the linked Transaction row + // when the on-chain event arrives. txHash is therefore not known yet at + // this point — RebalanceDetails.txHash is left undefined. + let txHash: string | undefined if (positionIds.length > 0) { const representativePosition = await db.position.findFirst({ @@ -230,20 +236,42 @@ export async function triggerRebalance( }) if (representativePosition) { - await db.transaction.create({ - data: { + const opId = await db.$transaction(async (tx) => { + const transaction = await tx.transaction.create({ + data: { + userId: representativePosition.userId, + positionId: representativePosition.id, + type: 'REBALANCE', + status: 'PENDING', + assetSymbol: representativePosition.assetSymbol, + amount, + network: representativePosition.user.network, + protocolName: toProtocol, + memo: `Agent rebalance from ${fromProtocol} to ${toProtocol}`, + } as any, + }) + + const op = await enqueueOutboxOp(tx, { + idempotencyKey: deriveIdempotencyKey( + 'REBALANCE', + representativePosition.userId, + transaction.id + ), userId: representativePosition.userId, - positionId: representativePosition.id, - txHash: onChainTransaction.hash, - type: 'REBALANCE', - status: 'PENDING', - assetSymbol: representativePosition.assetSymbol, - amount, - network: representativePosition.user.network, - protocolName: toProtocol, - memo: `Agent rebalance from ${fromProtocol} to ${toProtocol}`, - } as any, + kind: 'REBALANCE', + actor: 'AGENT', + payload: { + method: 'rebalance', + toProtocol, + expectedApyBasisPoints, + transactionId: transaction.id, + }, + }) + + return op.id }) + + dispatchInBackground(opId) } else { logger.warn('No position found to persist rebalance transaction', { fromProtocol, @@ -257,7 +285,7 @@ export async function triggerRebalance( fromProtocol, toProtocol, amount, - txHash: onChainTransaction.hash, + txHash, timestamp: new Date(), improvedBy: comparison.improvement, } @@ -302,8 +330,7 @@ export async function triggerRebalance( }) } - logger.info('Rebalance successful', { - txHash: onChainTransaction.hash, + logger.info('Rebalance queued (durable, dispatched asynchronously)', { duration, improvedBy: comparison.improvement.toFixed(2), }) diff --git a/src/config/env.ts b/src/config/env.ts index fe47ed5..efb9652 100644 --- a/src/config/env.ts +++ b/src/config/env.ts @@ -590,4 +590,46 @@ export const config = { process.env.RECURRING_DEPOSITS_INTERVAL_MS || '300000' ), }, + outbox: { + /** + * How often (ms) the background dispatcher sweeps for PENDING ops left + * behind by a crash (or whose backoff window has elapsed) and for + * SUBMITTED ops that have gone quiet long enough to need a fee-bump or + * escalation (default: 15 seconds). See docs/OUTBOX.md. + */ + dispatchIntervalMs: parseInt( + process.env.OUTBOX_DISPATCH_INTERVAL_MS || '15000' + ), + /** Attempts before a PENDING op is given up on and moved to FAILED. */ + maxAttempts: parseInt(process.env.OUTBOX_MAX_ATTEMPTS || '5'), + /** Full-jitter exponential backoff bounds (ms) between submit attempts. */ + backoffBaseMs: parseInt(process.env.OUTBOX_BACKOFF_BASE_MS || '2000'), + backoffMaxMs: parseInt(process.env.OUTBOX_BACKOFF_MAX_MS || '120000'), + /** + * How long (ms) a SUBMITTED op may sit unconfirmed before the dispatcher + * treats it as congested and resubmits at a higher fee (default: 90s — + * comfortably past normal Stellar ledger close time). + */ + submittedTimeoutMs: parseInt( + process.env.OUTBOX_SUBMITTED_TIMEOUT_MS || '90000' + ), + /** Fee multiplier applied on each fee-bump resubmission (compounds). */ + feeBumpMultiplier: parseFloat( + process.env.OUTBOX_FEE_BUMP_MULTIPLIER || '2' + ), + /** Hard cap on fee-bump resubmissions before a stuck op is escalated to FAILED. */ + feeBumpMaxAttempts: parseInt( + process.env.OUTBOX_FEE_BUMP_MAX_ATTEMPTS || '3' + ), + /** Global cap on ops in flight (claimed, not yet CONFIRMED/FAILED) at once. */ + globalMaxInFlight: parseInt( + process.env.OUTBOX_GLOBAL_MAX_IN_FLIGHT || '10' + ), + /** Per-signer (per Stellar account) cap on ops in flight at once. */ + perAccountMaxInFlight: parseInt( + process.env.OUTBOX_PER_ACCOUNT_MAX_IN_FLIGHT || '1' + ), + /** Ops claimed per dispatcher sweep, priority-ordered (see src/outbox/stateMachine.ts). */ + batchSize: parseInt(process.env.OUTBOX_BATCH_SIZE || '20'), + }, } diff --git a/src/controllers/transaction-controller.ts b/src/controllers/transaction-controller.ts index a313fc0..5e5c0ee 100644 --- a/src/controllers/transaction-controller.ts +++ b/src/controllers/transaction-controller.ts @@ -1,11 +1,104 @@ import { Request, Response } from 'express' import { Transaction } from '@prisma/client' import db from '../db' -import { depositForUser, withdrawForUser } from '../stellar/contract' import { formatDepositReply, formatWithdrawReply } from '../whatsapp/formatters' -import { sendNotFound, sendConflict, sendUnauthorized } from '../utils/errors' +import { sendNotFound, sendUnauthorized } from '../utils/errors' import { logger } from '../utils/logger' import { dispatchWebhookEvent } from '../services/webhookDispatcher' +import { enqueueOutboxOp } from '../outbox/service' +import { dispatchOne } from '../outbox/dispatcher' +import { deriveIdempotencyKey } from '../outbox/idempotency' +import { OutboxOpKind } from '../outbox/types' + +/** + * Persist the Transaction row (PENDING, no hash yet) and its outbox intent in + * the same DB transaction (#325) — "intent persisted" and "business state + * written" now commit or roll back together, then dispatch it inline so the + * HTTP/job caller still gets a synchronous CONFIRMED/FAILED result exactly as + * before this change. If the process crashes between commit and submission, + * the durable OutboxOp row survives for the background dispatcher + * (src/outbox/dispatcher.ts) to pick up on the next sweep. + */ +async function enqueueAndDispatch(params: { + kind: Extract + userId: string + userAddress: string + amount: number + assetSymbol: string + network: Transaction['network'] + type: 'DEPOSIT' | 'WITHDRAWAL' + protocolName?: string + memo?: string + actingAsUserId?: string | null +}): Promise { + const pending = await db.$transaction(async (tx) => { + const transaction = await tx.transaction.create({ + data: { + userId: params.userId, + actingAsUserId: params.actingAsUserId ?? null, + type: params.type, + status: 'PENDING', + assetSymbol: params.assetSymbol, + amount: params.amount, + network: params.network, + protocolName: params.protocolName, + memo: params.memo, + }, + }) + + const op = await enqueueOutboxOp(tx, { + idempotencyKey: deriveIdempotencyKey( + params.kind, + params.userId, + transaction.id + ), + userId: params.userId, + kind: params.kind, + actor: 'USER', + payload: + params.kind === 'DEPOSIT' + ? { + method: 'deposit', + userId: params.userId, + userAddress: params.userAddress, + amount: params.amount, + assetSymbol: params.assetSymbol, + transactionId: transaction.id, + } + : { + method: 'withdraw', + userId: params.userId, + userAddress: params.userAddress, + amount: params.amount, + assetSymbol: params.assetSymbol, + transactionId: transaction.id, + }, + }) + + return { transaction, opId: op.id } + }) + + try { + const result = await dispatchOne(pending.opId) + const succeeded = !result.status || result.status === 'success' + return db.transaction.update({ + where: { id: pending.transaction.id }, + data: { + txHash: result.hash, + status: succeeded ? 'CONFIRMED' : 'FAILED', + confirmedAt: succeeded ? new Date() : null, + }, + }) + } catch (err) { + await db.transaction + .update({ + where: { id: pending.transaction.id }, + data: { status: 'FAILED' }, + }) + .catch(() => {}) + throw err + } +} export interface ExecuteDepositParams { userId: string @@ -46,47 +139,25 @@ export async function executeDeposit( assetSymbol, }) - const onChainResult = await depositForUser( + const transaction = await enqueueAndDispatch({ + kind: 'DEPOSIT', userId, - walletAddress, + userAddress: walletAddress, amount, - assetSymbol - ) + assetSymbol, + network: user.network, + type: 'DEPOSIT', + memo, + actingAsUserId, + }) logger.info('On-chain deposit completed', { userId, - txHash: onChainResult.hash, - status: onChainResult.status, - }) - - const transactionStatus = - onChainResult.status === 'success' ? 'CONFIRMED' : 'FAILED' - - const existing = await db.transaction.findUnique({ - where: { txHash: onChainResult.hash }, - select: { id: true }, - }) - - if (existing) { - throw new Error('Duplicate transaction hash') - } - - const transaction = await db.transaction.create({ - data: { - userId, - actingAsUserId: actingAsUserId ?? null, - txHash: onChainResult.hash, - type: 'DEPOSIT', - status: transactionStatus, - assetSymbol, - amount, - network: user.network, - memo, - confirmedAt: transactionStatus === 'CONFIRMED' ? new Date() : null, - }, + txHash: transaction.txHash, + status: transaction.status, }) - if (transactionStatus === 'CONFIRMED') { + if (transaction.status === 'CONFIRMED') { dispatchWebhookEvent('transaction.confirmed', { txHash: transaction.txHash, type: 'DEPOSIT', @@ -97,7 +168,10 @@ export async function executeDeposit( }).catch(() => {}) } - return { transaction, status: transactionStatus } + return { + transaction, + status: transaction.status as 'CONFIRMED' | 'FAILED', + } } export async function processOnChainTransaction( @@ -137,50 +211,28 @@ export async function processOnChainTransaction( assetSymbol, }) - const onChainTransaction = await withdrawForUser( + const transaction = await enqueueAndDispatch({ + kind: 'WITHDRAW', userId, - req.auth!.walletAddress, + userAddress: req.auth!.walletAddress, amount, - assetSymbol - ) + assetSymbol, + network: user.network, + type, + protocolName, + memo, + actingAsUserId, + }) logger.info('On-chain withdrawal completed', { correlationId: req.correlationId, type, userId, - txHash: onChainTransaction.hash, - status: onChainTransaction.status, - }) - - const transactionStatus = - onChainTransaction.status === 'success' ? 'CONFIRMED' : 'FAILED' - - const existing = await db.transaction.findUnique({ - where: { txHash: onChainTransaction.hash }, - select: { id: true }, - }) - - if (existing) { - return sendConflict(res, 'Duplicate transaction hash') - } - - const transaction = await db.transaction.create({ - data: { - userId, - actingAsUserId, - txHash: onChainTransaction.hash, - type, - status: transactionStatus, - assetSymbol, - amount, - network: user.network, - protocolName, - memo, - confirmedAt: transactionStatus === 'CONFIRMED' ? new Date() : null, - }, + txHash: transaction.txHash, + status: transaction.status, }) - if (transactionStatus === 'CONFIRMED') { + if (transaction.status === 'CONFIRMED') { dispatchWebhookEvent('transaction.confirmed', { txHash: transaction.txHash, type, diff --git a/src/index.ts b/src/index.ts index 25d35f1..0b47cb0 100644 --- a/src/index.ts +++ b/src/index.ts @@ -53,6 +53,7 @@ import { scheduleAlertRules } from './jobs/alertRules' import { scheduleStrategyMetrics } from './jobs/strategyMetrics' import { scheduleAllocationSuggestions } from './jobs/allocationSuggestions' import { scheduleAttribution } from './jobs/attribution' +import { scheduleOutboxDispatcher } from './outbox/dispatcher' // Was never imported or started, so ProtocolRiskScore rows were never refreshed // after their first backfill. That matters beyond staleness: risk-ceiling // filtering is fail-closed (applyRiskCeiling treats an unknown score as @@ -119,6 +120,7 @@ let strategyMetricsHandle: NodeJS.Timeout | null = null let allocationSuggestionsHandle: NodeJS.Timeout | null = null let protocolRiskScoringHandle: NodeJS.Timeout | null = null let attributionHandle: NodeJS.Timeout | null = null +let outboxDispatcherHandle: NodeJS.Timeout | null = null function allServicesReady(): boolean { return Object.values(serviceStatus).every((s) => s.ready) @@ -399,6 +401,12 @@ async function gracefulShutdown(signal: string): Promise { logger.info('[Shutdown] Performance attribution timer cleared') } + if (outboxDispatcherHandle) { + clearInterval(outboxDispatcherHandle) + outboxDispatcherHandle = null + logger.info('[Shutdown] Outbox dispatcher timer cleared') + } + if (!httpServer) { logger.warn('[Shutdown] No HTTP server to close') process.exit(0) diff --git a/src/middleware/adminAuth.ts b/src/middleware/adminAuth.ts index b07d7ad..ef15697 100644 --- a/src/middleware/adminAuth.ts +++ b/src/middleware/adminAuth.ts @@ -28,6 +28,10 @@ export const ADMIN_SCOPES = [ 'keys:write', 'fiat:read', 'fiat:write', + // #325 — durable outbox admin tooling (list/inspect ops, force-retry FAILED, + // cancel unsent PENDING ops). + 'outbox:read', + 'outbox:write', 'super', ] as const export type AdminScope = (typeof ADMIN_SCOPES)[number] diff --git a/src/outbox/dispatcher.ts b/src/outbox/dispatcher.ts new file mode 100644 index 0000000..40efc65 --- /dev/null +++ b/src/outbox/dispatcher.ts @@ -0,0 +1,338 @@ +/** + * 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. + * + * 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 { dispatchWebhookEvent } from '../services/webhookDispatcher' +import { TransactionResult } from '../stellar/types' +import { getSignerLock } from './signerLock' +import { resolveSignerPublicKey, executeOutboxPayload } from './executors' +import { sortForDispatch } from './stateMachine' +import { OutboxOpRecord } from './types' +import { + claimOp, + findClaimableOps, + findStuckSubmittedOps, + getOp, + getQueueStats, + isUserHalted, + markConfirmed, + markFailedOrRetry, + markFailedTerminal, + mirrorLinkedTransaction, + returnStuckOpToPending, +} from './service' +import { + recordOutboxOp, + recordOutboxFeeBump, + recordOutboxLatency, + updateOutboxQueueDepth, + updateOutboxStuckSubmitted, +} from '../utils/metrics' + +function signerLock() { + return getSignerLock( + config.outbox.globalMaxInFlight, + config.outbox.perAccountMaxInFlight + ) +} + +/** + * 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 { + const bumps = Math.min( + Math.max(attempts - 1, 0), + config.outbox.feeBumpMaxAttempts + ) + return config.outbox.feeBumpMultiplier ** bumps +} + +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 dispatchWebhookEvent('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 { + const feeMultiplier = computeFeeMultiplier(op.attempts) + const lock = signerLock() + + try { + const result = await lock.withLock(op.signerPublicKey!, () => + executeOutboxPayload(op.payload, feeMultiplier) + ) + + if (!result.status || result.status === 'success') { + await markConfirmed(op.id, result.hash) + await mirrorLinkedTransaction(op.payload, { + txHash: result.hash, + status: 'CONFIRMED', + }) + recordOutboxOp(op.kind, op.priority, 'confirmed') + 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 { + 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) + const { terminal } = await markFailedOrRetry(op, message) + recordOutboxOp(op.kind, op.priority, terminal ? 'failed' : 'retry') + + if (terminal) { + 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, + }) + } + + throw err + } +} + +/** + * 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. + */ +export async function dispatchOne(opId: string): Promise { + const op = await getOp(opId) + if (!op) throw new Error(`Outbox op ${opId} not found`) + + 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})`) + } + + if (await isUserHalted(op.userId)) { + throw new Error( + `Outbox op ${opId} not dispatched: user ${op.userId} is frozen` + ) + } + + const signerPublicKey = await resolveSignerPublicKey(op.payload, op.userId) + const claimed = await claimOp(opId, signerPublicKey) + if (!claimed) { + 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 { + dispatchOne(opId).catch((err) => { + logger.warn('[Outbox] Background dispatch attempt failed (retriable)', { + opId, + error: err instanceof Error ? err.message : String(err), + }) + }) +} + +/** 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) + + for (const op of stuck) { + 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 + } + + 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 { + await reconcileStuckSubmitted() + + const stats = await getQueueStats() + updateOutboxQueueDepth(stats) + + const claimable = sortForDispatch( + await findClaimableOps(config.outbox.batchSize) + ) + + for (const op of claimable) { + if (await isUserHalted(op.userId)) { + logger.info('[Outbox] Skipping op for frozen user', { + opId: op.id, + userId: op.userId, + }) + continue + } + + 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 + } + + 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. + }) + } +} + +let dispatcherHandle: NodeJS.Timeout | null = null + +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), + }) + ) + }, config.outbox.dispatchIntervalMs) + + logger.info('[Outbox] Dispatcher scheduled', { + intervalMs: config.outbox.dispatchIntervalMs, + }) + return dispatcherHandle +} diff --git a/src/outbox/executors.ts b/src/outbox/executors.ts new file mode 100644 index 0000000..53c8547 --- /dev/null +++ b/src/outbox/executors.ts @@ -0,0 +1,88 @@ +/** + * The single point where a durable OutboxOp becomes a real on-chain + * submission (#325). + * + * 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. + */ + +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' + +/** + * 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. + */ +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() + } +} + +/** + * Perform the actual on-chain submission for a claimed op. `feeMultiplier` + * implements the dispatcher's fee-bump-on-congestion retry strategy (docs/OUTBOX.md). + */ +export async function executeOutboxPayload( + payload: OutboxPayload, + feeMultiplier: number = 1 +): Promise { + switch (payload.method) { + case 'deposit': + return depositForUser( + payload.userId, + payload.userAddress, + payload.amount, + payload.assetSymbol, + feeMultiplier + ) + case 'withdraw': + return withdrawForUser( + payload.userId, + payload.userAddress, + payload.amount, + payload.assetSymbol, + feeMultiplier + ) + case 'rebalance': + return submitRebalance( + payload.toProtocol, + payload.expectedApyBasisPoints, + feeMultiplier + ) + case 'referral_reward': + return payReferralReward( + payload.recipientAddress, + payload.amount, + payload.assetSymbol, + feeMultiplier + ) + } +} diff --git a/src/outbox/idempotency.ts b/src/outbox/idempotency.ts new file mode 100644 index 0000000..56096ea --- /dev/null +++ b/src/outbox/idempotency.ts @@ -0,0 +1,23 @@ +import { OutboxOpKind } from './types' + +/** + * Deterministic idempotency anchor for an outbox op: `kind:userId:businessRecordId`. + * + * businessRecordId is whatever row this op is the durable intent for — a + * Transaction id for DEPOSIT/WITHDRAW/REBALANCE, a ReferralConversion leg for + * REFERRAL_REWARD. Reusing the SAME businessRecordId (e.g. retrying against an + * already-PENDING Transaction) resolves to the same key, so + * src/outbox/service.ts's upsert-by-key never creates a second op for work + * that is already durably queued. + */ +export function deriveIdempotencyKey( + kind: OutboxOpKind, + userId: string, + businessRecordId: string +): string { + if (!userId) throw new Error('deriveIdempotencyKey: userId is required') + if (!businessRecordId) { + throw new Error('deriveIdempotencyKey: businessRecordId is required') + } + return `${kind}:${userId}:${businessRecordId}` +} diff --git a/src/outbox/service.ts b/src/outbox/service.ts new file mode 100644 index 0000000..8bb422e --- /dev/null +++ b/src/outbox/service.ts @@ -0,0 +1,427 @@ +/** + * DB-facing outbox operations (#325): transactional enqueue, atomic claim, + * and the terminal/retry transitions the dispatcher drives ops through. + * + * Every write here goes through the pure rules in src/outbox/stateMachine.ts + * first — this module is deliberately thin glue between that and Prisma. + */ + +import { Prisma, OutboxOpStatus as PrismaOutboxOpStatus } from '@prisma/client' +import db from '../db' +import { logger } from '../utils/logger' +import { config } from '../config/env' +import { assertTransition, computeBackoffMs } from './stateMachine' +import { + OutboxOpKind, + OutboxOpActor, + OutboxOpRecord, + OutboxOpStatus, + OutboxPayload, + OutboxPriority, + PRIORITY_BY_KIND, +} from './types' + +type DbClient = typeof db | Prisma.TransactionClient + +function toRecord(row: { + id: string + idempotencyKey: string + userId: string + kind: string + actor: string + payload: Prisma.JsonValue + priority: string + status: string + txHash: string | null + attempts: number + nextAttemptAt: Date | null + error: string | null + submittedAt: Date | null + confirmedAt: Date | null + createdAt: Date + updatedAt: Date + signerPublicKey: string | null +}): OutboxOpRecord { + return { + id: row.id, + idempotencyKey: row.idempotencyKey, + userId: row.userId, + kind: row.kind as OutboxOpKind, + actor: row.actor as OutboxOpActor, + payload: row.payload as unknown as OutboxPayload, + priority: row.priority as OutboxPriority, + status: row.status as OutboxOpStatus, + txHash: row.txHash, + attempts: row.attempts, + nextAttemptAt: row.nextAttemptAt, + error: row.error, + submittedAt: row.submittedAt, + confirmedAt: row.confirmedAt, + createdAt: row.createdAt, + updatedAt: row.updatedAt, + signerPublicKey: row.signerPublicKey, + } +} + +/** + * Write the durable intent. Call this INSIDE the same `db.$transaction` as + * whatever business row the caller derives (a Transaction row, a + * ReferralConversion leg, ...) — that is the atomicity guarantee #325 adds: + * "intent persisted" and "business state written" commit or roll back + * together. + * + * Idempotent by `idempotencyKey`: a caller that re-runs against the same + * business record (e.g. a retried job tick) gets back the SAME op row rather + * than a duplicate, so it is safe to call unconditionally. + */ +export async function enqueueOutboxOp( + tx: DbClient, + params: { + idempotencyKey: string + userId: string + kind: OutboxOpKind + actor: OutboxOpActor + payload: OutboxPayload + priority?: OutboxPriority + } +): Promise { + const existing = await tx.outboxOp.findUnique({ + where: { idempotencyKey: params.idempotencyKey }, + }) + if (existing) { + return toRecord(existing) + } + + const created = await tx.outboxOp.create({ + data: { + idempotencyKey: params.idempotencyKey, + userId: params.userId, + kind: params.kind, + actor: params.actor, + payload: params.payload as unknown as Prisma.InputJsonValue, + priority: params.priority ?? PRIORITY_BY_KIND[params.kind], + status: 'PENDING', + }, + }) + return toRecord(created) +} + +/** + * Atomically claim a PENDING op for submission: PENDING -> SUBMITTED via a + * conditional update. If two dispatchers (or a synchronous inline caller and + * a background sweep) race for the same op, exactly one `updateMany` call + * sees `count === 1`; the loser gets `null` back and simply moves on — see + * tests/integration/outbox/dispatcher.integration.test.ts. + */ +export async function claimOp( + opId: string, + signerPublicKey: string +): Promise { + assertTransition('PENDING', 'SUBMITTED') + + const result = await db.outboxOp.updateMany({ + where: { id: opId, status: 'PENDING' as PrismaOutboxOpStatus }, + data: { + status: 'SUBMITTED' as PrismaOutboxOpStatus, + attempts: { increment: 1 }, + submittedAt: new Date(), + signerPublicKey, + }, + }) + + if (result.count === 0) { + return null + } + + const row = await db.outboxOp.findUnique({ where: { id: opId } }) + return row ? toRecord(row) : null +} + +export async function markConfirmed( + opId: string, + txHash: string +): Promise { + assertTransition('SUBMITTED', 'CONFIRMED') + await db.outboxOp.update({ + where: { id: opId }, + data: { + status: 'CONFIRMED' as PrismaOutboxOpStatus, + txHash, + confirmedAt: new Date(), + error: null, + }, + }) +} + +/** + * A submit attempt failed. If attempts remain, return the op to PENDING with + * a full-jitter backoff `nextAttemptAt` so the dispatcher retries it later; + * once attempts are exhausted, move it to the terminal FAILED state instead. + * Returns whether the op is now terminal. + */ +export async function markFailedOrRetry( + op: Pick, + errorMessage: string +): Promise<{ terminal: boolean }> { + const terminal = op.attempts >= config.outbox.maxAttempts + + if (terminal) { + assertTransition('SUBMITTED', 'FAILED') + await db.outboxOp.update({ + where: { id: op.id }, + data: { + status: 'FAILED' as PrismaOutboxOpStatus, + error: errorMessage.slice(0, 2000), + }, + }) + return { terminal: true } + } + + assertTransition('SUBMITTED', 'PENDING') + const backoffMs = computeBackoffMs( + op.attempts, + config.outbox.backoffBaseMs, + config.outbox.backoffMaxMs + ) + await db.outboxOp.update({ + where: { id: op.id }, + data: { + status: 'PENDING' as PrismaOutboxOpStatus, + error: errorMessage.slice(0, 2000), + nextAttemptAt: new Date(Date.now() + backoffMs), + }, + }) + return { terminal: false } +} + +/** + * The on-chain submission resolved (no exception) but the network itself + * rejected the operation (TransactionResult.status === 'failed') — e.g. a + * vault-contract precondition failure. This is not a transient + * network/congestion error, so unlike markFailedOrRetry it does not retry: + * it records the txHash for audit and moves straight to the terminal FAILED + * state, matching the single-attempt behavior the deposit/withdraw routes + * had before #325. + */ +export async function markFailedTerminal( + opId: string, + txHash: string, + errorMessage: string +): Promise { + assertTransition('SUBMITTED', 'FAILED') + await db.outboxOp.update({ + where: { id: opId }, + data: { + status: 'FAILED' as PrismaOutboxOpStatus, + txHash, + error: errorMessage.slice(0, 2000), + }, + }) +} + +/** Admin-only: force a FAILED op back to PENDING, clearing backoff. */ +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`) + assertTransition(op.status as OutboxOpStatus, 'PENDING') + + const updated = await db.outboxOp.update({ + where: { id: opId }, + data: { + status: 'PENDING' as PrismaOutboxOpStatus, + error: null, + nextAttemptAt: null, + }, + }) + logger.info('[Outbox] Admin force-retry', { opId }) + return toRecord(updated) +} + +/** Cancel an unsent op. Only PENDING ops can be cancelled — once SUBMITTED, it's on-chain. */ +export async function cancelOp(opId: string): Promise { + const result = await db.outboxOp.updateMany({ + where: { id: opId, status: 'PENDING' as PrismaOutboxOpStatus }, + data: { status: 'CANCELLED' as PrismaOutboxOpStatus }, + }) + if (result.count === 0) { + const existing = await db.outboxOp.findUnique({ where: { id: opId } }) + if (!existing) throw new Error(`Outbox op ${opId} not found`) + throw new Error( + `Outbox op ${opId} is ${existing.status}, not PENDING — only unsent ops can be cancelled` + ) + } + const row = await db.outboxOp.findUnique({ where: { id: opId } }) + logger.info('[Outbox] Admin cancel', { opId }) + return toRecord(row!) +} + +export async function getOp(opId: string): Promise { + const row = await db.outboxOp.findUnique({ where: { id: opId } }) + return row ? toRecord(row) : null +} + +export async function findOpByTxHash( + txHash: string +): Promise { + const row = await db.outboxOp.findUnique({ where: { txHash } }) + return row ? toRecord(row) : null +} + +export interface ListOpsFilter { + status?: OutboxOpStatus + kind?: OutboxOpKind + priority?: OutboxPriority + userId?: string + limit?: number + offset?: number +} + +export async function listOps( + filter: ListOpsFilter = {} +): Promise<{ ops: OutboxOpRecord[]; total: number }> { + const where: Prisma.OutboxOpWhereInput = { + status: filter.status as PrismaOutboxOpStatus | undefined, + kind: filter.kind, + priority: filter.priority, + userId: filter.userId, + } + const limit = Math.min(Math.max(filter.limit ?? 50, 1), 500) + const offset = Math.max(filter.offset ?? 0, 0) + + const [rows, total] = await Promise.all([ + db.outboxOp.findMany({ + where, + orderBy: { createdAt: 'desc' }, + take: limit, + skip: offset, + }), + db.outboxOp.count({ where }), + ]) + + return { ops: rows.map(toRecord), total } +} + +/** Queue depth by status/priority — feeds the admin throughput view + Prometheus gauge. */ +export async function getQueueStats(): Promise< + Array<{ status: OutboxOpStatus; priority: OutboxPriority; count: number }> +> { + const grouped = await db.outboxOp.groupBy({ + by: ['status', 'priority'], + _count: { _all: true }, + }) + return grouped.map((g) => ({ + status: g.status as OutboxOpStatus, + priority: g.priority as OutboxPriority, + count: g._count._all, + })) +} + +/** + * PENDING ops eligible right now: never attempted, or past their backoff + * 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. + */ +export async function findClaimableOps( + limit: number +): Promise { + const now = new Date() + const rows = await db.outboxOp.findMany({ + where: { + status: 'PENDING' as PrismaOutboxOpStatus, + OR: [{ nextAttemptAt: null }, { nextAttemptAt: { lte: now } }], + }, + take: limit, + }) + return rows.map(toRecord) +} + +/** + * Escalation step for a SUBMITTED op stuck past the confirmation timeout: + * return it to PENDING, immediately eligible for reclaim. The next claim + * increments `attempts`, which is what drives the fee-bump multiplier on + * resubmission (src/outbox/dispatcher.ts#computeFeeMultiplier). + */ +export async function returnStuckOpToPending(opId: string): Promise { + assertTransition('SUBMITTED', 'PENDING') + await db.outboxOp.updateMany({ + where: { id: opId, status: 'SUBMITTED' as PrismaOutboxOpStatus }, + data: { + status: 'PENDING' as PrismaOutboxOpStatus, + nextAttemptAt: new Date(), + }, + }) +} + +/** SUBMITTED ops that have been quiet longer than the confirmation timeout. */ +export async function findStuckSubmittedOps( + timeoutMs: number +): Promise { + const cutoff = new Date(Date.now() - timeoutMs) + const rows = await db.outboxOp.findMany({ + where: { + status: 'SUBMITTED' as PrismaOutboxOpStatus, + submittedAt: { lte: cutoff }, + }, + }) + return rows.map(toRecord) +} + +/** + * Mirror a definitive outcome back onto the linked Transaction row + * (DEPOSIT/WITHDRAW/REBALANCE payloads carry `transactionId`; REFERRAL_REWARD + * does not and is a no-op here — its caller updates its Transaction row + * itself since it always awaits dispatchOne synchronously). + * + * This is what closes the loop for a NON-blocking dispatch (an agent + * rebalance via dispatchInBackground) and for the crash-recovery case where + * the background sweep — not the original synchronous caller, which may have + * already returned an error response — is what eventually confirms an op. + * Idempotent: safe to also run from a synchronous caller that just updated + * the same row itself. + */ +export async function mirrorLinkedTransaction( + payload: OutboxPayload, + outcome: { txHash?: string; status: 'CONFIRMED' | 'FAILED' } +): Promise { + if (!('transactionId' in payload)) return + await db.transaction.updateMany({ + where: { id: payload.transactionId }, + data: { + txHash: outcome.txHash, // undefined -> Prisma leaves the column untouched + status: outcome.status, + confirmedAt: outcome.status === 'CONFIRMED' ? new Date() : null, + }, + }) +} + +/** + * Confirmation fallback oracle (#325): called from src/stellar/events.ts on + * the SAME `tx` handle that just confirmed a Transaction row by txHash. If a + * SUBMITTED OutboxOp is still waiting on that hash — the dispatcher process + * crashed after submitting but before observing its own confirmation — this + * is what closes it out instead of leaving it SUBMITTED forever. A no-op + * (0 rows) for any txHash with no matching op, e.g. legacy/manual + * transactions never created through the outbox. + */ +export async function reconcileOutboxOpByTxHash( + tx: DbClient, + txHash: string +): Promise { + await tx.outboxOp.updateMany({ + where: { txHash, status: 'SUBMITTED' as PrismaOutboxOpStatus }, + data: { + status: 'CONFIRMED' as PrismaOutboxOpStatus, + confirmedAt: new Date(), + }, + }) +} + +/** The central freeze guard the dispatcher must consult before dispatching an op (#325). */ +export async function isUserHalted(userId: string): Promise { + const user = await db.user.findUnique({ + where: { id: userId }, + select: { isActive: true }, + }) + return !user || user.isActive === false +} diff --git a/src/outbox/signerLock.ts b/src/outbox/signerLock.ts new file mode 100644 index 0000000..34d7316 --- /dev/null +++ b/src/outbox/signerLock.ts @@ -0,0 +1,86 @@ +/** + * Per-signer serialization + in-flight concurrency caps for the dispatcher (#325). + * + * The agent-signed wallet (and any custodial user wallet) submits Stellar + * transactions using that account's sequence number: two ops signed by the + * same key must never be in flight at once, or the second submission's + * sequence number will already be stale. This is an in-process mutex — + * sufficient because a single dispatcher process claims ops (the atomic + * PENDING -> SUBMITTED conditional update in src/outbox/service.ts is what + * makes claiming itself safe across multiple dispatcher processes; running + * more than one dispatcher process is out of scope for this change, and is + * called out in docs/OUTBOX.md). + */ + +class SignerLock { + private queues = new Map>() + private globalInFlight = 0 + private perAccountInFlight = new Map() + + constructor( + private readonly globalMaxInFlight: number, + private readonly perAccountMaxInFlight: number + ) {} + + /** True if claiming one more op for this signer would exceed either cap. */ + hasCapacity(signerPublicKey: string): boolean { + if (this.globalInFlight >= this.globalMaxInFlight) return false + const accountInFlight = this.perAccountInFlight.get(signerPublicKey) ?? 0 + return accountInFlight < this.perAccountMaxInFlight + } + + /** + * Run `fn` holding the mutex for `signerPublicKey`. Ops for the same signer + * queue and run strictly one-at-a-time, in call order; ops for different + * signers run concurrently (subject to the global cap). + */ + async withLock(signerPublicKey: string, fn: () => Promise): Promise { + const previousTail = this.queues.get(signerPublicKey) ?? Promise.resolve() + + let releaseNext: () => void + const myTail = new Promise((resolve) => { + releaseNext = resolve + }) + // Every future caller for this signer waits on `previousTail` (our turn) + // followed by `myTail` (our own completion) before they get to run. + this.queues.set( + signerPublicKey, + previousTail.then(() => myTail) + ) + + await previousTail + + this.globalInFlight++ + this.perAccountInFlight.set( + signerPublicKey, + (this.perAccountInFlight.get(signerPublicKey) ?? 0) + 1 + ) + + try { + return await fn() + } finally { + this.globalInFlight-- + const remaining = (this.perAccountInFlight.get(signerPublicKey) ?? 1) - 1 + if (remaining <= 0) { + this.perAccountInFlight.delete(signerPublicKey) + } else { + this.perAccountInFlight.set(signerPublicKey, remaining) + } + releaseNext!() + } + } +} + +let instance: SignerLock | null = null + +export function getSignerLock( + globalMaxInFlight: number, + perAccountMaxInFlight: number +): SignerLock { + if (!instance) { + instance = new SignerLock(globalMaxInFlight, perAccountMaxInFlight) + } + return instance +} + +export { SignerLock } diff --git a/src/outbox/stateMachine.ts b/src/outbox/stateMachine.ts new file mode 100644 index 0000000..1a1248f --- /dev/null +++ b/src/outbox/stateMachine.ts @@ -0,0 +1,88 @@ +/** + * Pure state-machine + priority-ordering rules for the outbox (#325). + * + * No I/O, no Prisma import — this module is the single source of truth for + * "is this transition legal" and "in what order should PENDING ops be + * dispatched," and is exercised directly by + * tests/unit/outbox/stateMachine.test.ts without a database. + */ + +import { OutboxOpStatus, OutboxOpRecord, OutboxPriority } from './types' + +/** + * Legal transitions out of each status. PENDING is re-entrant (a transient + * 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). + */ +const VALID_TRANSITIONS: Record = { + PENDING: ['SUBMITTED', 'CANCELLED', 'FAILED'], + SUBMITTED: ['CONFIRMED', 'PENDING', 'FAILED'], + CONFIRMED: [], + FAILED: ['PENDING'], // admin force-retry only (src/outbox/service.ts) + CANCELLED: [], +} + +export function canTransition( + from: OutboxOpStatus, + to: OutboxOpStatus +): boolean { + return VALID_TRANSITIONS[from]?.includes(to) ?? false +} + +export function assertTransition( + from: OutboxOpStatus, + to: OutboxOpStatus +): void { + if (!canTransition(from, to)) { + throw new Error(`Illegal outbox transition: ${from} -> ${to}`) + } +} + +/** Lower number = dispatched first. */ +export const PRIORITY_WEIGHT: Record = { + CRITICAL: 0, + NORMAL: 1, + LOW: 2, +} + +/** + * Ordering for claiming PENDING ops: priority first, then FIFO (oldest + * createdAt first) within a priority tier. + * + * This is what keeps a CRITICAL withdrawal from starving behind a NORMAL or + * LOW wave (e.g. a burst of agent rebalances or recurring-deposit runs): a + * CRITICAL op sorts ahead of every NORMAL/LOW op regardless of how much + * longer they have been queued. See + * tests/unit/outbox/stateMachine.test.ts ("does not starve a CRITICAL op + * behind a wave of NORMAL ops"). + */ +export function compareForDispatch( + a: Pick, + b: Pick +): number { + const weightDiff = PRIORITY_WEIGHT[a.priority] - PRIORITY_WEIGHT[b.priority] + if (weightDiff !== 0) return weightDiff + return a.createdAt.getTime() - b.createdAt.getTime() +} + +export function sortForDispatch< + T extends Pick, +>(ops: T[]): T[] { + return [...ops].sort(compareForDispatch) +} + +/** + * 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. + */ +export function computeBackoffMs( + attempt: number, + baseMs: number, + maxMs: number, + random: () => number = Math.random +): number { + const capped = Math.min(maxMs, baseMs * 2 ** Math.max(0, attempt - 1)) + return Math.floor(random() * capped) +} diff --git a/src/outbox/types.ts b/src/outbox/types.ts new file mode 100644 index 0000000..a741e15 --- /dev/null +++ b/src/outbox/types.ts @@ -0,0 +1,88 @@ +/** + * Shared types for the durable outbox (#325). + * + * Kept free of Prisma/DB imports so src/outbox/stateMachine.ts and + * src/outbox/priority.ts stay pure and unit-testable without a database. + */ + +export type OutboxOpKind = + | 'DEPOSIT' + | 'WITHDRAW' + | 'REBALANCE' + | 'RECURRING_DEPOSIT' + | 'REFERRAL_REWARD' + | 'YIELD_CLAIM' + +export type OutboxOpActor = 'USER' | 'AGENT' | 'SYSTEM' + +export type OutboxPriority = 'CRITICAL' | 'NORMAL' | 'LOW' + +export type OutboxOpStatus = + 'PENDING' | 'SUBMITTED' | 'CONFIRMED' | 'FAILED' | 'CANCELLED' + +/** + * The exact, validated operation the dispatcher will submit. Mirrors the + * arguments the equivalent src/stellar/contract.ts write function already + * takes — never a key, never raw unvalidated request input. + */ +export type OutboxPayload = + | { + method: 'deposit' + userId: string + userAddress: string + amount: number + assetSymbol: string + transactionId: string + } + | { + method: 'withdraw' + userId: string + userAddress: string + amount: number + assetSymbol: string + transactionId: string + } + | { + method: 'rebalance' + toProtocol: string + expectedApyBasisPoints: number + transactionId: string + } + | { + method: 'referral_reward' + recipientAddress: string + amount: number + assetSymbol: string + conversionId: string + leg: 'owner' | 'referred' + } + +export interface OutboxOpRecord { + id: string + idempotencyKey: string + userId: string + kind: OutboxOpKind + actor: OutboxOpActor + payload: OutboxPayload + priority: OutboxPriority + status: OutboxOpStatus + txHash: string | null + attempts: number + nextAttemptAt: Date | null + error: string | null + submittedAt: Date | null + confirmedAt: Date | null + createdAt: Date + updatedAt: Date + signerPublicKey: string | null +} + +/** Priority classification for each kind, per the #325 design. */ +export const PRIORITY_BY_KIND: Record = { + WITHDRAW: 'CRITICAL', + DEPOSIT: 'NORMAL', + RECURRING_DEPOSIT: 'NORMAL', + REFERRAL_REWARD: 'NORMAL', + YIELD_CLAIM: 'NORMAL', + REBALANCE: 'LOW', +} diff --git a/src/referral/service.ts b/src/referral/service.ts index 907b036..cad002f 100644 --- a/src/referral/service.ts +++ b/src/referral/service.ts @@ -27,8 +27,10 @@ import db from '../db' import { config } from '../config' import { logger } from '../utils/logger' import { alertingService } from '../services/alerting' -import { payReferralReward } from '../stellar/contract' import { getWalletByUserId } from '../stellar/wallet' +import { enqueueOutboxOp } from '../outbox/service' +import { dispatchOne } from '../outbox/dispatcher' +import { deriveIdempotencyKey } from '../outbox/idempotency' type Db = typeof db | Prisma.TransactionClient @@ -223,14 +225,22 @@ async function resolveRewardAddress(userId: string): Promise { } /** - * Pay one reward leg and record it as a distinctly-typed REFERRAL_REWARD - * Transaction. Returns the Transaction.id, or throws so the caller leaves the - * conversion retriable. + * Pay one reward leg through the durable outbox (#325) and record it as a + * distinctly-typed REFERRAL_REWARD Transaction. Returns the Transaction.id, + * or throws so the caller leaves the conversion retriable. + * + * The Transaction row and its OutboxOp intent are written in the same DB + * transaction — a crash between "persisted" and "submitted" leaves a durable, + * retriable record instead of nothing (idempotency key: + * REFERRAL_REWARD:::, so re-running this + * for the same conversion leg is safe). */ async function payOneReward( recipientUserId: string, amount: number, - network: Network + network: Network, + conversionId: string, + leg: 'owner' | 'referred' ): Promise { const address = await resolveRewardAddress(recipientUserId) if (!address) { @@ -238,22 +248,68 @@ async function payOneReward( } const asset = config.referral.rewardAsset - const result = await payReferralReward(address, amount, asset) - const tx = await db.transaction.create({ - data: { + const pending = await db.$transaction(async (tx) => { + const transaction = await tx.transaction.create({ + data: { + userId: recipientUserId, + type: TransactionType.REFERRAL_REWARD, + status: TransactionStatus.PENDING, + assetSymbol: asset, + amount: new Decimal(amount), + network, + memo: 'Referral reward', + }, + }) + + const op = await enqueueOutboxOp(tx, { + idempotencyKey: deriveIdempotencyKey( + 'REFERRAL_REWARD', + recipientUserId, + `${conversionId}:${leg}` + ), userId: recipientUserId, - txHash: result.hash, - type: TransactionType.REFERRAL_REWARD, - status: TransactionStatus.CONFIRMED, - assetSymbol: asset, - amount: new Decimal(amount), - network, - memo: 'Referral reward', - confirmedAt: new Date(), - }, + kind: 'REFERRAL_REWARD', + actor: 'SYSTEM', + payload: { + method: 'referral_reward', + recipientAddress: address, + amount, + assetSymbol: asset, + conversionId, + leg, + }, + }) + + return { transaction, opId: op.id } }) - return tx.id + + try { + const result = await dispatchOne(pending.opId) + const succeeded = !result.status || result.status === 'success' + await db.transaction.update({ + where: { id: pending.transaction.id }, + data: { + txHash: result.hash, + status: succeeded + ? TransactionStatus.CONFIRMED + : TransactionStatus.FAILED, + confirmedAt: succeeded ? new Date() : null, + }, + }) + if (!succeeded) { + throw new Error('On-chain reward submission returned status=failed') + } + return pending.transaction.id + } catch (err) { + await db.transaction + .update({ + where: { id: pending.transaction.id }, + data: { status: TransactionStatus.FAILED }, + }) + .catch(() => {}) + throw err + } } /** @@ -296,7 +352,9 @@ export async function payoutActivatedConversions(): Promise<{ ownerRewardTxId = await payOneReward( ownerUserId, config.referral.ownerReward, - network + network, + conversion.id, + 'owner' ) await db.referralConversion.update({ where: { id: conversion.id }, @@ -314,7 +372,9 @@ export async function payoutActivatedConversions(): Promise<{ referredRewardTxId = await payOneReward( referredUserId, config.referral.referredReward, - network + network, + conversion.id, + 'referred' ) await db.referralConversion.update({ where: { id: conversion.id }, diff --git a/src/routes/admin.ts b/src/routes/admin.ts index 13bc071..ff6026b 100644 --- a/src/routes/admin.ts +++ b/src/routes/admin.ts @@ -910,4 +910,158 @@ router.post( } ) +// ── Durable outbox admin tooling (#325) ───────────────────────────────────── + +/** + * GET /api/admin/outbox + * List/query outbox ops with filters. Required scope: outbox:read + * + * Query params: status, kind, priority, userId, limit (max 500), offset + */ +router.get( + '/outbox', + requireAdminScope('outbox:read'), + async (req: Request, res: Response) => { + try { + const { status, kind, priority, userId, limit, offset } = req.query + const { listOps } = await import('../outbox/service') + + const result = await listOps({ + status: status as any, + kind: kind as any, + priority: priority as any, + userId: userId as string | undefined, + limit: limit ? parseInt(limit as string, 10) : undefined, + offset: offset ? parseInt(offset as string, 10) : undefined, + }) + + auditLog(req, res, 'OUTBOX_LIST', 'success', { + status, + kind, + priority, + userId, + returned: result.ops.length, + }) + + res.status(200).json({ + success: true, + data: result, + timestamp: new Date().toISOString(), + }) + } catch (error) { + const message = error instanceof Error ? error.message : 'Unknown error' + auditLog(req, res, 'OUTBOX_LIST', 'failure', { error: message }) + res + .status(500) + .json({ success: false, error: 'Failed to list outbox ops' }) + } + } +) + +/** + * GET /api/admin/outbox/stats + * Queue depth by status/priority — throughput view. Required scope: outbox:read + */ +router.get( + '/outbox/stats', + requireAdminScope('outbox:read'), + async (req: Request, res: Response) => { + try { + const { getQueueStats } = await import('../outbox/service') + const stats = await getQueueStats() + auditLog(req, res, 'OUTBOX_STATS', 'success') + res.status(200).json({ + success: true, + data: { stats }, + timestamp: new Date().toISOString(), + }) + } catch (error) { + const message = error instanceof Error ? error.message : 'Unknown error' + auditLog(req, res, 'OUTBOX_STATS', 'failure', { error: message }) + res + .status(500) + .json({ success: false, error: 'Failed to get outbox stats' }) + } + } +) + +/** + * GET /api/admin/outbox/:id + * Inspect a single outbox op. Required scope: outbox:read + */ +router.get( + '/outbox/:id', + requireAdminScope('outbox:read'), + async (req: Request, res: Response) => { + try { + const { getOp } = await import('../outbox/service') + const op = await getOp(req.params.id) + if (!op) { + auditLog(req, res, 'OUTBOX_GET', 'failure', { + opId: req.params.id, + error: 'not_found', + }) + return res + .status(404) + .json({ success: false, error: 'Outbox op not found' }) + } + auditLog(req, res, 'OUTBOX_GET', 'success', { opId: op.id }) + res.status(200).json({ success: true, data: op }) + } catch (error) { + const message = error instanceof Error ? error.message : 'Unknown error' + auditLog(req, res, 'OUTBOX_GET', 'failure', { error: message }) + res.status(500).json({ success: false, error: 'Failed to get outbox op' }) + } + } +) + +/** + * POST /api/admin/outbox/:id/retry + * Force a FAILED op back to PENDING for the dispatcher to re-attempt. + * Required scope: outbox:write + */ +router.post( + '/outbox/:id/retry', + requireAdminScope('outbox:write'), + async (req: Request, res: Response) => { + try { + const { forceRetry } = await import('../outbox/service') + const op = await forceRetry(req.params.id) + auditLog(req, res, 'OUTBOX_FORCE_RETRY', 'success', { opId: op.id }) + res.status(200).json({ success: true, data: op }) + } catch (error) { + const message = error instanceof Error ? error.message : 'Unknown error' + auditLog(req, res, 'OUTBOX_FORCE_RETRY', 'failure', { + opId: req.params.id, + error: message, + }) + res.status(400).json({ success: false, error: message }) + } + } +) + +/** + * POST /api/admin/outbox/:id/cancel + * Cancel an unsent (PENDING only) op. Required scope: outbox:write + */ +router.post( + '/outbox/:id/cancel', + requireAdminScope('outbox:write'), + async (req: Request, res: Response) => { + try { + const { cancelOp } = await import('../outbox/service') + const op = await cancelOp(req.params.id) + auditLog(req, res, 'OUTBOX_CANCEL', 'success', { opId: op.id }) + res.status(200).json({ success: true, data: op }) + } catch (error) { + const message = error instanceof Error ? error.message : 'Unknown error' + auditLog(req, res, 'OUTBOX_CANCEL', 'failure', { + opId: req.params.id, + error: message, + }) + res.status(400).json({ success: false, error: message }) + } + } +) + export default router diff --git a/src/stellar/contract.ts b/src/stellar/contract.ts index ef84364..7df3070 100644 --- a/src/stellar/contract.ts +++ b/src/stellar/contract.ts @@ -39,19 +39,30 @@ function getVaultContract(): Contract { } /** - * Build contract invocation transaction + * Build contract invocation transaction. + * + * `feeMultiplier` scales the base fee (default 1x). The outbox dispatcher + * (#325) resubmits a congested-network op at a higher multiplier — fee-bump + * strategy documented in docs/OUTBOX.md — up to a configured cap. */ async function buildContractCall( method: string, args: xdr.ScVal[], - sourcePublicKey: string = getAgentKeypair().publicKey() + sourcePublicKey: string = getAgentKeypair().publicKey(), + feeMultiplier: number = 1 ): Promise { const server = getRpcServer() const contract = getVaultContract() const account = await getAccount(sourcePublicKey) + const fee = + feeMultiplier === 1 + ? BASE_FEE + : String( + BigInt(BASE_FEE) * BigInt(Math.max(1, Math.round(feeMultiplier))) + ) const tx = new TransactionBuilder(account, { - fee: BASE_FEE, + fee, networkPassphrase: getNetworkPassphrase(), }) .addOperation(contract.call(method, ...args)) @@ -71,9 +82,15 @@ function toContractAmount(amount: number): bigint { async function executeWriteContractCall( method: string, args: xdr.ScVal[], - signer: Keypair + signer: Keypair, + feeMultiplier: number = 1 ): Promise { - const tx = await buildContractCall(method, args, signer.publicKey()) + const tx = await buildContractCall( + method, + args, + signer.publicKey(), + feeMultiplier + ) // Pre-Transaction Simulation & Validation (Issue #58) const simulation = await simulateTransaction(tx) @@ -114,7 +131,8 @@ async function executeCustodialVaultOperation( userId: string, userAddress: string, amount: number, - assetSymbol: string + assetSymbol: string, + feeMultiplier: number = 1 ): Promise { const signer = await getKeypairForUser(userId) const userScVal = nativeToScVal(userAddress, { type: 'address' }) @@ -124,7 +142,8 @@ async function executeCustodialVaultOperation( return executeWriteContractCall( method, [userScVal, amountScVal, assetScVal], - signer + signer, + feeMultiplier ) } @@ -185,7 +204,8 @@ export async function getActiveProtocol(): Promise { */ export async function triggerRebalance( protocol: string, - expectedApyBasisPoints: number + expectedApyBasisPoints: number, + feeMultiplier: number = 1 ): Promise { const protocolScVal = nativeToScVal(protocol, { type: 'string' }) const apyScVal = nativeToScVal(expectedApyBasisPoints, { type: 'u32' }) @@ -194,7 +214,8 @@ export async function triggerRebalance( return executeWriteContractCall( 'rebalance', [protocolScVal, apyScVal], - keypair + keypair, + feeMultiplier ) } @@ -229,7 +250,8 @@ export async function updateTotalAssets( export async function payReferralReward( recipientAddress: string, amount: number, - assetSymbol: string + assetSymbol: string, + feeMultiplier: number = 1 ): Promise { const recipientScVal = nativeToScVal(recipientAddress, { type: 'address' }) const amountScVal = nativeToScVal(toContractAmount(amount), { type: 'i128' }) @@ -239,7 +261,8 @@ export async function payReferralReward( return executeWriteContractCall( config.referral.rewardContractMethod, [recipientScVal, amountScVal, assetScVal], - keypair + keypair, + feeMultiplier ) } @@ -259,14 +282,16 @@ export async function depositForUser( userId: string, userAddress: string, amount: number, - assetSymbol: string + assetSymbol: string, + feeMultiplier: number = 1 ): Promise { return executeCustodialVaultOperation( 'deposit', userId, userAddress, amount, - assetSymbol + assetSymbol, + feeMultiplier ) } @@ -286,14 +311,16 @@ export async function withdrawForUser( userId: string, userAddress: string, amount: number, - assetSymbol: string + assetSymbol: string, + feeMultiplier: number = 1 ): Promise { return executeCustodialVaultOperation( 'withdraw', userId, userAddress, amount, - assetSymbol + assetSymbol, + feeMultiplier ) } diff --git a/src/stellar/events.ts b/src/stellar/events.ts index 6039f03..ba85486 100644 --- a/src/stellar/events.ts +++ b/src/stellar/events.ts @@ -34,6 +34,7 @@ import { } from '../utils/metrics' import { dispatchWebhookEvent } from '../services/webhookDispatcher' import { checkAndActivateOnDeposit } from '../referral/service' +import { reconcileOutboxOpByTxHash } from '../outbox/service' import { createLotForDeposit, recordDisposalsForWithdrawal, @@ -256,6 +257,11 @@ async function handleDepositEvent( }) )) as any + // #325 confirmation fallback: closes out a SUBMITTED OutboxOp left behind + // by a dispatcher crash between submission and its own confirmation. No-op + // for txHashes with no matching op. + await reconcileOutboxOpByTxHash(tx, event.txHash) + const position = (await timedDbOperation(() => tx.position.findFirst({ where: { @@ -366,6 +372,11 @@ async function handleWithdrawEvent( }) )) as any + // #325 confirmation fallback: closes out a SUBMITTED OutboxOp left behind + // by a dispatcher crash between submission and its own confirmation. No-op + // for txHashes with no matching op. + await reconcileOutboxOpByTxHash(tx, event.txHash) + const position = (await timedDbOperation(() => tx.position.findFirst({ where: { diff --git a/src/utils/metrics.ts b/src/utils/metrics.ts index d796411..ecc3545 100644 --- a/src/utils/metrics.ts +++ b/src/utils/metrics.ts @@ -574,6 +574,75 @@ export function recordFiatRateDrift( fiatRateDriftPct.observe({ provider, direction }, absDriftPct) } +// ── Outbox Metrics (#325) ──────────────────────────────────────────────────── + +export const outboxOpsTotal = new client.Counter({ + name: 'outbox_ops_total', + help: 'Total outbox op submit attempts, by kind/priority/outcome', + labelNames: ['kind', 'priority', 'outcome'] as const, // outcome: confirmed|retry|failed + registers: [register], +}) + +export const outboxQueueDepth = new client.Gauge({ + name: 'outbox_queue_depth', + help: 'Current outbox op count by status and priority', + labelNames: ['status', 'priority'] as const, + registers: [register], +}) + +export const outboxOpLatencySeconds = new client.Histogram({ + name: 'outbox_op_latency_seconds', + help: 'Time from outbox op creation to confirmation, in seconds', + labelNames: ['kind'] as const, + buckets: [0.5, 1, 2, 5, 10, 30, 60, 120, 300], + registers: [register], +}) + +export const outboxFeeBumpTotal = new client.Counter({ + name: 'outbox_fee_bump_total', + help: 'Total fee-bump resubmissions triggered by unconfirmed-too-long ops', + labelNames: ['kind'] as const, + registers: [register], +}) + +export const outboxStuckSubmitted = new client.Gauge({ + name: 'outbox_stuck_submitted', + help: 'Ops SUBMITTED but unconfirmed longer than the configured timeout — the "lost in flight" alarm', + registers: [register], +}) + +export function recordOutboxOp( + kind: string, + priority: string, + outcome: 'confirmed' | 'retry' | 'failed' +): void { + outboxOpsTotal.inc({ kind, priority, outcome }) +} + +export function updateOutboxQueueDepth( + rows: Array<{ status: string; priority: string; count: number }> +): void { + outboxQueueDepth.reset() + for (const row of rows) { + outboxQueueDepth.set( + { status: row.status, priority: row.priority }, + row.count + ) + } +} + +export function recordOutboxLatency(kind: string, seconds: number): void { + outboxOpLatencySeconds.observe({ kind }, seconds) +} + +export function recordOutboxFeeBump(kind: string): void { + outboxFeeBumpTotal.inc({ kind }) +} + +export function updateOutboxStuckSubmitted(count: number): void { + outboxStuckSubmitted.set(count) +} + /** * Get metrics for Prometheus scraping */ diff --git a/src/validators/webhook-validators.ts b/src/validators/webhook-validators.ts index 58a09c0..52acefb 100644 --- a/src/validators/webhook-validators.ts +++ b/src/validators/webhook-validators.ts @@ -42,6 +42,9 @@ const WEBHOOK_EVENTS = [ // is affected. Never carries the publisher's identity. 'strategy.updated', 'strategy.unpublished', + // Durable outbox (#325): a money-moving op exhausted its retries and moved + // to the terminal FAILED state — see docs/OUTBOX.md. + 'outbox.op_failed', ] as const export const createWebhookSchema = z.object({ diff --git a/tests/integration/agent/rebalance.integration.test.ts b/tests/integration/agent/rebalance.integration.test.ts index c3d4a36..b75774f 100644 --- a/tests/integration/agent/rebalance.integration.test.ts +++ b/tests/integration/agent/rebalance.integration.test.ts @@ -44,7 +44,9 @@ const agentLogStore: Array<{ const mockPositionFindFirst = jest.fn() const mockPositionFindMany = jest.fn() -const mockTransactionCreate = jest.fn().mockResolvedValue({}) +const mockTransactionCreate = jest + .fn() + .mockImplementation(({ data }: any) => ({ id: 'txn-1', ...data })) const mockAgentLogCreate = jest .fn() .mockImplementation(({ data }: { data: any }) => { @@ -57,9 +59,16 @@ const mockAgentLogCreate = jest return Promise.resolve({ id: `log-${agentLogStore.length}` }) }) -jest.mock('../../../src/db', () => ({ - __esModule: true, - default: { +// #325: triggerRebalance now enqueues a durable OutboxOp (transactionally +// with the Transaction row) instead of calling the Stellar contract inline — +// see src/outbox/service.ts#enqueueOutboxOp. The outbox itself has its own +// coverage (tests/unit/outbox/, tests/integration/outbox/); here it only +// needs to look like a real Prisma transaction client. +const mockOutboxOpCreate = jest.fn().mockResolvedValue({ id: 'op-1' }) +const mockOutboxOpFindUnique = jest.fn().mockResolvedValue(null) + +jest.mock('../../../src/db', () => { + const client: any = { agentLog: { create: (...args: unknown[]) => mockAgentLogCreate(...args), }, @@ -70,6 +79,10 @@ jest.mock('../../../src/db', () => ({ transaction: { create: (...args: unknown[]) => mockTransactionCreate(...args), }, + outboxOp: { + create: (...args: unknown[]) => mockOutboxOpCreate(...args), + findUnique: (...args: unknown[]) => mockOutboxOpFindUnique(...args), + }, user: { // Should NOT be called by logAgentAction anymore findMany: jest @@ -78,7 +91,17 @@ jest.mock('../../../src/db', () => ({ new Error('db.user.findMany should not be called for agent logging') ), }, - }, + } + client.$transaction = (fn: (tx: unknown) => unknown) => fn(client) + return { __esModule: true, default: client } +}) + +// dispatchInBackground fires an async dispatchOne(opId) that this test does +// not need to exercise (the outbox's own dispatch behavior is covered +// elsewhere) — stub it to a no-op so no unhandled background call touches +// mocks this suite doesn't set up. +jest.mock('../../../src/outbox/dispatcher', () => ({ + dispatchInBackground: jest.fn(), })) // ------------------------------------------------------------------------ diff --git a/tests/integration/deposit-withdraw.integration.test.ts b/tests/integration/deposit-withdraw.integration.test.ts index c800832..28eec74 100644 --- a/tests/integration/deposit-withdraw.integration.test.ts +++ b/tests/integration/deposit-withdraw.integration.test.ts @@ -55,6 +55,12 @@ jest.mock('../../src/utils/metrics', () => ({ recordHttpRequest: jest.fn(), recordRequestTimeout: jest.fn(), recordRejectedRequest: jest.fn(), + // #325 — the outbox dispatcher records these on every submit attempt. + recordOutboxOp: jest.fn(), + updateOutboxQueueDepth: jest.fn(), + recordOutboxLatency: jest.fn(), + recordOutboxFeeBump: jest.fn(), + updateOutboxStuckSubmitted: jest.fn(), })) // Avoid external alerting side effects diff --git a/tests/integration/outbox/dispatcher.integration.test.ts b/tests/integration/outbox/dispatcher.integration.test.ts new file mode 100644 index 0000000..05350c4 --- /dev/null +++ b/tests/integration/outbox/dispatcher.integration.test.ts @@ -0,0 +1,254 @@ +/** + * Integration tests for the durable outbox dispatcher (#325), against a real + * Postgres database — same convention as + * tests/integration/deposit-withdraw.integration.test.ts: excluded from the + * default `npm test` run (see jest.config.js) because it needs a live DB. + * + * Run manually: + * DATABASE_URL=postgresql://user:pass@localhost:5432/db \ + * npx jest --testPathIgnorePatterns='/node_modules/' --testPathPattern='integration/outbox' + * + * Covers the acceptance-criteria scenarios that only a real DB can prove: + * - atomic claim: two concurrent claimers for the same op, exactly one wins + * - kill-the-worker: a SUBMITTED op with no confirmation (simulated crash) + * is recovered by the stuck-submitted sweep and completes exactly once + * - priority ordering: a CRITICAL op dispatches ahead of a NORMAL wave + * + * The actual Stellar RPC call is mocked (src/outbox/executors.ts) — this + * suite exercises the outbox's own DB state machine and concurrency control, + * not real network I/O. + */ +import db from '../../../src/db' +import { + enqueueOutboxOp, + claimOp, + getOp, + findStuckSubmittedOps, + returnStuckOpToPending, +} from '../../../src/outbox/service' +import { dispatchOne, runDispatchSweep } from '../../../src/outbox/dispatcher' +import { deriveIdempotencyKey } from '../../../src/outbox/idempotency' + +jest.mock('../../../src/utils/logger', () => ({ + logger: { + info: jest.fn(), + warn: jest.fn(), + error: jest.fn(), + debug: jest.fn(), + }, +})) +jest.mock('../../../src/services/alerting', () => ({ + alertingService: { emit: jest.fn().mockResolvedValue(undefined) }, +})) +jest.mock('../../../src/services/webhookDispatcher', () => ({ + dispatchWebhookEvent: jest.fn().mockResolvedValue(undefined), +})) + +const mockExecuteOutboxPayload = jest.fn() +const mockResolveSignerPublicKey = jest.fn() +jest.mock('../../../src/outbox/executors', () => ({ + executeOutboxPayload: (...args: unknown[]) => + mockExecuteOutboxPayload(...args), + resolveSignerPublicKey: (...args: unknown[]) => + mockResolveSignerPublicKey(...args), +})) + +function uuid(): string { + return `it-${Date.now()}-${Math.random().toString(36).slice(2)}` +} + +async function seedUser(overrides: { isActive?: boolean } = {}) { + return db.user.create({ + data: { + walletAddress: `G${uuid().replace(/-/g, '').slice(0, 47)}`.slice(0, 56), + network: 'TESTNET', + email: `${uuid()}@example.com`, + isActive: overrides.isActive ?? true, + }, + }) +} + +async function seedDepositOp(userId: string, businessRecordId: string) { + return db.$transaction((tx) => + enqueueOutboxOp(tx, { + idempotencyKey: deriveIdempotencyKey('DEPOSIT', userId, businessRecordId), + userId, + kind: 'DEPOSIT', + actor: 'USER', + payload: { + method: 'deposit', + userId, + userAddress: 'GADDRESS', + amount: 10, + assetSymbol: 'USDC', + transactionId: businessRecordId, + }, + }) + ) +} + +describe('outbox dispatcher — atomic claim (no double-submission)', () => { + beforeEach(() => { + jest.clearAllMocks() + mockResolveSignerPublicKey.mockResolvedValue('SIGNER_A') + }) + + it('two concurrent claimers for the same op: exactly one wins', async () => { + const user = await seedUser() + const op = await seedDepositOp(user.id, uuid()) + + const [a, b] = await Promise.all([ + claimOp(op.id, 'SIGNER_A'), + claimOp(op.id, 'SIGNER_A'), + ]) + + const winners = [a, b].filter((r) => r !== null) + const losers = [a, b].filter((r) => r === null) + expect(winners).toHaveLength(1) + expect(losers).toHaveLength(1) + + const finalRow = await getOp(op.id) + expect(finalRow?.status).toBe('SUBMITTED') + // Exactly one claim incremented attempts, not two. + expect(finalRow?.attempts).toBe(1) + }) + + it('two concurrent dispatchOne calls for the same op: the network is only touched once', async () => { + const user = await seedUser() + const op = await seedDepositOp(user.id, uuid()) + + mockExecuteOutboxPayload.mockResolvedValue({ + hash: `tx-${uuid()}`, + status: 'success', + }) + + const results = await Promise.allSettled([ + dispatchOne(op.id), + dispatchOne(op.id), + ]) + + // The atomic PENDING -> SUBMITTED claim means only one of the two + // concurrent calls could ever reach executeOutboxPayload — the other + // loses the claim race and rejects without touching the network. + const fulfilled = results.filter((r) => r.status === 'fulfilled') + const rejected = results.filter((r) => r.status === 'rejected') + expect(fulfilled).toHaveLength(1) + expect(rejected).toHaveLength(1) + expect(mockExecuteOutboxPayload).toHaveBeenCalledTimes(1) + }) +}) + +describe('outbox dispatcher — kill-the-worker recovery', () => { + beforeEach(() => { + jest.clearAllMocks() + mockResolveSignerPublicKey.mockResolvedValue('SIGNER_B') + }) + + it('a SUBMITTED op with no observed confirmation is recovered, reclaimed, and confirmed exactly once', async () => { + const user = await seedUser() + const op = await seedDepositOp(user.id, uuid()) + + // Simulate the dispatcher process claiming the op and then crashing + // before executeOutboxPayload ever resolves — SUBMITTED, no txHash. + const claimed = await claimOp(op.id, 'SIGNER_B') + expect(claimed).not.toBeNull() + + // Backdate submittedAt past the timeout window (can't wait real time). + await db.outboxOp.update({ + where: { id: op.id }, + data: { submittedAt: new Date(Date.now() - 10 * 60 * 1000) }, + }) + + const stuck = await findStuckSubmittedOps(60_000) + expect(stuck.map((o) => o.id)).toContain(op.id) + + await returnStuckOpToPending(op.id) + const recovered = await getOp(op.id) + expect(recovered?.status).toBe('PENDING') + expect(recovered?.attempts).toBe(1) // the crashed attempt is not forgotten + + // Now the op is reclaimable and completes normally — no op is lost, and + // it is not double-executed: the mocked network call fires exactly once + // for this (the only) successful attempt. + const txHash = `tx-${uuid()}` + mockExecuteOutboxPayload.mockResolvedValueOnce({ + hash: txHash, + status: 'success', + }) + + const result = await dispatchOne(op.id) + expect(result.hash).toBe(txHash) + expect(mockExecuteOutboxPayload).toHaveBeenCalledTimes(1) + + const final = await getOp(op.id) + expect(final?.status).toBe('CONFIRMED') + expect(final?.attempts).toBe(2) // crashed attempt + the one that confirmed + }) + + it('a frozen user halts dispatch of their queued op', async () => { + const user = await seedUser({ isActive: false }) + const op = await seedDepositOp(user.id, uuid()) + + await expect(dispatchOne(op.id)).rejects.toThrow(/frozen/) + expect(mockExecuteOutboxPayload).not.toHaveBeenCalled() + + const row = await getOp(op.id) + expect(row?.status).toBe('PENDING') // never claimed + }) +}) + +describe('outbox dispatcher — priority ordering under the real claim path', () => { + beforeEach(() => { + jest.clearAllMocks() + mockResolveSignerPublicKey.mockImplementation( + async (_payload, userId) => userId + ) + // A fresh hash per call — txHash is unique, and this sweep submits + // several ops in the same test. + mockExecuteOutboxPayload.mockImplementation(async () => ({ + hash: `tx-${uuid()}`, + status: 'success', + })) + }) + + it('a CRITICAL withdrawal claims ahead of a wave of NORMAL deposits', async () => { + const users = await Promise.all(Array.from({ length: 5 }, () => seedUser())) + + // Five NORMAL deposits, all older than the CRITICAL withdrawal below. + for (const user of users) { + await seedDepositOp(user.id, uuid()) + } + + const withdrawUser = await seedUser() + const withdrawOp = await db.$transaction((tx) => + enqueueOutboxOp(tx, { + idempotencyKey: deriveIdempotencyKey( + 'WITHDRAW', + withdrawUser.id, + uuid() + ), + userId: withdrawUser.id, + kind: 'WITHDRAW', + actor: 'USER', + payload: { + method: 'withdraw', + userId: withdrawUser.id, + userAddress: 'GADDRESS', + amount: 5, + assetSymbol: 'USDC', + transactionId: uuid(), + }, + }) + ) + + // A single sweep claims config.outbox.batchSize (20) ops, priority-ordered + // — 6 total here fit in one batch, so if the CRITICAL op were starved + // behind the NORMAL wave (wrong ordering) it simply wouldn't be claimed + // this sweep. Asserting it's CONFIRMED after exactly one sweep proves the + // ordering, not just that it eventually gets there. + await runDispatchSweep() + + const withdrawRow = await getOp(withdrawOp.id) + expect(withdrawRow?.status).toBe('CONFIRMED') + }) +}) diff --git a/tests/unit/outbox/idempotency.test.ts b/tests/unit/outbox/idempotency.test.ts new file mode 100644 index 0000000..ac62d54 --- /dev/null +++ b/tests/unit/outbox/idempotency.test.ts @@ -0,0 +1,35 @@ +import { deriveIdempotencyKey } from '../../../src/outbox/idempotency' + +describe('src/outbox/idempotency', () => { + it('derives a deterministic kind:userId:businessRecordId key', () => { + expect(deriveIdempotencyKey('DEPOSIT', 'user-1', 'txn-1')).toBe( + 'DEPOSIT:user-1:txn-1' + ) + }) + + it('is deterministic — the same inputs always produce the same key', () => { + const a = deriveIdempotencyKey('WITHDRAW', 'user-2', 'txn-2') + const b = deriveIdempotencyKey('WITHDRAW', 'user-2', 'txn-2') + expect(a).toBe(b) + }) + + it('different kinds for the same record never collide', () => { + const deposit = deriveIdempotencyKey('DEPOSIT', 'user-1', 'record-1') + const withdraw = deriveIdempotencyKey('WITHDRAW', 'user-1', 'record-1') + expect(deposit).not.toBe(withdraw) + }) + + it('different users for the same record never collide', () => { + const a = deriveIdempotencyKey('REFERRAL_REWARD', 'user-a', 'conv:owner') + const b = deriveIdempotencyKey('REFERRAL_REWARD', 'user-b', 'conv:owner') + expect(a).not.toBe(b) + }) + + it('throws without a userId', () => { + expect(() => deriveIdempotencyKey('DEPOSIT', '', 'record-1')).toThrow() + }) + + it('throws without a businessRecordId', () => { + expect(() => deriveIdempotencyKey('DEPOSIT', 'user-1', '')).toThrow() + }) +}) diff --git a/tests/unit/outbox/stateMachine.test.ts b/tests/unit/outbox/stateMachine.test.ts new file mode 100644 index 0000000..166b585 --- /dev/null +++ b/tests/unit/outbox/stateMachine.test.ts @@ -0,0 +1,162 @@ +import { + canTransition, + assertTransition, + compareForDispatch, + sortForDispatch, + computeBackoffMs, + PRIORITY_WEIGHT, +} from '../../../src/outbox/stateMachine' +import { OutboxOpRecord } from '../../../src/outbox/types' + +function op( + overrides: Partial> +): Pick & { id?: string } { + return { + priority: 'NORMAL', + createdAt: new Date('2026-01-01T00:00:00.000Z'), + ...overrides, + } +} + +describe('src/outbox/stateMachine — transitions', () => { + it('allows the documented PENDING -> SUBMITTED -> CONFIRMED happy path', () => { + expect(canTransition('PENDING', 'SUBMITTED')).toBe(true) + expect(canTransition('SUBMITTED', 'CONFIRMED')).toBe(true) + }) + + it('allows a transient submit failure to return SUBMITTED -> PENDING for retry', () => { + expect(canTransition('SUBMITTED', 'PENDING')).toBe(true) + }) + + it('allows SUBMITTED -> FAILED (terminal or fee-bump-cap escalation)', () => { + expect(canTransition('SUBMITTED', 'FAILED')).toBe(true) + }) + + it('allows PENDING -> CANCELLED (admin cancels an unsent op)', () => { + expect(canTransition('PENDING', 'CANCELLED')).toBe(true) + }) + + it('allows FAILED -> PENDING only for admin force-retry', () => { + expect(canTransition('FAILED', 'PENDING')).toBe(true) + }) + + it('CONFIRMED and CANCELLED are terminal — no transitions out', () => { + expect(canTransition('CONFIRMED', 'PENDING')).toBe(false) + expect(canTransition('CONFIRMED', 'FAILED')).toBe(false) + expect(canTransition('CANCELLED', 'PENDING')).toBe(false) + }) + + it('rejects illegal jumps, e.g. PENDING -> CONFIRMED directly', () => { + expect(canTransition('PENDING', 'CONFIRMED')).toBe(false) + }) + + it('assertTransition throws with a clear message on an illegal transition', () => { + expect(() => assertTransition('CONFIRMED', 'PENDING')).toThrow( + /Illegal outbox transition: CONFIRMED -> PENDING/ + ) + }) + + it('assertTransition does not throw on a legal transition', () => { + expect(() => assertTransition('PENDING', 'SUBMITTED')).not.toThrow() + }) +}) + +describe('src/outbox/stateMachine — priority ordering', () => { + it('orders CRITICAL before NORMAL before LOW', () => { + expect(PRIORITY_WEIGHT.CRITICAL).toBeLessThan(PRIORITY_WEIGHT.NORMAL) + expect(PRIORITY_WEIGHT.NORMAL).toBeLessThan(PRIORITY_WEIGHT.LOW) + }) + + it('within the same priority, orders oldest createdAt first (FIFO)', () => { + const older = op({ + priority: 'NORMAL', + createdAt: new Date('2026-01-01T00:00:00Z'), + }) + const newer = op({ + priority: 'NORMAL', + createdAt: new Date('2026-01-01T00:05:00Z'), + }) + expect(compareForDispatch(older, newer)).toBeLessThan(0) + expect(compareForDispatch(newer, older)).toBeGreaterThan(0) + }) + + it('does not starve a CRITICAL op behind a wave of NORMAL ops', () => { + // Simulates the scenario called out in issue #325: a burst of NORMAL + // recurring-deposit/referral ops queued well before a CRITICAL withdrawal + // must still let the withdrawal dispatch first. + const normalWave = Array.from({ length: 100 }, (_, i) => + op({ + id: `normal-${i}`, + priority: 'NORMAL', + createdAt: new Date(Date.now() - (100 - i) * 1000), // all older than the CRITICAL op + }) + ) + const criticalWithdrawal = op({ + id: 'critical-1', + priority: 'CRITICAL', + createdAt: new Date(), // youngest of the batch + }) + + const sorted = sortForDispatch([...normalWave, criticalWithdrawal]) + expect(sorted[0]).toBe(criticalWithdrawal) + }) + + it('does not starve a CRITICAL op behind a wave of LOW-priority rebalances', () => { + const lowWave = Array.from({ length: 50 }, (_, i) => + op({ + id: `low-${i}`, + priority: 'LOW', + createdAt: new Date(Date.now() - (50 - i) * 1000), + }) + ) + const criticalWithdrawal = op({ id: 'critical-1', priority: 'CRITICAL' }) + + const sorted = sortForDispatch([...lowWave, criticalWithdrawal]) + expect(sorted[0]).toBe(criticalWithdrawal) + }) + + it('sortForDispatch does not mutate the input array', () => { + const a = op({ priority: 'LOW', createdAt: new Date(1000) }) + const b = op({ priority: 'CRITICAL', createdAt: new Date(2000) }) + const input = [a, b] + const sorted = sortForDispatch(input) + expect(input[0]).toBe(a) // original order preserved + expect(sorted[0]).toBe(b) + }) + + it('a full priority mix sorts as CRITICAL, NORMAL, LOW', () => { + const low = op({ id: 'low', priority: 'LOW' }) + const normal = op({ id: 'normal', priority: 'NORMAL' }) + const critical = op({ id: 'critical', priority: 'CRITICAL' }) + + const sorted = sortForDispatch([low, normal, critical]) + expect(sorted.map((o) => o.priority)).toEqual(['CRITICAL', 'NORMAL', 'LOW']) + }) +}) + +describe('src/outbox/stateMachine — backoff', () => { + it('is bounded by [0, min(base * 2^(attempt-1), max)]', () => { + const random = () => 1 // force the upper bound + expect(computeBackoffMs(1, 1000, 60000, random)).toBe(1000) + expect(computeBackoffMs(2, 1000, 60000, random)).toBe(2000) + expect(computeBackoffMs(3, 1000, 60000, random)).toBe(4000) + }) + + it('caps at maxMs regardless of how large attempt grows', () => { + const random = () => 1 + expect(computeBackoffMs(20, 1000, 60000, random)).toBe(60000) + }) + + it('at random()=0, backoff is always 0 (full jitter floor)', () => { + const random = () => 0 + expect(computeBackoffMs(5, 1000, 60000, random)).toBe(0) + }) + + it('never exceeds maxMs across many random samples', () => { + for (let i = 0; i < 200; i++) { + const ms = computeBackoffMs(10, 2000, 30000, Math.random) + expect(ms).toBeGreaterThanOrEqual(0) + expect(ms).toBeLessThanOrEqual(30000) + } + }) +}) diff --git a/tests/unit/outbox/structural.test.ts b/tests/unit/outbox/structural.test.ts new file mode 100644 index 0000000..b926e93 --- /dev/null +++ b/tests/unit/outbox/structural.test.ts @@ -0,0 +1,134 @@ +/** + * Structural guarantee for the durable outbox (#325). + * + * THIS IS THE ACCEPTANCE CRITERION, not a style check. The whole point of the + * outbox is that it is the single choke point every on-chain money movement + * passes through — durable, retriable, priority-ordered, observable. That + * property is worth nothing the first time someone adds a route or job that + * calls `depositForUser`/`withdrawForUser`/`triggerRebalance`/ + * `payReferralReward` directly instead of going through + * src/outbox/service.ts + src/outbox/dispatcher.ts. + * + * Uses the same specifier-parsing approach as + * tests/unit/analytics/structural.test.ts and + * tests/integration/agent/strategy-follow.integration.test.ts. + */ + +import fs from 'fs' +import path from 'path' + +const SRC_DIR = path.join(__dirname, '../../../src') + +// The raw write functions in src/stellar/contract.ts that actually move +// money on-chain. Read functions (getOnChainBalance, getOnChainAPY, ...) are +// exempt — they never submit a transaction. +const RAW_WRITE_FUNCTIONS = [ + 'depositForUser', + 'withdrawForUser', + 'triggerRebalance', + 'payReferralReward', + // Legacy aliases exported by contract.ts for the same write paths. + 'deposit', + 'withdraw', +] + +// Files allowed to import them directly: +// - src/stellar/contract.ts itself (it defines them) +// - src/outbox/executors.ts (the ONE place the dispatcher actually submits) +const ALLOWED_IMPORTERS = new Set([ + path.join(SRC_DIR, 'stellar', 'contract.ts'), + path.join(SRC_DIR, 'outbox', 'executors.ts'), +]) + +function walk(dir: string, out: string[] = []): string[] { + for (const entry of fs.readdirSync(dir, { withFileTypes: true })) { + const full = path.join(dir, entry.name) + if (entry.isDirectory()) { + walk(full, out) + } else if (entry.name.endsWith('.ts') && !entry.name.endsWith('.d.ts')) { + out.push(full) + } + } + return out +} + +function resolveSpecifier(fromFile: string, specifier: string): string | null { + if (!specifier.startsWith('.')) return null + const resolved = path.resolve(path.dirname(fromFile), specifier) + return resolved.endsWith('.ts') ? resolved : `${resolved}.ts` +} + +/** Named imports from a given module specifier, e.g. `from '../stellar/contract'`. */ +function namedImportsFrom(source: string, specifierSuffix: string): string[] { + const names: string[] = [] + const importRegex = + /import\s*(?:type\s*)?\{([^}]+)\}\s*from\s*['"]([^'"]+)['"]/g + let match: RegExpExecArray | null + while ((match = importRegex.exec(source)) !== null) { + const [, namedBlock, specifier] = match + if (!specifier.includes('stellar/contract')) continue + if (specifierSuffix && !specifier.endsWith(specifierSuffix)) continue + for (const raw of namedBlock.split(',')) { + const name = raw + .trim() + .split(/\s+as\s+/)[0] + .trim() + if (name) names.push(name) + } + } + return names +} + +describe('src/outbox/ — no-bypass structural guarantee', () => { + const contractFile = path.join(SRC_DIR, 'stellar', 'contract.ts') + + it('src/stellar/contract.ts still exports every raw write function this test guards', () => { + const source = fs.readFileSync(contractFile, 'utf8') + for (const fn of [ + 'depositForUser', + 'withdrawForUser', + 'triggerRebalance', + 'payReferralReward', + ]) { + expect(source).toMatch( + new RegExp(`export\\s+async\\s+function\\s+${fn}\\b`) + ) + } + }) + + it('src/outbox/executors.ts exists and imports the raw write functions', () => { + const executorsFile = path.join(SRC_DIR, 'outbox', 'executors.ts') + expect(fs.existsSync(executorsFile)).toBe(true) + const source = fs.readFileSync(executorsFile, 'utf8') + const imports = namedImportsFrom(source, '') + expect(imports.length).toBeGreaterThan(0) + }) + + const allFiles = walk(SRC_DIR) + // Sanity: this test must actually be scanning a non-trivial tree, or a + // regex that silently stopped matching could make every assertion below + // pass vacuously. + it('scans a non-trivial number of source files', () => { + expect(allFiles.length).toBeGreaterThan(50) + }) + + it.each(allFiles.filter((f) => !ALLOWED_IMPORTERS.has(f)))( + '%s does not import a raw on-chain write function from stellar/contract', + (file) => { + const source = fs.readFileSync(file, 'utf8') + const imports = namedImportsFrom(source, '') + const forbidden = imports.filter((name) => + RAW_WRITE_FUNCTIONS.includes(name) + ) + expect(forbidden).toEqual([]) + } + ) + + it('resolves relative "stellar/contract" specifiers to the same file across the tree (regex sanity)', () => { + // Guards the resolver helper itself: if it stopped resolving correctly, + // the "does not import" assertions above could pass for the wrong reason. + const sample = path.join(SRC_DIR, 'controllers', 'sample.ts') + const resolved = resolveSpecifier(sample, '../stellar/contract') + expect(resolved).toBe(contractFile) + }) +}) diff --git a/tests/unit/referral/service.test.ts b/tests/unit/referral/service.test.ts index 9ef97ea..2419db3 100644 --- a/tests/unit/referral/service.test.ts +++ b/tests/unit/referral/service.test.ts @@ -7,7 +7,6 @@ // the conversion ACTIVATED + retriable, and REWARDED requires every leg paid import db from '../../../src/db' import { alertingService } from '../../../src/services/alerting' -import { payReferralReward } from '../../../src/stellar/contract' import { getWalletByUserId } from '../../../src/stellar/wallet' import { attributeSignup, @@ -27,9 +26,6 @@ jest.mock('../../../src/utils/logger', () => ({ jest.mock('../../../src/services/alerting', () => ({ alertingService: { emit: jest.fn().mockResolvedValue({ sent: true }) }, })) -jest.mock('../../../src/stellar/contract', () => ({ - payReferralReward: jest.fn(), -})) jest.mock('../../../src/stellar/wallet', () => ({ getWalletByUserId: jest.fn(), })) @@ -46,9 +42,22 @@ jest.mock('../../../src/config', () => ({ }, })) +// The outbox is a separate module with its own unit/integration coverage +// (tests/unit/outbox/, tests/integration/outbox/) — here it is mocked as the +// dependency boundary so this suite stays focused on referral bookkeeping: +// did we enqueue/dispatch a durable op, not "does the dispatcher's retry +// machinery work." +const mockEnqueueOutboxOp = jest.fn() +const mockDispatchOne = jest.fn() +jest.mock('../../../src/outbox/service', () => ({ + enqueueOutboxOp: (...args: unknown[]) => mockEnqueueOutboxOp(...args), +})) +jest.mock('../../../src/outbox/dispatcher', () => ({ + dispatchOne: (...args: unknown[]) => mockDispatchOne(...args), +})) + const mockDb = db as any const mockEmit = alertingService.emit as jest.Mock -const mockPay = payReferralReward as jest.Mock const mockGetWallet = getWalletByUserId as jest.Mock jest.mock('@prisma/client', () => { @@ -88,7 +97,8 @@ beforeEach(() => { update: jest.fn(), } mockDb.user = { findUnique: jest.fn() } - mockDb.transaction = { create: jest.fn() } + mockDb.transaction = { create: jest.fn(), update: jest.fn() } + mockDb.$transaction = jest.fn((fn: (tx: any) => unknown) => fn(mockDb)) }) describe('attributeSignup', () => { @@ -232,10 +242,15 @@ describe('payoutActivatedConversions', () => { beforeEach(() => { mockGetWallet.mockResolvedValue({ publicKey: 'GADDRESS' }) mockDb.user.findUnique.mockResolvedValue({ network: 'MAINNET' }) - mockPay.mockResolvedValue({ hash: 'onchainhash', status: 'success' }) + mockEnqueueOutboxOp.mockResolvedValue({ id: 'op-1' }) + mockDispatchOne.mockResolvedValue({ + hash: 'onchainhash', + status: 'success', + }) mockDb.transaction.create.mockImplementation(({ data }: any) => ({ id: `tx-${data.userId}`, })) + mockDb.transaction.update.mockResolvedValue({}) mockDb.referralConversion.update.mockResolvedValue({}) }) @@ -247,19 +262,24 @@ describe('payoutActivatedConversions', () => { const res = await payoutActivatedConversions() expect(res.rewarded).toBe(1) - expect(mockPay).toHaveBeenCalledTimes(2) + expect(mockDispatchOne).toHaveBeenCalledTimes(2) const finalUpdate = mockDb.referralConversion.update.mock.calls.at(-1)[0] expect(finalUpdate.data.status).toBe('REWARDED') }) - it('records payout as a distinct REFERRAL_REWARD transaction', async () => { + it('enqueues a distinct REFERRAL_REWARD transaction, then confirms it via the outbox', async () => { mockDb.referralConversion.findMany.mockResolvedValue([ activatedConversion(), ]) await payoutActivatedConversions() + const createArg = mockDb.transaction.create.mock.calls[0][0] expect(createArg.data.type).toBe('REFERRAL_REWARD') - expect(createArg.data.status).toBe('CONFIRMED') + expect(createArg.data.status).toBe('PENDING') + + const updateArg = mockDb.transaction.update.mock.calls[0][0] + expect(updateArg.data.status).toBe('CONFIRMED') + expect(updateArg.data.txHash).toBe('onchainhash') }) it('is idempotent — skips a leg already paid', async () => { @@ -267,15 +287,15 @@ describe('payoutActivatedConversions', () => { activatedConversion({ ownerRewardTxId: 'already-paid' }), ]) await payoutActivatedConversions() - // Only the referred leg should be paid. - expect(mockPay).toHaveBeenCalledTimes(1) + // Only the referred leg should be dispatched. + expect(mockDispatchOne).toHaveBeenCalledTimes(1) }) it('leaves conversion ACTIVATED + retriable when a leg fails, and does NOT reward', async () => { mockDb.referralConversion.findMany.mockResolvedValue([ activatedConversion(), ]) - mockPay.mockRejectedValueOnce(new Error('rpc timeout')) + mockDispatchOne.mockRejectedValueOnce(new Error('rpc timeout')) const res = await payoutActivatedConversions() @@ -306,6 +326,6 @@ describe('payoutActivatedConversions', () => { const res = await payoutActivatedConversions() expect(res.rewarded).toBe(0) - expect(mockPay).not.toHaveBeenCalled() + expect(mockDispatchOne).not.toHaveBeenCalled() }) })