perf(api): bound worker pools and add telemetry

Share one configured PostgreSQL pool across worker repositories and make API and worker connection budgets explicit.

Add privacy-safe aggregate request, query, pool, sync, and outbox telemetry, a loopback-only representative load fixture, bounded worker shutdown, and focused regression and integration coverage.

Document the database-wide budget formula and defer pg_stat_statements activation to a reviewed migration and planned restart.

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

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",

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 {

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 {

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)));

apps/api/src/domains/operations/repository.ts

@@ -15,9 +15,19 @@ export interface OperationsRepository {
   message(id: string): Promise<MessageForClassification | null>;
   classifierConfig(): Promise<ClassifierRuntimeConfig | null>;
   classify(id: string, status: 'safe' | 'flagged', reason: string | null, category: string): Promise<void>;
+  outboxState(): Promise<{ due: number; oldestDueAgeSeconds: number }>;
   close(): Promise<void>;
 }
 
+/** 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<now()-interval '5 minutes')`;
+
 function item(row: OutboxRow): OperationItem {
   return { id: row.id, kind: row.kind, dedupeKey: row.dedupe_key, recipientId: row.recipient_id,
     pushToken: row.push_token, payload: row.payload, attempts: row.attempts };
@@ -280,6 +290,30 @@ export class PostgresOperationsRepository implements OperationsRepository {
       return result.rows.map(item);
     });
   }
+  async outboxState(): Promise<{ due: number; oldestDueAgeSeconds: number }> {
+    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)]); }

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};
   }
 }
-

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<void> {
   if (shuttingDown) return;
   shuttingDown = true;
+  clearInterval(poolTelemetryTimer);
   console.log(`Received ${signal}; shutting down`);
   const forcedExit = setTimeout(() => {
     console.error('Graceful shutdown timed out');

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;
+}

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<number> {
+  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<void> {
+  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();

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<string, TelemetryRecord>();
+
+function aggregate(record: Extract<TelemetryRecord, { event: 'request' | 'query' | 'sync' }>): 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<TelemetryRecord, { event: 'request' | 'query' | 'sync' }>): 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<PoolClient>();
+  const instrumentQuery = <T extends { query: Pool['query'] }>(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<unknown>).then === 'function') return (result as Promise<unknown>).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<PoolClient>).then === 'function') return (result as Promise<PoolClient>).then((client) => { if (!clients.has(client)) { clients.add(client); instrumentQuery(client); } return client; });
+    return result;
+  }) as Pool['connect'];
+  return pool;
+}

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};

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();

apps/api/src/workers/lifecycle.ts

@@ -0,0 +1,47 @@
+export interface WorkerLifecycleDependencies {
+  pollIntervalMs: number;
+  runTick: () => Promise<void>;
+  closePool: () => Promise<void>;
+  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<void> | undefined;
+  private shutdownPromise: Promise<number> | 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<number> {
+    if (!this.shutdownPromise) this.shutdownPromise = this.stop();
+    return this.shutdownPromise;
+  }
+
+  private async tick(): Promise<void> {
+    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<number> {
+    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<number>((resolve) => { timeout = setTimeout(() => resolve(1), this.graceMs); });
+    const result = await Promise.race([work, timedOut]);
+    if (timeout) clearTimeout(timeout);
+    return result;
+  }
+}

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');
+  });
 });

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');
+  });
+});

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(); }
+  });
 });

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);
+  });
+});

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});
+  });
 });

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();
+  });
+});

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<T>() { let resolve!: (value: T) => void; const promise = new Promise<T>((next) => { resolve = next; }); return { promise, resolve }; }
+
+describe('WorkerLifecycle', () => {
+  it('reuses one shutdown promise, awaits the active tick, and closes once', async () => {
+    const tick = deferred<void>(); 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<void>(() => 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<void>(() => 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();
+  });
+});