diff --git a/apps/api/.env.example b/apps/api/.env.example index a923cd8..76a26bd 100644 --- a/apps/api/.env.example +++ b/apps/api/.env.example @@ -5,8 +5,9 @@ SUPABASE_AUTH_AUDIENCE=authenticated # Optional connection-pool bounds. The API process opens one pool and shares it # across every repository, so these are the process-wide limits, not per-domain -# ones. Raising DATABASE_POOL_MAX raises the server connections this instance can -# hold; keep the total across instances inside the database's own limit. +# ones. The worker has its own separately bounded process pool. Raising either +# maximum raises the server connections this instance can hold; keep the total +# across instances inside the database's own limit. # DATABASE_POOL_MAX=10 # DATABASE_POOL_IDLE_TIMEOUT_MS=30000 # DATABASE_POOL_CONNECTION_TIMEOUT_MS=10000 diff --git a/apps/api/package.json b/apps/api/package.json index bc15148..d4cd82d 100644 --- a/apps/api/package.json +++ b/apps/api/package.json @@ -17,6 +17,7 @@ "db:migrate:prod": "node dist/db/migrate.js", "db:seed": "tsx src/db/seed.ts", "migration:cutover": "tsx src/migration/run.ts", + "load:representative": "tsx src/load/representative.ts", "test": "vitest run", "test:integration": "vitest run --config vitest.integration.config.ts", "typecheck": "tsc --noEmit", diff --git a/apps/api/src/app.ts b/apps/api/src/app.ts index 55f6ac8..b3b5425 100644 --- a/apps/api/src/app.ts +++ b/apps/api/src/app.ts @@ -27,6 +27,7 @@ import type { FilesystemMediaStorage } from './domains/media/storage.js'; import type { AdminRequestAuth } from './domains/admin/auth.js'; import type { AdminRepository } from './domains/admin/repository.js'; import { registerAdminRoutes } from './domains/admin/routes.js'; +import { recordRequest } from './telemetry.js'; type Variables = { identity: AuthIdentity }; @@ -57,6 +58,18 @@ function error(code: string, message: string): ApiErrorResponse { export function createApp(deps: AppDependencies): Hono<{ Variables: Variables }> { const app = new Hono<{ Variables: Variables }>(); + app.use('*', async (c, next) => { + const startedAt = performance.now(); + try { + await next(); + } finally { + // Hono returns the matched template after dispatch, never the raw path. + const candidate = c.req.routePath; + const route = candidate && !/[0-9a-f]{8}-(?:[0-9a-f]{4}-){3}[0-9a-f]{12}/i.test(candidate) && !candidate.includes('?') ? candidate : 'unmatched'; + recordRequest(c.req.method, route, c.res.status, startedAt); + } + }); + app.get('/health/live', (c) => c.json({ status: 'ok' })); app.get('/health/ready', async (c) => { try { diff --git a/apps/api/src/db/pool.ts b/apps/api/src/db/pool.ts index 0db68bd..ecb22b7 100644 --- a/apps/api/src/db/pool.ts +++ b/apps/api/src/db/pool.ts @@ -1,4 +1,5 @@ import { Pool } from 'pg'; +import { instrumentPool, type TelemetryProcess } from '../telemetry.js'; /** * A repository is constructed either from a connection string, in which case it @@ -23,13 +24,14 @@ export interface DatabasePoolOptions { */ export const DEFAULT_POOL_MAX = 10; -export function createDatabasePool(databaseUrl: string, options: DatabasePoolOptions = {}): Pool { - return new Pool({ +export function createDatabasePool(databaseUrl: string, options: DatabasePoolOptions = {}, telemetryProcess?: TelemetryProcess): Pool { + const pool = new Pool({ connectionString: databaseUrl, max: options.max ?? DEFAULT_POOL_MAX, ...(options.idleTimeoutMillis === undefined ? {} : { idleTimeoutMillis: options.idleTimeoutMillis }), ...(options.connectionTimeoutMillis === undefined ? {} : { connectionTimeoutMillis: options.connectionTimeoutMillis }), }); + return telemetryProcess ? instrumentPool(telemetryProcess, pool) : pool; } export interface ResolvedPool { diff --git a/apps/api/src/domains/app-data/routes.ts b/apps/api/src/domains/app-data/routes.ts index 4eaf7eb..c48346f 100644 --- a/apps/api/src/domains/app-data/routes.ts +++ b/apps/api/src/domains/app-data/routes.ts @@ -3,6 +3,7 @@ import type { Hono } from 'hono'; import { z } from 'zod'; import type { AuthIdentity } from '../../auth.js'; import { AppDataError, type AppDataRepository } from './repository.js'; +import { recordSync } from '../../telemetry.js'; type App = Hono<{ Variables: { identity: AuthIdentity } }>; const uuid = z.string().uuid(); @@ -46,7 +47,9 @@ export function registerAppDataRoutes(app: App, repository: AppDataRepository): }); // The client polls this instead of re-fetching every domain; see SyncSummary. app.get('/v1/me/sync', async c => { + const startedAt = performance.now(); const result = await run(() => repository.syncSummary(c.get('identity').id)); + recordSync('api',result.ok?'ok':'error',startedAt); return result.ok ? c.json(result.value) : c.json(error(result.cause.code, result.cause.message), result.cause.status); }); app.get('/v1/me/nudge-status', async c => c.json(await repository.nudgeStatus(c.get('identity').id))); diff --git a/apps/api/src/domains/operations/repository.ts b/apps/api/src/domains/operations/repository.ts index 8792c5b..3635815 100644 --- a/apps/api/src/domains/operations/repository.ts +++ b/apps/api/src/domains/operations/repository.ts @@ -15,9 +15,19 @@ export interface OperationsRepository { message(id: string): Promise; classifierConfig(): Promise; classify(id: string, status: 'safe' | 'flagged', reason: string | null, category: string): Promise; + outboxState(): Promise<{ due: number; oldestDueAgeSeconds: number }>; close(): Promise; } +/** Matches `claim`: only currently due, retryable, and unlocked/expired leases. */ +export const OUTBOX_STATE_SQL = `select count(*)::int due, + coalesce(floor(extract(epoch from now()-min(available_at))),0)::int oldest_due_age_seconds + from operations_outbox +where sent_at is null + and attempts<5 + and available_at<=now() + and (locked_at is null or locked_at { + const client = await this.pool.connect(); + let released = false; + try { + await client.query('begin'); + // A server-side timeout cancels the query and transaction; unlike a + // Promise race it cannot leave an unobserved query or borrowed client. + await client.query("set local statement_timeout = '250ms'"); + const result = await client.query<{ due: number; oldest_due_age_seconds: number }>(OUTBOX_STATE_SQL); + await client.query('commit'); + const row = result.rows[0]; + return { due: row?.due ?? 0, oldestDueAgeSeconds: row?.oldest_due_age_seconds ?? 0 }; + } catch (cause) { + try { await client.query('rollback'); } + catch (rollbackCause) { + // A failed rollback leaves transaction state uncertain. Evict this + // client instead of returning it to the shared pool, but retain the + // original observation failure for callers and logs. + client.release(rollbackCause instanceof Error ? rollbackCause : new Error('outbox observation rollback failed')); + released = true; + } + throw cause; + } finally { if (!released) client.release(); } + } async complete(id:string) { await this.pool.query('update operations_outbox set sent_at=now(),locked_at=null,last_error=null where id=$1',[id]); } async fail(id:string,message:string) { await this.pool.query( `update operations_outbox set locked_at=null,last_error=$2,available_at=now()+make_interval(secs=>least(3600,power(2,attempts)::int*30)) where id=$1`,[id,message.slice(0,500)]); } diff --git a/apps/api/src/domains/operations/runner.ts b/apps/api/src/domains/operations/runner.ts index 8c8fc2f..1943308 100644 --- a/apps/api/src/domains/operations/runner.ts +++ b/apps/api/src/domains/operations/runner.ts @@ -1,6 +1,7 @@ import type { MediaJanitor } from '../media/janitor.js'; import type { OperationsRepository } from './repository.js'; import type { DiscordProvider, ExpoPushProvider, MessageClassifier } from './providers.js'; +import { emitTelemetry } from '../../telemetry.js'; export class OperationsRunner { constructor(private readonly repository:OperationsRepository,private readonly expo:ExpoPushProvider,private readonly discord:DiscordProvider,private readonly classifier:MessageClassifier,private readonly mediaJanitor?:MediaJanitor) {} @@ -13,7 +14,11 @@ export class OperationsRunner { await this.repository.complete(item.id);completed++; } catch(cause) { await this.repository.fail(item.id,cause instanceof Error?cause.message:'unknown error');failed++; } } + // Observation never blocks useful work. It runs after claiming/completing and + // has a short caller-side deadline; the claim lock predicate remains the + // source of truth when several worker instances race. + const after=await this.repository.outboxState().catch(()=>null); + if(after)emitTelemetry({event:'outbox',process:'worker',due_after:after.due,oldest_due_age_seconds:after.oldestDueAgeSeconds,claimed:items.length,completed,failed,count:1}); return {produced,claimed:items.length,completed,failed}; } } - diff --git a/apps/api/src/index.ts b/apps/api/src/index.ts index 0121673..5bd56d8 100644 --- a/apps/api/src/index.ts +++ b/apps/api/src/index.ts @@ -3,6 +3,7 @@ import { createApp } from './app.js'; import { createAuthVerifier } from './auth.js'; import { loadConfig } from './config.js'; import { createDatabasePool } from './db/pool.js'; +import { recordPool } from './telemetry.js'; import { PostgresRepository } from './db/repository.js'; import { PostgresAppDataRepository } from './domains/app-data/repository.js'; import { AdminRequestAuth } from './domains/admin/auth.js'; @@ -30,7 +31,7 @@ const authVerifier = createAuthVerifier(config); // One deliberately bounded pool for the whole process. Repositories used to open // a default-sized pool each, so a single API instance could hold twelve times the // connections a self-hosted PostgreSQL was sized for. -const pool = createDatabasePool(config.databaseUrl, config.databasePool); +const pool = createDatabasePool(config.databaseUrl, config.databasePool, 'api'); const repository = new PostgresRepository(pool); const coupleRepository = new PostgresCoupleRepository(pool); const packsRepository = new PostgresPacksRepository(pool); @@ -81,6 +82,10 @@ const app = createApp({ ), }); +// Sample pool state on the telemetry cadence rather than once per request. +const poolTelemetryTimer = setInterval(() => recordPool('api', pool), 60_000); +poolTelemetryTimer.unref(); + const server = serve({ fetch: app.fetch, port: config.port }, (info) => { console.log(`Sauci API listening on http://127.0.0.1:${info.port}`); }); @@ -90,6 +95,7 @@ let shuttingDown = false; async function shutdown(signal: string): Promise { if (shuttingDown) return; shuttingDown = true; + clearInterval(poolTelemetryTimer); console.log(`Received ${signal}; shutting down`); const forcedExit = setTimeout(() => { console.error('Graceful shutdown timed out'); diff --git a/apps/api/src/load/loopback.ts b/apps/api/src/load/loopback.ts new file mode 100644 index 0000000..cc6ad8c --- /dev/null +++ b/apps/api/src/load/loopback.ts @@ -0,0 +1,16 @@ +/** Refuse fixture targets outside the local development boundary. */ +export function requireLoopback(url: string, name: string): URL { + const parsed = new URL(url); + if (!['127.0.0.1', 'localhost', '::1'].includes(parsed.hostname)) { + throw new Error(`${name} must resolve to loopback; refusing non-local load`); + } + return parsed; +} + +export function requireLocalApi(url: string, expectedPort = 3003): URL { + const parsed = requireLoopback(url, 'SAUCI_LOAD_API_URL'); + if (parsed.protocol !== 'http:' || Number(parsed.port || 80) !== expectedPort) { + throw new Error(`SAUCI_LOAD_API_URL must be http loopback on port ${expectedPort}`); + } + return parsed; +} diff --git a/apps/api/src/load/representative.ts b/apps/api/src/load/representative.ts new file mode 100644 index 0000000..f33e3aa --- /dev/null +++ b/apps/api/src/load/representative.ts @@ -0,0 +1,43 @@ +import { createDatabasePool } from '../db/pool.js'; +import { PostgresOperationsRepository } from '../domains/operations/repository.js'; +import { emitTelemetry, recordPool } from '../telemetry.js'; +import { requireLocalApi, requireLoopback } from './loopback.js'; + +function percentile(samples: number[], fraction: number): number { + const sorted = [...samples].sort((a, b) => a - b); + return Math.round((sorted[Math.min(sorted.length - 1, Math.ceil(sorted.length * fraction) - 1)] ?? 0) * 100) / 100; +} +async function read(url: URL, headers: HeadersInit): Promise { + const startedAt = performance.now(); + const response = await fetch(url, { headers }); + if (!response.ok) throw new Error(`representative API read failed with ${response.status}`); + return performance.now() - startedAt; +} + +async function main(): Promise { + const expectedPort = Number(process.env.SAUCI_LOAD_API_PORT ?? '3003'); + const apiUrl = requireLocalApi(process.env.SAUCI_LOAD_API_URL ?? `http://127.0.0.1:${expectedPort}`, expectedPort); + const databaseUrl = process.env.DATABASE_URL; + const authorization = process.env.SAUCI_LOAD_AUTHORIZATION; + const matchId = process.env.SAUCI_LOAD_MATCH_ID; + const expectedUserId = process.env.SAUCI_LOAD_USER_ID; + if (!databaseUrl || !authorization || !matchId || !expectedUserId) throw new Error('DATABASE_URL, SAUCI_LOAD_AUTHORIZATION, SAUCI_LOAD_MATCH_ID, and SAUCI_LOAD_USER_ID are required'); + requireLoopback(databaseUrl, 'DATABASE_URL'); + if (!/^[0-9a-f]{8}-(?:[0-9a-f]{4}-){3}[0-9a-f]{12}$/i.test(matchId) || !/^[0-9a-f]{8}-(?:[0-9a-f]{4}-){3}[0-9a-f]{12}$/i.test(expectedUserId)) throw new Error('SAUCI_LOAD_MATCH_ID and SAUCI_LOAD_USER_ID must be UUIDs'); + + const health = await fetch(new URL('/health/live', apiUrl)); + if (!health.ok || (await health.json() as { status?: string }).status !== 'ok') throw new Error('local Sauci API health preflight failed'); + const headers = { authorization }; + const me = await fetch(new URL('/v1/me', apiUrl), { headers }); + const profile = me.ok ? await me.json() as { profile?: { id?: string } } : undefined; + if (profile?.profile?.id !== expectedUserId) throw new Error('local fixture identity preflight failed'); + + const sync = await Promise.all(Array.from({ length: 5 }, () => read(new URL('/v1/me/sync', apiUrl), headers))); + const chat = await Promise.all(Array.from({ length: 5 }, () => read(new URL(`/v1/matches/${matchId}/messages?typing=true`, apiUrl), headers))); + for (const [operation, samples] of [['sync', sync], ['chat', chat]] as const) emitTelemetry({ event: 'load', process: 'load-fixture', operation, count: samples.length, p50_ms: percentile(samples, .5), p95_ms: percentile(samples, .95), p99_ms: percentile(samples, .99) }); + + const pool = createDatabasePool(databaseUrl, { max: 1 }, 'load-fixture'); + try { await new PostgresOperationsRepository(pool).outboxState(); recordPool('load-fixture', pool); } + finally { await pool.end(); } +} +void main(); diff --git a/apps/api/src/telemetry.ts b/apps/api/src/telemetry.ts new file mode 100644 index 0000000..a0932ff --- /dev/null +++ b/apps/api/src/telemetry.ts @@ -0,0 +1,64 @@ +import type { Pool, PoolClient } from 'pg'; + +/** Fixed, privacy-safe JSON log vocabulary. Do not add free-form fields. */ +export type TelemetryProcess = 'api' | 'worker' | 'load-fixture'; +type Outcome = 'ok' | 'error'; +type StatusClass = '2xx' | '3xx' | '4xx' | '5xx'; + +export type TelemetryRecord = + | { event: 'request'; process: 'api'; method: string; route: string; status_class: StatusClass; count: number; duration_ms: number } + | { event: 'query'; process: TelemetryProcess; operation: 'pool_query'; outcome: Outcome; count: number; duration_ms: number } + | { event: 'pool'; process: TelemetryProcess; total: number; idle: number; waiting: number } + | { event: 'sync'; process: 'api' | 'worker'; outcome: Outcome; count: number; duration_ms: number } + | { event: 'outbox'; process: 'worker'; due_after: number; oldest_due_age_seconds: number; claimed: number; completed: number; failed: number; count: number } + | { event: 'load'; process: 'load-fixture'; operation: 'sync' | 'chat'; count: number; p50_ms: number; p95_ms: number; p99_ms: number }; + +type TelemetrySink = (record: TelemetryRecord) => void; +let sink: TelemetrySink = (record) => console.log(JSON.stringify(record)); +let pending = new Map(); + +function aggregate(record: Extract): void { + const key = record.event === 'request' ? `${record.event}:${record.method}:${record.route}:${record.status_class}` : `${record.event}:${record.process}:${record.outcome}`; + const prior = pending.get(key) as typeof record | undefined; + if (!prior) { pending.set(key, { ...record }); return; } + prior.count += record.count; + prior.duration_ms += record.duration_ms; +} + +/** Emits bounded aggregates each minute; workers also flush after a tick. */ +export function flushTelemetry(): void { for (const record of pending.values()) sink(record); pending = new Map(); } +const flushTimer = setInterval(flushTelemetry, 60_000); +flushTimer.unref(); + +export function emitTelemetry(record: Exclude): void { sink(record); } +export function recordRequest(method: string, route: string, status: number, startedAt: number): void { + const statusClass: StatusClass = status >= 500 ? '5xx' : status >= 400 ? '4xx' : status >= 300 ? '3xx' : '2xx'; + aggregate({ event: 'request', process: 'api', method, route, status_class: statusClass, duration_ms: durationMs(startedAt), count: 1 }); +} +export function recordSync(process: 'api' | 'worker', outcome: Outcome, startedAt: number): void { aggregate({ event: 'sync', process, outcome, duration_ms: durationMs(startedAt), count: 1 }); } +/** Test-only hook; production code always uses the JSON console sink. */ +export function setTelemetrySinkForTests(next: TelemetrySink): () => void { const previous = sink; sink = next; return () => { sink = previous; pending = new Map(); }; } +export function durationMs(startedAt: number): number { return Math.max(0, Math.round((performance.now() - startedAt) * 100) / 100); } +export function recordPool(process: TelemetryProcess, pool: Pool): void { emitTelemetry({ event: 'pool', process, total: pool.totalCount, idle: pool.idleCount, waiting: pool.waitingCount }); } + +export function instrumentPool(process: TelemetryProcess, pool: Pool): Pool { + const clients = new WeakSet(); + const instrumentQuery = (target: T): T => { + const query = target.query.bind(target) as (...args: unknown[]) => unknown; + target.query = ((...args: unknown[]) => { + const startedAt = performance.now(); + const finish = (outcome: Outcome) => aggregate({ event: 'query', process, operation: 'pool_query', outcome, duration_ms: durationMs(startedAt), count: 1 }); + try { const result = query(...args); if (result && typeof (result as Promise).then === 'function') return (result as Promise).then((value) => { finish('ok'); return value; }, (error) => { finish('error'); throw error; }); finish('ok'); return result; } + catch (error) { finish('error'); throw error; } + }) as Pool['query']; + return target; + }; + instrumentQuery(pool); + const connect = pool.connect.bind(pool) as (...args: unknown[]) => unknown; + pool.connect = ((...args: unknown[]) => { + const result = connect(...args); + if (result && typeof (result as Promise).then === 'function') return (result as Promise).then((client) => { if (!clients.has(client)) { clients.add(client); instrumentQuery(client); } return client; }); + return result; + }) as Pool['connect']; + return pool; +} diff --git a/apps/api/src/workers/config.ts b/apps/api/src/workers/config.ts index f4edd30..f32eb69 100644 --- a/apps/api/src/workers/config.ts +++ b/apps/api/src/workers/config.ts @@ -4,6 +4,7 @@ const optionalString=z.preprocess(value=>value===''?undefined:value,z.string().m const optionalUrl=z.preprocess(value=>value===''?undefined:value,z.string().url().optional()); const schema=z.object({ NODE_ENV:z.enum(['development','test','production']).default('development'), DATABASE_URL:z.string().url(), + DATABASE_POOL_MAX:z.coerce.number().int().min(1).max(200).default(4), DATABASE_POOL_IDLE_TIMEOUT_MS:z.coerce.number().int().min(0).max(3_600_000).default(30_000), DATABASE_POOL_CONNECTION_TIMEOUT_MS:z.coerce.number().int().min(0).max(60_000).default(10_000), WORKER_POLL_INTERVAL_MS:z.coerce.number().int().min(1000).max(300000).default(30000), WORKER_BATCH_SIZE:z.coerce.number().int().min(1).max(100).default(25), MEDIA_ROOT:z.string().min(1).default('/data/media'), MEDIA_SIGNING_SECRET:z.string().min(32).default('local-media-signing-secret-change-me-123456'), MEDIA_PUBLIC_BASE_URL:z.string().url().default('http://127.0.0.1:3003'), CLASSIFIER_ENABLED:z.enum(['true','false']).default('true').transform(v=>v==='true'), @@ -15,7 +16,7 @@ export function loadWorkerConfig(env:NodeJS.ProcessEnv=process.env) { const value=schema.parse(env); if(value.NODE_ENV==='production'&&value.CLASSIFIER_ENABLED&&!value.OPENROUTER_API_KEY)throw new Error('OPENROUTER_API_KEY is required when classification is enabled'); if(value.ADMIN_PRIVATE_KEY_JWK){try{JSON.parse(value.ADMIN_PRIVATE_KEY_JWK);}catch{throw new Error('ADMIN_PRIVATE_KEY_JWK must be valid JSON');}} - return {nodeEnv:value.NODE_ENV,databaseUrl:value.DATABASE_URL,pollIntervalMs:value.WORKER_POLL_INTERVAL_MS,batchSize:value.WORKER_BATCH_SIZE, + return {nodeEnv:value.NODE_ENV,databaseUrl:value.DATABASE_URL,databasePool:{max:value.DATABASE_POOL_MAX,idleTimeoutMillis:value.DATABASE_POOL_IDLE_TIMEOUT_MS,connectionTimeoutMillis:value.DATABASE_POOL_CONNECTION_TIMEOUT_MS},pollIntervalMs:value.WORKER_POLL_INTERVAL_MS,batchSize:value.WORKER_BATCH_SIZE, mediaRoot:value.MEDIA_ROOT,mediaSigningSecret:value.MEDIA_SIGNING_SECRET,mediaPublicBaseUrl:value.MEDIA_PUBLIC_BASE_URL,classifierEnabled:value.CLASSIFIER_ENABLED, openRouterApiKey:value.OPENROUTER_API_KEY,classifierModel:value.CLASSIFIER_MODEL,classifierPrompt:env.CLASSIFIER_PROMPT||defaultPrompt, adminPrivateKeyJwk:value.ADMIN_PRIVATE_KEY_JWK,discordWebhookUrl:value.DISCORD_WEBHOOK_URL}; diff --git a/apps/api/src/workers/index.ts b/apps/api/src/workers/index.ts index 89a4d04..5973a7e 100644 --- a/apps/api/src/workers/index.ts +++ b/apps/api/src/workers/index.ts @@ -3,14 +3,16 @@ import { PostgresMediaRepository } from '../domains/media/repository.js'; import { FilesystemMediaStorage } from '../domains/media/storage.js'; import { DiscordProvider, ExpoPushProvider, MessageClassifier, OperationsRunner, PostgresOperationsRepository } from '../domains/operations/index.js'; import { loadWorkerConfig } from './config.js'; +import { createDatabasePool } from '../db/pool.js'; +import { flushTelemetry, recordPool, recordSync } from '../telemetry.js'; +import { WorkerLifecycle } from './lifecycle.js'; -const config=loadWorkerConfig(); const operations=new PostgresOperationsRepository(config.databaseUrl); const mediaRepository=new PostgresMediaRepository(config.databaseUrl); +const config=loadWorkerConfig(); const pool=createDatabasePool(config.databaseUrl,config.databasePool,'worker'); const operations=new PostgresOperationsRepository(pool); const mediaRepository=new PostgresMediaRepository(pool); const mediaStorage=new FilesystemMediaStorage(config.mediaRoot,config.mediaSigningSecret,config.mediaPublicBaseUrl); const runner=new OperationsRunner(operations,new ExpoPushProvider(),new DiscordProvider(config.discordWebhookUrl),new MessageClassifier(operations, {enabled:config.classifierEnabled,apiKey:config.openRouterApiKey,model:config.classifierModel,prompt:config.classifierPrompt,adminPrivateKeyJwk:config.adminPrivateKeyJwk,mediaRoot:config.mediaRoot}), new MediaJanitor(mediaRepository,mediaStorage)); -let stopped=false;let timer:NodeJS.Timeout|undefined; -async function tick(){if(stopped)return;try{const result=await runner.runOnce(new Date(),config.batchSize);if(result.claimed||result.failed)console.log('Operations worker tick',result);}catch(cause){console.error('Operations worker tick failed',cause);}finally{if(!stopped){timer=setTimeout(()=>void tick(),config.pollIntervalMs);}}} -async function shutdown(signal:string){if(stopped)return;stopped=true;if(timer)clearTimeout(timer);console.log(`Received ${signal}; stopping operations worker`);await Promise.all([operations.close(),mediaRepository.close()]);} -process.on('SIGINT',()=>void shutdown('SIGINT').then(()=>process.exit(0))); process.on('SIGTERM',()=>void shutdown('SIGTERM').then(()=>process.exit(0))); -console.log('Sauci operations worker started');void tick(); +const lifecycle=new WorkerLifecycle({pollIntervalMs:config.pollIntervalMs,closePool:()=>pool.end(),onTickFailure:cause=>console.error('Operations worker tick failed',cause),runTick:async()=>{const startedAt=performance.now();try{const result=await runner.runOnce(new Date(),config.batchSize);recordSync('worker','ok',startedAt);recordPool('worker',pool);flushTelemetry();if(result.claimed||result.failed)console.log('Operations worker tick',result);}catch(cause){recordSync('worker','error',startedAt);recordPool('worker',pool);flushTelemetry();throw cause;}}}); +function shutdown(signal:string){console.log(`Received ${signal}; stopping operations worker`);return lifecycle.shutdown();} +process.on('SIGINT',()=>void shutdown('SIGINT').then(code=>process.exit(code))); process.on('SIGTERM',()=>void shutdown('SIGTERM').then(code=>process.exit(code))); +console.log('Sauci operations worker started');lifecycle.start(); diff --git a/apps/api/src/workers/lifecycle.ts b/apps/api/src/workers/lifecycle.ts new file mode 100644 index 0000000..837d47f --- /dev/null +++ b/apps/api/src/workers/lifecycle.ts @@ -0,0 +1,47 @@ +export interface WorkerLifecycleDependencies { + pollIntervalMs: number; + runTick: () => Promise; + closePool: () => Promise; + onTickFailure: (cause: unknown) => void; + graceMs?: number; +} + +/** Owns worker tick scheduling and makes shutdown single-flight and bounded. */ +export class WorkerLifecycle { + private stopped = false; + private timer: NodeJS.Timeout | undefined; + private activeTick: Promise | undefined; + private shutdownPromise: Promise | undefined; + private readonly graceMs: number; + + constructor(private readonly deps: WorkerLifecycleDependencies) { this.graceMs = deps.graceMs ?? 25_000; } + + start(): void { if (!this.activeTick && !this.stopped) this.activeTick = this.tick(); } + + shutdown(): Promise { + if (!this.shutdownPromise) this.shutdownPromise = this.stop(); + return this.shutdownPromise; + } + + private async tick(): Promise { + try { await this.deps.runTick(); } + catch (cause) { this.deps.onTickFailure(cause); } + finally { + if (!this.stopped) this.timer = setTimeout(() => { this.activeTick = this.tick(); }, this.deps.pollIntervalMs); + } + } + + private async stop(): Promise { + this.stopped = true; + if (this.timer) clearTimeout(this.timer); + const work = (async () => { + try { if (this.activeTick) await this.activeTick; await this.deps.closePool(); return 0; } + catch { return 1; } + })(); + let timeout: NodeJS.Timeout | undefined; + const timedOut = new Promise((resolve) => { timeout = setTimeout(() => resolve(1), this.graceMs); }); + const result = await Promise.race([work, timedOut]); + if (timeout) clearTimeout(timeout); + return result; + } +} diff --git a/apps/api/test/app.test.ts b/apps/api/test/app.test.ts index a0b393c..f19f1a4 100644 --- a/apps/api/test/app.test.ts +++ b/apps/api/test/app.test.ts @@ -5,6 +5,7 @@ import type { ApiRepository, MobileCompatibleProfile, } from '../src/db/repository.js'; +import { flushTelemetry, setTelemetrySinkForTests, type TelemetryRecord } from '../src/telemetry.js'; const alice: AuthIdentity = { id: '11111111-1111-4111-8111-111111111111', @@ -119,4 +120,15 @@ describe('Sauci API', () => { }); expect(response.status).toBe(400); }); + + it('logs only a route template and status class, never raw UUIDs or auth values', async () => { + const records: TelemetryRecord[] = []; const restore = setTelemetrySinkForTests((record) => records.push(record)); + const app = createApp({ auth: verifier(), repository: new MemoryRepository() }); + await app.request('/v1/me/11111111-1111-4111-8111-111111111111?token=private-token', { headers: { authorization: 'Bearer private-token' } }); + flushTelemetry(); restore(); + const serialized = JSON.stringify(records); + expect(records).toContainEqual(expect.objectContaining({ event: 'request', status_class: '4xx' })); + expect(serialized).not.toContain('11111111-1111-4111-8111-111111111111'); + expect(serialized).not.toContain('private-token'); + }); }); diff --git a/apps/api/test/load.loopback.test.ts b/apps/api/test/load.loopback.test.ts new file mode 100644 index 0000000..0b4b12b --- /dev/null +++ b/apps/api/test/load.loopback.test.ts @@ -0,0 +1,17 @@ +import { describe, expect, it } from 'vitest'; +import { requireLocalApi, requireLoopback } from '../src/load/loopback.js'; + +describe('representative load fixture boundary', () => { + it('accepts only loopback API and PostgreSQL URLs', () => { + expect(requireLoopback('http://127.0.0.1:3003', 'API').hostname).toBe('127.0.0.1'); + expect(requireLoopback('postgresql://sauci:test@localhost:5432/sauci', 'DATABASE').hostname).toBe('localhost'); + expect(() => requireLoopback('https://api.sauci.example', 'API')).toThrow('must resolve to loopback'); + expect(() => requireLoopback('postgresql://sauci:test@db.internal:5432/sauci', 'DATABASE')).toThrow('must resolve to loopback'); + }); + + it('allows only the controlled local HTTP API port', () => { + expect(requireLocalApi('http://127.0.0.1:3003').port).toBe('3003'); + expect(() => requireLocalApi('https://127.0.0.1:3003')).toThrow('must be http loopback'); + expect(() => requireLocalApi('http://127.0.0.1:3004')).toThrow('must be http loopback'); + }); +}); diff --git a/apps/api/test/operations.integration.test.ts b/apps/api/test/operations.integration.test.ts index 820297c..b76e9c8 100644 --- a/apps/api/test/operations.integration.test.ts +++ b/apps/api/test/operations.integration.test.ts @@ -2,7 +2,7 @@ import { randomUUID } from 'node:crypto'; import { readFile } from 'node:fs/promises'; import { Pool } from 'pg'; import { afterAll,beforeAll,describe,expect,it } from 'vitest'; -import { PostgresOperationsRepository } from '../src/domains/operations/repository.js'; +import { OUTBOX_STATE_SQL, PostgresOperationsRepository } from '../src/domains/operations/repository.js'; const databaseUrl=process.env.DATABASE_URL;const local=databaseUrl?['127.0.0.1','localhost','::1'].includes(new URL(databaseUrl).hostname):false; if(databaseUrl&&!local)throw new Error('Operations integration tests only permit a localhost DATABASE_URL'); @@ -111,4 +111,13 @@ describe.skipIf(!databaseUrl||!local)('PostgresOperationsRepository',()=>{ const [a,b]=await Promise.all([repo.claim(100),repo2.claim(100)]);const ids=[...a,...b].map(item=>item.id);expect(new Set(ids).size).toBe(ids.length);expect(ids.length).toBeGreaterThan(0); expect([...a,...b].find(item=>item.dedupeKey.startsWith('message:'))?.pushToken).toBeNull(); }); + + it('keeps due/unlocked outbox observation eligible for the existing due index',async()=>{ + const client=await pool.connect(); + try { + await client.query('begin'); await client.query('set local enable_seqscan=off'); + const plan=await client.query<{"QUERY PLAN":string}>(`explain (costs off) ${OUTBOX_STATE_SQL}`); + expect(plan.rows.map(row=>row['QUERY PLAN']).join('\n')).toContain('operations_outbox_due_idx'); + } finally { await client.query('rollback').catch(()=>undefined); client.release(); } + }); }); diff --git a/apps/api/test/operations.repository.test.ts b/apps/api/test/operations.repository.test.ts new file mode 100644 index 0000000..ce8670a --- /dev/null +++ b/apps/api/test/operations.repository.test.ts @@ -0,0 +1,29 @@ +import { describe, expect, it, vi } from 'vitest'; +import { OUTBOX_STATE_SQL, PostgresOperationsRepository } from '../src/domains/operations/repository.js'; + +describe('operations outbox observation', () => { + it('uses a transaction-scoped statement timeout and always releases on query failure', async () => { + const query = vi.fn(async (sql: string) => { + if (sql === OUTBOX_STATE_SQL) throw new Error('statement timeout'); + return { rows: [], rowCount: 0 }; + }); + const release = vi.fn(); + const pool = { connect: vi.fn(async () => ({ query, release })) }; + const repository = new PostgresOperationsRepository(pool as never); + await expect(repository.outboxState()).rejects.toThrow('statement timeout'); + expect(query).toHaveBeenNthCalledWith(1, 'begin'); + expect(query).toHaveBeenNthCalledWith(2, "set local statement_timeout = '250ms'"); + expect(query).toHaveBeenNthCalledWith(3, OUTBOX_STATE_SQL); + expect(query).toHaveBeenCalledWith('rollback'); + expect(release).toHaveBeenCalledOnce(); + }); + + it('evicts the client when rollback itself fails while preserving the original error', async () => { + const original = new Error('statement timeout'); const rollback = new Error('rollback failed'); + const query = vi.fn(async (sql: string) => { if (sql === OUTBOX_STATE_SQL) throw original; if (sql === 'rollback') throw rollback; return { rows: [], rowCount: 0 }; }); + const release = vi.fn(); const repository = new PostgresOperationsRepository({ connect: vi.fn(async () => ({ query, release })) } as never); + await expect(repository.outboxState()).rejects.toBe(original); + expect(release).toHaveBeenCalledTimes(1); + expect(release).toHaveBeenCalledWith(rollback); + }); +}); diff --git a/apps/api/test/operations.runner.test.ts b/apps/api/test/operations.runner.test.ts index c06900d..f1d3e30 100644 --- a/apps/api/test/operations.runner.test.ts +++ b/apps/api/test/operations.runner.test.ts @@ -7,7 +7,7 @@ import { loadWorkerConfig } from '../src/workers/config.js'; const summary={releasedPacks:0,streakMilestones:0,digests:0,packChanges:0,weeklySummaries:0,unpairedReminders:0,catchupReminders:0,streakReminders:0,daresExpired:0}; function repository(items:OperationItem[]):OperationsRepository { - return {produce:vi.fn(async()=>summary),claim:vi.fn(async()=>items),complete:vi.fn(async()=>undefined),fail:vi.fn(async()=>undefined), + return {produce:vi.fn(async()=>summary),claim:vi.fn(async()=>items),complete:vi.fn(async()=>undefined),fail:vi.fn(async()=>undefined),outboxState:vi.fn(async()=>({due:items.length,oldestDueAgeSeconds:0})), message:vi.fn(async()=>null),classifierConfig:vi.fn(async()=>null),classify:vi.fn(async()=>undefined),close:vi.fn(async()=>undefined)}; } @@ -71,4 +71,11 @@ describe('worker configuration',()=>{ discordWebhookUrl:undefined, }); }); + + it('uses a distinct bounded pool configuration for the worker process',()=>{ + expect(loadWorkerConfig({ + DATABASE_URL:'postgresql://sauci:test@localhost:5432/sauci', + DATABASE_POOL_MAX:'3', DATABASE_POOL_IDLE_TIMEOUT_MS:'1200', DATABASE_POOL_CONNECTION_TIMEOUT_MS:'3400', + }).databasePool).toEqual({max:3,idleTimeoutMillis:1200,connectionTimeoutMillis:3400}); + }); }); diff --git a/apps/api/test/telemetry.test.ts b/apps/api/test/telemetry.test.ts new file mode 100644 index 0000000..1f4a8ba --- /dev/null +++ b/apps/api/test/telemetry.test.ts @@ -0,0 +1,50 @@ +import { Pool } from 'pg'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { createDatabasePool } from '../src/db/pool.js'; +import { flushTelemetry, recordPool, setTelemetrySinkForTests, type TelemetryRecord } from '../src/telemetry.js'; + +const url = 'postgres://user:pass@127.0.0.1:5432/sauci_test'; + +describe('privacy-safe telemetry', () => { + let restore: () => void = () => {}; + afterEach(() => { restore(); vi.restoreAllMocks(); }); + + it('records query duration without SQL or bound values', async () => { + const records: TelemetryRecord[] = []; + restore = setTelemetrySinkForTests((record) => records.push(record)); + vi.spyOn(Pool.prototype, 'query').mockResolvedValueOnce({ rows: [], rowCount: 0 } as never); + const pool = createDatabasePool(url, {}, 'api'); + await pool.query('select secret from profiles where id=$1', ['private-user-id']); + flushTelemetry(); + const serialized = JSON.stringify(records); + expect(records).toContainEqual(expect.objectContaining({ event: 'query', process: 'api', operation: 'pool_query', outcome: 'ok', count: 1 })); + expect(serialized).not.toContain('secret'); + expect(serialized).not.toContain('private-user-id'); + await pool.end(); + }); + + it('emits only numeric pool state', async () => { + const records: TelemetryRecord[] = []; + restore = setTelemetrySinkForTests((record) => records.push(record)); + const pool = new Pool({ connectionString: url }); + recordPool('worker', pool); + expect(records).toEqual([{ event: 'pool', process: 'worker', total: 0, idle: 0, waiting: 0 }]); + await pool.end(); + }); + + it('also instruments transaction-client queries', async () => { + const records: TelemetryRecord[] = []; + restore = setTelemetrySinkForTests((record) => records.push(record)); + const client = { query: vi.fn(async () => ({ rows: [], rowCount: 0 })), release: vi.fn() }; + vi.spyOn(Pool.prototype, 'connect').mockResolvedValueOnce(client as never); + const pool = createDatabasePool(url, {}, 'worker'); + const borrowed = await pool.connect(); + await borrowed.query('select message_body from messages where id=$1', ['private-message-id']); + flushTelemetry(); + const serialized = JSON.stringify(records); + expect(records).toContainEqual(expect.objectContaining({ event: 'query', process: 'worker', outcome: 'ok', count: 1 })); + expect(serialized).not.toContain('message_body'); + expect(serialized).not.toContain('private-message-id'); + await pool.end(); + }); +}); diff --git a/apps/api/test/worker.lifecycle.test.ts b/apps/api/test/worker.lifecycle.test.ts new file mode 100644 index 0000000..fff9eec --- /dev/null +++ b/apps/api/test/worker.lifecycle.test.ts @@ -0,0 +1,30 @@ +import { describe, expect, it, vi } from 'vitest'; +import { WorkerLifecycle } from '../src/workers/lifecycle.js'; + +function deferred() { let resolve!: (value: T) => void; const promise = new Promise((next) => { resolve = next; }); return { promise, resolve }; } + +describe('WorkerLifecycle', () => { + it('reuses one shutdown promise, awaits the active tick, and closes once', async () => { + const tick = deferred(); const closePool = vi.fn(async () => undefined); + const lifecycle = new WorkerLifecycle({ pollIntervalMs: 60_000, runTick: () => tick.promise, closePool, onTickFailure: vi.fn(), graceMs: 50 }); + lifecycle.start(); const first = lifecycle.shutdown(); const second = lifecycle.shutdown(); + expect(second).toBe(first); expect(closePool).not.toHaveBeenCalled(); + tick.resolve(); await expect(first).resolves.toBe(0); expect(closePool).toHaveBeenCalledTimes(1); + }); + + it('returns nonzero when an active tick exceeds the one total deadline', async () => { + vi.useFakeTimers(); + const lifecycle = new WorkerLifecycle({ pollIntervalMs: 60_000, runTick: () => new Promise(() => undefined), closePool: vi.fn(async () => undefined), onTickFailure: vi.fn(), graceMs: 10 }); + lifecycle.start(); const shutdown = lifecycle.shutdown(); await vi.advanceTimersByTimeAsync(10); + await expect(shutdown).resolves.toBe(1); vi.useRealTimers(); + }); + + it('returns nonzero and closes once when pool shutdown rejects or hangs', async () => { + const rejected = new WorkerLifecycle({ pollIntervalMs: 60_000, runTick: async () => undefined, closePool: vi.fn(async () => { throw new Error('close failed'); }), onTickFailure: vi.fn(), graceMs: 50 }); + rejected.start(); await expect(rejected.shutdown()).resolves.toBe(1); + vi.useFakeTimers(); const closePool = vi.fn(() => new Promise(() => undefined)); + const hung = new WorkerLifecycle({ pollIntervalMs: 60_000, runTick: async () => undefined, closePool, onTickFailure: vi.fn(), graceMs: 10 }); + hung.start(); const shutdown = hung.shutdown(); await vi.advanceTimersByTimeAsync(10); + await expect(shutdown).resolves.toBe(1); expect(closePool).toHaveBeenCalledTimes(1); vi.useRealTimers(); + }); +}); diff --git a/docs/architecture/client-data-freshness.md b/docs/architecture/client-data-freshness.md index 5745b1d..f447af9 100644 --- a/docs/architecture/client-data-freshness.md +++ b/docs/architecture/client-data-freshness.md @@ -189,7 +189,10 @@ default-sized pool per domain. Shutdown ends that pool once. Repositories still accept a connection string in tests, in which case they own and end the pool they created — see `resolvePool` / `closeResolvedPool` in `apps/api/src/db/pool.ts`. -Bounds come from `DATABASE_POOL_MAX` (default 10), +Bounds come from `DATABASE_POOL_MAX` (default 10 for API, 4 for worker), `DATABASE_POOL_IDLE_TIMEOUT_MS` (30000) and `DATABASE_POOL_CONNECTION_TIMEOUT_MS` (10000). They are process-wide; keep the total across running instances inside the database's own connection limit. +Production Compose exposes these as separate `API_DATABASE_POOL_MAX` and +`WORKER_DATABASE_POOL_MAX` budgets; `infra/backend/README.md` documents the +database-wide formula and deferred `pg_stat_statements` prerequisite. diff --git a/infra/backend/.env.production.example b/infra/backend/.env.production.example index 847fcde..91f40c9 100644 --- a/infra/backend/.env.production.example +++ b/infra/backend/.env.production.example @@ -4,6 +4,13 @@ POSTGRES_USER=sauci POSTGRES_PASSWORD=generate-a-long-random-password DATABASE_URL=postgresql://sauci:URL_ENCODED_PASSWORD@postgres:5432/sauci +# Keep this total below PostgreSQL max_connections: +# (API replicas * API_DATABASE_POOL_MAX) + (worker replicas * WORKER_DATABASE_POOL_MAX) + migration/admin headroom. +API_DATABASE_POOL_MAX=10 +WORKER_DATABASE_POOL_MAX=4 +DATABASE_POOL_IDLE_TIMEOUT_MS=30000 +DATABASE_POOL_CONNECTION_TIMEOUT_MS=10000 + # Hosted Supabase Auth only. The service-role key is used solely by the # server-side Auth Admin endpoint for account deletion and admin identity # enrichment. It belongs on the API only, never the worker, MCP, or a client. diff --git a/infra/backend/README.md b/infra/backend/README.md index 9cd6833..9bd3568 100644 --- a/infra/backend/README.md +++ b/infra/backend/README.md @@ -111,6 +111,53 @@ docker compose --env-file infra/backend/.env.production.local \ -f infra/backend/compose.production.yaml config ``` +## PostgreSQL connection and telemetry contract + +The API and worker each create exactly one process-owned PostgreSQL pool and +hand it to every repository. They have separate deployment variables: +`API_DATABASE_POOL_MAX` (default 10) and `WORKER_DATABASE_POOL_MAX` (default +4). Set PostgreSQL `max_connections` above: + +```text +(API replicas * API_DATABASE_POOL_MAX) + (worker replicas * WORKER_DATABASE_POOL_MAX) + migration/admin headroom +``` + +The headroom covers the one-shot migrator, an operator session, and any +backup/monitoring connection; allocating the full server limit to application +pools is unsafe. `DATABASE_POOL_IDLE_TIMEOUT_MS` and +`DATABASE_POOL_CONNECTION_TIMEOUT_MS` apply to both process pools. + +API and worker write privacy-safe JSON telemetry to normal logs. Fixed event +types are request, query, pool, sync, and outbox; fields are route templates, +method/status/outcome, durations, counts, and pool/outbox sizes. They never +include SQL, query values, raw paths, identities, tokens, content, dedupe keys, +or request bodies. There is deliberately no public `/metrics` endpoint. +Request, query, and sync events are aggregated into one fixed-interval JSON +record per label set (and worker flushes after its tick); pool state is sampled +once per minute rather than per request. This prevents telemetry itself from +becoming a connection or log-volume amplifier. Outbox observation runs after +work with a 250ms best-effort deadline and uses the same due/unlocked predicate +as claims, so a slow observation cannot delay the next lease. + +`pg_stat_statements` is useful for aggregate query investigation but is not +enabled by this repository change. A database operator must first confirm the +PostgreSQL image/provider supports it, add `shared_preload_libraries = +'pg_stat_statements'`, and restart PostgreSQL in a planned window. After that, +the extension itself must be introduced by a reviewed, committed migration in +`apps/api/drizzle/`; do not run ad hoc provider/operator DDL. This changes +server startup and remains deferred from ordinary API deployments. + +For a controlled local observation, `npm run load:representative -w @sauci/api` +performs five authenticated sync reads, five authenticated chat reads, and a +read-only outbox-state query. It requires `DATABASE_URL`, +`SAUCI_LOAD_AUTHORIZATION`, `SAUCI_LOAD_USER_ID`, and `SAUCI_LOAD_MATCH_ID`. +It accepts only `http` loopback API port 3003 (or the explicit local +`SAUCI_LOAD_API_PORT`) and loopback PostgreSQL, verifies `/health/live` and the +authenticated fixture identity before load, and emits counts plus p50/p95/p99 +summaries without any pass/fail latency threshold. Use disposable local or +designated non-production Auth credentials only; never pass a staging or +production token to this fixture. + Run the repeatable disposable container check before changing the Dokploy application. It uses hard-coded test-only credentials, creates a unique Compose project, verifies migration/startup/security/shutdown behaviour, and removes its diff --git a/infra/backend/compose.production.yaml b/infra/backend/compose.production.yaml index c3c0287..eee0a83 100644 --- a/infra/backend/compose.production.yaml +++ b/infra/backend/compose.production.yaml @@ -8,6 +8,9 @@ x-common-environment: &common-environment x-api-environment: &api-environment <<: *common-environment PORT: 3003 + DATABASE_POOL_MAX: ${API_DATABASE_POOL_MAX:-10} + DATABASE_POOL_IDLE_TIMEOUT_MS: ${DATABASE_POOL_IDLE_TIMEOUT_MS:-30000} + DATABASE_POOL_CONNECTION_TIMEOUT_MS: ${DATABASE_POOL_CONNECTION_TIMEOUT_MS:-10000} SUPABASE_AUTH_URL: ${SUPABASE_AUTH_URL:?set SUPABASE_AUTH_URL} SUPABASE_AUTH_AUDIENCE: ${SUPABASE_AUTH_AUDIENCE:-authenticated} SUPABASE_AUTH_SERVICE_ROLE_KEY: ${SUPABASE_AUTH_SERVICE_ROLE_KEY:-} @@ -20,6 +23,9 @@ x-api-environment: &api-environment x-worker-environment: &worker-environment <<: *common-environment + DATABASE_POOL_MAX: ${WORKER_DATABASE_POOL_MAX:-4} + DATABASE_POOL_IDLE_TIMEOUT_MS: ${DATABASE_POOL_IDLE_TIMEOUT_MS:-30000} + DATABASE_POOL_CONNECTION_TIMEOUT_MS: ${DATABASE_POOL_CONNECTION_TIMEOUT_MS:-10000} OPENROUTER_API_KEY: ${OPENROUTER_API_KEY:-} CLASSIFIER_ENABLED: ${CLASSIFIER_ENABLED:-true} CLASSIFIER_MODEL: ${CLASSIFIER_MODEL:-openai/gpt-4o}