From 1f2069676c251a13dcddc7abe17bc55f636f0f47 Mon Sep 17 00:00:00 2001 From: AbelOsaretin Date: Tue, 20 Oct 2026 13:17:41 +0100 Subject: [PATCH 1/4] fix: read treasury native balance from RPC account entry rpc.Server.getAccount returns an Account with only a sequence number, so the old balances-array parse always yielded 0 and no sweep ever planned. Read the on-chain account entry instead and convert stroops to XLM. Closes #654 --- src/jobs/treasurySweep.ts | 11 +--- src/stellar/client.ts | 15 ++++++ tests/unit/jobs/treasurySweep.test.ts | 14 ++--- tests/unit/stellar/nativeBalance.test.ts | 67 ++++++++++++++++++++++++ 4 files changed, 89 insertions(+), 18 deletions(-) create mode 100644 tests/unit/stellar/nativeBalance.test.ts diff --git a/src/jobs/treasurySweep.ts b/src/jobs/treasurySweep.ts index 8a2a839..49f2a8b 100644 --- a/src/jobs/treasurySweep.ts +++ b/src/jobs/treasurySweep.ts @@ -17,7 +17,7 @@ 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 +36,6 @@ const TIER_PAIRS: Array<[TreasuryTier, TreasuryTier]> = [ ['WARM', 'COLD'], ] -async function getNativeBalance(publicKey: string): Promise { - 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 +58,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 diff --git a/src/stellar/client.ts b/src/stellar/client.ts index a8f8137..77d578a 100644 --- a/src/stellar/client.ts +++ b/src/stellar/client.ts @@ -270,6 +270,21 @@ export async function getAccount(publicKey: string): Promise { ) } +/** + * Native (XLM) balance for an account, in XLM. + * + * `rpc.Server.getAccount` returns an Account with only a sequence number, so + * the balance must be read from the on-chain account entry that + * `getAccountEntry` resolves (`balance()` is in stroops). + */ +export async function getNativeBalance(publicKey: string): Promise { + const entry = await getResilientClient().execute( + (server) => server.getAccountEntry(publicKey), + 'stellar.getNativeBalance' + ) + return Number(entry.balance().toString()) / 1e7 +} + const MAX_EXTERNAL_WALLET_RESPONSE_BYTES = 128 * 1024 function horizonBaseUrl(): string { diff --git a/tests/unit/jobs/treasurySweep.test.ts b/tests/unit/jobs/treasurySweep.test.ts index ba433f6..bd71fce 100644 --- a/tests/unit/jobs/treasurySweep.test.ts +++ b/tests/unit/jobs/treasurySweep.test.ts @@ -6,7 +6,7 @@ import { validateHysteresis, } from '../../../src/jobs/treasurySweep' import db from '../../../src/db' -import { getAccount } from '../../../src/stellar/client' +import { getNativeBalance } from '../../../src/stellar/client' import { enqueueOutboxOp } from '../../../src/outbox/service' import { alertingService } from '../../../src/services/alerting' @@ -21,7 +21,7 @@ jest.mock('../../../src/db', () => ({ }, })) jest.mock('../../../src/stellar/client', () => ({ - getAccount: jest.fn(), + getNativeBalance: jest.fn(), })) jest.mock('../../../src/outbox/service', () => ({ enqueueOutboxOp: jest.fn(), @@ -34,7 +34,7 @@ jest.mock('../../../src/utils/logger', () => ({ })) const mockDb = db as any -const mockGetAccount = getAccount as jest.Mock +const mockGetNativeBalance = getNativeBalance as jest.Mock const mockEnqueue = enqueueOutboxOp as jest.Mock beforeEach(() => { @@ -75,9 +75,7 @@ describe('evaluateTreasuryBalances', () => { } return Promise.resolve([]) }) - mockGetAccount.mockResolvedValue({ - balances: [{ asset_type: 'native', balance: '8000' }], - }) + mockGetNativeBalance.mockResolvedValue(8000) const plans = await evaluateTreasuryBalances() @@ -228,9 +226,7 @@ describe('previewTreasurySweep (#620)', () => { Promise.resolve(where.fromTier === 'HOT' ? policy : null) ) mockDb.treasuryAccount.findMany.mockResolvedValue([{ publicKey: 'GHOT' }]) - mockGetAccount.mockResolvedValue({ - balances: [{ asset_type: 'native', balance: '6000' }], - }) + mockGetNativeBalance.mockResolvedValue(6000) }) it('reports plans for a disabled policy without writing anything', async () => { diff --git a/tests/unit/stellar/nativeBalance.test.ts b/tests/unit/stellar/nativeBalance.test.ts new file mode 100644 index 0000000..883acac --- /dev/null +++ b/tests/unit/stellar/nativeBalance.test.ts @@ -0,0 +1,67 @@ +/** + * #654 — getNativeBalance must read the on-chain account entry. + * + * `rpc.Server.getAccount` returns an Account with only a sequence number, so + * the balance has to come from `getAccountEntry`, whose `balance()` is in + * stroops. The fixture below is the shape that call actually returns. + */ +jest.mock('../../../src/utils/rpc-metrics', () => ({ + rpcAttemptCounter: { inc: jest.fn() }, + rpcFailoverCounter: { inc: jest.fn() }, + rpcCircuitOpenCounter: { inc: jest.fn() }, + rpcLatencyHistogram: { startTimer: jest.fn(() => jest.fn()) }, +})) + +jest.mock('../../../src/config', () => ({ + config: { + stellar: { network: 'testnet', rpcUrl: 'https://rpc.example.com' }, + httpClient: { + timeoutMs: 5_000, + maxRetries: 1, + baseDelayMs: 0, + maxDelayMs: 0, + circuitBreakerThreshold: 5, + circuitBreakerResetMs: 60_000, + }, + }, +})) + +jest.mock('../../../src/utils/logger', () => ({ + logger: { info: jest.fn(), warn: jest.fn(), error: jest.fn() }, +})) + +import { + getNativeBalance, + getResilientClient, +} from '../../../src/stellar/client' + +/** Shape of `rpc.Server.getAccountEntry(...)` — an xdr.AccountEntry. */ +function rpcAccountEntryFixture(balanceStroops: string, subentryCount = 2) { + return { + balance: () => ({ toString: () => balanceStroops }), + subentryCount: () => subentryCount, + seqNum: () => ({ toString: () => '9182736450' }), + } +} + +describe('getNativeBalance', () => { + it('converts the account entry balance from stroops to XLM', async () => { + const execute = jest + .spyOn(getResilientClient(), 'execute') + .mockResolvedValue(rpcAccountEntryFixture('80000000000') as never) + + await expect(getNativeBalance('GHOT')).resolves.toBe(8000) + expect(execute).toHaveBeenCalledWith( + expect.any(Function), + 'stellar.getNativeBalance' + ) + }) + + it('returns 0 for an account with no balance', async () => { + jest + .spyOn(getResilientClient(), 'execute') + .mockResolvedValue(rpcAccountEntryFixture('0') as never) + + await expect(getNativeBalance('GHOT')).resolves.toBe(0) + }) +}) From 09eae24b49f20b3d5c0de294a1f1ab35d5582a1e Mon Sep 17 00:00:00 2001 From: AbelOsaretin Date: Tue, 20 Oct 2026 13:20:01 +0100 Subject: [PATCH 2/4] fix: refuse large treasury sweeps instead of stranding a MultisigEnvelope MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The multisig envelope lifecycle stops at PENDING — nothing signs, assembles or enqueues it — so sweeps above requiresApprovalAbove sat in PLANNED with funds never moved. Refuse those amounts with a thrown error and a critical alert before any sweep row is written. Closes #655 --- src/jobs/treasurySweep.ts | 56 +++++++++++---------------- tests/unit/jobs/treasurySweep.test.ts | 34 ++++++++-------- 2 files changed, 39 insertions(+), 51 deletions(-) diff --git a/src/jobs/treasurySweep.ts b/src/jobs/treasurySweep.ts index 49f2a8b..dd581ce 100644 --- a/src/jobs/treasurySweep.ts +++ b/src/jobs/treasurySweep.ts @@ -6,10 +6,11 @@ * 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' @@ -141,6 +142,24 @@ async function runSweep(plan: SweepPlan): Promise { ) } + 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, @@ -152,35 +171,6 @@ async function runSweep(plan: SweepPlan): Promise { }, }) - 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', diff --git a/tests/unit/jobs/treasurySweep.test.ts b/tests/unit/jobs/treasurySweep.test.ts index bd71fce..3c5d2c6 100644 --- a/tests/unit/jobs/treasurySweep.test.ts +++ b/tests/unit/jobs/treasurySweep.test.ts @@ -134,33 +134,31 @@ describe('executeSweep', () => { ) }) - it('opens a MultisigEnvelope instead of enqueueing when amount exceeds requiresApprovalAbove', async () => { + it('refuses a sweep above requiresApprovalAbove with an operator-visible error (#655)', async () => { mockDb.treasurySweepPolicy.findFirst.mockResolvedValue({ fromTier: 'HOT', toTier: 'WARM', requiresApprovalAbove: '500', sweepsEnabled: true, }) - mockDb.treasurySweep.create.mockResolvedValue({ id: 'sweep-2' }) - mockDb.multisigEnvelope.create.mockResolvedValue({ id: 'env-1' }) - await executeSweep({ - fromTier: 'HOT', - toTier: 'WARM', - asset: 'XLM', - amount: '1000', - reason: 'hot_over_high_band', - }) - - expect(mockDb.multisigEnvelope.create).toHaveBeenCalledWith( - expect.objectContaining({ - data: expect.objectContaining({ - sweepId: 'sweep-2', - status: 'PENDING', - }), + await expect( + executeSweep({ + fromTier: 'HOT', + toTier: 'WARM', + asset: 'XLM', + amount: '1000', + reason: 'hot_over_high_band', }) - ) + ).rejects.toThrow(/exceeds requiresApprovalAbove/) + + expect(mockDb.treasurySweep.create).not.toHaveBeenCalled() + expect(mockDb.multisigEnvelope.create).not.toHaveBeenCalled() expect(mockEnqueue).not.toHaveBeenCalled() + expect(alertingService.emit).toHaveBeenCalledWith( + expect.objectContaining({ severity: 'critical' }), + expect.stringContaining('treasury-sweep-failed') + ) }) }) From c59f2e0b62fe7cb2f0f7d0f9eba9a7c856fa1b95 Mon Sep 17 00:00:00 2001 From: AbelOsaretin Date: Tue, 20 Oct 2026 13:27:50 +0100 Subject: [PATCH 3/4] fix: take a per-tick advisory lock so one replica runs each job MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every pod ran every scheduled job on the same tick, so money-moving and destructive work depended on each job being idempotent by convention. Each tick now takes a pg_try_advisory_xact_lock keyed by job name — one winner, the rest skip — and every job is classified in docs/job-observability.md. Closes #636 --- docs/job-observability.md | 38 +++++++++ src/jobs/jobClassification.ts | 40 +++++++++ src/jobs/resilientScheduler.ts | 67 +++++++++++++-- tests/unit/jobs/jobClassification.test.ts | 61 ++++++++++++++ tests/unit/jobs/resilientScheduler.test.ts | 95 ++++++++++++++++++++++ 5 files changed, 294 insertions(+), 7 deletions(-) create mode 100644 src/jobs/jobClassification.ts create mode 100644 tests/unit/jobs/jobClassification.test.ts diff --git a/docs/job-observability.md b/docs/job-observability.md index 0fe678a..5baded9 100644 --- a/docs/job-observability.md +++ b/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: diff --git a/src/jobs/jobClassification.ts b/src/jobs/jobClassification.ts new file mode 100644 index 0000000..2294bd3 --- /dev/null +++ b/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 = { + 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', +} diff --git a/src/jobs/resilientScheduler.ts b/src/jobs/resilientScheduler.ts index c102bec..20b8401 100644 --- a/src/jobs/resilientScheduler.ts +++ b/src/jobs/resilientScheduler.ts @@ -7,10 +7,17 @@ * 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 @@ -18,6 +25,42 @@ 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 +) => Promise + +/** + * 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 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 } diff --git a/tests/unit/jobs/jobClassification.test.ts b/tests/unit/jobs/jobClassification.test.ts new file mode 100644 index 0000000..23b4650 --- /dev/null +++ b/tests/unit/jobs/jobClassification.test.ts @@ -0,0 +1,61 @@ +/** + * #636 — every scheduled job must appear in the concurrency classification + * table. An unclassified job fails review: adding `jobName: 'new_thing'` to + * src/jobs/ without a row here breaks this suite. + */ +import * as fs from 'fs' +import * as path from 'path' +import { JOB_CLASSIFICATION } from '../../../src/jobs/jobClassification' + +const JOBS_DIR = path.resolve(__dirname, '../../../src/jobs') + +function collectScheduledJobNames(): string[] { + const names = new Set() + for (const file of fs.readdirSync(JOBS_DIR)) { + if (!file.endsWith('.ts')) continue + const source = fs.readFileSync(path.join(JOBS_DIR, file), 'utf8') + // jobName: 'literal' + for (const match of source.matchAll(/jobName:\s*'([^']+)'/g)) { + names.add(match[1]) + } + // jobName: CONSTANT → resolve `export const CONSTANT = 'value'` + for (const match of source.matchAll(/jobName:\s*([A-Z][A-Z0-9_]*)\b/g)) { + const constName = match[1] + const def = new RegExp( + `(?:export\\s+)?const\\s+${constName}\\s*=\\s*'([^']+)'` + ).exec(source) + if (def) names.add(def[1]) + } + } + return [...names].sort() +} + +describe('job classification (#636)', () => { + it('classifies every job scheduled through scheduleResilientJob', () => { + const scheduled = collectScheduledJobNames() + expect(scheduled.length).toBeGreaterThan(15) + + const unclassified = scheduled.filter((n) => !(n in JOB_CLASSIFICATION)) + expect(unclassified).toEqual([]) + }) + + it('has no classification rows for jobs that are not scheduled', () => { + const scheduled = new Set(collectScheduledJobNames()) + const orphans = Object.keys(JOB_CLASSIFICATION).filter( + (n) => !scheduled.has(n) + ) + expect(orphans).toEqual([]) + }) + + it('labels every entry with a known concurrency class', () => { + const allowed = new Set([ + 'locked', + 'idempotent-by-design', + 'safe-to-run-concurrently', + ]) + for (const [name, cls] of Object.entries(JOB_CLASSIFICATION)) { + expect({ name, cls }).toEqual({ name, cls: expect.any(String) }) + expect(allowed.has(cls)).toBe(true) + } + }) +}) diff --git a/tests/unit/jobs/resilientScheduler.test.ts b/tests/unit/jobs/resilientScheduler.test.ts index 344b92d..d42dbd5 100644 --- a/tests/unit/jobs/resilientScheduler.test.ts +++ b/tests/unit/jobs/resilientScheduler.test.ts @@ -172,3 +172,98 @@ describe('scheduleResilientJob', () => { clearInterval(handle) }) }) + +describe('cross-pod tick guard (#636)', () => { + afterEach(() => { + jest.useRealTimers() + }) + + /** + * Shared mutex standing in for `pg_try_advisory_xact_lock(hashtext(job))`: + * only one holder at a time, release returns the key. + */ + function createSharedLock(): { + guard: (jobName: string, task: () => Promise) => Promise + holders: Set + } { + const holders = new Set() + return { + holders, + guard: async (jobName, task) => { + if (holders.has(jobName)) return + holders.add(jobName) + try { + await task() + } finally { + holders.delete(jobName) + } + }, + } + } + + it('runs a job once per tick when two scheduler instances share one lock', async () => { + jest.useFakeTimers() + let release: () => void = () => undefined + let calls = 0 + const task = () => { + calls += 1 + return new Promise((resolve) => { + release = resolve + }) + } + + const lock = createSharedLock() + + // Two "replicas" — two schedulers for the same job name against one lock. + const handleA = scheduleResilientJob({ + jobName: 'shared-tick-job', + task, + intervalMs: 1000, + sleep: async () => undefined, + guard: lock.guard, + }) + const handleB = scheduleResilientJob({ + jobName: 'shared-tick-job', + task, + intervalMs: 1000, + sleep: async () => undefined, + guard: lock.guard, + }) + + // Both schedulers fire their immediate tick; the lock lets only one through. + for (let i = 0; i < 10; i++) await Promise.resolve() + expect(calls).toBe(1) + + release() + for (let i = 0; i < 10; i++) await Promise.resolve() + + clearInterval(handleA) + clearInterval(handleB) + }) + + it('skips a tick when another replica already holds the lock', async () => { + let calls = 0 + const task = async () => { + calls += 1 + } + + const holders = new Set(['locked-job']) + const guard = async (jobName: string, t: () => Promise) => { + if (holders.has(jobName)) return + await t() + } + + const handle = scheduleResilientJob({ + jobName: 'locked-job', + task, + intervalMs: 60_000, + sleep: async () => undefined, + guard, + }) + + for (let i = 0; i < 10; i++) await Promise.resolve() + expect(calls).toBe(0) + + clearInterval(handle) + }) +}) From 8332baed657db21720467492e7ac247051225c7b Mon Sep 17 00:00:00 2001 From: AbelOsaretin Date: Tue, 20 Oct 2026 14:17:06 +0100 Subject: [PATCH 4/4] fix: back express-rate-limit with a shared Redis counter MemoryStore counted per pod, so n replicas enforced n x the documented tier limits. Counters now live in Redis shared by every replica; NODE_ENV=test keeps the in-process store so unit tests never need Redis. Retry-After and RateLimit-* headers derive from the shared counter. Closes #635 --- docs/RATE_LIMITING.md | 18 +++ src/middleware/rateLimitStore.ts | 103 +++++++++++++++ src/middleware/rateLimiter.ts | 5 + tests/unit/middleware/rateLimitStore.test.ts | 130 +++++++++++++++++++ 4 files changed, 256 insertions(+) create mode 100644 src/middleware/rateLimitStore.ts create mode 100644 tests/unit/middleware/rateLimitStore.test.ts diff --git a/docs/RATE_LIMITING.md b/docs/RATE_LIMITING.md index 2219638..78a6581 100644 --- a/docs/RATE_LIMITING.md +++ b/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** diff --git a/src/middleware/rateLimitStore.ts b/src/middleware/rateLimitStore.ts new file mode 100644 index 0000000..c963da3 --- /dev/null +++ b/src/middleware/rateLimitStore.ts @@ -0,0 +1,103 @@ +/** + * Shared (cross-replica) hit counter for express-rate-limit (#635). + * + * The default MemoryStore counts per process, so n replicas behind a load + * balancer enforce n x the documented budget. This store keeps the counter in + * the same Redis the cache/event-bridge already use, so every replica draws + * on one number. NODE_ENV=test keeps the in-process store: unit tests assert + * header behaviour and must not need Redis. + */ +import type { + ClientRateLimitInfo, + IncrementResponse, + Options, + Store, +} from 'express-rate-limit' +import { getRedisClient } from '../config/redis' +import { logger } from '../utils/logger' + +type RedisLike = { + incr(key: string): Promise + pttl(key: string): Promise + pexpire(key: string, ms: number): Promise + del(key: string): Promise +} + +export class RedisRateLimitStore implements Store { + prefix: string + localKeys = false + + private windowMs = 60_000 + private readonly clientOverride?: RedisLike + + constructor(prefix = 'rl:', clientOverride?: RedisLike) { + this.prefix = prefix + this.clientOverride = clientOverride + } + + init(options: Options): void { + this.windowMs = options.windowMs + } + + private client(): RedisLike | null { + return (this.clientOverride ?? getRedisClient()) as RedisLike | null + } + + async increment(key: string): Promise { + const client = this.client() + if (!client) { + throw new Error('Redis unavailable for rate-limit store') + } + const redisKey = this.prefix + key + const totalHits = await client.incr(redisKey) + if (totalHits === 1) { + await client.pexpire(redisKey, this.windowMs) + } + const pttl = await client.pttl(redisKey) + return { + totalHits, + resetTime: new Date(Date.now() + (pttl > 0 ? pttl : this.windowMs)), + } + } + + async get(key: string): Promise { + const client = this.client() + if (!client) return undefined + const redisKey = this.prefix + key + const pttl = await client.pttl(redisKey) + if (pttl < 0) return undefined + return { + totalHits: 0, + resetTime: new Date(Date.now() + pttl), + } + } + + async decrement(): Promise { + // express-rate-limit only calls this to correct double-counting; the + // increment above is the source of truth and is not rolled back. + } + + async resetKey(key: string): Promise { + const client = this.client() + if (!client) return + await client.del(this.prefix + key) + } +} + +/** + * Store used by every limiter. In tests this is left undefined so + * express-rate-limit falls back to its in-process MemoryStore; everywhere + * else it is the shared Redis counter. If Redis is not configured the + * limiter still works, but per-pod — a warning says so. + */ +export function resolveRateLimitStore(limiterType: string): Store | undefined { + if (process.env.NODE_ENV === 'test') return undefined + if (!getRedisClient()) { + logger.warn( + '[rate-limit] REDIS_URL not set — counters are per-pod; multi-replica deployments will exceed documented limits', + { limiterType } + ) + return undefined + } + return new RedisRateLimitStore(`rl:${limiterType}:`) +} diff --git a/src/middleware/rateLimiter.ts b/src/middleware/rateLimiter.ts index a179b7a..fae2573 100644 --- a/src/middleware/rateLimiter.ts +++ b/src/middleware/rateLimiter.ts @@ -1,5 +1,6 @@ import { type Request, type Response, type NextFunction } from 'express' import rateLimit, { type RateLimitRequestHandler } from 'express-rate-limit' +import { resolveRateLimitStore } from './rateLimitStore' import { config } from '../config/env' import { recordRateLimitHit, updateRateLimitViolations } from '../utils/metrics' import { logger } from '../utils/logger' @@ -311,6 +312,9 @@ export function buildRateLimiter( const limiter: RateLimitRequestHandler = rateLimit({ windowMs: opts.windowMs, max: resolveMax, + // #635 — shared Redis counter in multi-replica deployments; in-process + // MemoryStore only under NODE_ENV=test (and when Redis is unconfigured). + store: resolveRateLimitStore(opts.limiterType), ...(usePrincipal && { keyGenerator: (req: Request) => resolvePrincipal(req).key, }), @@ -373,6 +377,7 @@ export const tieredRateLimiter = (() => { const limiter: RateLimitRequestHandler = rateLimit({ windowMs, + store: resolveRateLimitStore('tiered'), max: async (req: Request) => { // An API key's own allowance always wins over both defaults. const token = req.header('Authorization')?.replace(/^Bearer\s+/, '') diff --git a/tests/unit/middleware/rateLimitStore.test.ts b/tests/unit/middleware/rateLimitStore.test.ts new file mode 100644 index 0000000..a692e62 --- /dev/null +++ b/tests/unit/middleware/rateLimitStore.test.ts @@ -0,0 +1,130 @@ +/** + * #635 — rate-limit counters must be shared across replicas. + * + * Two express apps stand in for two pods. Each gets its own store instance + * (each pod has its own memory) but both talk to one Redis. A client that + * exhausts the budget on pod A must be throttled on pod B, and the + * RateLimit and Retry-After headers must reflect the shared counter. + */ +import express from 'express' +import request from 'supertest' +import rateLimit from 'express-rate-limit' +import { RedisRateLimitStore } from '../../../src/middleware/rateLimitStore' + +/** Minimal ioredis surface the store uses, backed by one shared Map. */ +function createFakeRedis() { + const data = new Map() + const now = () => Date.now() + + function entry(key: string) { + const row = data.get(key) + if (!row) return null + if (row.expiresAt !== null && row.expiresAt <= now()) { + data.delete(key) + return null + } + return row + } + + return { + async incr(key: string) { + const row = entry(key) ?? { value: 0, expiresAt: null } + row.value += 1 + data.set(key, row) + return row.value + }, + async pttl(key: string) { + const row = entry(key) + if (!row) return -2 + if (row.expiresAt === null) return -1 + return row.expiresAt - now() + }, + async pexpire(key: string, ms: number) { + const row = entry(key) + if (!row) return 0 + row.expiresAt = now() + ms + data.set(key, row) + return 1 + }, + async del(key: string) { + return data.delete(key) ? 1 : 0 + }, + } +} + +function buildApp(store: RedisRateLimitStore, windowMs: number, max: number) { + const app = express() + app.use( + rateLimit({ + windowMs, + max, + store, + standardHeaders: true, + legacyHeaders: false, + message: { error: 'Too many requests.' }, + }) + ) + app.get('/test', (_req, res) => res.status(200).json({ ok: true })) + return app +} + +describe('RedisRateLimitStore (#635)', () => { + it('enforces one global budget across two app instances against one Redis', async () => { + const redis = createFakeRedis() + // Two "pods": separate store objects, one Redis. + const podA = new RedisRateLimitStore('rl:test:', redis as never) + const podB = new RedisRateLimitStore('rl:test:', redis as never) + + const appA = buildApp(podA, 60_000, 5) + const appB = buildApp(podB, 60_000, 5) + + // Exhaust the budget on pod A: 5 x 200. + for (let i = 0; i < 5; i++) { + const res = await request(appA).get('/test') + expect(res.status).toBe(200) + } + + // The 6th request on pod A is throttled. + const blockedA = await request(appA).get('/test') + expect(blockedA.status).toBe(429) + + // pod B shares the counter — it must also throttle immediately. + const blockedB = await request(appB).get('/test') + expect(blockedB.status).toBe(429) + + // Headers derive from the shared counter. + expect(blockedB.headers['ratelimit-limit']).toBeDefined() + expect(blockedB.headers['ratelimit-remaining']).toBe('0') + expect(blockedB.headers['retry-after']).toBeDefined() + expect(Number(blockedB.headers['retry-after'])).toBeGreaterThan(0) + }) + + it('decrements the shared remaining count on every response', async () => { + const redis = createFakeRedis() + const podA = new RedisRateLimitStore('rl:hdr:', redis as never) + const podB = new RedisRateLimitStore('rl:hdr:', redis as never) + + const appA = buildApp(podA, 60_000, 10) + const appB = buildApp(podB, 60_000, 10) + + const first = await request(appA).get('/test') + expect(first.headers['ratelimit-remaining']).toBe('9') + + const second = await request(appB).get('/test') + expect(second.headers['ratelimit-remaining']).toBe('8') + }) + + it('keeps counters separate per limiter prefix', async () => { + const redis = createFakeRedis() + const auth = new RedisRateLimitStore('rl:auth:', redis as never) + const admin = new RedisRateLimitStore('rl:admin:', redis as never) + + const authApp = buildApp(auth, 60_000, 1) + const adminApp = buildApp(admin, 60_000, 1) + + expect((await request(authApp).get('/test')).status).toBe(200) + expect((await request(authApp).get('/test')).status).toBe(429) + // A different limiter type has its own bucket. + expect((await request(adminApp).get('/test')).status).toBe(200) + }) +})