fix: cross-replica rate limits, job tick locks, treasury sweep balance and large-sweep guard

docs/RATE_LIMITING.md

@@ -12,6 +12,24 @@ is part of the change — not a follow-up.
 
 ---
 
+## Distributed enforcement (#635)
+
+Rate-limit counters live in **Redis** (`src/middleware/rateLimitStore.ts`), the
+same Redis already used for cache invalidation and the event bridge. Every
+replica behind the load balancer draws on one number per bucket, so the
+documented tier budgets hold at the fleet level — `n` pods do **not** mean `n x`
+the limit.
+
+- `NODE_ENV=test` keeps express-rate-limit's in-process `MemoryStore`, so unit
+  tests never need Redis.
+- If `REDIS_URL` is unset the limiter falls back to the per-pod store and logs a
+  warning. A multi-replica deployment without Redis will exceed the documented
+  limits; that is a deployment bug, not a tuning option.
+- `Retry-After` and the `RateLimit-*` headers are derived from the shared
+  counter, so a client throttled on one pod is throttled on all of them.
+
+---
+
 ## The two layers every request passes through
 
 Before a request reaches a route, it passes through the **app-wide chain**

docs/job-observability.md

@@ -38,6 +38,7 @@ This document covers the dedicated observability metrics for scheduled backgroun
 
 Scheduled maintenance jobs go through `scheduleResilientJob` in `src/jobs/resilientScheduler.ts`.
 
+- **Cross-pod tick lock (#636):** each tick takes a `pg_try_advisory_xact_lock` keyed by the job name before running. In a multi-replica deployment exactly one pod wins each tick; the others skip (log: `skipped; another replica holds the tick lock`). The lock is released when the wrapping transaction ends.
 - Each job runs once at process start, so work that was due while the process was down is resumed immediately.
 - A thrown failure is retried up to 3 times with full-jitter exponential backoff (1s base, 30s cap).
 - A tick that arrives while a run is still in flight does not start a second run. If that run succeeds, the skipped tick is resumed once. If retries were exhausted, the next interval is the resume point. The scheduler does not tight-loop a job that is still failing.
@@ -81,6 +82,43 @@ Scheduled maintenance jobs go through `scheduleResilientJob` in `src/jobs/resili
 | `reserve_reconciliation`       | Compares reserve sponsorship rows with on-chain sponsor state                    |
 | `pool_metrics`                 | Mirrors Prisma pool gauges into Prometheus                                       |
 
+## Concurrency classification (#636)
+
+Every scheduled job is classified for what would happen *without* the tick lock.
+Source of truth: `src/jobs/jobClassification.ts` — `tests/unit/jobs/jobClassification.test.ts`
+fails if a `jobName` in `src/jobs/` is missing a row.
+
+| Class | Meaning |
+| --- | --- |
+| `locked` | Money-moving or destructive; must never run twice. |
+| `idempotent-by-design` | Safe to run twice (upsert / recompute). |
+| `safe-to-run-concurrently` | Designed for parallel runs (read + emit only). |
+
+| Job | Class | Notes |
+| --- | --- | --- |
+| `alert_rules` | idempotent-by-design | Re-evaluates rules; alerts are deduped |
+| `allocation_suggestions` | idempotent-by-design | Recomputes suggestions |
+| `approval_expiry_sweep` | locked | Expires approvals; double-run races the expiry write |
+| `fiat_reconciliation` | locked | Settles fiat orders; money-moving |
+| `linked_external_wallet_sync` | idempotent-by-design | Refreshes wallet snapshots |
+| `loan-accrual` | idempotent-by-design | Accrual is safe to apply twice on two replicas by design |
+| `loan-liquidation-monitor` | locked | Triggers liquidations; money-moving |
+| `message_delivery_sweep` | idempotent-by-design | Re-delivers undelivered messages |
+| `performance_attribution` | idempotent-by-design | Recomputes attribution windows |
+| `pool_metrics` | safe-to-run-concurrently | Read-only gauge mirror |
+| `portfolio_risk` | idempotent-by-design | Upserts risk aggregates |
+| `protocol_risk_scoring` | idempotent-by-design | Recomputes scores |
+| `recurring_deposits` | locked | Submits deposits; money-moving |
+| `recurring_withdrawals` | locked | Submits withdrawals; money-moving |
+| `referral_payout` | locked | Pays referral rewards; money-moving |
+| `reserve_reconciliation` | idempotent-by-design | Reconciles sponsor state |
+| `retention_all_jobs` | locked | Destructive deletes |
+| `risk_metadata_staleness` | idempotent-by-design | Marks stale metadata |
+| `session_cleanup` | locked | Destructive deletes |
+| `stellar_signer_rotation` | locked | Rotates signers; must not interleave |
+| `strategy_metrics` | idempotent-by-design | Refreshes marketplace metrics |
+| `support_ticket_sla` | idempotent-by-design | Updates SLA fields |
+
 ## Accessing Metrics
 
 The `/metrics` endpoint is internal and requires a valid bearer token:

src/jobs/jobClassification.ts

@@ -0,0 +1,40 @@
+/**
+ * Cross-pod concurrency classification for every scheduled job (#636).
+ *
+ * `scheduleResilientJob` takes a per-tick advisory lock so exactly one replica
+ * runs each job per tick. This table records what would happen without that
+ * lock, so a new job cannot ship unclassified: tests/unit/jobs/jobClassification.test.ts
+ * fails if a `jobName` appears in src/jobs/ that is not listed here.
+ *
+ * - `locked`                     — money-moving or destructive; must never run twice.
+ * - `idempotent-by-design`       — safe to run twice (upsert / recompute).
+ * - `safe-to-run-concurrently`   — designed for parallel runs (read + emit only).
+ */
+
+export type JobConcurrencyClass =
+  'locked' | 'idempotent-by-design' | 'safe-to-run-concurrently'
+
+export const JOB_CLASSIFICATION: Record<string, JobConcurrencyClass> = {
+  alert_rules: 'idempotent-by-design',
+  allocation_suggestions: 'idempotent-by-design',
+  approval_expiry_sweep: 'locked',
+  fiat_reconciliation: 'locked',
+  linked_external_wallet_sync: 'idempotent-by-design',
+  'loan-accrual': 'idempotent-by-design',
+  'loan-liquidation-monitor': 'locked',
+  message_delivery_sweep: 'idempotent-by-design',
+  performance_attribution: 'idempotent-by-design',
+  pool_metrics: 'safe-to-run-concurrently',
+  portfolio_risk: 'idempotent-by-design',
+  protocol_risk_scoring: 'idempotent-by-design',
+  recurring_deposits: 'locked',
+  recurring_withdrawals: 'locked',
+  referral_payout: 'locked',
+  reserve_reconciliation: 'idempotent-by-design',
+  retention_all_jobs: 'locked',
+  risk_metadata_staleness: 'idempotent-by-design',
+  session_cleanup: 'locked',
+  stellar_signer_rotation: 'locked',
+  strategy_metrics: 'idempotent-by-design',
+  support_ticket_sla: 'idempotent-by-design',
+}

src/jobs/resilientScheduler.ts

@@ -7,17 +7,60 @@
  * in flight at a crash is picked up again. Overlapping ticks do not start a
  * second run; a tick that arrives mid-run is resumed once the current attempt
  * finishes, unless that attempt already exhausted its retries.
+ *
+ * Cross-pod safety (#636): each tick first takes a Postgres advisory lock
+ * keyed by the job name, so in a multi-replica deployment exactly one pod
+ * runs the job per tick. The lock is a `pg_try_advisory_xact_lock` held for
+ * the duration of the run and released when the wrapping transaction ends.
+ * A tick that loses the lock contest is skipped, not queued.
  */
 import { logger } from '../utils/logger'
 import { computeBackoffMs } from '../outbox/stateMachine'
 import { recordJobCompletion, recordJobRetry } from '../utils/job-metrics'
+import db from '../db'
 
 export const JOB_MAX_ATTEMPTS = 3
 export const JOB_BACKOFF_BASE_MS = 1_000
 export const JOB_BACKOFF_MAX_MS = 30_000
 
 export type JobRunStatus = 'success' | 'failed'
 
+/**
+ * Wraps one tick so at most one replica executes the job. The guard simply
+ * declines to call `task` when another replica already holds the lock.
+ */
+export type JobRunGuard = (
+  jobName: string,
+  task: () => Promise<void>
+) => Promise<void>
+
+/**
+ * Per-tick cross-pod lock (#636). `pg_try_advisory_xact_lock` is used rather
+ * than session-scoped `pg_advisory_lock` because Prisma hands pooled
+ * connections back between calls — a session lock would leak across queries.
+ * The xact lock is released when the transaction commits or rolls back.
+ */
+export const withPostgresJobLock: JobRunGuard = async (jobName, task) => {
+  await db.$transaction(
+    async (tx) => {
+      const rows: Array<{ locked: boolean }> = await tx.$queryRaw`
+        SELECT pg_try_advisory_xact_lock(hashtext(${jobName})) AS locked
+      `
+      if (!rows[0]?.locked) {
+        logger.info(
+          `[JobScheduler] ${jobName} skipped; another replica holds the tick lock`
+        )
+        return
+      }
+      await task()
+    },
+    { timeout: 600_000, maxWait: 2_000 }
+  )
+}
+
+/** Escape hatch for tests and single-replica local runs. */
+export const withoutJobLock: JobRunGuard = (_jobName, task) => task()
+
 export type JobRetryOptions = {
   maxAttempts?: number
   baseDelayMs?: number
@@ -121,10 +164,18 @@ export function scheduleResilientJob(
     task: () => Promise<void>
     intervalMs: number
     unref?: boolean
+    /**
+     * Cross-pod tick guard. Defaults to the Postgres advisory lock outside
+     * tests; pass `withoutJobLock` or a custom shared lock in tests.
+     */
+    guard?: JobRunGuard
   } & JobRetryOptions
 ): NodeJS.Timeout {
   const slot: JobSlot = { running: false, resume: false }
   slots.set(opts.jobName, slot)
+  const guard: JobRunGuard =
+    opts.guard ??
+    (process.env.NODE_ENV === 'test' ? withoutJobLock : withPostgresJobLock)
 
   const tick = (): void => {
     if (slot.running) {
@@ -138,13 +189,15 @@ export function scheduleResilientJob(
     slot.running = true
     void (async () => {
       try {
-        do {
-          slot.resume = false
-          const status = await runJobWithRetry(opts.jobName, opts.task, opts)
-          // A terminal failure must wait for the next interval. Honouring
-          // resume here would tight-loop a job that is still broken.
-          if (status === 'failed') slot.resume = false
-        } while (slot.resume)
+        await guard(opts.jobName, async () => {
+          do {
+            slot.resume = false
+            const status = await runJobWithRetry(opts.jobName, opts.task, opts)
+            // A terminal failure must wait for the next interval. Honouring
+            // resume here would tight-loop a job that is still broken.
+            if (status === 'failed') slot.resume = false
+          } while (slot.resume)
+        })
       } finally {
         slot.running = false
       }

src/jobs/treasurySweep.ts

@@ -6,18 +6,19 @@
  * evaluateTreasuryBalances reads each active TreasuryAccount and its tier
  * pair's active TreasurySweepPolicy, planning a sweep whenever the hot-side
  * balance exceeds the policy's maxHotBalance by more than minSweepAmount.
- * executeSweep gates on requiresApprovalAbove via a MultisigEnvelope (the
- * treasury-specific multisig primitive already in schema) rather than the
- * sub-account ApprovalPolicy engine, which has no concept of a
- * non-user-owned system sweep — see docs/TREASURY.md.
+ * executeSweep refuses amounts above requiresApprovalAbove outright: the
+ * MultisigEnvelope lifecycle (sign -> assemble -> enqueue) is not implemented,
+ * so opening an envelope would strand the sweep in PLANNED forever. The
+ * refusal is loud — a thrown error and a critical alert — so an operator
+ * either lowers the amount or runs the emergency sweep path. See docs/TREASURY.md.
  */
 
 import { TreasuryTier } from '@prisma/client'
 import db from '../db'
 import { logger } from '../utils/logger'
 import { getActivePolicy } from '../treasury/policy'
 import { enqueueOutboxOp } from '../outbox/service'
-import { getAccount } from '../stellar/client'
+import { getNativeBalance } from '../stellar/client'
 import { alertingService } from '../services/alerting'
 
 const TREASURY_SWEEP_INTERVAL_MS = 300_000 // 5 minutes
@@ -36,13 +37,6 @@ const TIER_PAIRS: Array<[TreasuryTier, TreasuryTier]> = [
   ['WARM', 'COLD'],
 ]
 
-async function getNativeBalance(publicKey: string): Promise<number> {
-  const account: any = await getAccount(publicKey).catch(() => null)
-  if (!account?.balances) return 0
-  const native = account.balances.find((b: any) => b.asset_type === 'native')
-  return native ? parseFloat(native.balance) : 0
-}
-
 export interface SweepPreviewItem extends SweepPlan {
   policyId: string
   policyVersion: number
@@ -65,7 +59,7 @@ async function planSweeps(
     })
 
     for (const account of accounts) {
-      const balance = await getNativeBalance(account.publicKey)
+      const balance = await getNativeBalance(account.publicKey).catch(() => 0)
       const over = balance - Number(policy.maxHotBalance)
       if (over <= Number(policy.minSweepAmount)) continue
 
@@ -148,6 +142,24 @@ async function runSweep(plan: SweepPlan): Promise<void> {
     )
   }
 
+  const requiresApproval =
+    policy?.requiresApprovalAbove != null &&
+    Number(plan.amount) > Number(policy.requiresApprovalAbove)
+
+  if (requiresApproval) {
+    // #655 — the MultisigEnvelope lifecycle stops at PENDING: nothing signs,
+    // assembles or enqueues it. Opening one would strand the sweep in PLANNED
+    // with funds never moved, so large sweeps are refused until that lifecycle
+    // exists. Operators see this via the thrown error and the critical alert
+    // executeSweep emits on failure.
+    throw new Error(
+      `Sweep of ${plan.amount} ${plan.asset} exceeds requiresApprovalAbove ` +
+        `(${policy.requiresApprovalAbove}) and is disabled: the multisig ` +
+        `envelope lifecycle is not implemented. Lower the amount or use the ` +
+        `emergency sweep path.`
+    )
+  }
+
   const sweep = await db.treasurySweep.create({
     data: {
       fromTier: plan.fromTier,
@@ -159,35 +171,6 @@ async function runSweep(plan: SweepPlan): Promise<void> {
     },
   })
 
-  const requiresApproval =
-    policy?.requiresApprovalAbove != null &&
-    Number(plan.amount) > Number(policy.requiresApprovalAbove)
-
-  if (requiresApproval) {
-    // Large sweeps are held behind the treasury's own multisig primitive
-    // rather than the sub-account ApprovalPolicy engine (which has no
-    // concept of a system-initiated, non-user-owned operation) — an admin
-    // must collect signatures via src/stellar/multisig.ts before this
-    // envelope reaches READY and a follow-up call enqueues the outbox op.
-    const envelope = await db.multisigEnvelope.create({
-      data: {
-        sweepId: sweep.id,
-        publicKey: '',
-        threshold: 1,
-        signatures: [],
-        status: 'PENDING',
-        expiresAt: new Date(Date.now() + 24 * 60 * 60 * 1000),
-      },
-    })
-
-    logger.info('[TreasurySweep] Sweep held for multisig approval', {
-      sweepId: sweep.id,
-      envelopeId: envelope.id,
-      amount: plan.amount,
-    })
-    return
-  }
-
   const op = await enqueueOutboxOp(db, {
     idempotencyKey: `TREASURY_SWEEP:${sweep.id}`,
     userId: 'SYSTEM',