From a2a221bf45361f12d5b7d55b53dfc8ca36104efd Mon Sep 17 00:00:00 2001 From: hardcordev Date: Sat, 17 Oct 2026 04:10:46 -0700 Subject: [PATCH] feat: propagate request and job correlation ids through logs, agent logs and errors --- .env.example | 3 + docs/api.md | 4 +- docs/architecture.md | 4 + src/agents/market-maker.agent.ts | 14 +-- src/agents/trader.agent.ts | 5 +- src/app.ts | 60 ++++++--- src/config/index.ts | 1 + src/controllers/market.controller.ts | 20 +-- src/cron/ingestion.cron.ts | 35 +++--- src/cron/monitor.cron.ts | 33 ++--- src/lib/agent-log.ts | 17 +++ src/lib/context.ts | 80 ++++++++++++ src/lib/logger.ts | 8 ++ src/middlewares/auth.middleware.ts | 13 +- src/middlewares/error.middleware.ts | 8 +- src/middlewares/rate-limit.middleware.ts | 11 +- src/services/hedging.service.ts | 35 +++--- src/services/indexer.service.ts | 18 +-- src/services/market.service.ts | 5 +- src/services/trade.service.ts | 5 +- src/types/index.ts | 1 + src/utils/response.util.ts | 6 +- tests/context.test.ts | 150 +++++++++++++++++++++++ 23 files changed, 431 insertions(+), 105 deletions(-) create mode 100644 src/lib/agent-log.ts create mode 100644 src/lib/context.ts create mode 100644 tests/context.test.ts diff --git a/.env.example b/.env.example index 0b2c5a4..0cf4368 100644 --- a/.env.example +++ b/.env.example @@ -74,6 +74,9 @@ RESOLUTION_SIGNERS= RESOLUTION_THRESHOLD=1 RESOLUTION_DISPUTE_WINDOW_SECS=86400 +# Accept a well-formed inbound X-Request-Id; set false without a trusted proxy. +TRUST_INBOUND_REQUEST_ID=true + # Event indexer (Stellar RPC getEvents poller). INDEXER_ENABLED=true INDEXER_POLL_MS=5000 diff --git a/docs/api.md b/docs/api.md index d9e7313..8319183 100644 --- a/docs/api.md +++ b/docs/api.md @@ -14,9 +14,11 @@ Every response, success or error: ```json { "ok": true, "data": { }, "error": null, "meta": { } } -{ "ok": false, "data": null, "error": { "code": "SOME_CODE", "message": "Readable text", "details": { } } } +{ "ok": false, "data": null, "error": { "code": "SOME_CODE", "message": "Readable text", "details": { }, "requestId": "req_018f..." } } ``` +Every response carries an `X-Request-Id` header (exposed via CORS), and error envelopes repeat it as `error.requestId`; quote it when reporting a problem. A client may send its own `X-Request-Id` (8 to 64 characters of `A-Z a-z 0-9 . _ : -`); any other value is ignored and a new id is generated. + `meta` is optional. List endpoints put pagination there: `{ "page": 1, "limit": 20, "total": 57, "totalPages": 3 }`. diff --git a/docs/architecture.md b/docs/architecture.md index fc3989f..d7c74d3 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -6,6 +6,10 @@ The backend is the API, indexer and LLM pipeline around the OracleDesk Soroban c Express 5 with helmet, hpp, an env CORS allowlist, rate limits (global, `/auth/*`, `/markets/generate`), JSON bodies capped at 1 MB, and pino logging with secrets redacted. BigInts serialise as decimal strings through the app's JSON replacer. Every response uses the `{ ok, data, error, meta }` envelope. The full contract with the frontend is [api.md](api.md). +## Correlation ids + +`src/lib/context.ts` keeps an `AsyncLocalStorage` context with a `requestId` and, where relevant, a `cycleId` or `jobId`. The first Express middleware assigns it: a well-formed inbound `X-Request-Id` (`[A-Za-z0-9._:-]{8,64}`, only when `TRUST_INBOUND_REQUEST_ID=true`) is kept, anything else is replaced and never logged. The id is echoed in the `X-Request-Id` response header and in `error.requestId`. Each cron tick and indexer poll runs in its own context with a fresh `cycleId`, and `POST /markets/generate` runs the cycle with a `jobId` plus the originating `requestId`. A pino mixin adds these ids to every log line, and `writeAgentLog()` (`src/lib/agent-log.ts`, the only writer of `AgentLog` rows) stores them in `data.correlation`. To follow one request or cycle, search the logs for `requestId=` (or the JSON field). + ## Login (SEP-10 style) `auth.service.ts` builds a challenge transaction with `WebAuth.buildChallengeTx`, signed by `AUTH_SIGNING_SECRET` (a key that holds nothing and is not the agent key). It stores the transaction hash in Redis for 5 minutes. `/auth/verify` reads the challenge, enforces its time bounds without the SDK's grace period, consumes it atomically (`GETDEL`), checks the user's signature with `verifyChallengeTxSigners`, and issues a JWT. Admin rights are an env allowlist of G-addresses checked server-side. diff --git a/src/agents/market-maker.agent.ts b/src/agents/market-maker.agent.ts index b73e741..9dc7db9 100644 --- a/src/agents/market-maker.agent.ts +++ b/src/agents/market-maker.agent.ts @@ -8,7 +8,7 @@ import { generateReasoningTrace } from '../services/reasoning.service'; import { createMarketOnChain } from '../services/market-chain.service'; import { enqueueTracePublication } from '../services/trace-publish.service'; import { TRACE_ACTIONS } from '../lib/trace-actions'; -import { prisma } from '../lib/prisma'; +import { writeAgentLog } from '../lib/agent-log'; import { logger } from '../lib/logger'; import { config } from '../config'; import { deadline } from '../lib/resilience'; @@ -135,13 +135,11 @@ export async function runMarketMakerCycle(opts?: { signal?: AbortSignal }): Prom } catch (err) { logger.error({ err }, 'Market Maker Agent: unhandled cycle error'); - await prisma.agentLog.create({ - data: { - agentType: 'MARKET_MAKER', - level: 'ERROR', - action: 'CYCLE_FAILED', - error: String(err), - }, + await writeAgentLog({ + agentType: 'MARKET_MAKER', + level: 'ERROR', + action: 'CYCLE_FAILED', + error: String(err), }).catch(() => {}); throw err; // Re-throw so the job tracker records FAILED status diff --git a/src/agents/trader.agent.ts b/src/agents/trader.agent.ts index 6c86132..68c2049 100644 --- a/src/agents/trader.agent.ts +++ b/src/agents/trader.agent.ts @@ -5,6 +5,7 @@ import { checkCorrelation } from '../services/correlation.service'; import { generateReasoningTrace } from '../services/reasoning.service'; import { detectHedgeOpportunities, autoHedge, checkDailyDrawdown, enforceStopLoss } from '../services/hedging.service'; import { prisma } from '../lib/prisma'; +import { writeAgentLog } from '../lib/agent-log'; import { logger } from '../lib/logger'; import { config } from '../config'; import { deadline } from '../lib/resilience'; @@ -99,9 +100,7 @@ export async function runTraderCycle(opts?: { signal?: AbortSignal }): Promise (typeof value === 'bigint' ? value.toString() : value)); +// ─── Correlation id ─── +// First, so even requests rejected by helmet/CORS/body parsing carry an id. +app.use((req, res, next) => { + const raw = req.get('x-request-id'); + const inbound = acceptInboundId(raw); + if (raw && !inbound) logger.debug({ length: raw.length }, 'Ignoring malformed X-Request-Id'); // never log the value + const requestId = inbound ?? newId('req'); + res.setHeader('X-Request-Id', requestId); + runWithContext({ requestId, kind: 'http' }, () => { + res.locals.requestContext = currentContext(); + next(); + }); +}); + // ─── Security ─── app.use(helmet()); app.use(hpp()); @@ -35,7 +50,7 @@ app.use(cors({ credentials: true, // #123: let browsers send Idempotency-Key and read Idempotent-Replayed. allowedHeaders: ['Content-Type', 'Authorization', 'Idempotency-Key'], - exposedHeaders: ['Idempotent-Replayed', 'Retry-After'], + exposedHeaders: ['Idempotent-Replayed', 'Retry-After', 'X-Request-Id'], })); // ─── Rate Limiting ─── @@ -44,15 +59,34 @@ app.use(globalLimiter); // ─── Body Parsing + Compression ─── app.use(compression()); app.use(express.json({ limit: '1mb' })); +// Stream-based body parsing can resume handlers outside the request's async +// context; re-enter it so everything downstream stays correlated. +app.use((_req, res, next) => runInContext(res.locals.requestContext, next)); // ─── HTTP Logging ─── -// Paths only: morgan's "combined" format would log query strings and referrers. -app.use( - morgan(':method :url :status :res[content-length] - :response-time ms', { - stream: { write: (msg) => logger.info(msg.trim()) }, - skip: () => config.NODE_ENV === 'test', - }), -); +// Paths only: the route template, never the query string or referrer. +app.use((req, res, next) => { + if (config.NODE_ENV !== 'test') { + const ctx = currentContext(); + res.on('finish', () => { + const log = () => + logger.info( + { + method: req.method, + route: req.route ? req.baseUrl + req.route.path : 'unmatched', + status: res.statusCode, + durationMs: Date.now() - (ctx?.startedAt ?? Date.now()), + contentLength: res.getHeader('content-length'), + }, + 'http', + ); + // 'finish' listeners may not run inside the request's context. + if (ctx) runInContext(ctx, log); + else log(); + }); + } + next(); +}); // ─── Request Timeout & Abort Signal ─── app.use(abortMiddleware); @@ -69,7 +103,7 @@ app.get('/readyz', async (_req, res) => { await prisma.$queryRaw`SELECT 1`; res.json({ ok: true, data: { status: 'ready' }, error: null }); } catch { - res.status(503).json({ ok: false, data: null, error: { code: 'NOT_READY', message: 'Database is not reachable' } }); + sendError(res, 503, 'NOT_READY', 'Database is not reachable'); } }); @@ -78,11 +112,7 @@ app.use('/api/v1', router); // ─── 404 Handler ─── app.use((_req, res) => { - res.status(404).json({ - ok: false, - data: null, - error: { code: 'NOT_FOUND', message: 'Route not found' }, - }); + sendError(res, 404, 'NOT_FOUND', 'Route not found'); }); // ─── Global Error Handler ─── diff --git a/src/config/index.ts b/src/config/index.ts index 1864342..5a287f0 100644 --- a/src/config/index.ts +++ b/src/config/index.ts @@ -98,6 +98,7 @@ const envSchema = z RESOLUTION_THRESHOLD: z.string().default('1').transform(Number), RESOLUTION_DISPUTE_WINDOW_SECS: z.string().default('86400').transform(Number), + TRUST_INBOUND_REQUEST_ID: z.enum(['true', 'false']).default('true').transform((v) => v === 'true'), INDEXER_ENABLED: z.enum(['true', 'false']).default('true').transform((v) => v === 'true'), INDEXER_POLL_MS: z.string().default('5000').transform(Number), diff --git a/src/controllers/market.controller.ts b/src/controllers/market.controller.ts index 62cddf3..01cc2ac 100644 --- a/src/controllers/market.controller.ts +++ b/src/controllers/market.controller.ts @@ -1,5 +1,7 @@ import type { Request, Response } from 'express'; import { prisma } from '../lib/prisma'; +import { writeAgentLog } from '../lib/agent-log'; +import { bindDetached } from '../lib/context'; import { v4 as uuidv4 } from 'uuid'; import type { JobRecord } from '../types'; import { JobStatus } from '../types'; @@ -219,25 +221,23 @@ export async function triggerMarketGeneration( // ── Persist audit log ──────────────────────────────────────── - prisma.agentLog - .create({ - data: { + bindDetached( + () => + writeAgentLog({ agentType: 'MARKET_MAKER', level: 'INFO', action: 'GENERATION_JOB_STARTED', - - data: { - jobId, - }, - }, - }) + data: { jobId }, + }), + { jobId }, + ) .catch(() => { // Non-blocking }); // ── Run generation in background ───────────────────────────── - runMarketMakerCycle() + bindDetached(() => runMarketMakerCycle(), { jobId, kind: 'job' }) .then((result: any) => { const existingJob = jobTracker.get(jobId); diff --git a/src/cron/ingestion.cron.ts b/src/cron/ingestion.cron.ts index 6f8ccc2..aef4330 100644 --- a/src/cron/ingestion.cron.ts +++ b/src/cron/ingestion.cron.ts @@ -1,13 +1,14 @@ import cron from 'node-cron'; import { runMarketMakerCycle } from '../agents/market-maker.agent'; import { logger } from '../lib/logger'; +import { newId, runWithContext } from '../lib/context'; -let isRunning = false; +let runningCycleId: string | null = null; /** * Schedules the Market Maker Agent to run every 15 minutes. * - * The `isRunning` guard prevents overlapping executions. + * The `runningCycleId` guard prevents overlapping executions. * If a cycle is still running when the next one is scheduled, * the new trigger is skipped. * @@ -15,20 +16,24 @@ let isRunning = false; * Runs at :00, :15, :30, :45 of every hour */ export function startIngestionCron(): void { - cron.schedule('0,15,30,45 * * * *', async () => { - if (isRunning) { - logger.warn('Ingestion cron: previous cycle still running — skipping'); - return; - } + // The context is entered inside the callback: node-cron fires it from its own timer. + cron.schedule('0,15,30,45 * * * *', () => { + const cycleId = newId('cyc'); + return runWithContext({ requestId: cycleId, cycleId, kind: 'cron', job: 'market_maker' }, async () => { + if (runningCycleId) { + logger.warn({ skippedJob: 'market_maker', runningCycleId }, 'Ingestion cron: previous cycle still running — skipping'); + return; + } - isRunning = true; - try { - await runMarketMakerCycle(); - } catch (err) { - logger.error({ err }, 'Ingestion cron: cycle failed'); - } finally { - isRunning = false; - } + runningCycleId = cycleId; + try { + await runMarketMakerCycle(); + } catch (err) { + logger.error({ err }, 'Ingestion cron: cycle failed'); + } finally { + runningCycleId = null; + } + }); }); logger.info('Ingestion cron started (every 15 minutes)'); diff --git a/src/cron/monitor.cron.ts b/src/cron/monitor.cron.ts index 9a2bb49..ae410db 100644 --- a/src/cron/monitor.cron.ts +++ b/src/cron/monitor.cron.ts @@ -2,9 +2,16 @@ import cron from 'node-cron'; import { runTraderCycle } from '../agents/trader.agent'; import { expireStaleSubscriptions } from '../services/subscription.service'; import { logger } from '../lib/logger'; +import { newId, runWithContext } from '../lib/context'; import { purgeExpiredIdempotencyKeys } from '../middlewares/idempotency.middleware'; -let traderRunning = false; +let runningTraderCycleId: string | null = null; + +/** One correlated context per cron tick. */ +function tick(job: string, fn: () => Promise): Promise { + const cycleId = newId('cyc'); + return runWithContext({ requestId: cycleId, cycleId, kind: 'cron', job }, fn); +} /** * Position Monitor Cron — runs every 15 minutes. @@ -17,45 +24,43 @@ let traderRunning = false; */ export function startMonitorCron(): void { // Trader Agent — every 15 minutes - cron.schedule('5,20,35,50 * * * *', async () => { - if (traderRunning) { - logger.warn('Monitor cron: trader cycle still running — skipping'); + cron.schedule('5,20,35,50 * * * *', () => tick('trader', async () => { + if (runningTraderCycleId) { + logger.warn({ skippedJob: 'trader', runningCycleId: runningTraderCycleId }, 'Monitor cron: trader cycle still running — skipping'); return; } - traderRunning = true; + runningTraderCycleId = newId('cyc'); try { await runTraderCycle(); } catch (err) { logger.error({ err }, 'Monitor cron: trader cycle failed'); } finally { - traderRunning = false; + runningTraderCycleId = null; } - }); + })); // Subscription expiry — every hour - cron.schedule('0 * * * *', async () => { + cron.schedule('0 * * * *', () => tick('subscription_expiry', async () => { try { await expireStaleSubscriptions(); } catch (err) { logger.error({ err }, 'Monitor cron: subscription expiry failed'); } - }); + })); // Expired Idempotency-Key rows (24h TTL, #123) — hourly at :10, in batches of 5 000 - cron.schedule('10 * * * *', async () => { + cron.schedule('10 * * * *', () => tick('idempotency_purge', async () => { try { const deleted = await purgeExpiredIdempotencyKeys(); if (deleted > 0) logger.info({ deleted }, 'Monitor cron: purged expired idempotency keys'); } catch (err) { logger.error({ err }, 'Monitor cron: idempotency key purge failed'); } - }); + })); // Copy trade sweep — every 5 minutes - cron.schedule('*/5 * * * *', async () => { - await sweepCopyTrades(); - }); + cron.schedule('*/5 * * * *', () => tick('copy_trade_sweep', sweepCopyTrades)); logger.info('Monitor cron started'); } diff --git a/src/lib/agent-log.ts b/src/lib/agent-log.ts new file mode 100644 index 0000000..c847401 --- /dev/null +++ b/src/lib/agent-log.ts @@ -0,0 +1,17 @@ +import type { AgentLog, Prisma, PrismaClient } from '@prisma/client'; +import { prisma } from './prisma'; +import { correlation } from './context'; + +export type AgentLogRow = Omit & { + data?: Record; +}; + +/** + * The only place that writes AgentLog rows: stamps the current request/cycle/job + * ids into `data.correlation` so a row can be tied back to its log lines. + */ +export function writeAgentLog(row: AgentLogRow, db: Pick = prisma): Promise { + return db.agentLog.create({ + data: { ...row, data: { ...row.data, correlation: correlation() } as any }, + }); +} diff --git a/src/lib/context.ts b/src/lib/context.ts new file mode 100644 index 0000000..6a311be --- /dev/null +++ b/src/lib/context.ts @@ -0,0 +1,80 @@ +import { AsyncLocalStorage } from 'node:async_hooks'; +import { v7 as uuidv7 } from 'uuid'; +import { config } from '../config'; + +export interface RequestContext { + requestId: string; + kind: 'http' | 'cron' | 'job' | 'indexer'; + cycleId?: string; + jobId?: string; + userId?: string; + job?: string; + startedAt: number; +} + +const als = new AsyncLocalStorage(); + +/** The context of the current request, cron tick, poll or job, if any. */ +export function currentContext(): RequestContext | undefined { + return als.getStore(); +} + +/** Re-enter an existing context object (e.g. after a library dropped it). */ +export function runInContext(ctx: RequestContext, fn: () => T): T { + return als.run(ctx, fn); +} + +/** + * Run `fn` with a fresh context. The store object is shared by everything + * running inside, so auth middleware may set `userId` on it; each call gets + * its own object, so nothing leaks across requests. + */ +export function runWithContext(ctx: Omit, fn: () => T): T { + return als.run({ ...ctx, startedAt: Date.now() }, fn); +} + +/** Time-sortable id (uuid v7), e.g. `req_018f...`. */ +export function newId(prefix: 'req' | 'cyc' | 'job'): string { + return `${prefix}_${uuidv7()}`; +} + +/** The ids to stamp onto log lines and AgentLog rows. */ +export function correlation(): { requestId?: string; cycleId?: string; jobId?: string } { + const ctx = als.getStore(); + if (!ctx) return {}; + return { + requestId: ctx.requestId, + ...(ctx.cycleId ? { cycleId: ctx.cycleId } : {}), + ...(ctx.jobId ? { jobId: ctx.jobId } : {}), + }; +} + +/** + * Run a fire-and-forget task in a child context: it keeps the originating + * requestId (if any) and adds `extra`, so it stays correlated after the + * response has finished. + */ +export function bindDetached(fn: () => Promise, extra: Partial> = {}): Promise { + const parent = als.getStore(); + return runWithContext( + { + kind: 'job', + requestId: parent?.requestId ?? newId('req'), + cycleId: parent?.cycleId, + userId: parent?.userId, + ...extra, + }, + fn, + ); +} + +const INBOUND_ID = /^[A-Za-z0-9._:-]{8,64}$/; + +/** + * Accept an inbound X-Request-Id only if it is well formed. The caller must + * never log a rejected value (log injection); log only its length. + */ +export function acceptInboundId(value: string | undefined): string | undefined { + if (!config.TRUST_INBOUND_REQUEST_ID || !value) return undefined; + return INBOUND_ID.test(value) ? value : undefined; +} diff --git a/src/lib/logger.ts b/src/lib/logger.ts index 77ad02a..d683c7a 100644 --- a/src/lib/logger.ts +++ b/src/lib/logger.ts @@ -1,5 +1,6 @@ import pino from 'pino'; import { config } from '../config'; +import { correlation, currentContext } from './context'; export const logger = pino({ level: config.NODE_ENV === 'production' ? 'info' : 'debug', @@ -11,9 +12,16 @@ export const logger = pino({ colorize: true, translateTime: 'SYS:standard', ignore: 'pid,hostname', + messageFormat: '{if requestId}[{requestId}] {end}{msg}', }, } : undefined, + // Every line emitted inside a request/cron/job context carries its ids. + mixin() { + const ctx = currentContext(); + if (!ctx) return {}; + return { ...correlation(), kind: ctx.kind, ...(ctx.userId ? { userId: ctx.userId } : {}) }; + }, // Never log credentials, even if a caller passes a whole request or config. redact: { paths: [ diff --git a/src/middlewares/auth.middleware.ts b/src/middlewares/auth.middleware.ts index 1750dbc..7a76088 100644 --- a/src/middlewares/auth.middleware.ts +++ b/src/middlewares/auth.middleware.ts @@ -3,6 +3,7 @@ import jwt from 'jsonwebtoken'; import { config } from '../config'; import { previousSecret, secret, secretEntry } from '../config/secrets'; import { sendError } from '../utils/response.util'; +import { currentContext } from '../lib/context'; export interface JwtPayload { userId: string; @@ -41,6 +42,12 @@ function decode(req: Request): { user?: JwtPayload; error?: 'missing' | 'expired } } +/** Tag the request's log context with the caller (mutates this request's own store object). */ +function setContextUser(userId: string): void { + const ctx = currentContext(); + if (ctx) ctx.userId = userId; +} + /** * Require a valid JWT Bearer token. Attaches decoded payload to req.user. */ @@ -48,6 +55,7 @@ export function requireAuth(req: Request, res: Response, next: NextFunction): vo const { user, error } = decode(req); if (user) { req.user = user; + setContextUser(user.userId); next(); return; } @@ -59,7 +67,10 @@ export function requireAuth(req: Request, res: Response, next: NextFunction): vo /** Attaches req.user when a valid token is present; never rejects. */ export function optionalAuth(req: Request, _res: Response, next: NextFunction): void { const { user } = decode(req); - if (user) req.user = user; + if (user) { + req.user = user; + setContextUser(user.userId); + } next(); } diff --git a/src/middlewares/error.middleware.ts b/src/middlewares/error.middleware.ts index c7251e2..8f45fcb 100644 --- a/src/middlewares/error.middleware.ts +++ b/src/middlewares/error.middleware.ts @@ -1,6 +1,7 @@ import type { Request, Response, NextFunction } from 'express'; import { Prisma } from '@prisma/client'; import { logger } from '../lib/logger'; +import { currentContext } from '../lib/context'; import { sendError } from '../utils/response.util'; export class AppError extends Error { @@ -39,7 +40,7 @@ export function errorMiddleware( // AppError instances if (err instanceof AppError) { logger.warn( - { code: err.code, path: req.path, details: err.details }, + { code: err.code, path: req.path, details: err.details, requestId: currentContext()?.requestId }, err.message, ); sendError(res, err.statusCode, err.code, err.message, err.details); @@ -74,6 +75,7 @@ export function errorMiddleware( } // Unexpected errors — log the full stack - logger.error({ err, path: req.path, method: req.method }, 'Unhandled error'); - sendError(res, 500, 'INTERNAL_ERROR', 'An unexpected error occurred'); + const requestId = currentContext()?.requestId ?? res.getHeader('X-Request-Id'); + logger.error({ err, path: req.path, method: req.method, requestId }, 'Unhandled error'); + sendError(res, 500, 'INTERNAL_ERROR', `An unexpected error occurred${requestId ? ` (reference: ${requestId})` : ''}`); } \ No newline at end of file diff --git a/src/middlewares/rate-limit.middleware.ts b/src/middlewares/rate-limit.middleware.ts index 89e66dd..ea5716e 100644 --- a/src/middlewares/rate-limit.middleware.ts +++ b/src/middlewares/rate-limit.middleware.ts @@ -1,7 +1,11 @@ import rateLimit from 'express-rate-limit'; import { config } from '../config'; -const body = { ok: false, data: null, error: { code: 'RATE_LIMITED', message: 'Too many requests. Try again shortly.' } }; +const body = (requestId?: string) => ({ + ok: false, + data: null, + error: { code: 'RATE_LIMITED', message: 'Too many requests. Try again shortly.', requestId }, +}); const limiter = (windowMs: number, limit: number) => rateLimit({ @@ -9,7 +13,10 @@ const limiter = (windowMs: number, limit: number) => limit, standardHeaders: true, legacyHeaders: false, - message: body, + message: (_req: unknown, res: { getHeader(name: string): unknown }) => { + const id = res.getHeader('X-Request-Id'); + return body(typeof id === 'string' ? id : undefined); + }, skip: () => config.NODE_ENV === 'test', }); diff --git a/src/services/hedging.service.ts b/src/services/hedging.service.ts index 94c5f5d..2787aff 100644 --- a/src/services/hedging.service.ts +++ b/src/services/hedging.service.ts @@ -1,5 +1,6 @@ import { prisma } from '../lib/prisma'; import { logger } from '../lib/logger'; +import { writeAgentLog } from '../lib/agent-log'; import { calculateKellySize } from './trade.service'; import { AppError } from '../middlewares/error.middleware'; import type { HedgeRecommendation } from '../types'; @@ -110,19 +111,17 @@ export async function autoHedge(recommendation: HedgeRecommendation): Promise void { } let stopped = false; let timer: NodeJS.Timeout | null = null; - const tick = async () => { - try { - await backfillEvents(undefined, { source: rpcServer, emit }); - } catch (err) { - logger.warn({ err: err instanceof Error ? err.message : err }, 'Indexer poll failed; will retry'); - } - if (!stopped) timer = setTimeout(tick, config.INDEXER_POLL_MS); + const tick = (): Promise => { + const cycleId = newId('cyc'); + return runWithContext({ requestId: cycleId, cycleId, kind: 'indexer' }, async () => { + try { + await backfillEvents(undefined, { source: rpcServer, emit }); + } catch (err) { + logger.warn({ err: err instanceof Error ? err.message : err }, 'Indexer poll failed; will retry'); + } + if (!stopped) timer = setTimeout(tick, config.INDEXER_POLL_MS); + }); }; void tick(); logger.info({ pollMs: config.INDEXER_POLL_MS }, 'Stellar event indexer started'); diff --git a/src/services/market.service.ts b/src/services/market.service.ts index d4338b6..0e8e176 100644 --- a/src/services/market.service.ts +++ b/src/services/market.service.ts @@ -2,6 +2,7 @@ import { z } from "zod"; import { prisma } from "../lib/prisma"; +import { writeAgentLog } from "../lib/agent-log"; import { callLLMJSON } from '../lib/llm'; import { logger } from "../lib/logger"; import type { @@ -374,9 +375,7 @@ async function logAgentAction( data?: Record, ): Promise { try { - await prisma.agentLog.create({ - data: { agentType, level, action, marketId: marketId ?? undefined, data: data as any }, - }); + await writeAgentLog({ agentType, level, action, marketId: marketId ?? undefined, data }); } catch (err) { logger.error({ err }, "Failed to write agent log"); } diff --git a/src/services/trade.service.ts b/src/services/trade.service.ts index 09c67a9..d2705a7 100644 --- a/src/services/trade.service.ts +++ b/src/services/trade.service.ts @@ -1,4 +1,5 @@ import { prisma } from '../lib/prisma'; +import { writeAgentLog } from '../lib/agent-log'; import { logger } from '../lib/logger'; import { AppError } from '../middlewares/error.middleware'; import type { KellyResult, TradePayload, KellyInput } from '../types'; @@ -328,8 +329,6 @@ async function logAgentAction( data?: Record, ): Promise { try { - await prisma.agentLog.create({ - data: { agentType, level, action, marketId: marketId ?? undefined, data: data as any }, - }); + await writeAgentLog({ agentType, level, action, marketId: marketId ?? undefined, data }); } catch {} } diff --git a/src/types/index.ts b/src/types/index.ts index 9a037a2..3402a93 100644 --- a/src/types/index.ts +++ b/src/types/index.ts @@ -177,6 +177,7 @@ export interface ApiError { code: string; message: string; details?: Record; + requestId?: string; } export interface PaginationMeta { diff --git a/src/utils/response.util.ts b/src/utils/response.util.ts index 5f01f2a..4b3bdcc 100644 --- a/src/utils/response.util.ts +++ b/src/utils/response.util.ts @@ -1,4 +1,5 @@ import type { Response } from 'express'; +import { currentContext } from '../lib/context'; import type { ApiResponse, ApiError, PaginationMeta } from '../types'; /** @@ -24,7 +25,10 @@ export function sendError( message: string, details?: Record, ): void { - const error: ApiError = { code, message, details }; + // The header is set first in the chain, so it survives lost async context. + const header = res.getHeader('X-Request-Id'); + const requestId = typeof header === 'string' ? header : currentContext()?.requestId; + const error: ApiError = { code, message, details, requestId }; const body: ApiResponse = { ok: false, data: null, error }; res.status(statusCode).json(body); } diff --git a/tests/context.test.ts b/tests/context.test.ts new file mode 100644 index 0000000..f21f9b1 --- /dev/null +++ b/tests/context.test.ts @@ -0,0 +1,150 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { Writable } from 'node:stream'; +import { readdirSync, readFileSync, statSync } from 'node:fs'; +import { join } from 'node:path'; +import type { AddressInfo } from 'node:net'; +import pino from 'pino'; +import { setupTestEnv } from './helpers/env'; + +setupTestEnv(); + +async function withServer(fn: (base: string) => Promise) { + const { default: app } = await import('../src/app'); + const server = app.listen(0); + try { + await fn(`http://127.0.0.1:${(server.address() as AddressInfo).port}`); + } finally { + server.close(); + } +} + +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); + +test('generates a request id and echoes it in X-Request-Id', async () => { + await withServer(async (base) => { + const res = await fetch(`${base}/health`); + assert.match(res.headers.get('x-request-id') ?? '', /^req_[0-9a-f-]{36}$/); + }); +}); + +test('accepts a well-formed inbound X-Request-Id', async () => { + await withServer(async (base) => { + const res = await fetch(`${base}/health`, { headers: { 'x-request-id': 'client-abc_123:4' } }); + assert.equal(res.headers.get('x-request-id'), 'client-abc_123:4'); + }); +}); + +test('rejects malformed inbound ids', async () => { + const { acceptInboundId } = await import('../src/lib/context'); + assert.equal(acceptInboundId('short'), undefined); + assert.equal(acceptInboundId('a'.repeat(65)), undefined); + assert.equal(acceptInboundId('valid-id-1\nforged line'), undefined); + assert.equal(acceptInboundId(undefined), undefined); + assert.equal(acceptInboundId('valid-id-1'), 'valid-id-1'); + await withServer(async (base) => { + const res = await fetch(`${base}/health`, { headers: { 'x-request-id': 'x'.repeat(65) } }); + assert.match(res.headers.get('x-request-id') ?? '', /^req_/); + }); +}); + +test('error envelope includes requestId for 404s and AppErrors', async () => { + await withServer(async (base) => { + const notFound = await fetch(`${base}/nope`, { headers: { 'x-request-id': 'client-id-0001' } }); + assert.equal(notFound.status, 404); + assert.equal((await notFound.json()).error.requestId, 'client-id-0001'); + const unauth = await fetch(`${base}/api/v1/admin/trace-audit`); + assert.equal(unauth.status, 401); + assert.equal((await unauth.json()).error.requestId, unauth.headers.get('x-request-id')); + }); +}); + +test('body-parser errors still carry a request id', async () => { + await withServer(async (base) => { + const res = await fetch(`${base}/api/v1/markets/generate`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: '{bad' }); + assert.ok(res.headers.get('x-request-id')); + }); +}); + +test('X-Request-Id is listed in Access-Control-Expose-Headers for allowed origins', async () => { + await withServer(async (base) => { + const res = await fetch(`${base}/health`, { headers: { origin: 'http://localhost:3000' } }); + assert.match(res.headers.get('access-control-expose-headers') ?? '', /X-Request-Id/i); + }); +}); + +test('log lines inside a context include the correlation ids', async () => { + const { runWithContext, correlation, currentContext } = await import('../src/lib/context'); + const lines: string[] = []; + const dest = new Writable({ write(chunk, _enc, cb) { lines.push(String(chunk)); cb(); } }); + const log = pino({ + mixin() { + const ctx = currentContext(); + return ctx ? { ...correlation(), kind: ctx.kind } : {}; + }, + }, dest); + runWithContext({ requestId: 'req_test', cycleId: 'cyc_test', kind: 'cron' }, () => log.info('inside')); + log.info('outside'); + const [inside, outside] = lines.map((l) => JSON.parse(l)); + assert.equal(inside.requestId, 'req_test'); + assert.equal(inside.cycleId, 'cyc_test'); + assert.equal(inside.kind, 'cron'); + assert.equal(outside.requestId, undefined); +}); + +test('context survives awaited calls, timers and Promise.all', async () => { + const { runWithContext, currentContext } = await import('../src/lib/context'); + await runWithContext({ requestId: 'req_async', kind: 'http' }, async () => { + await sleep(5); + assert.equal(currentContext()?.requestId, 'req_async'); + const ids = await Promise.all([1, 2, 3].map(async (n) => { await sleep(n); return currentContext()?.requestId; })); + assert.deepEqual(ids, ['req_async', 'req_async', 'req_async']); + const fromTimer = await new Promise((resolve) => setTimeout(() => resolve(currentContext()?.requestId), 1)); + assert.equal(fromTimer, 'req_async'); + }); +}); + +test('bindDetached keeps requestId and adds jobId after the request context has ended', async () => { + const { runWithContext, bindDetached, currentContext } = await import('../src/lib/context'); + let seen: { requestId?: string; jobId?: string; kind?: string } = {}; + let task: Promise = Promise.resolve(); + runWithContext({ requestId: 'req_origin', kind: 'http' }, () => { + task = bindDetached(async () => { + await sleep(10); + const c = currentContext(); + seen = { requestId: c?.requestId, jobId: c?.jobId, kind: c?.kind }; + }, { jobId: 'job_1' }); + }); + assert.equal(currentContext(), undefined); + await task; + assert.deepEqual(seen, { requestId: 'req_origin', jobId: 'job_1', kind: 'job' }); +}); + +test('each context gets its own store, so userId does not leak between requests', async () => { + const { runWithContext, currentContext } = await import('../src/lib/context'); + runWithContext({ requestId: 'req_a', kind: 'http' }, () => { currentContext()!.userId = 'u1'; }); + runWithContext({ requestId: 'req_b', kind: 'http' }, () => assert.equal(currentContext()!.userId, undefined)); +}); + +test('AgentLog writes include data.correlation', async () => { + const { writeAgentLog } = await import('../src/lib/agent-log'); + const { runWithContext } = await import('../src/lib/context'); + const created: any[] = []; + const db = { agentLog: { create: async (args: any) => { created.push(args); return args.data; } } }; + await runWithContext({ requestId: 'req_log', cycleId: 'cyc_log', kind: 'cron' }, () => + writeAgentLog({ agentType: 'TRADER', level: 'INFO', action: 'TEST', data: { a: 1 } }, db as any)); + assert.deepEqual(created[0].data.data, { a: 1, correlation: { requestId: 'req_log', cycleId: 'cyc_log' } }); + await writeAgentLog({ agentType: 'TRADER', level: 'INFO', action: 'TEST' }, db as any); + assert.deepEqual(created[1].data.data, { correlation: {} }); +}); + +test('no direct prisma.agentLog.create calls outside the helper', () => { + const walk = (dir: string): string[] => + readdirSync(dir).flatMap((f) => { + const p = join(dir, f); + return statSync(p).isDirectory() ? walk(p) : p.endsWith('.ts') ? [p] : []; + }); + const root = join(__dirname, '..', '..', 'src'); + const offenders = walk(root).filter((f) => !f.endsWith('agent-log.ts') && /prisma\.agentLog\s*\.create\(/.test(readFileSync(f, 'utf8'))); + assert.deepEqual(offenders, []); +});