From 17d11352872514bba8c5a3902bc76d507e398456 Mon Sep 17 00:00:00 2001 From: kingksjo Date: Sun, 18 Oct 2026 14:17:47 +0100 Subject: [PATCH] fix: harden cache failover, idempotency leases, outbox relay, and route prefetch - backend(cache): expose stale fail-open reads with headers/metrics and a bypass path (#1553) - backend(idempotency): add a processing lease timeout with safe reclaim (#1556) - backend(outbox): add lease expiry, fencing tokens, and a stuck-relay reaper (#1557) - frontend(prefetch): cover heavy routes, wire nav hover/focus, and record LCP metrics (#1569) Repairs committed merge artifacts in idempotency.ts and eventOutbox.ts and aligns the idempotency store's dry-run return contract. --- backend/src/__tests__/cacheFailover.test.ts | 288 ++++++ .../src/__tests__/eventOutboxLease.test.ts | 153 +++ .../__tests__/idempotencyPendingLease.test.ts | 224 +++++ backend/src/eventOutbox.ts | 444 ++++++--- backend/src/idempotency.ts | 897 ++++-------------- backend/src/idempotencyStore.ts | 4 +- backend/src/middleware/cache.ts | 148 ++- backend/src/redisCache.ts | 168 +++- frontend/src/App.tsx | 33 +- frontend/src/components/Navbar.tsx | 40 +- frontend/src/i18n/locales/en.ts | 2 + frontend/src/i18n/locales/es.ts | 2 + frontend/src/lib/routePrefetch.test.ts | 53 ++ frontend/src/lib/routePrefetch.ts | 230 ++++- 14 files changed, 1761 insertions(+), 925 deletions(-) create mode 100644 backend/src/__tests__/cacheFailover.test.ts create mode 100644 backend/src/__tests__/eventOutboxLease.test.ts create mode 100644 backend/src/__tests__/idempotencyPendingLease.test.ts create mode 100644 frontend/src/lib/routePrefetch.test.ts diff --git a/backend/src/__tests__/cacheFailover.test.ts b/backend/src/__tests__/cacheFailover.test.ts new file mode 100644 index 000000000..dc34cbae1 --- /dev/null +++ b/backend/src/__tests__/cacheFailover.test.ts @@ -0,0 +1,288 @@ +/** + * @file cacheFailover.test.ts + * Regression tests for issue #1553. + * + * Covers the Redis fail-open path: + * - fallback reads are reported as stale (header + metadata + metrics) + * - stale responses are never cached downstream (`Cache-Control: no-store`) + * - `X-Cache-Bypass` / `Cache-Control: no-cache` force a fresh lookup + */ + +import { EventEmitter } from 'events'; +import type { Request, Response } from 'express'; + +class FakeRedis extends EventEmitter { + private store = new Map(); + + async ping() { + return 'PONG'; + } + + async get(key: string): Promise { + return this.store.get(key) ?? null; + } + + async setex(key: string, _ttl: number, value: string) { + this.store.set(key, value); + return 'OK'; + } + + async del(...keys: string[]) { + let removed = 0; + for (const key of keys) if (this.store.delete(key)) removed++; + return removed; + } + + async scan(cursor: string): Promise<[string, string[]]> { + return ['0', cursor === '0' ? [...this.store.keys()] : []]; + } + + async quit() {} +} + +import { + redisCacheClient, + getFromCache, + getFromCacheDetailed, + setInCache, + redisCacheFallbackCount, + redisCacheStaleServeCount, + redisCacheBypassCount, +} from '../redisCache'; +import { + responseCache, + cacheMiddleware, + isCacheBypassRequested, +} from '../middleware/cache'; + +let fakeRedis: FakeRedis; + +function makeEntry(data: unknown, ttlMs = 60_000) { + return { + data, + statusCode: 200, + headers: {} as Record, + expiresAt: Date.now() + ttlMs, + ttl: ttlMs, + lastUsed: Date.now(), + }; +} + +function makeFakeRequest( + path: string, + headers: Record = {}, +): Partial { + return { + method: 'GET', + path, + baseUrl: '', + query: {}, + headers, + route: { path }, + } as unknown as Partial; +} + +function makeFakeResponse(): { + res: Partial; + headers: Record; + getCaptured: () => unknown; + getStatus: () => number; +} { + const headers: Record = {}; + let captured: unknown; + let status = 200; + + const res = { + statusCode: 200, + setHeader(name: string, value: string | number) { + headers[name.toLowerCase()] = value; + }, + status(code: number) { + status = code; + (this as any).statusCode = code; + return this; + }, + json(data: unknown) { + captured = data; + return this; + }, + } as unknown as Partial; + + return { + res, + headers, + getCaptured: () => captured, + getStatus: () => status, + }; +} + +async function counterValue( + counter: import('prom-client').Counter, + labels: Record, +): Promise { + const values = await counter.get(); + const found = values.values.find((v) => + Object.entries(labels).every(([k, val]) => v.labels[k] === val), + ); + return found?.value ?? 0; +} + +beforeEach(() => { + fakeRedis = new FakeRedis(); + (redisCacheClient as any).client = fakeRedis; + (redisCacheClient as any)._isReady = true; + responseCache.clear(); +}); + +afterEach(() => { + (redisCacheClient as any)._isReady = false; + responseCache.clear(); +}); + +describe('Redis fail-open staleness (issue #1553)', () => { + it('marks an LRU hit as stale when Redis is configured but down', async () => { + responseCache.set('GET:/api/v1/vault/summary', makeEntry({ totalAssets: '10' })); + (redisCacheClient as any)._isReady = false; + + const result = await getFromCacheDetailed('GET:/api/v1/vault/summary'); + + expect(result.entry).not.toBeNull(); + expect(result.backend).toBe('memory'); + expect(result.fromFallback).toBe(true); + expect(result.stale).toBe(true); + expect(result.ageMs).not.toBeNull(); + }); + + it('does not mark an LRU hit as stale when Redis was never configured', async () => { + (redisCacheClient as any).client = null; + (redisCacheClient as any)._isReady = false; + responseCache.set('GET:/api/v1/vault/summary', makeEntry({ totalAssets: '10' })); + + const result = await getFromCacheDetailed('GET:/api/v1/vault/summary'); + + expect(result.entry).not.toBeNull(); + expect(result.fromFallback).toBe(true); + expect(result.stale).toBe(false); + }); + + it('keeps the backward-compatible getFromCache() return shape', async () => { + responseCache.set('GET:/api/v1/vault/summary', makeEntry({ totalAssets: '10' })); + (redisCacheClient as any)._isReady = false; + + const entry = await getFromCache('GET:/api/v1/vault/summary'); + expect((entry?.data as { totalAssets: string }).totalAssets).toBe('10'); + }); + + it('increments fallback and stale-serve metrics', async () => { + const route = 'GET:/api/v1/vault/metrics'; + responseCache.set(route, makeEntry({ apy: '8' })); + (redisCacheClient as any)._isReady = false; + + const beforeFallback = await counterValue(redisCacheFallbackCount, { + route, + reason: 'redis_unavailable', + }); + const beforeStale = await counterValue(redisCacheStaleServeCount, { + route, + backend: 'memory', + }); + + await getFromCacheDetailed(route); + + expect( + await counterValue(redisCacheFallbackCount, { route, reason: 'redis_unavailable' }), + ).toBe(beforeFallback + 1); + expect( + await counterValue(redisCacheStaleServeCount, { route, backend: 'memory' }), + ).toBe(beforeStale + 1); + }); + + it('serves fallback hits with staleness headers and no-store', async () => { + const route = '/api/v1/vault/summary'; + const key = `GET:${route}`; + responseCache.set(key, makeEntry({ totalAssets: '42' })); + (redisCacheClient as any)._isReady = false; + + const req = makeFakeRequest(route); + const { res, headers } = makeFakeResponse(); + + await new Promise((resolve) => { + (res as any).json = function () { + resolve(); + return res; + }; + + const middleware = cacheMiddleware({ ttl: 60_000 }); + middleware(req as Request, res as Response, () => { + throw new Error('next() must not be called for a fallback hit'); + }); + }); + + expect(headers['x-cache-hit']).toBe('true'); + expect(headers['x-cache-backend']).toBe('memory'); + expect(headers['x-cache-stale']).toBe('true'); + expect(headers['warning']).toContain('110'); + expect(headers['cache-control']).toBe('no-store'); + }); + + it('detects bypass requests from either header', () => { + expect( + isCacheBypassRequested(makeFakeRequest('/x', { 'x-cache-bypass': 'true' }) as Request), + ).toBe(true); + expect( + isCacheBypassRequested( + makeFakeRequest('/x', { 'cache-control': 'no-cache, no-store' }) as Request, + ), + ).toBe(true); + expect( + isCacheBypassRequested(makeFakeRequest('/x', { 'cache-control': 'max-age=60' }) as Request), + ).toBe(false); + }); + + it('bypasses a warm cache and re-populates it', async () => { + const route = '/api/v1/vault/summary'; + const key = `GET:${route}`; + responseCache.set(key, makeEntry({ totalAssets: 'stale' })); + + const req = makeFakeRequest(route, { 'x-cache-bypass': 'true' }); + const { res, headers, getCaptured } = makeFakeResponse(); + const beforeBypass = await counterValue(redisCacheBypassCount, { route: key }); + + let nextCalled = false; + await new Promise((resolve) => { + (res as any).json = function () { + resolve(); + return res; + }; + + const middleware = cacheMiddleware({ ttl: 60_000 }); + middleware(req as Request, res as Response, () => { + nextCalled = true; + (res as any).json({ totalAssets: 'fresh' }); + }); + }); + + expect(nextCalled).toBe(true); + expect(headers['x-cache-bypass']).toBe('true'); + expect(headers['x-cache-hit']).toBe('false'); + expect((getCaptured() as { totalAssets: string }).totalAssets).toBe('fresh'); + expect( + await counterValue(redisCacheBypassCount, { route: key }), + ).toBe(beforeBypass + 1); + + // The fresh response replaced the stale entry. + expect( + (responseCache.get(key)?.data as { totalAssets: string }).totalAssets, + ).toBe('fresh'); + }); + + it('reports Redis hits as non-stale', async () => { + const route = 'GET:/api/v1/vault/summary'; + await setInCache(route, makeEntry({ totalAssets: '99' }), 60_000); + + const result = await getFromCacheDetailed(route); + expect(result.backend).toBe('redis'); + expect(result.stale).toBe(false); + expect(result.fromFallback).toBe(false); + }); +}); diff --git a/backend/src/__tests__/eventOutboxLease.test.ts b/backend/src/__tests__/eventOutboxLease.test.ts new file mode 100644 index 000000000..02f49b7ed --- /dev/null +++ b/backend/src/__tests__/eventOutboxLease.test.ts @@ -0,0 +1,153 @@ +/** + * @file eventOutboxLease.test.ts + * Regression tests for issue #1557: crashed relay workers must not strand + * outbox rows, and a worker whose lease expired must not overwrite a newer + * claim (fencing). + */ + +jest.mock('../webhookDelivery', () => ({ + emitTransactionEvent: jest.fn(async () => 1), +})); + +import { prisma } from '../prisma'; +import { + eventOutboxService, + buildOutboxLeaseToken, + getOutboxLeaseMs, +} from '../eventOutbox'; +import { emitTransactionEvent } from '../webhookDelivery'; + +const emitMock = emitTransactionEvent as unknown as jest.Mock; + +const originalLease = process.env.OUTBOX_LOCK_TIMEOUT_MS; + +function payload() { + return { + transactionId: 'tx-lease-1', + amount: '100', + asset: 'USDC', + walletAddress: `G${'A'.repeat(55)}`, + transactionHash: '0xabc', + status: 'completed', + timestamp: new Date().toISOString(), + }; +} + +async function seed(overrides: Record = {}) { + return prisma.eventOutbox.create({ + data: { + id: `obx-${Math.random().toString(36).slice(2, 10)}`, + eventType: 'transaction.deposit.created', + payload: JSON.stringify(payload()), + status: 'pending', + aggregateType: 'transaction', + aggregateId: 'tx-lease-1', + vaultId: 'vault-lease-1', + sequence: 1, + attemptCount: 0, + maxAttempts: 3, + createdAt: new Date(), + updatedAt: new Date(), + ...overrides, + }, + }); +} + +beforeEach(async () => { + jest.clearAllMocks(); + emitMock.mockResolvedValue(1); + process.env.OUTBOX_LOCK_TIMEOUT_MS = '60000'; + await prisma.eventOutbox.deleteMany({}); +}); + +afterAll(() => { + if (originalLease === undefined) delete process.env.OUTBOX_LOCK_TIMEOUT_MS; + else process.env.OUTBOX_LOCK_TIMEOUT_MS = originalLease; +}); + +describe('outbox lease configuration', () => { + it('defaults to a 60s lease and honours the env override', () => { + delete process.env.OUTBOX_LOCK_TIMEOUT_MS; + expect(getOutboxLeaseMs()).toBe(60_000); + process.env.OUTBOX_LOCK_TIMEOUT_MS = '1234'; + expect(getOutboxLeaseMs()).toBe(1234); + }); + + it('derives a fencing token that changes on every re-claim', () => { + expect(buildOutboxLeaseToken(null, null)).toBeNull(); + const first = buildOutboxLeaseToken('worker-a', new Date(1000)); + const second = buildOutboxLeaseToken('worker-a', new Date(2000)); + expect(first).toBe('worker-a:1000'); + expect(second).toBe('worker-a:2000'); + expect(first).not.toBe(second); + }); +}); + +describe('reapStuckRelays', () => { + it('releases only expired leases and leaves active leases alone', async () => { + const stale = await seed({ + id: 'obx-stale', + lockedAt: new Date(Date.now() - 10 * 60 * 1000), + lockedBy: 'crashed-worker', + }); + const active = await seed({ + id: 'obx-active', + vaultId: 'vault-lease-2', + lockedAt: new Date(), + lockedBy: 'live-worker', + }); + + const released = await eventOutboxService.reapStuckRelays(); + + expect(released).toBe(1); + const staleRow = await prisma.eventOutbox.findUnique({ where: { id: stale.id } }); + const activeRow = await prisma.eventOutbox.findUnique({ where: { id: active.id } }); + expect(staleRow?.lockedAt).toBeNull(); + expect(staleRow?.lockedBy).toBeNull(); + expect(activeRow?.lockedBy).toBe('live-worker'); + }); + + it('recovers a crashed relay so the next cycle delivers it', async () => { + const crashed = await seed({ + id: 'obx-crashed', + lockedAt: new Date(Date.now() - 10 * 60 * 1000), + lockedBy: 'crashed-worker', + }); + + const result = await eventOutboxService.processOutbox(10); + + expect(result.relayed).toBe(1); + expect(emitMock).toHaveBeenCalledTimes(1); + const row = await prisma.eventOutbox.findUnique({ where: { id: crashed.id } }); + expect(row?.status).toBe('relayed'); + expect(row?.lockedAt).toBeNull(); + }); +}); + +describe('lease fencing', () => { + it('does not let a stale worker overwrite a row re-claimed mid-delivery', async () => { + const row = await seed({ id: 'obx-fenced' }); + + // Simulate another worker re-claiming the row while this worker is + // delivering the event. The claim timestamp changes, invalidating the + // original fencing token. + emitMock.mockImplementationOnce(async () => { + await prisma.eventOutbox.update({ + where: { id: row.id }, + data: { + lockedAt: new Date(Date.now() + 5_000), + lockedBy: 'other-worker', + }, + }); + return 1; + }); + + const result = await eventOutboxService.processOutbox(10); + + // The stale worker must discard its result instead of marking the row. + expect(result.relayed).toBe(0); + const after = await prisma.eventOutbox.findUnique({ where: { id: row.id } }); + expect(after?.status).toBe('pending'); + expect(after?.lockedBy).toBe('other-worker'); + }); +}); diff --git a/backend/src/__tests__/idempotencyPendingLease.test.ts b/backend/src/__tests__/idempotencyPendingLease.test.ts new file mode 100644 index 000000000..c83ea4738 --- /dev/null +++ b/backend/src/__tests__/idempotencyPendingLease.test.ts @@ -0,0 +1,224 @@ +/** + * @file idempotencyPendingLease.test.ts + * Regression tests for issue #1556: a hung/crashed worker must not block + * retries for the full 24h retention TTL. + */ + +jest.mock('../prisma', () => ({ + prisma: { + idempotencyKey: { + findFirst: jest.fn(), + deleteMany: jest.fn(), + }, + }, +})); + +jest.mock('../middleware/structuredLogging', () => ({ + logger: { log: jest.fn() }, +})); + +import type { Request, Response } from 'express'; +import { prisma } from '../prisma'; +import { + DEFAULT_PROCESSING_LEASE_MS, + hashRequestBody, + isProcessingLeaseExpired, + processingLeaseRemainingSeconds, + reclaimExpiredIdempotencyRecord, + reclaimExpiredIdempotencyRecords, + enforceIdempotency, +} from '../idempotency'; + +const idempotencyKeyDelegate = prisma.idempotencyKey as unknown as { + findFirst: jest.Mock; + deleteMany: jest.Mock; +}; + +function makeRecord(overrides: Record = {}) { + return { + keyId: 'a'.repeat(32), + tenantId: 'tenant-1', + walletAddress: null, + operation: 'POST /v1/vault/deposit', + status: 'pending', + requestHash: hashRequestBody({ amount: '1000' }), + responseHash: null, + responseBody: null, + statusCode: null, + createdAt: new Date(), + completedAt: null, + expiresAt: new Date(Date.now() + 24 * 60 * 60 * 1000), + ...overrides, + }; +} + +function makeReq(body: unknown = { amount: '1000' }): Partial { + return { + get: ((header: string) => + header === 'Idempotency-Key' ? 'a'.repeat(32) : undefined) as Request['get'], + tenantId: 'tenant-1', + body, + } as unknown as Partial; +} + +function makeRes() { + const json = jest.fn(); + const status = jest.fn().mockReturnValue({ json }); + return { res: { status, json } as unknown as Partial, status, json }; +} + +const originalLease = process.env.IDEMPOTENCY_PROCESSING_LEASE_MS; + +beforeEach(() => { + jest.clearAllMocks(); + process.env.IDEMPOTENCY_PROCESSING_LEASE_MS = '60000'; +}); + +afterAll(() => { + if (originalLease === undefined) delete process.env.IDEMPOTENCY_PROCESSING_LEASE_MS; + else process.env.IDEMPOTENCY_PROCESSING_LEASE_MS = originalLease; +}); + +describe('processing-lease helpers', () => { + it('defaults the lease to two minutes', () => { + delete process.env.IDEMPOTENCY_PROCESSING_LEASE_MS; + expect(DEFAULT_PROCESSING_LEASE_MS).toBe(2 * 60 * 1000); + }); + + it('treats only old pending records as expired', () => { + const now = Date.now(); + expect( + isProcessingLeaseExpired({ status: 'pending', createdAt: new Date(now - 1_000) }, now), + ).toBe(false); + expect( + isProcessingLeaseExpired({ status: 'pending', createdAt: new Date(now - 120_000) }, now), + ).toBe(true); + // Completed records are governed by the retention TTL, not the lease. + expect( + isProcessingLeaseExpired({ status: 'completed', createdAt: new Date(now - 120_000) }, now), + ).toBe(false); + }); + + it('reports at least one second remaining for a live lease', () => { + const now = Date.now(); + expect( + processingLeaseRemainingSeconds({ createdAt: new Date(now - 1_000) }, now), + ).toBe(59); + }); +}); + +describe('reclaimExpiredIdempotencyRecord', () => { + it('deletes only the guarded stale pending row', async () => { + idempotencyKeyDelegate.deleteMany.mockResolvedValue({ count: 1 }); + + const reclaimed = await reclaimExpiredIdempotencyRecord('key-1', 'tenant-1', Date.now()); + + expect(reclaimed).toBe(true); + expect(idempotencyKeyDelegate.deleteMany).toHaveBeenCalledWith({ + where: { + keyId: 'key-1', + tenantId: 'tenant-1', + status: 'pending', + createdAt: { lte: expect.any(Date) }, + }, + }); + }); + + it('reports false when another worker already completed the key', async () => { + idempotencyKeyDelegate.deleteMany.mockResolvedValue({ count: 0 }); + await expect( + reclaimExpiredIdempotencyRecord('key-2', 'tenant-1', Date.now()), + ).resolves.toBe(false); + }); +}); + +describe('reclaimExpiredIdempotencyRecords', () => { + it('reaps all abandoned pending rows in bulk', async () => { + idempotencyKeyDelegate.deleteMany.mockResolvedValue({ count: 3 }); + await expect(reclaimExpiredIdempotencyRecords(Date.now())).resolves.toBe(3); + expect(idempotencyKeyDelegate.deleteMany).toHaveBeenCalledWith({ + where: { + status: 'pending', + createdAt: { lte: expect.any(Date) }, + }, + }); + }); +}); + +describe('enforceIdempotency hung-key recovery', () => { + it('reclaims an expired pending lease and lets the retry proceed', async () => { + idempotencyKeyDelegate.findFirst.mockResolvedValue( + makeRecord({ createdAt: new Date(Date.now() - 10 * 60 * 1000) }), + ); + idempotencyKeyDelegate.deleteMany.mockResolvedValue({ count: 1 }); + + const next = jest.fn(); + const { res } = makeRes(); + const middleware = enforceIdempotency(); + + await middleware(makeReq() as Request, res as Response, next); + + expect(idempotencyKeyDelegate.deleteMany).toHaveBeenCalledTimes(1); + expect(next).toHaveBeenCalledTimes(1); + expect(res.status).not.toHaveBeenCalled(); + }); + + it('keeps blocking while the lease is still live and reports retryAfter', async () => { + idempotencyKeyDelegate.findFirst.mockResolvedValue( + makeRecord({ createdAt: new Date(Date.now() - 5_000) }), + ); + + const next = jest.fn(); + const { res, status } = makeRes(); + const middleware = enforceIdempotency(); + + await middleware(makeReq() as Request, res as Response, next); + + expect(idempotencyKeyDelegate.deleteMany).not.toHaveBeenCalled(); + expect(next).not.toHaveBeenCalled(); + expect(status).toHaveBeenCalledWith(409); + expect(res.json).toHaveBeenCalledWith( + expect.objectContaining({ + code: 'IDEMPOTENCY_PENDING', + retryAfter: expect.any(Number), + }), + ); + }); + + it('does not reclaim a pending row when the guarded delete matches nothing', async () => { + idempotencyKeyDelegate.findFirst.mockResolvedValue( + makeRecord({ createdAt: new Date(Date.now() - 10 * 60 * 1000) }), + ); + idempotencyKeyDelegate.deleteMany.mockResolvedValue({ count: 0 }); + + const next = jest.fn(); + const { res, status } = makeRes(); + const middleware = enforceIdempotency(); + + await middleware(makeReq() as Request, res as Response, next); + + expect(next).not.toHaveBeenCalled(); + expect(status).toHaveBeenCalledWith(409); + }); + + it('replays a completed response without touching the lease', async () => { + idempotencyKeyDelegate.findFirst.mockResolvedValue( + makeRecord({ + status: 'completed', + statusCode: 201, + responseBody: JSON.stringify({ id: 'tx-1' }), + }), + ); + + const next = jest.fn(); + const { res, status } = makeRes(); + const middleware = enforceIdempotency(); + + await middleware(makeReq() as Request, res as Response, next); + + expect(next).not.toHaveBeenCalled(); + expect(idempotencyKeyDelegate.deleteMany).not.toHaveBeenCalled(); + expect(status).toHaveBeenCalledWith(201); + expect(res.json).toHaveBeenCalledWith({ id: 'tx-1' }); + }); +}); diff --git a/backend/src/eventOutbox.ts b/backend/src/eventOutbox.ts index 7b7fe922a..f74a7894e 100644 --- a/backend/src/eventOutbox.ts +++ b/backend/src/eventOutbox.ts @@ -12,6 +12,21 @@ * * This guarantees that events are NEVER lost when the publishing process * crashes between persisting business state and dispatching the event. + * + * Lease & fencing (issue #1557) + * ----------------------------- + * `lockedAt`/`lockedBy` form a *lease*, not a permanent claim. A relay worker + * claims rows by writing `lockedAt = now` and `lockedBy = instanceId`; the + * lease expires `OUTBOX_LOCK_TIMEOUT_MS` (default 60s) later. A crashed worker + * therefore cannot strand a row forever: + * + * - `reapStuckRelays()` clears leases older than the timeout (and runs at the + * start of every relay cycle as well as on its own timer). + * - Completion writes are conditional on the exact `(lockedBy, lockedAt)` + * pair observed when the row was claimed. If another worker re-claimed the + * row in the meantime, the update matches zero rows and the stale worker + * skips the write. The claim timestamp is the fencing token: it changes on + * every re-claim, so a stale worker cannot overwrite a newer attempt. */ import crypto from 'crypto'; @@ -25,21 +40,29 @@ import { // ─── Prisma Access ─────────────────────────────────────────────────────────── /** - * Lazily resolves the Prisma client and its `eventOutbox` delegate. - * - * Importing `prisma` at module load time is fragile: depending on the runtime - * path (e.g. test harnesses, hot-reload, or a partially initialized client), - * the imported binding can be undefined or lack the `eventOutbox` delegate, - * causing `prisma.eventOutbox.create` to throw - * "Cannot read properties of undefined (reading 'create')". - * - * Resolving the delegate at call time (with a clear error if unavailable) - * makes the failure mode explicit and robust across runtime paths. + * Lazily resolves the Prisma client. Importing `prisma` at module load time is + * fragile: depending on the runtime path (tests, hot-reload, partially + * initialized client) the binding can be undefined, causing confusing + * `Cannot read properties of undefined` errors deep inside fire-and-forget + * writes. Resolving at call time makes the failure mode explicit. */ -function getEventOutboxDelegate() { +function getPrisma() { // eslint-disable-next-line @typescript-eslint/no-var-requires const { prisma } = require('./prisma') as typeof import('./prisma'); - const delegate = prisma?.eventOutbox; + if (!prisma) { + throw new Error( + 'Prisma client is not initialized. Ensure `prisma generate` has run and the client is instantiated before use.', + ); + } + return prisma; +} + +/** + * Resolves the `eventOutbox` delegate and fails with a clear message when the + * generated client is stale or a mocked client is missing the model. + */ +function getEventOutboxDelegate() { + const delegate = getPrisma()?.eventOutbox; if (!delegate) { throw new Error( 'Prisma client is not initialized or missing the `eventOutbox` delegate. ' + @@ -62,12 +85,19 @@ export interface EventOutboxRecord { status: OutboxEventStatus; aggregateType: OutboxAggregateType; aggregateId: string; + vaultId: string | null; sequence: number; attemptCount: number; maxAttempts: number; lastError: string | null; lockedAt: string | null; lockedBy: string | null; + /** + * Fencing token for the current lease (`${lockedBy}:${lockedAt epoch ms}`). + * `null` when the row is not leased. Changes on every re-claim, so consumers + * can detect a stale worker attempting to finish an old lease. + */ + leaseToken: string | null; createdAt: string; updatedAt: string; relayedAt: string | null; @@ -110,11 +140,20 @@ function getBatchSize(): number { return Number.isFinite(parsed) && parsed > 0 ? parsed : 50; } -function getLockTimeoutMs(): number { +/** + * Lease duration for a claimed row. Exposed so operators/tests can shorten it + * and so the reaper and claim logic always agree on the boundary. + */ +export function getOutboxLeaseMs(): number { const parsed = parseInt(process.env.OUTBOX_LOCK_TIMEOUT_MS || '', 10); return Number.isFinite(parsed) && parsed > 0 ? parsed : 60000; } +/** Backwards-compatible alias for the lease duration. */ +function getLockTimeoutMs(): number { + return getOutboxLeaseMs(); +} + function getMaxAttempts(): number { const parsed = parseInt(process.env.OUTBOX_MAX_ATTEMPTS || '', 10); return Number.isFinite(parsed) && parsed > 0 ? parsed : 3; @@ -125,19 +164,42 @@ function getRetentionMs(): number { return Number.isFinite(parsed) && parsed > 0 ? parsed : 7 * 24 * 60 * 60 * 1000; // 7 days } +function getReaperIntervalMs(): number { + const parsed = parseInt(process.env.OUTBOX_REAPER_INTERVAL_MS || '', 10); + if (Number.isFinite(parsed) && parsed > 0) return parsed; + // Reap at least twice per lease window so stuck rows recover promptly. + return Math.max(1000, Math.floor(getOutboxLeaseMs() / 2)); +} + /** Unique instance identifier for distributed locking. */ function getInstanceId(): string { return process.env.OUTBOX_INSTANCE_ID || `instance-${crypto.randomUUID().slice(0, 8)}`; } +/** + * Builds the fencing token for a lease. The claim timestamp (millisecond + * precision) is included so every re-claim produces a different token. + */ +export function buildOutboxLeaseToken( + lockedBy: string | null | undefined, + lockedAt: Date | string | null | undefined, +): string | null { + if (!lockedBy || !lockedAt) return null; + const millis = lockedAt instanceof Date ? lockedAt.getTime() : Date.parse(String(lockedAt)); + if (!Number.isFinite(millis)) return null; + return `${lockedBy}:${millis}`; +} + /** * Throws if the Prisma client has no `eventOutbox` model delegate (e.g. a stale * generated client or a mocked client missing the model). Without this guard * the failure surfaces as a `Cannot read properties of undefined` TypeError * deep inside fire-and-forget writes, where callers only log it. */ -export function assertEventOutboxModelAvailable(client: unknown = prisma): void { - const delegate = (client as { eventOutbox?: { create?: unknown } } | null | undefined)?.eventOutbox; +export function assertEventOutboxModelAvailable(client?: unknown): void { + const resolved = client ?? getPrisma(); + const delegate = (resolved as { eventOutbox?: { create?: unknown } } | null | undefined) + ?.eventOutbox; if (!delegate || typeof delegate.create !== 'function') { throw new Error( 'Prisma client is missing the eventOutbox model. Run `prisma generate` and ensure the EventOutbox migration is applied.', @@ -151,6 +213,7 @@ class EventOutboxService { private isRunning = false; private pollTimer: NodeJS.Timeout | null = null; private cleanupTimer: NodeJS.Timeout | null = null; + private reaperTimer: NodeJS.Timeout | null = null; private instanceId: string; private inFlight: Promise | null = null; private abortCleanup: (() => void) | null = null; @@ -160,24 +223,22 @@ class EventOutboxService { this.instanceId = getInstanceId(); } + /** Instance identifier used as the lease owner. */ + getInstanceId(): string { + return this.instanceId; + } + // ─── Public API ──────────────────────────────────────────────────────────── /** * Writes an event to the outbox table. * This should be called within the same transaction boundary as the business * operation (e.g., inside a Prisma $transaction) to ensure atomic persistence. - * - * Returns the created outbox record. */ async writeEvent(input: OutboxWriteInput): Promise { - if (!prisma.eventOutbox) { - throw new Error('prisma.eventOutbox is not initialized. Ensure Prisma schema includes eventOutbox model.'); - } - assertEventOutboxModelAvailable(); const now = new Date(); const vaultId = input.vaultId ?? input.aggregateId; - const record = await prisma.eventOutbox.create({ const record = await getEventOutboxDelegate().create({ data: { id: `obx-${crypto.randomUUID()}`, @@ -200,8 +261,11 @@ class EventOutboxService { /** * Processes pending events from the outbox table. - * Uses instance-level locking to prevent duplicate processing in multi-pod deployments. - * Can be called on a schedule or manually. + * + * The cycle first reaps expired leases, then claims eligible rows with a + * conditional update. Completion writes are fenced on the claim timestamp so + * a worker whose lease expired mid-delivery cannot overwrite the state of the + * worker that re-claimed the row. */ async processOutbox(batchSize?: number, signal?: AbortSignal): Promise { const limit = batchSize ?? getBatchSize(); @@ -216,27 +280,23 @@ class EventOutboxService { }; try { - // 1. Find eligible entries: pending or failed entries whose lock is expired. - // Order by (vaultId, sequence, createdAt) so per-vault ordering is - // preserved and events for different vaults do not block each other. - // We select the head-of-line event per vault to enforce strict - // per-vault ordering (a vault's next event is only processed after - // its predecessor has been relayed or dead-lettered). - const candidates = await prisma.eventOutbox.findMany({ - // 1. Find eligible entries: pending or failed entries whose lock is expired + // 0. Return crashed workers' leases to the pool before selecting work. + await this.reapStuckRelays(now).catch((err) => { + logger.log('warn', 'Outbox lease reaper failed during relay cycle', { + error: err instanceof Error ? err.message : String(err), + }); + }); + + // 1. Find eligible entries: pending or failed entries whose lock is + // expired. Order by (vaultId, sequence, createdAt) so per-vault + // ordering is preserved and events for different vaults do not block + // each other. const candidates = await getEventOutboxDelegate().findMany({ where: { status: { in: ['pending', 'failed'] }, - OR: [ - { lockedAt: null }, - { lockedAt: { lt: lockExpiry } }, - ], + OR: [{ lockedAt: null }, { lockedAt: { lt: lockExpiry } }], }, - orderBy: [ - { vaultId: 'asc' }, - { sequence: 'asc' }, - { createdAt: 'asc' }, - ], + orderBy: [{ vaultId: 'asc' }, { sequence: 'asc' }, { createdAt: 'asc' }], take: limit, }); @@ -248,12 +308,12 @@ class EventOutboxService { // process the lowest-sequence eligible event. This guarantees a // consumer tracking vault.version never observes sequence N+1 // before sequence N for the same vault. - const headOfLine = new Map(); - for (const c of candidates) { - const key = c.vaultId ?? c.aggregateId; + const headOfLine = new Map(); + for (const candidate of candidates) { + const key = candidate.vaultId ?? candidate.aggregateId; const existing = headOfLine.get(key); - if (!existing || c.sequence < existing.sequence) { - headOfLine.set(key, c); + if (!existing || candidate.sequence < existing.sequence) { + headOfLine.set(key, candidate); } } const orderedCandidates = Array.from(headOfLine.values()).sort((a, b) => { @@ -263,17 +323,14 @@ class EventOutboxService { return a.sequence - b.sequence; }); - // 2. Lock the claimed entries by updating lockedAt/lockedBy in bulk - const candidateIds = orderedCandidates.map((e) => e.id); - await prisma.eventOutbox.updateMany({ - const candidateIds = candidates.map((e) => e.id); + // 2. Claim the selected entries. The guard re-checks that nobody else + // claimed them between the read and the write. + const candidateIds = orderedCandidates.map((entry) => entry.id); await getEventOutboxDelegate().updateMany({ where: { id: { in: candidateIds }, - OR: [ - { lockedAt: null }, - { lockedAt: { lt: lockExpiry } }, - ], + status: { in: ['pending', 'failed'] }, + OR: [{ lockedAt: null }, { lockedAt: { lt: lockExpiry } }], }, data: { lockedAt: now, @@ -281,60 +338,57 @@ class EventOutboxService { }, }); - // 3. Re-fetch the entries we successfully locked (some may have been - // concurrently claimed by another instance) + // 3. Re-fetch only the rows this instance actually claimed. `lockedAt` + // is the fencing token: a row claimed by another instance is skipped. const entries = await getEventOutboxDelegate().findMany({ where: { id: { in: candidateIds }, lockedBy: this.instanceId, lockedAt: now, }, - orderBy: [ - { vaultId: 'asc' }, - { sequence: 'asc' }, - { createdAt: 'asc' }, - ], + orderBy: [{ vaultId: 'asc' }, { sequence: 'asc' }, { createdAt: 'asc' }], }); if (entries.length === 0) { return result; } - // 4. Relay each entry + // 4. Relay each entry. for (const entry of entries) { - // Stop claiming work once shutdown begins; unprocessed entries stay - // locked until the lock timeout expires and are picked up on restart. if (signal?.aborted) break; + const claimToken = this.leaseTokenFor(entry.lockedBy, entry.lockedAt); try { const payload = JSON.parse(entry.payload) as TransactionEventPayload; // emitTransactionEvent schedules webhook deliveries asynchronously. - // We await it here so relayed/retry decisions happen after the event - // is queued for delivery, not before. Note that actual HTTP delivery - // happens in the background via setTimeout — if the process crashes - // after marking relayed but before delivery completes, the webhook - // delivery's own retry mechanism still applies. + // Awaiting it ensures relayed/retry decisions happen after the event + // is queued for delivery, not before. const deliveredCount = await emitTransactionEvent( entry.eventType as TransactionEventType, payload, ); - await getEventOutboxDelegate().update({ // Do not write the mark-as-sent update after abort; the event will be // re-relayed after restart (at-least-once delivery). if (signal?.aborted) break; - await prisma.eventOutbox.update({ - where: { id: entry.id }, - data: { - status: 'relayed', - attemptCount: { increment: 1 }, - relayedAt: new Date(), - lockedAt: null, - lockedBy: null, - lastError: null, - }, + const fenced = await this.fencedUpdate(entry.id, claimToken, { + status: 'relayed', + attemptCount: { increment: 1 }, + relayedAt: new Date(), + lockedAt: null, + lockedBy: null, + lastError: null, }); + if (!fenced) { + logger.log('warn', 'Outbox relay result discarded: lease was re-claimed (fencing)', { + outboxId: entry.id, + eventType: entry.eventType, + staleLeaseToken: claimToken, + }); + continue; + } + result.relayed++; logger.log('info', 'Outbox event relayed', { outboxId: entry.id, @@ -343,6 +397,7 @@ class EventOutboxService { aggregateId: entry.aggregateId, sequence: entry.sequence, deliveredCount, + leaseToken: claimToken, }); } catch (error) { const errorMessage = error instanceof Error ? error.message : String(error); @@ -351,17 +406,16 @@ class EventOutboxService { if (signal?.aborted) break; if (nextAttempt >= entry.maxAttempts) { - // Exhausted retries — move to dead_letter - await getEventOutboxDelegate().update({ - where: { id: entry.id }, - data: { - status: 'dead_letter', - attemptCount: nextAttempt, - lastError: errorMessage, - lockedAt: null, - lockedBy: null, - }, + const fenced = await this.fencedUpdate(entry.id, claimToken, { + status: 'dead_letter', + attemptCount: nextAttempt, + lastError: errorMessage, + lockedAt: null, + lockedBy: null, }); + + if (!fenced) continue; + result.deadLettered++; result.errors.push(`[${entry.id}] Dead-letter: ${errorMessage}`); @@ -374,19 +428,20 @@ class EventOutboxService { }); void this.sendDeadLetterAlert(entry.id, entry.eventType, nextAttempt, errorMessage); } else { - // Mark as failed for retry - await getEventOutboxDelegate().update({ - where: { id: entry.id }, - data: { - status: 'failed', - attemptCount: nextAttempt, - lastError: errorMessage, - lockedAt: null, - lockedBy: null, - }, + const fenced = await this.fencedUpdate(entry.id, claimToken, { + status: 'failed', + attemptCount: nextAttempt, + lastError: errorMessage, + lockedAt: null, + lockedBy: null, }); + + if (!fenced) continue; + result.failed++; - result.errors.push(`[${entry.id}] Failed (${nextAttempt}/${entry.maxAttempts}): ${errorMessage}`); + result.errors.push( + `[${entry.id}] Failed (${nextAttempt}/${entry.maxAttempts}): ${errorMessage}`, + ); logger.log('warn', 'Outbox event relay failed, will retry', { outboxId: entry.id, @@ -410,7 +465,93 @@ class EventOutboxService { } } - private async sendDeadLetterAlert(outboxId: string, eventType: string, attempts: number, error: string): Promise { + /** + * Releases leases held past their expiry, returning stranded rows to the + * pending/failed pool. Safe to run concurrently: the `lockedAt < cutoff` + * guard means only genuinely expired leases are touched. + * + * @returns number of rows whose lease was released. + */ + async reapStuckRelays(now: Date = new Date()): Promise { + const cutoff = new Date(now.getTime() - getOutboxLeaseMs()); + + const stuck = await getEventOutboxDelegate().findMany({ + where: { + status: { in: ['pending', 'failed'] }, + lockedAt: { lt: cutoff }, + }, + select: { id: true, eventType: true, lockedBy: true, lockedAt: true }, + }); + + if (stuck.length === 0) return 0; + + const result = await getEventOutboxDelegate().updateMany({ + where: { + id: { in: stuck.map((row) => row.id) }, + status: { in: ['pending', 'failed'] }, + lockedAt: { lt: cutoff }, + }, + data: { + lockedAt: null, + lockedBy: null, + }, + }); + + if (result.count > 0) { + logger.log('warn', 'Reaped stuck outbox relay leases', { + action: 'outbox_lease_reaper', + reclaimed: result.count, + candidates: stuck.length, + leaseMs: getOutboxLeaseMs(), + releasedLeases: stuck.map((row) => ({ + outboxId: row.id, + eventType: row.eventType, + lockedBy: row.lockedBy, + leaseToken: this.leaseTokenFor(row.lockedBy, row.lockedAt), + })), + }); + } + + return result.count; + } + + /** + * Applies an update only if this worker still owns the exact lease it + * claimed. Returns `false` when the lease was stolen/reclaimed (fencing). + */ + private async fencedUpdate( + id: string, + claimToken: string | null, + // eslint-disable-next-line @typescript-eslint/no-explicit-any + data: any, + ): Promise { + if (!claimToken) return false; + + // Encode both halves of the token in the guard. Extract the timestamp from + // the token so Prisma can compare the DateTime column. + const separator = claimToken.lastIndexOf(':'); + const lockedBy = claimToken.slice(0, separator); + const lockedAtMs = Number(claimToken.slice(separator + 1)); + if (!lockedBy || !Number.isFinite(lockedAtMs)) return false; + + const result = await getEventOutboxDelegate().updateMany({ + where: { + id, + lockedBy, + lockedAt: new Date(lockedAtMs), + }, + data, + }); + + return result.count === 1; + } + + private async sendDeadLetterAlert( + outboxId: string, + eventType: string, + attempts: number, + error: string, + ): Promise { const webhookUrl = process.env.SLACK_WEBHOOK_URL || process.env.ALERT_WEBHOOK_URL; if (!webhookUrl) return; @@ -419,14 +560,14 @@ class EventOutboxService { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ - text: `:rotating_light: *YieldVault Outbox Dead-Lettered*\nEvent \`${eventType}\` (ID: ${outboxId}) failed after ${attempts} attempts.\nLatest error: \`${error}\`` - }) + text: `:rotating_light: *YieldVault Outbox Dead-Lettered*\nEvent \`${eventType}\` (ID: ${outboxId}) failed after ${attempts} attempts.\nLatest error: \`${error}\``, + }), }); } catch (err) { logger.log('error', 'Failed to send outbox dead-letter alert', { error: err instanceof Error ? err.message : String(err), outboxId, - eventType + eventType, }); } } @@ -484,11 +625,10 @@ class EventOutboxService { /** * Replays pending events from the outbox on startup. - * Used to recover any events that were written but not relayed before a crash. + * Used to recover events that were written but not relayed before a crash. */ async replayOnStartup(): Promise { assertEventOutboxModelAvailable(); - const pendingCount = await prisma.eventOutbox.count({ const pendingCount = await getEventOutboxDelegate().count({ where: { status: { in: ['pending', 'failed'] } }, }); @@ -508,8 +648,7 @@ class EventOutboxService { // ─── Background Processing ───────────────────────────────────────────────── /** - * Starts the background outbox processor. - * Polls for pending events on a configured interval and relays them. + * Starts the background outbox processor and its lease reaper. */ start(options: { signal?: AbortSignal } = {}): void { if (this.isRunning) { @@ -533,6 +672,7 @@ class EventOutboxService { logger.log('info', 'Starting outbox processor', { pollIntervalMs: intervalMs, batchSize: getBatchSize(), + leaseMs: getOutboxLeaseMs(), instanceId: this.instanceId, }); @@ -556,13 +696,13 @@ class EventOutboxService { this.pollTimer.unref(); } - // Start periodic cleanup of old processed entries (every hour) this.startCleanupTimer(); + this.startReaperTimer(); } /** - * Stops the background outbox processor: clears timers and resolves once - * any in-flight polling cycle has finished. + * Stops the background processor: clears timers and resolves once any + * in-flight polling cycle has finished. */ async stop(): Promise { if (this.isRunning) { @@ -575,6 +715,10 @@ class EventOutboxService { clearInterval(this.cleanupTimer); this.cleanupTimer = null; } + if (this.reaperTimer) { + clearInterval(this.reaperTimer); + this.reaperTimer = null; + } this.abortCleanup?.(); this.abortCleanup = null; this.signal = undefined; @@ -587,9 +731,7 @@ class EventOutboxService { } } - /** - * Runs one polling cycle unless stopped/aborted or a cycle is already running. - */ + /** Runs one polling cycle unless stopped/aborted or a cycle is already running. */ private runCycle(label: string): void { if (!this.isRunning || this.signal?.aborted || this.inFlight) return; @@ -605,18 +747,14 @@ class EventOutboxService { this.inFlight = cycle; } - /** - * Returns whether the processor is currently running. - */ + /** Returns whether the processor is currently running. */ get isActive(): boolean { return this.isRunning; } // ─── Metrics & Health ────────────────────────────────────────────────────── - /** - * Returns a snapshot of outbox metrics for monitoring/administration. - */ + /** Returns a snapshot of outbox metrics for monitoring/administration. */ async getMetrics(): Promise { const [pending, relayed, failed, deadLettered, locked, total] = await Promise.all([ getEventOutboxDelegate().count({ where: { status: 'pending' } }), @@ -635,22 +773,21 @@ class EventOutboxService { return { pending, relayed, failed, deadLettered, locked, total }; } - /** - * Lists outbox entries with optional filters. - */ - async listEntries(filters: { - status?: OutboxEventStatus | OutboxEventStatus[]; - aggregateType?: OutboxAggregateType; - aggregateId?: string; - limit?: number; - offset?: number; - } = {}): Promise { + /** Lists outbox entries with optional filters. */ + async listEntries( + filters: { + status?: OutboxEventStatus | OutboxEventStatus[]; + aggregateType?: OutboxAggregateType; + aggregateId?: string; + vaultId?: string; + limit?: number; + offset?: number; + } = {}, + ): Promise { const where: Record = {}; if (filters.status) { - where.status = Array.isArray(filters.status) - ? { in: filters.status } - : filters.status; + where.status = Array.isArray(filters.status) ? { in: filters.status } : filters.status; } if (filters.aggregateType) { where.aggregateType = filters.aggregateType; @@ -658,6 +795,9 @@ class EventOutboxService { if (filters.aggregateId) { where.aggregateId = filters.aggregateId; } + if (filters.vaultId) { + where.vaultId = filters.vaultId; + } const rows = await getEventOutboxDelegate().findMany({ where, @@ -671,6 +811,13 @@ class EventOutboxService { // ─── Private Helpers ─────────────────────────────────────────────────────── + private leaseTokenFor( + lockedBy: string | null | undefined, + lockedAt: Date | string | null | undefined, + ): string | null { + return buildOutboxLeaseToken(lockedBy, lockedAt); + } + /** * Starts a periodic cleanup timer to remove old processed entries. * Runs every hour by default, configurable via OUTBOX_CLEANUP_INTERVAL_MS. @@ -689,12 +836,30 @@ class EventOutboxService { }); }, cleanupIntervalMs); - // Unref so the timer doesn't keep the process alive if (this.cleanupTimer && typeof this.cleanupTimer === 'object' && 'unref' in this.cleanupTimer) { this.cleanupTimer.unref(); } } + /** + * Starts the standalone stuck-lease reaper. Runs every + * `OUTBOX_REAPER_INTERVAL_MS` (default half the lease duration). + */ + private startReaperTimer(): void { + this.reaperTimer = setInterval(() => { + if (this.signal?.aborted) return; + this.reapStuckRelays().catch((err) => { + logger.log('error', 'Outbox lease reaper cycle failed', { + error: err instanceof Error ? err.message : String(err), + }); + }); + }, getReaperIntervalMs()); + + if (this.reaperTimer && typeof this.reaperTimer === 'object' && 'unref' in this.reaperTimer) { + this.reaperTimer.unref(); + } + } + private toRecord(row: { id: string; eventType: string; @@ -702,6 +867,7 @@ class EventOutboxService { status: string; aggregateType: string; aggregateId: string; + vaultId?: string | null; sequence: number; attemptCount: number; maxAttempts: number; @@ -719,12 +885,14 @@ class EventOutboxService { status: row.status as OutboxEventStatus, aggregateType: row.aggregateType as OutboxAggregateType, aggregateId: row.aggregateId, + vaultId: row.vaultId ?? null, sequence: row.sequence, attemptCount: row.attemptCount, maxAttempts: row.maxAttempts, lastError: row.lastError, lockedAt: row.lockedAt?.toISOString() ?? null, lockedBy: row.lockedBy, + leaseToken: this.leaseTokenFor(row.lockedBy, row.lockedAt), createdAt: row.createdAt.toISOString(), updatedAt: row.updatedAt.toISOString(), relayedAt: row.relayedAt?.toISOString() ?? null, @@ -733,12 +901,10 @@ class EventOutboxService { /** * Computes the next monotonic sequence number for a given vault. - * Sequences are per-vault and increment by exactly 1. Uses a transaction - * with a row-level lock on the vault's outbox rows to avoid races between - * concurrent writers. + * Sequences are per-vault and increment by exactly 1. */ private async nextSequenceForVault(vaultId: string): Promise { - const result = await prisma.$transaction(async (tx) => { + const result = await getPrisma().$transaction(async (tx) => { const latest = await tx.eventOutbox.findFirst({ where: { vaultId }, orderBy: { sequence: 'desc' }, diff --git a/backend/src/idempotency.ts b/backend/src/idempotency.ts index cf111e0d6..1f677ffab 100644 --- a/backend/src/idempotency.ts +++ b/backend/src/idempotency.ts @@ -6,11 +6,26 @@ * withdrawals, transfers). Ensures that retried requests produce the same result * without creating duplicate side effects. * - * Acceptance Criteria: - * ✓ Accept idempotency keys for critical mutation endpoints - * ✓ Reject or reuse repeated submissions safely - * ✓ Track pending and completed keys with expiry - * ✓ Document API expectations for clients + * TTL semantics (issue #1556) + * --------------------------- + * Two independent clocks govern a key: + * + * 1. **Replay retention TTL** (`DEFAULT_IDEMPOTENCY_TTL_MS`, 24h by default): + * how long a *completed* or *failed* record is kept so a retry can replay + * the stored response. It never blocks retries; it only bounds how long a + * replay is available. Expired rows are removed by the cleanup task. + * + * 2. **Processing lease** (`IDEMPOTENCY_PROCESSING_LEASE_MS`, 2 minutes by + * default): how long a *pending* record may remain in flight. The lease is + * measured from `createdAt`. If the worker that created a pending record + * crashes or hangs, a retry after the lease expires may safely reclaim the + * slot (`reclaimExpiredIdempotencyRecord`) instead of being blocked with a + * 409 for the full 24h retention window. + * + * The lease MUST be shorter than the retention TTL. It should also exceed the + * longest expected mutation latency (the transfer orchestrator times out at + * 45s by default), otherwise a genuinely slow request could be reclaimed and + * executed twice. * * Usage: * POST /v1/vault/deposit @@ -22,18 +37,19 @@ */ import crypto from 'crypto'; -import NodeCache from 'node-cache'; import { prisma } from './prisma'; -import { redisClientManager } from './rateLimiter'; import { logger } from './middleware/structuredLogging'; import type { Request, Response, NextFunction } from 'express'; -import NodeCache from 'node-cache'; -import { redisClientManager } from './rateLimiter'; /** * The replay cache used by money-moving operations (deposits, withdrawals, * transfers) lives in `idempotencyStore.ts`; it is re-exported here so callers * keep a single import surface for the whole idempotency feature. + * + * The Prisma-backed helpers below are the durable, per-key record lifecycle + * used by `enforceIdempotency`. Keeping a single implementation (rather than a + * second, inline copy) avoids the duplicate declarations that previously made + * this module un-loadable. */ export { idempotencyStore, @@ -43,6 +59,7 @@ export { getIdempotencyHashThreshold, type IdempotentOperationResult, type IdempotencyKeyInfo, + type IdempotencyKeyMetadata, type IdempotencyMetrics, } from './idempotencyStore'; @@ -63,18 +80,17 @@ export interface IdempotencyRecord { operation: string; } -export interface IdempotencyKeyMetadata { - keyId: string; - keyFormat: 'uuid' | 'nonce' | 'custom'; - maxAge: number; - hint?: string; -} - // ─── Constants ────────────────────────────────────────────────────────────── -/** Default idempotency key TTL: 24 hours */ +/** Default idempotency key replay retention TTL: 24 hours. */ export const DEFAULT_IDEMPOTENCY_TTL_MS = 24 * 60 * 60 * 1000; +/** + * Default processing lease for a pending record. A pending record older than + * this is considered abandoned (crashed/hung worker) and may be reclaimed. + */ +export const DEFAULT_PROCESSING_LEASE_MS = 2 * 60 * 1000; + /** Minimum idempotency key length to prevent brute-force */ export const MIN_KEY_LENGTH = 16; @@ -91,6 +107,50 @@ export const IDEMPOTENT_ENDPOINTS = [ 'DELETE /admin/allowlist/remove', ] as const; +/** + * Reads the processing-lease duration at call time so operators and tests can + * tune it without re-importing the module. Values below 1s are ignored. + */ +export function getProcessingLeaseMs(): number { + const parsed = parseInt(process.env.IDEMPOTENCY_PROCESSING_LEASE_MS || '', 10); + return Number.isFinite(parsed) && parsed >= 1_000 ? parsed : DEFAULT_PROCESSING_LEASE_MS; +} + +/** + * True when a pending record has outlived its processing lease and may be + * reclaimed. Completed/failed records are never lease-reclaimed: only their + * retention TTL applies. + */ +export function isProcessingLeaseExpired( + record: Pick, + now: number = Date.now(), +): boolean { + if (record.status !== 'pending') return false; + + const createdAt = + record.createdAt instanceof Date + ? record.createdAt.getTime() + : Date.parse(String(record.createdAt)); + + if (!Number.isFinite(createdAt)) return false; + return now - createdAt >= getProcessingLeaseMs(); +} + +/** Seconds a caller should wait before retrying a still-leased pending key. */ +export function processingLeaseRemainingSeconds( + record: Pick, + now: number = Date.now(), +): number { + const createdAt = + record.createdAt instanceof Date + ? record.createdAt.getTime() + : Date.parse(String(record.createdAt)); + if (!Number.isFinite(createdAt)) return Math.ceil(getProcessingLeaseMs() / 1000); + + const remainingMs = createdAt + getProcessingLeaseMs() - now; + return Math.max(1, Math.ceil(remainingMs / 1000)); +} + // ─── Validation ───────────────────────────────────────────────────────────── /** @@ -138,7 +198,7 @@ export function hashResponseBody(body: unknown): string { */ export async function getIdempotencyRecord( keyId: string, - tenantId: string + tenantId: string, ): Promise { const record = await prisma.idempotencyKey.findFirst({ where: { @@ -175,7 +235,7 @@ export async function createIdempotencyRecord( walletAddress: string | undefined, operation: string, requestBody: unknown, - ttlMs = DEFAULT_IDEMPOTENCY_TTL_MS + ttlMs = DEFAULT_IDEMPOTENCY_TTL_MS, ): Promise { const now = new Date(); const expiresAt = new Date(now.getTime() + ttlMs); @@ -213,7 +273,7 @@ export async function completeIdempotencyRecord( keyId: string, tenantId: string, statusCode: number, - responseBody: unknown + responseBody: unknown, ): Promise { const now = new Date(); const responseHash = hashResponseBody(responseBody); @@ -252,7 +312,7 @@ export async function failIdempotencyRecord( keyId: string, tenantId: string, statusCode: number, - errorBody: unknown + errorBody: unknown, ): Promise { const now = new Date(); @@ -282,6 +342,73 @@ export async function failIdempotencyRecord( return record; } +// ─── Processing-lease reclaim (issue #1556) ───────────────────────────────── + +/** + * Atomically reclaims a *single* expired pending record. + * + * The delete is guarded by `status: 'pending'` and `createdAt <= cutoff`, so a + * concurrent writer that has already completed the record (or created a newer + * attempt) is never clobbered. Returns `true` only when this call actually + * removed the stale row. + */ +export async function reclaimExpiredIdempotencyRecord( + keyId: string, + tenantId: string, + now: number = Date.now(), +): Promise { + const cutoff = new Date(now - getProcessingLeaseMs()); + + const result = await prisma.idempotencyKey.deleteMany({ + where: { + keyId, + tenantId, + status: 'pending', + createdAt: { lte: cutoff }, + }, + }); + + if (result.count > 0) { + logger.log('warn', 'Reclaimed expired idempotency processing lease', { + action: 'idempotency_lease_reclaim', + keyId, + tenantId, + leaseMs: getProcessingLeaseMs(), + reclaimed: result.count, + }); + return true; + } + + return false; +} + +/** + * Bulk reaper for abandoned pending records. Intended to run periodically so a + * crashed worker does not leave keys blocking retries until the 24h TTL. + */ +export async function reclaimExpiredIdempotencyRecords( + now: number = Date.now(), +): Promise { + const cutoff = new Date(now - getProcessingLeaseMs()); + + const result = await prisma.idempotencyKey.deleteMany({ + where: { + status: 'pending', + createdAt: { lte: cutoff }, + }, + }); + + if (result.count > 0) { + logger.log('warn', 'Reaped expired idempotency processing leases', { + action: 'idempotency_lease_reaper', + reclaimed: result.count, + leaseMs: getProcessingLeaseMs(), + }); + } + + return result.count; +} + // ─── Middleware ───────────────────────────────────────────────────────────── /** @@ -294,7 +421,12 @@ export async function failIdempotencyRecord( * Returns: * - 400 if Idempotency-Key is missing or invalid * - 409 if request is different from pending submission - * - Same response if resubmitting identical request + * - 409 with `retryAfter` while an unexpired processing lease is held + * - Same response if resubmitting an identical completed/failed request + * + * A pending record whose processing lease has expired is reclaimed and the + * retry is allowed to proceed, so a crashed worker cannot block the key for the + * full retention TTL. */ export function enforceIdempotency(options: { optional?: boolean } = {}) { return async (req: Request, res: Response, next: NextFunction): Promise => { @@ -336,7 +468,7 @@ export function enforceIdempotency(options: { optional?: boolean } = {}) { // Check for existing idempotency record const existingRecord = await getIdempotencyRecord( idempotencyKey, - req.tenantId || '' + req.tenantId || '', ).catch(() => null); if (existingRecord) { @@ -354,12 +486,33 @@ export function enforceIdempotency(options: { optional?: boolean } = {}) { return; } - // Still pending - return 409 Conflict + // Still pending. If the worker's processing lease has expired, the + // original attempt is presumed crashed/hung — reclaim the slot and let + // this retry run. Otherwise report how long to wait. + if (isProcessingLeaseExpired(existingRecord)) { + const reclaimed = await reclaimExpiredIdempotencyRecord( + idempotencyKey, + existingRecord.tenantId, + ).catch(() => false); + + if (reclaimed) { + logger.log('warn', 'Allowing retry after reclaiming expired idempotency lease', { + action: 'idempotency_lease_retry', + keyId: idempotencyKey, + tenantId: existingRecord.tenantId, + ageMs: Date.now() - existingRecord.createdAt.getTime(), + }); + next(); + return; + } + } + + // Still pending - return 409 Conflict with the remaining lease time. res.status(409).json({ error: 'Conflict', message: 'Request is still being processed. Please wait or retry with a new Idempotency-Key.', code: 'IDEMPOTENCY_PENDING', - retryAfter: 30, + retryAfter: processingLeaseRemainingSeconds(existingRecord), }); return; } @@ -396,306 +549,6 @@ declare global { } } -// ─── Error Classes ────────────────────────────────────────────────────── - -export class IdempotencyConflictError extends Error { - constructor(message = 'Idempotency key already used for a different request body') { - super(message); - this.name = 'IdempotencyConflictError'; - } -} - -// ─── Internal Types ───────────────────────────────────────────────────── - -export interface IdempotentOperationResult { - statusCode: number; - body: T; -} - -interface StoredResponse extends IdempotentOperationResult { - fingerprint: string; - metadata: IdempotencyKeyMetadata; -} - -interface PendingOperation { - fingerprint: string; - promise: Promise>; - metadata: IdempotencyKeyMetadata; -} - -export interface IdempotencyKeyInfo { - key: string; - metadata: IdempotencyKeyMetadata; -} - -export interface IdempotencyMetrics { - hits: number; - conflicts: number; - evictions: number; - activeKeys: number; - pendingKeys: number; -} - -// ─── Redis key prefix ─────────────────────────────────────────────────── - -const REDIS_PREFIX = 'idempotency:'; - -// ─── Fingerprint helper ───────────────────────────────────────────────── - -export function getIdempotencyHashThreshold(): number { - return parseInt(process.env.IDEMPOTENCY_HASH_THRESHOLD_BYTES || '4096', 10); -} - -export function buildIdempotencyFingerprint(payload: unknown): string { - const stable = stableStringify(payload); - const byteLength = Buffer.byteLength(stable, 'utf-8'); - if (byteLength > getIdempotencyHashThreshold()) { - return `hashv1:${crypto.createHash('sha256').update(stable).digest('hex')}`; - } - return stable; -} - -function stableStringify(value: unknown): string { - if (value === null) return 'null'; - if (value instanceof Date) return JSON.stringify(value.toISOString()); - if (typeof value !== 'object') return JSON.stringify(value); - - if (Array.isArray(value)) { - return `[${value.map((item) => stableStringify(item)).join(',')}]`; - } - - const record = value as Record; - const keys = Object.keys(record).sort(); - const serialized = keys.map((key) => `${JSON.stringify(key)}:${stableStringify(record[key])}`); - return `{${serialized.join(',')}}`; -} - -// ─── IdempotencyStore ─────────────────────────────────────────────────── - -export class IdempotencyStore { - private readonly completedResponses: NodeCache; - private readonly pendingResponses = new Map>(); - - private _hits = 0; - private _conflicts = 0; - private _evictions = 0; - - constructor(private readonly ttlMs = 24 * 60 * 60 * 1000) { - const ttlSeconds = Math.max(1, Math.ceil(this.ttlMs / 1000)); - this.completedResponses = new NodeCache({ stdTTL: ttlSeconds, checkperiod: ttlSeconds }); - this.completedResponses.on('expired', (_key: string, _value: unknown) => { - this._evictions++; - }); - } - - private redisKey(key: string): string { - return `${REDIS_PREFIX}${key}`; - } - - private get redis() { - const client = redisClientManager.getClient(); - return redisClientManager.isReady() && client ? client : null; - } - - private async redisGet(key: string): Promise | null> { - const r = this.redis; - if (!r) return null; - try { - const raw = await r.get(this.redisKey(key)); - return raw ? (JSON.parse(raw) as StoredResponse) : null; - } catch { - return null; - } - } - - private async redisSet(key: string, value: StoredResponse): Promise { - const r = this.redis; - if (!r) return; - try { - const ttlSeconds = Math.max(1, Math.ceil(this.ttlMs / 1000)); - await r.set(this.redisKey(key), JSON.stringify(value), 'EX', ttlSeconds); - } catch (err) { - console.log(JSON.stringify({ level: 'warn', event: 'idempotency_redis_write_fail', key, reason: (err as Error).message })); - } - } - - private async redisDel(key: string): Promise { - const r = this.redis; - if (!r) return false; - try { - return (await r.del(this.redisKey(key))) > 0; - } catch { - return false; - } - } - - async execute( - key: string, - fingerprint: string, - operation: () => Promise> - ): Promise<{ result: IdempotentOperationResult; replayed: boolean }> { - const now = new Date().toISOString(); - - let completed = await this.redisGet(key); - if (!completed) { - completed = this.completedResponses.get>(key) ?? null; - } - - if (completed) { - if (completed.fingerprint !== fingerprint) { - this._conflicts++; - throw new IdempotencyConflictError(); - } - this._hits++; - completed.metadata.lastAccessedAt = now; - completed.metadata.replayCount++; - await this.redisSet(key, completed); - this.completedResponses.set(key, completed); - return { result: { statusCode: completed.statusCode, body: completed.body }, replayed: true }; - } - - const pendingOperation = this.pendingResponses.get(key) as PendingOperation | undefined; - if (pendingOperation) { - if (pendingOperation.fingerprint !== fingerprint) { - this._conflicts++; - throw new IdempotencyConflictError(); - } - this._hits++; - pendingOperation.metadata.lastAccessedAt = now; - pendingOperation.metadata.replayCount++; - const replayed = await pendingOperation.promise; - return { result: { statusCode: replayed.statusCode, body: replayed.body }, replayed: true }; - } - - const metadata: IdempotencyKeyMetadata = { createdAt: now, lastAccessedAt: now, replayCount: 0, status: 'pending' }; - - const operationPromise = (async () => { - const result = await operation(); - const stored: StoredResponse = { - ...result, - fingerprint, - metadata: { ...metadata, status: 'completed', lastAccessedAt: new Date().toISOString() }, - }; - await this.redisSet(key, stored); - this.completedResponses.set(key, stored, this.ttlMs / 1000); - return stored; - })(); - - this.pendingResponses.set(key, { fingerprint, promise: operationPromise, metadata }); - - try { - const stored = await operationPromise; - return { result: { statusCode: stored.statusCode, body: stored.body }, replayed: false }; - } finally { - this.pendingResponses.delete(key); - } - } - - inspectKeys(prefix?: string): IdempotencyKeyInfo[] { - const results: IdempotencyKeyInfo[] = []; - for (const key of this.completedResponses.keys()) { - if (prefix && !key.startsWith(prefix)) continue; - const entry = this.completedResponses.get>(key); - if (entry) results.push({ key, metadata: { ...entry.metadata } }); - } - for (const [key, pending] of this.pendingResponses.entries()) { - if (prefix && !key.startsWith(prefix)) continue; - if (!results.some((r) => r.key === key)) { - results.push({ key, metadata: { ...pending.metadata } }); - } - } - return results; - } - - async deleteKey(key: string): Promise { - const deletedLocal = this.completedResponses.del(key) > 0; - const deletedPending = this.pendingResponses.delete(key); - const deletedRedis = await this.redisDel(key); - if (deletedLocal || deletedPending || deletedRedis) { - this._evictions++; - return true; - } - return false; - } - - clear(): void { - const count = this.completedResponses.keys().length + this.pendingResponses.size; - this._evictions += count; - this.completedResponses.flushAll(); - this.pendingResponses.clear(); - } - - getMetrics(): IdempotencyMetrics { - return { - hits: this._hits, - conflicts: this._conflicts, - evictions: this._evictions, - activeKeys: this.completedResponses.keys().length, - pendingKeys: this.pendingResponses.size, - }; - } - - async pruneStaleKeys( - retentionMs: number, - dryRun = false, - ): Promise<{ pruned: number; localPruned: number; redisPruned: number }> { - const cutoff = Date.now() - retentionMs; - let localPruned = 0; - let redisPruned = 0; - - for (const key of this.completedResponses.keys()) { - const entry = this.completedResponses.get>(key); - if (!entry) continue; - const createdAt = Date.parse(entry.metadata.createdAt); - if (Number.isNaN(createdAt) || createdAt >= cutoff) continue; - if (!dryRun) { - this.completedResponses.del(key); - this._evictions++; - } - localPruned++; - } - - const r = this.redis; - if (r) { - let cursor = '0'; - do { - const [nextCursor, keys] = await r.scan(cursor, 'MATCH', `${REDIS_PREFIX}*`, 'COUNT', 100); - cursor = nextCursor; - for (const redisKey of keys) { - try { - const raw = await r.get(redisKey); - if (!raw) continue; - const entry = JSON.parse(raw) as StoredResponse; - const createdAt = Date.parse(entry.metadata?.createdAt ?? ''); - const ttl = await r.ttl(redisKey); - const isStale = (!Number.isNaN(createdAt) && createdAt < cutoff) || ttl === 0; - if (!isStale) continue; - if (!dryRun) { - await r.del(redisKey); - this._evictions++; - } - redisPruned++; - } catch { - if (!dryRun) { - await r.del(redisKey); - this._evictions++; - } - redisPruned++; - } - } - } while (cursor !== '0'); - } - - return { pruned: localPruned + redisPruned, localPruned, redisPruned }; - } -} - -// ─── Singleton ────────────────────────────────────────────────────────── - -export const idempotencyStore = new IdempotencyStore( - parseInt(process.env.IDEMPOTENCY_KEY_TTL_MS || '86400000', 10) -); - // ─── Cleanup ──────────────────────────────────────────────────────────────── /** @@ -740,383 +593,25 @@ export function startIdempotencyCleanupTask(intervalMs = 3600000): NodeJS.Timer }, intervalMs); } -// ─── In-flight store (Redis + NodeCache) ───────────────────────────────────── -// -// Issue #811: multi-instance deployments lost idempotency guarantees on pod -// recycle because responses were stored only in-process. This revision -// persists completed responses to Redis using SET … EX so all replicas share -// the same store, falling back to NodeCache when Redis is unavailable -// (fail-open, with a warning). -// -// The Prisma helpers above (enforceIdempotency / *IdempotencyRecord) are the -// newer, DB-backed design but are not yet wired into any route. The mutation -// endpoints (deposit/withdrawal/transfer), the admin key routes and the -// retention sweeper all still call the store below, so both surfaces live here -// until those call sites are migrated. Restoring this store is what keeps -// `idempotencyStore`, `IdempotencyStore` and `IdempotencyConflictError` -// resolvable for those consumers. -// ─── Response Store (Redis + NodeCache) ───────────────────────────────────── -// -// Shared response store used by vault endpoints and the transfer orchestrator. -// Completed responses are persisted to Redis (when available) with a NodeCache -// in-process fallback. Retention sweeps live in idempotencyRetention.ts. - -// ─── Public Types ───────────────────────────────────────────────────────────── - -export interface IdempotentOperationResult { - statusCode: number; - body: T; -} - -/** Metadata attached to every idempotency key entry. */ -export interface IdempotencyEntryMetadata { -export interface IdempotencyStoreKeyMetadata { - /** ISO-8601 timestamp when the key was first stored. */ - createdAt: string; - /** ISO-8601 timestamp of the most recent access (read or write). */ - lastAccessedAt: string; - /** Number of times this key has been replayed (returned cached result). */ - replayCount: number; - /** Current state of the entry. */ - status: 'pending' | 'completed'; -} - -/** Summary returned by GET /admin/idempotency/keys. */ -export interface IdempotencyKeyInfo { - key: string; - metadata: IdempotencyEntryMetadata; - metadata: IdempotencyStoreKeyMetadata; -} - -/** Snapshot of store-wide observability counters. */ -export interface IdempotencyMetrics { - hits: number; - conflicts: number; - evictions: number; - activeKeys: number; - pendingKeys: number; -} - -interface StoredResponse extends IdempotentOperationResult { - fingerprint: string; - metadata: IdempotencyEntryMetadata; -// ─── Internal Types ─────────────────────────────────────────────────────────── - -interface StoredResponse extends IdempotentOperationResult { - fingerprint: string; - metadata: IdempotencyStoreKeyMetadata; -} - -interface PendingOperation { - fingerprint: string; - promise: Promise>; - metadata: IdempotencyEntryMetadata; -} - - metadata: IdempotencyStoreKeyMetadata; -} - -// ─── Errors ─────────────────────────────────────────────────────────────────── - -export class IdempotencyConflictError extends Error { - constructor(message = 'Idempotency key already used for a different request body') { - super(message); - this.name = 'IdempotencyConflictError'; - } -} - -const REDIS_PREFIX = 'idempotency:'; - -// ─── Redis key prefix ───────────────────────────────────────────────────────── - -const REDIS_PREFIX = 'idempotency:'; - -// ─── Store ──────────────────────────────────────────────────────────────────── - -export class IdempotencyStore { - /** Fallback in-process store used when Redis is unavailable. */ - private readonly localCache: NodeCache; - private readonly pendingResponses = new Map>(); - - // Observability counters - private _hits = 0; - private _conflicts = 0; - private _evictions = 0; - - constructor(private readonly ttlMs = 24 * 60 * 60 * 1000) { - const ttlSeconds = Math.max(1, Math.ceil(this.ttlMs / 1000)); - this.localCache = new NodeCache({ stdTTL: ttlSeconds, checkperiod: ttlSeconds }); - this.localCache.on('expired', () => { this._evictions++; }); - } - - // ─── Redis helpers ───────────────────────────────────────────────────────── - - private redisKey(key: string): string { - return `${REDIS_PREFIX}${key}`; - } - - private get redis() { - const client = redisClientManager.getClient(); - return redisClientManager.isReady() && client ? client : null; - } - - private async redisGet(key: string): Promise | null> { - const r = this.redis; - if (!r) return null; - try { - const raw = await r.get(this.redisKey(key)); - return raw ? (JSON.parse(raw) as StoredResponse) : null; - } catch { - return null; - } - } - - private async redisSet(key: string, value: StoredResponse): Promise { - const r = this.redis; - if (!r) return; - try { - const ttlSeconds = Math.max(1, Math.ceil(this.ttlMs / 1000)); - await r.set(this.redisKey(key), JSON.stringify(value), 'EX', ttlSeconds); - } catch (err) { - console.log(JSON.stringify({ level: 'warn', event: 'idempotency_redis_write_fail', key, reason: (err as Error).message })); - } - } - - private async redisDel(key: string): Promise { - const r = this.redis; - if (!r) return false; - try { - return (await r.del(this.redisKey(key))) > 0; - } catch { - return false; - } - } - - // ─── Core execute ────────────────────────────────────────────────────────── - - async execute( - key: string, - fingerprint: string, - operation: () => Promise> - ): Promise<{ result: IdempotentOperationResult; replayed: boolean }> { - const now = new Date().toISOString(); - - // 1. Check Redis first, then local cache - let completed = await this.redisGet(key); - if (!completed) { - completed = this.localCache.get>(key) ?? null; - } - - if (completed) { - if (completed.fingerprint !== fingerprint) { - this._conflicts++; - throw new IdempotencyConflictError(); - } - this._hits++; - completed.metadata.lastAccessedAt = now; - completed.metadata.replayCount++; - // Refresh in both backends; errors are non-fatal - await this.redisSet(key, completed); - this.localCache.set(key, completed); - return { result: { statusCode: completed.statusCode, body: completed.body }, replayed: true }; - } - - // 2. Currently in-flight - const pendingOperation = this.pendingResponses.get(key) as PendingOperation | undefined; - if (pendingOperation) { - if (pendingOperation.fingerprint !== fingerprint) { - this._conflicts++; - throw new IdempotencyConflictError(); - } - this._hits++; - pendingOperation.metadata.lastAccessedAt = now; - pendingOperation.metadata.replayCount++; - const replayed = await pendingOperation.promise; - return { result: { statusCode: replayed.statusCode, body: replayed.body }, replayed: true }; - } - - // 3. First execution - const metadata: IdempotencyEntryMetadata = { createdAt: now, lastAccessedAt: now, replayCount: 0, status: 'pending' }; - const metadata: IdempotencyStoreKeyMetadata = { createdAt: now, lastAccessedAt: now, replayCount: 0, status: 'pending' }; - - const operationPromise = (async () => { - const result = await operation(); - const stored: StoredResponse = { - ...result, - fingerprint, - metadata: { ...metadata, status: 'completed', lastAccessedAt: new Date().toISOString() }, - }; - // Persist to Redis (primary) and local cache (fallback/fast-path) - await this.redisSet(key, stored); - this.localCache.set(key, stored, this.ttlMs / 1000); - return stored; - })(); - - this.pendingResponses.set(key, { fingerprint, promise: operationPromise, metadata }); - - try { - const stored = await operationPromise; - return { result: { statusCode: stored.statusCode, body: stored.body }, replayed: false }; - } finally { - this.pendingResponses.delete(key); - } - } - - // ─── Inspection ──────────────────────────────────────────────────────────── - - inspectKeys(prefix?: string): IdempotencyKeyInfo[] { - const results: IdempotencyKeyInfo[] = []; - for (const key of this.localCache.keys()) { - if (prefix && !key.startsWith(prefix)) continue; - const entry = this.localCache.get>(key); - if (entry) results.push({ key, metadata: { ...entry.metadata } }); - } - for (const [key, pending] of this.pendingResponses.entries()) { - if (prefix && !key.startsWith(prefix)) continue; - if (!results.some((r) => r.key === key)) { - results.push({ key, metadata: { ...pending.metadata } }); - } - } - return results; - } - - // ─── Targeted deletion ───────────────────────────────────────────────────── - - async deleteKey(key: string): Promise { - const deletedLocal = this.localCache.del(key) > 0; - const deletedPending = this.pendingResponses.delete(key); - const deletedRedis = await this.redisDel(key); - if (deletedLocal || deletedPending || deletedRedis) { - this._evictions++; - return true; - } - return false; - } - - // ─── Global clear (admin only) ───────────────────────────────────────────── - - clear(): void { - const count = this.localCache.keys().length + this.pendingResponses.size; - this._evictions += count; - this.localCache.flushAll(); - this.pendingResponses.clear(); - // Note: Redis keys are prefixed with REDIS_PREFIX; a full Redis FLUSHDB is intentionally - // not issued here to avoid clearing unrelated keys. Use deleteKey() per-key when needed. - } - - // ─── Observability ───────────────────────────────────────────────────────── - - getMetrics(): IdempotencyMetrics { - return { - hits: this._hits, - conflicts: this._conflicts, - evictions: this._evictions, - activeKeys: this.localCache.keys().length, - pendingKeys: this.pendingResponses.size, - }; - } - - // ─── Retention cleanup ───────────────────────────────────────────────────── - - async pruneStaleKeys( - retentionMs: number, - dryRun = false, - ): Promise<{ pruned: number; localPruned: number; redisPruned: number }> { - /** - * Removes entries older than `retentionMs` from the local cache and Redis. - * With `dryRun`, only counts the entries that would be removed: nothing is - * deleted and the eviction counter is left untouched (Issue #1375). - */ - - async pruneStaleKeys( - retentionMs: number, - dryRun = false, - ): Promise<{ pruned: number; localPruned: number; redisPruned: number; dryRun: boolean }> { - const cutoff = Date.now() - retentionMs; - let localPruned = 0; - let redisPruned = 0; - - for (const key of this.localCache.keys()) { - const entry = this.localCache.get>(key); - if (!entry) continue; - const createdAt = Date.parse(entry.metadata.createdAt); - if (Number.isNaN(createdAt) || createdAt >= cutoff) continue; - if (!dryRun) { - this.localCache.del(key); - this._evictions++; - } - localPruned++; - } - - const r = this.redis; - if (r) { - let cursor = '0'; - do { - const [nextCursor, keys] = await r.scan(cursor, 'MATCH', `${REDIS_PREFIX}*`, 'COUNT', 100); - cursor = nextCursor; - for (const redisKey of keys) { - try { - const raw = await r.get(redisKey); - if (!raw) continue; - const entry = JSON.parse(raw) as StoredResponse; - const createdAt = Date.parse(entry.metadata?.createdAt ?? ''); - const ttl = await r.ttl(redisKey); - const isStale = (!Number.isNaN(createdAt) && createdAt < cutoff) || ttl === 0; - if (!isStale) continue; - if (!dryRun) { - await r.del(redisKey); - this._evictions++; - } - redisPruned++; - } catch { - if (!dryRun) { - await r.del(redisKey); - this._evictions++; - } - redisPruned++; - } - } - } while (cursor !== '0'); - } - - return { pruned: localPruned + redisPruned, localPruned, redisPruned }; - return { pruned: localPruned + redisPruned, localPruned, redisPruned, dryRun }; - } -} - -// ─── Singleton ──────────────────────────────────────────────────────────────── - -export const idempotencyStore = new IdempotencyStore( - parseInt(process.env.IDEMPOTENCY_KEY_TTL_MS || '86400000', 10) -); - -// ─── Fingerprint helper ─────────────────────────────────────────────────────── - -export function getIdempotencyHashThreshold(): number { - return parseInt(process.env.IDEMPOTENCY_HASH_THRESHOLD_BYTES || '4096', 10); -} - -export function buildIdempotencyFingerprint(payload: unknown): string { - const stable = stableStringify(payload); - const byteLength = Buffer.byteLength(stable, 'utf-8'); - if (byteLength > getIdempotencyHashThreshold()) { - return `hashv1:${crypto.createHash('sha256').update(stable).digest('hex')}`; - } - return stable; -} - -function stableStringify(value: unknown): string { - if (value === null) return 'null'; - if (value instanceof Date) return JSON.stringify(value.toISOString()); - if (typeof value !== 'object') return JSON.stringify(value); +/** + * Starts a periodic reaper for abandoned pending records (expired processing + * leases). Runs every minute by default; the timer is unref'd so it never keeps + * the process alive on its own. + */ +export function startIdempotencyLeaseReaper(intervalMs = 60_000): NodeJS.Timer { + const timer = setInterval(() => { + reclaimExpiredIdempotencyRecords().catch((error) => { + logger.log('error', 'Idempotency lease reaper failed', { + action: 'idempotency_lease_reaper_error', + error: error instanceof Error ? error.message : String(error), + }); + }); + }, Math.max(1_000, intervalMs)); - if (Array.isArray(value)) { - return `[${value.map((item) => stableStringify(item)).join(',')}]`; + const maybeUnref = timer as unknown as { unref?: () => void }; + if (typeof maybeUnref.unref === 'function') { + maybeUnref.unref(); } - const record = value as Record; - const keys = Object.keys(record).sort(); - const serialized = keys.map((key) => `${JSON.stringify(key)}:${stableStringify(record[key])}`); - return `{${serialized.join(',')}}`; + return timer; } - diff --git a/backend/src/idempotencyStore.ts b/backend/src/idempotencyStore.ts index 41e4e6d3e..4f91d88cc 100644 --- a/backend/src/idempotencyStore.ts +++ b/backend/src/idempotencyStore.ts @@ -271,7 +271,7 @@ export class IdempotencyStore { async pruneStaleKeys( retentionMs: number, dryRun = false, - ): Promise<{ pruned: number; localPruned: number; redisPruned: number }> { + ): Promise<{ pruned: number; localPruned: number; redisPruned: number; dryRun: boolean }> { const cutoff = Date.now() - retentionMs; let localPruned = 0; let redisPruned = 0; @@ -319,7 +319,7 @@ export class IdempotencyStore { } while (cursor !== '0'); } - return { pruned: localPruned + redisPruned, localPruned, redisPruned }; + return { pruned: localPruned + redisPruned, localPruned, redisPruned, dryRun }; } } diff --git a/backend/src/middleware/cache.ts b/backend/src/middleware/cache.ts index c080b4ca1..028ea48a1 100644 --- a/backend/src/middleware/cache.ts +++ b/backend/src/middleware/cache.ts @@ -1,7 +1,12 @@ import type { Request, Response, NextFunction } from 'express'; import { cacheHitCount, cacheMissCount, cacheEvictionCount } from '../metrics'; import { latencyMonitoringService } from '../latencyMonitoring'; -import { getFromCache, setInCache, redisCacheClient } from '../redisCache'; +import { + getFromCacheDetailed, + setInCache, + redisCacheClient, + redisCacheBypassCount, +} from '../redisCache'; interface CacheEntry { data: unknown; @@ -129,6 +134,68 @@ export function buildCacheKey(req: Request): string { return `${req.method}:${pathname}:${pairs.join('&')}`; } +// ── Cache Headers ──────────────────────────────────────────────────────────── + +/** + * Requests can force a cache refresh with either `X-Cache-Bypass: true` or a + * standard `Cache-Control: no-cache` / `no-store` directive. The lookup is + * skipped and the freshly computed response replaces the stored entry, so a + * client can recover from a stale fail-open response without waiting for TTL. + */ +export function isCacheBypassRequested(req: Request): boolean { + const bypassHeader = req.headers['x-cache-bypass']; + const bypassValue = Array.isArray(bypassHeader) ? bypassHeader[0] : bypassHeader; + if (bypassValue === 'true' || bypassValue === '1') return true; + + const cacheControl = req.headers['cache-control']; + const value = Array.isArray(cacheControl) ? cacheControl.join(',') : (cacheControl ?? ''); + return /\bno-cache\b|\bno-store\b/i.test(value); +} + +const CACHE_META_HEADERS = new Set([ + 'cache-control', + 'x-cache-hit', + 'x-cache-backend', + 'x-cache-stale', + 'x-cache-age', + 'x-cache-bypass', + 'warning', +]); + +interface CacheHeaderOptions { + hit: boolean; + backend: 'redis' | 'memory' | 'none'; + stale: boolean; + ageMs: number | null; + maxAgeSec: number; + bypassed: boolean; +} + +/** + * Applies the cache metadata headers. When a response is served from the + * fail-open LRU fallback we mark it stale and disable downstream caching so a + * stale vault summary cannot be re-cached by a CDN or browser indefinitely. + */ +function applyCacheHeaders(res: Response, options: CacheHeaderOptions): void { + res.setHeader('X-Cache-Hit', String(options.hit)); + res.setHeader('X-Cache-Backend', options.backend); + res.setHeader('X-Cache-Stale', String(options.stale)); + + if (options.ageMs !== null) { + res.setHeader('X-Cache-Age', String(Math.max(0, Math.floor(options.ageMs / 1000)))); + } + if (options.bypassed) { + res.setHeader('X-Cache-Bypass', 'true'); + } + + if (options.stale) { + res.setHeader('Warning', '110 - "Response is stale"'); + res.setHeader('Cache-Control', 'no-store'); + } else { + res.setHeader('Cache-Control', `public, max-age=${Math.max(options.maxAgeSec, 0)}`); + } +} + // ── Middleware ──────────────────────────────────────────────────────────────── export interface CacheOptions { @@ -163,6 +230,13 @@ export function cacheMiddleware(options: CacheOptions) { const route = (req.route?.path as string | undefined) ?? req.path; const cacheKey = buildCacheKey(req); + const bypassRequested = isCacheBypassRequested(req); + + // The synchronous path below does not go through getFromCacheDetailed, so + // record bypasses there explicitly. + if (bypassRequested && !redisCacheClient.isConfigured) { + redisCacheBypassCount.inc({ route: cacheKey }); + } // ── Redis-aware async path ─────────────────────────────────────────────── // When Redis is configured we use the async Redis+LRU path; otherwise we @@ -170,21 +244,28 @@ export function cacheMiddleware(options: CacheOptions) { if (redisCacheClient.isConfigured) { const hitStart = Date.now(); - getFromCache(cacheKey) - .then((cached) => { - if (cached && cached.expiresAt > Date.now()) { + getFromCacheDetailed(cacheKey, { bypass: bypassRequested }) + .then((lookup) => { + const cached = lookup.entry; + const now = Date.now(); + + if (cached && cached.expiresAt > now) { // Cache HIT (Redis or LRU fallback) cacheHitCount.inc({ method: req.method, route }); - const remainingMs = cached.expiresAt - Date.now(); - const remainingSec = Math.ceil(remainingMs / 1000); + const remainingSec = Math.max(0, Math.ceil((cached.expiresAt - now) / 1000)); - res.setHeader('X-Cache-Hit', 'true'); - res.setHeader('X-Cache-Backend', redisCacheClient.isReady ? 'redis' : 'memory'); - res.setHeader('Cache-Control', `public, max-age=${Math.max(remainingSec, 0)}`); + applyCacheHeaders(res, { + hit: true, + backend: lookup.backend, + stale: lookup.stale, + ageMs: lookup.ageMs, + maxAgeSec: remainingSec, + bypassed: false, + }); for (const [name, value] of Object.entries(cached.headers)) { - if (name !== 'Cache-Control' && name !== 'X-Cache-Hit' && name !== 'X-Cache-Backend') { + if (!CACHE_META_HEADERS.has(name.toLowerCase())) { res.setHeader(name, value); } } @@ -205,9 +286,14 @@ export function cacheMiddleware(options: CacheOptions) { if (status >= 200 && status < 300) { const maxAgeSec = Math.ceil(options.ttl / 1000); - res.setHeader('X-Cache-Hit', 'false'); - res.setHeader('X-Cache-Backend', redisCacheClient.isReady ? 'redis' : 'memory'); - res.setHeader('Cache-Control', `public, max-age=${maxAgeSec}`); + applyCacheHeaders(res, { + hit: false, + backend: redisCacheClient.isReady ? 'redis' : 'memory', + stale: false, + ageMs: null, + maxAgeSec, + bypassed: bypassRequested, + }); const entry = { data, @@ -240,23 +326,32 @@ export function cacheMiddleware(options: CacheOptions) { } // ── Synchronous in-memory-only path (Redis not configured) ─────────────── - const cached = responseCache.get(cacheKey); + const cached = bypassRequested ? undefined : responseCache.get(cacheKey); if (cached && cached.expiresAt > Date.now()) { // Cache HIT const hitStart = Date.now(); cacheHitCount.inc({ method: req.method, route }); - const remainingMs = cached.expiresAt - Date.now(); - const remainingSec = Math.ceil(remainingMs / 1000); - - res.setHeader('X-Cache-Hit', 'true'); - res.setHeader('X-Cache-Backend', 'memory'); - res.setHeader('Cache-Control', `public, max-age=${Math.max(remainingSec, 0)}`); + const now = Date.now(); + const remainingSec = Math.ceil((cached.expiresAt - now) / 1000); + const ageMs = + Number.isFinite(cached.ttl) && Number.isFinite(cached.expiresAt) + ? Math.max(0, now - (cached.expiresAt - cached.ttl)) + : null; + + applyCacheHeaders(res, { + hit: true, + backend: 'memory', + stale: false, + ageMs, + maxAgeSec: remainingSec, + bypassed: false, + }); // Restore any other cached response headers for (const [name, value] of Object.entries(cached.headers)) { - if (name !== 'Cache-Control' && name !== 'X-Cache-Hit' && name !== 'X-Cache-Backend') { + if (!CACHE_META_HEADERS.has(name.toLowerCase())) { res.setHeader(name, value); } } @@ -281,9 +376,14 @@ export function cacheMiddleware(options: CacheOptions) { // R3 C3: only cache 2xx responses if (status >= 200 && status < 300) { const maxAgeSec = Math.ceil(options.ttl / 1000); - res.setHeader('X-Cache-Hit', 'false'); - res.setHeader('X-Cache-Backend', 'memory'); - res.setHeader('Cache-Control', `public, max-age=${maxAgeSec}`); + applyCacheHeaders(res, { + hit: false, + backend: 'memory', + stale: false, + ageMs: null, + maxAgeSec, + bypassed: bypassRequested, + }); responseCache.set(cacheKey, { data, diff --git a/backend/src/redisCache.ts b/backend/src/redisCache.ts index 1ac01a2b2..c5fe8981b 100644 --- a/backend/src/redisCache.ts +++ b/backend/src/redisCache.ts @@ -14,6 +14,14 @@ * Prometheus counters are exported so operators can track Redis vs in-memory * hit/miss ratios from the /metrics endpoint. * + * Fail-open staleness (issue #1553): + * When Redis is configured but unavailable, reads fall back to the in-memory + * LRU. Those reads are reported as `stale` by `getFromCacheDetailed`, counted + * by `redis_cache_fallback_total` / `redis_cache_stale_serve_total`, and the + * cache middleware emits `X-Cache-Stale` + `Warning` and disables downstream + * caching. A request may force a refresh with `X-Cache-Bypass: true` or + * `Cache-Control: no-cache`. + * * Environment variables: * REDIS_URL – Redis connection URL (required to enable Redis) * REDIS_CACHE_KEY_PREFIX – Key namespace prefix (default: "cache:") @@ -105,6 +113,48 @@ export const redisCacheConnectionStatus = createOrGet( 'redis_cache_connection_status', ); +/** + * Counts reads that were served from the in-memory LRU because Redis was + * configured but unreachable. A non-zero rate means responses may be stale. + */ +export const redisCacheFallbackCount = createOrGet( + () => + new Counter({ + name: 'redis_cache_fallback_total', + help: 'Total cache reads served from the in-memory LRU fallback while Redis was unavailable', + labelNames: ['route', 'reason'] as const, + registers: [defaultRegister], + }), + 'redis_cache_fallback_total', +); + +/** + * Counts stale responses actually returned to clients from the fallback store. + * Alert on this counter: a sustained rate means the vault summary can drift. + */ +export const redisCacheStaleServeCount = createOrGet( + () => + new Counter({ + name: 'redis_cache_stale_serve_total', + help: 'Total stale cache entries served from the in-memory fallback', + labelNames: ['route', 'backend'] as const, + registers: [defaultRegister], + }), + 'redis_cache_stale_serve_total', +); + +/** Counts requests that explicitly bypassed the cache (force refresh). */ +export const redisCacheBypassCount = createOrGet( + () => + new Counter({ + name: 'redis_cache_bypass_total', + help: 'Total cache reads skipped because the caller requested a force refresh', + labelNames: ['route'] as const, + registers: [defaultRegister], + }), + 'redis_cache_bypass_total', +); + // ─── Redis Client ───────────────────────────────────────────────────────────── /** @@ -373,36 +423,128 @@ export const redisCacheClient = new RedisCacheClient(); // ─── High-level Cache Operations ───────────────────────────────────────────── +/** Which store produced a cache lookup result. */ +export type CacheBackend = 'redis' | 'memory' | 'none'; + +/** Options accepted by {@link getFromCacheDetailed}. */ +export interface CacheReadOptions { + /** + * Force a cache miss so the caller re-computes and re-populates the entry. + * Used to implement `Cache-Control: no-cache` / `X-Cache-Bypass` requests. + */ + bypass?: boolean; +} + /** - * Read a cache entry by key, consulting Redis first then falling back to the - * in-memory LRU store. + * Rich result of a cache lookup. The middleware uses the extra metadata to + * emit staleness headers and metrics for the in-memory fail-open path. + */ +export interface CacheLookupResult { + entry: RedisCacheEntry | null; + backend: CacheBackend; + /** True when a fallback (LRU) entry was served while Redis was configured but unavailable. */ + stale: boolean; + /** True when the entry came from the in-memory fallback, even if Redis is not configured. */ + fromFallback: boolean; + /** Approximate age of the entry in milliseconds, when known. */ + ageMs: number | null; +} + +/** Age of an entry derived from its TTL and expiry timestamp. */ +function entryAgeMs(entry: RedisCacheEntry, now: number): number | null { + if (!Number.isFinite(entry.ttl) || !Number.isFinite(entry.expiresAt)) return null; + const writtenAt = entry.expiresAt - entry.ttl; + return Math.max(0, now - writtenAt); +} + +function lookupResult( + entry: RedisCacheEntry | null, + backend: CacheBackend, + stale: boolean, + fromFallback: boolean, + now: number, +): CacheLookupResult { + return { + entry, + backend, + stale, + fromFallback, + ageMs: entry ? entryAgeMs(entry, now) : null, + }; +} + +/** + * Read a cache entry together with failover metadata. * - * Returns `null` when neither source has a valid (non-expired) entry. + * Order of preference: + * 1. Redis (when configured and healthy) + * 2. In-memory LRU (fail-open) + * + * A fallback hit is marked `stale: true` whenever Redis was configured but + * unavailable: the entry may be arbitrarily old and the caller MUST surface + * that to clients (see the cache middleware's `X-Cache-Stale` header). */ -export async function getFromCache(key: string): Promise { +export async function getFromCacheDetailed( + key: string, + options: CacheReadOptions = {}, +): Promise { + const now = Date.now(); + + if (options.bypass) { + redisCacheBypassCount.inc({ route: key }); + return lookupResult(null, 'none', false, false, now); + } + // Try Redis first if (redisCacheClient.isReady) { const entry = await redisCacheClient.get(key); - if (entry && entry.expiresAt > Date.now()) { + if (entry && entry.expiresAt > now) { redisCacheHitCount.inc({ route: key }); - return entry; + return lookupResult(entry, 'redis', false, false, now); } if (entry) { // Stale entry – Redis should have expired it but hadn't yet; treat as miss redisCacheMissCount.inc({ route: key }); - return null; + return lookupResult(null, 'redis', false, false, now); } redisCacheMissCount.inc({ route: key }); - return null; + return lookupResult(null, 'redis', false, false, now); } - // Fall back to in-memory LRU - const lruEntry = responseCache.get(key); - if (lruEntry && lruEntry.expiresAt > Date.now()) { - return lruEntry as RedisCacheEntry; + // Redis is configured but not ready (outage) -> fail open to LRU. + // When Redis is not configured at all the LRU is the primary store and a hit + // is not stale. + const redisConfigured = redisCacheClient.isConfigured; + const reason = redisConfigured ? 'redis_unavailable' : 'redis_not_configured'; + + const lruEntry = responseCache.get(key) as RedisCacheEntry | undefined; + if (lruEntry && lruEntry.expiresAt > now) { + redisCacheHitCount.inc({ route: key }); + if (redisConfigured) { + redisCacheFallbackCount.inc({ route: key, reason }); + redisCacheStaleServeCount.inc({ route: key, backend: 'memory' }); + } + return lookupResult(lruEntry, 'memory', redisConfigured, true, now); } - return null; + redisCacheMissCount.inc({ route: key }); + if (redisConfigured) { + redisCacheFallbackCount.inc({ route: key, reason }); + } + return lookupResult(null, 'memory', false, true, now); +} + +/** + * Read a cache entry by key, consulting Redis first then falling back to the + * in-memory LRU store. + * + * Returns `null` when neither source has a valid (non-expired) entry. + * Callers that need to know whether a hit came from the fail-open fallback + * should use {@link getFromCacheDetailed} instead. + */ +export async function getFromCache(key: string): Promise { + const result = await getFromCacheDetailed(key); + return result.entry; } /** diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index 9df000a83..dc8ae5dfa 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -46,7 +46,10 @@ import { LazyTransactionHistory, LazyUIPreview, LazyVaultComparison, + LazyVaultHealth, + observeLcp, prefetchDashboardRoutes, + recordRouteNavigation, } from "./lib/routePrefetch"; import NetworkBannerStack from "./components/NetworkBannerStack"; import { useVault, VaultProvider } from "./context/VaultContext"; @@ -60,7 +63,6 @@ const SentryRoutes = Sentry.withSentryReactRouterV6Routing(Routes); const TransactionReceipt = lazy(() => import("./pages/TransactionReceipt")); const StrategyDetail = lazy(() => import("./pages/StrategyDetail")); const Admin = lazy(() => import("./pages/Admin")); -const VaultHealthDashboard = lazy(() => import("./pages/VaultHealthDashboard")); const AuditLog = lazy(() => import("./pages/AuditLog")); // Removed simple fallback in favor of components/ErrorFallback @@ -112,6 +114,15 @@ function AppContent() { return () => globalThis.clearTimeout(timeoutId); }, [location.pathname]); + // Observe Largest Contentful Paint and record each navigation together with + // its prefetch status. Comparing warm vs cold samples is how the issue's LCP + // improvement is measured in the field. + useEffect(() => observeLcp(), []); + + useEffect(() => { + recordRouteNavigation(location.pathname); + }, [location.pathname]); + const handleConnect = useCallback((address: string) => { renewSession(); clearSessionExpired(); @@ -221,24 +232,6 @@ function AppContent() { } /> - } /> - } /> - } /> - } /> - } /> - } /> - } /> - } /> - - - - - - } - /> - + } /> diff --git a/frontend/src/components/Navbar.tsx b/frontend/src/components/Navbar.tsx index ccf220433..ec9d802d7 100644 --- a/frontend/src/components/Navbar.tsx +++ b/frontend/src/components/Navbar.tsx @@ -132,9 +132,16 @@ const Navbar: FC = ({ {t("nav.portfolio")} - + {t("nav.compare")} + + {t("nav.vaultHealth")} + {t("nav.analytics")} @@ -157,8 +164,11 @@ const Navbar: FC = ({ )} + + {t("nav.settings")} + {role === "admin" && ( - + {t("nav.admin")} )} @@ -245,9 +255,12 @@ const Navbar: FC = ({ setIsMobileMenuOpen(false)} {...getRoutePrefetchHandlers("/portfolio")}> {t("nav.portfolio")} - setIsMobileMenuOpen(false)}> + setIsMobileMenuOpen(false)} {...getRoutePrefetchHandlers("/compare")}> {t("nav.compare")} + setIsMobileMenuOpen(false)} {...getRoutePrefetchHandlers("/vault-health")}> + {t("nav.vaultHealth")} + setIsMobileMenuOpen(false)} {...getRoutePrefetchHandlers("/analytics")}> {t("nav.analytics")} @@ -259,19 +272,18 @@ const Navbar: FC = ({ )} + setIsMobileMenuOpen(false)} {...getRoutePrefetchHandlers("/settings")}> + {t("nav.settings")} + {role === "admin" && ( - setIsMobileMenuOpen(false)}> + setIsMobileMenuOpen(false)} {...getRoutePrefetchHandlers("/admin")}> {t("nav.admin")} )} -
- - - {walletAddress && {networkLabel}} -
+ {walletAddress && {networkLabel}}
@@ -285,9 +297,12 @@ const Navbar: FC = ({ setMenuOpen(false)} {...getRoutePrefetchHandlers("/portfolio")}> {t("nav.portfolio")} - setMenuOpen(false)}> + setMenuOpen(false)} {...getRoutePrefetchHandlers("/compare")}> {t("nav.compare")} + setMenuOpen(false)} {...getRoutePrefetchHandlers("/vault-health")}> + {t("nav.vaultHealth")} + setMenuOpen(false)} {...getRoutePrefetchHandlers("/analytics")}> {t("nav.analytics")} @@ -299,8 +314,11 @@ const Navbar: FC = ({ )} + setMenuOpen(false)} {...getRoutePrefetchHandlers("/settings")}> + {t("nav.settings")} + {role === "admin" && ( - setMenuOpen(false)}> + setMenuOpen(false)} {...getRoutePrefetchHandlers("/admin")}> {t("nav.admin")} )} diff --git a/frontend/src/i18n/locales/en.ts b/frontend/src/i18n/locales/en.ts index 6fc01f5df..3cb1eebeb 100644 --- a/frontend/src/i18n/locales/en.ts +++ b/frontend/src/i18n/locales/en.ts @@ -20,6 +20,8 @@ export const en = { compare: "Compare", analytics: "Analytics", transactions: "Transactions", + vaultHealth: "Vault Health", + settings: "Settings", admin: "Admin", }, admin: { diff --git a/frontend/src/i18n/locales/es.ts b/frontend/src/i18n/locales/es.ts index 36366823b..1d87e883b 100644 --- a/frontend/src/i18n/locales/es.ts +++ b/frontend/src/i18n/locales/es.ts @@ -20,6 +20,8 @@ export const es = { compare: "Comparar", analytics: "Analica", transactions: "Transacciones", + vaultHealth: "Salud de la Bóveda", + settings: "Configuración", admin: "Administración", }, admin: { diff --git a/frontend/src/lib/routePrefetch.test.ts b/frontend/src/lib/routePrefetch.test.ts new file mode 100644 index 000000000..7350e796a --- /dev/null +++ b/frontend/src/lib/routePrefetch.test.ts @@ -0,0 +1,53 @@ +import { beforeEach, describe, expect, it } from "vitest"; +import { + HEAVY_ROUTES, + PREFETCHABLE_ROUTES, + PRIMARY_ROUTES, + getRoutePrefetchHandlers, + getRoutePrefetchMetrics, + recordRouteNavigation, + resetRoutePrefetchStateForTests, +} from "./routePrefetch"; + +describe("route prefetch coverage (issue #1569)", () => { + it("includes the previously cold heavy routes", () => { + expect(HEAVY_ROUTES).toEqual( + expect.arrayContaining(["/compare", "/vault-health", "/settings"]), + ); + }); + + it("exposes every primary and heavy route as prefetchable", () => { + for (const route of [...PRIMARY_ROUTES, ...HEAVY_ROUTES]) { + expect(PREFETCHABLE_ROUTES).toContain(route); + } + }); + + it("provides hover, focus and touch handlers for nav links", () => { + const handlers = getRoutePrefetchHandlers("/compare"); + expect(typeof handlers.onMouseEnter).toBe("function"); + expect(typeof handlers.onPointerEnter).toBe("function"); + expect(typeof handlers.onFocus).toBe("function"); + expect(typeof handlers.onTouchStart).toBe("function"); + }); +}); + +describe("route prefetch measurement", () => { + beforeEach(() => { + resetRoutePrefetchStateForTests(); + }); + + it("records navigation samples with prefetch status", () => { + recordRouteNavigation("/vault-health"); + + const metrics = getRoutePrefetchMetrics(); + expect(metrics.navigations).toHaveLength(1); + expect(metrics.navigations[0]).toMatchObject({ + path: "/vault-health", + prefetched: false, + }); + }); + + it("starts with no prefetched routes after a reset", () => { + expect(getRoutePrefetchMetrics().prefetched).toEqual([]); + }); +}); diff --git a/frontend/src/lib/routePrefetch.ts b/frontend/src/lib/routePrefetch.ts index 76a11b188..8c5c98636 100644 --- a/frontend/src/lib/routePrefetch.ts +++ b/frontend/src/lib/routePrefetch.ts @@ -6,40 +6,239 @@ export const routeImports = { "/analytics": () => import("../pages/Analytics"), "/transactions": () => import("../pages/TransactionHistory"), "/compare": () => import("../pages/VaultComparison"), + "/vault-health": () => import("../pages/VaultHealthDashboard"), "/settings": () => import("../pages/Settings"), "/ui-kit": () => import("../pages/UIPreview"), } as const; export type PrefetchableRoute = keyof typeof routeImports; +/** + * Routes on the critical dashboard path. Prefetched during the first idle + * window after mount so the initial navigation is instant. + */ +export const PRIMARY_ROUTES: readonly PrefetchableRoute[] = [ + "/", + "/portfolio", + "/analytics", + "/transactions", +]; + +/** + * Heavy routes (charts, comparison tables, settings forms) that used to stay + * cold until the user clicked them. Issue #1569: now covered by the idle + * prefetch and by nav hover/focus handlers. + */ +export const HEAVY_ROUTES: readonly PrefetchableRoute[] = [ + "/compare", + "/vault-health", + "/settings", +]; + +/** Every route eligible for prefetching. */ +export const PREFETCHABLE_ROUTES: readonly PrefetchableRoute[] = [ + ...PRIMARY_ROUTES, + ...HEAVY_ROUTES, + "/ui-kit", +]; + const prefetchedRoutes = new Set(); -export function prefetchRoute(path: PrefetchableRoute): void { - if (prefetchedRoutes.has(path)) return; +export interface PrefetchStat { + path: PrefetchableRoute; + status: "pending" | "fulfilled" | "rejected"; + startedAt: number; + durationMs: number | null; + error?: string; +} + +/** One sample recorded when the router commits to a new location. */ +export interface NavigationMetric { + path: string; + at: number; + /** Latest observed Largest Contentful Paint, in ms, when available. */ + lcpMs: number | null; + /** Whether the route's chunk was prefetched before navigation. */ + prefetched: boolean; +} + +export interface RoutePrefetchMetrics { + latestLcpMs: number | null; + prefetched: PrefetchableRoute[]; + stats: PrefetchStat[]; + navigations: NavigationMetric[]; +} + +const prefetchStats = new Map(); +const navigationMetrics: NavigationMetric[] = []; +const NAVIGATION_METRIC_LIMIT = 50; + +let latestLcpMs: number | null = null; +let lcpObserver: PerformanceObserver | null = null; + +function now(): number { + if (typeof performance !== "undefined" && typeof performance.now === "function") { + return performance.now(); + } + return Date.now(); +} + +/** + * Prefetches a route chunk exactly once. Best-effort: failures are recorded and + * the route is removed from the in-flight set so a later hover can retry. + */ +export function prefetchRoute(path: PrefetchableRoute): Promise { + if (prefetchedRoutes.has(path)) return Promise.resolve(); + if (typeof window === "undefined") return Promise.resolve(); + prefetchedRoutes.add(path); - void routeImports[path](); + const startedAt = now(); + const stat: PrefetchStat = { + path, + status: "pending", + startedAt, + durationMs: null, + }; + prefetchStats.set(path, stat); + + return routeImports[path]() + .then(() => { + stat.status = "fulfilled"; + stat.durationMs = now() - startedAt; + }) + .catch((error: unknown) => { + stat.status = "rejected"; + stat.durationMs = now() - startedAt; + stat.error = error instanceof Error ? error.message : String(error); + // Allow a later hover/focus to retry a failed chunk load. + prefetchedRoutes.delete(path); + }); } -export function prefetchDashboardRoutes(excludePath?: string): void { - const primaryRoutes: PrefetchableRoute[] = [ - "/", - "/portfolio", - "/analytics", - "/transactions", - ]; - - for (const path of primaryRoutes) { +function prefetchRoutes( + routes: readonly PrefetchableRoute[], + excludePath?: string, +): void { + for (const path of routes) { if (path !== excludePath) { - prefetchRoute(path); + void prefetchRoute(path); } } } +/** + * Prefetches all dashboard routes, including the heavy ones that were + * previously left cold. Called from an idle callback in `App.tsx`. + */ +export function prefetchDashboardRoutes(excludePath?: string): void { + prefetchRoutes(PRIMARY_ROUTES, excludePath); + prefetchRoutes(HEAVY_ROUTES, excludePath); +} + +/** Prefetches only the heavy routes (useful for a second, longer idle slice). */ +export function prefetchHeavyRoutes(excludePath?: string): void { + prefetchRoutes(HEAVY_ROUTES, excludePath); +} + +/** + * Handlers that warm a route on the first sign of user intent. Attach these to + * nav links (``). + * + * `onPointerEnter` covers mouse/pen, `onTouchStart` covers touch devices that + * never fire hover, and `onFocus` covers keyboard navigation. + */ export function getRoutePrefetchHandlers(path: PrefetchableRoute) { + const warm = () => { + void prefetchRoute(path); + }; + return { - onMouseEnter: () => prefetchRoute(path), - onFocus: () => prefetchRoute(path), + onMouseEnter: warm, + onPointerEnter: warm, + onFocus: warm, + onTouchStart: warm, + }; +} + +/** + * Starts observing Largest Contentful Paint so prefetch effectiveness can be + * measured in the field. Returns a cleanup function. + */ +export function observeLcp(): () => void { + if (typeof PerformanceObserver === "undefined") { + return () => undefined; + } + + try { + lcpObserver?.disconnect(); + lcpObserver = new PerformanceObserver((list) => { + const entries = list.getEntries(); + const last = entries[entries.length - 1] as + | (PerformanceEntry & { renderTime?: number; loadTime?: number }) + | undefined; + if (!last) return; + latestLcpMs = last.renderTime ?? last.loadTime ?? last.startTime; + }); + lcpObserver.observe({ + type: "largest-contentful-paint", + buffered: true, + } as PerformanceObserverInit); + } catch { + // Unsupported browser / restricted environment — measurement is optional. + return () => undefined; + } + + return () => { + lcpObserver?.disconnect(); + lcpObserver = null; + }; +} + +/** + * Records a navigation together with the latest LCP sample and whether the + * destination chunk was already prefetched. Comparing `prefetched: true` vs + * `false` samples gives the LCP improvement attributable to prefetching. + */ +export function recordRouteNavigation(path: string): void { + const metric: NavigationMetric = { + path, + at: now(), + lcpMs: latestLcpMs, + prefetched: prefetchedRoutes.has(path), }; + + navigationMetrics.push(metric); + if (navigationMetrics.length > NAVIGATION_METRIC_LIMIT) { + navigationMetrics.shift(); + } + + if (typeof performance !== "undefined" && typeof performance.mark === "function") { + try { + performance.mark(`route-nav:${path}`); + } catch { + // performance.mark is best-effort only. + } + } +} + +/** Snapshot used by tests and by an optional analytics beacon. */ +export function getRoutePrefetchMetrics(): RoutePrefetchMetrics { + return { + latestLcpMs, + prefetched: Array.from(prefetchedRoutes) as PrefetchableRoute[], + stats: Array.from(prefetchStats.values()), + navigations: [...navigationMetrics], + }; +} + +/** Clears measurement state. Exposed for tests. */ +export function resetRoutePrefetchStateForTests(): void { + prefetchedRoutes.clear(); + prefetchStats.clear(); + navigationMetrics.length = 0; + latestLcpMs = null; + lcpObserver?.disconnect(); + lcpObserver = null; } export const LazyHome = lazy(routeImports["/"]); @@ -47,5 +246,6 @@ export const LazyPortfolio = lazy(routeImports["/portfolio"]); export const LazyAnalytics = lazy(routeImports["/analytics"]); export const LazyTransactionHistory = lazy(routeImports["/transactions"]); export const LazyVaultComparison = lazy(routeImports["/compare"]); +export const LazyVaultHealth = lazy(routeImports["/vault-health"]); export const LazySettings = lazy(routeImports["/settings"]); export const LazyUIPreview = lazy(routeImports["/ui-kit"]);