Repository Public
1/** Kernel handle (pool, data dir, logger) and the transaction helper every state write goes through. */2import os from 'node:os';3import path from 'node:path';4import pg from 'pg';5import { pino, type Logger } from 'pino';6import type { EventKind, Trust } from '@aos/shared';7import { KernelError, KernelStorageError } from './errors.js';89export const EVENTS_CHANNEL = 'aos_events';10/** Tiny control doorbell: 'q' = something became claimable, 'c:<runId>' = cancel requested. */11export const WAKE_CHANNEL = 'aos_wake';12// pg_advisory locks (two-int form). Class 0x414f53 = "AOS".13export const LOCK_CLASS = 0x414f53;14export const LOCK_EVENTS = 1;15export const LOCK_MIGRATE = 2;1617export interface KernelOptions {18 /** Default: env DATABASE_URL. Ignored when `pool` is given. */19 databaseUrl?: string;20 pool?: pg.Pool;21 poolMax?: number;22 /** Default: env AOS_DATA_DIR, else <tmpdir>/aos-data. Artifacts live under <dataDir>/artifacts. */23 dataDir?: string;24 logger?: Logger;25}2627export interface Kernel {28 readonly pool: pg.Pool;29 readonly dataDir: string;30 readonly log: Logger;31 /** Ends the pool if the kernel created it. */32 close(): Promise<void>;33}3435export function createLogger(name = 'kernel'): Logger {36 return pino({ name, level: process.env.AOS_LOG_LEVEL ?? 'info' });37}3839export function createKernel(opts: KernelOptions = {}): Kernel {40 const log = opts.logger ?? createLogger();41 let pool = opts.pool;42 let owns = false;43 if (!pool) {44 const url = opts.databaseUrl ?? process.env.DATABASE_URL;45 if (!url) throw new KernelError('createKernel: no databaseUrl / DATABASE_URL');46 pool = new pg.Pool({ connectionString: url, max: opts.poolMax ?? 10, idleTimeoutMillis: 30_000 });47 // Idle-client errors surface on the next query; never crash the process.48 pool.on('error', (err) => log.warn({ err: err.message }, 'idle pg client error'));49 owns = true;50 }51 const dataDir = path.resolve(opts.dataDir ?? process.env.AOS_DATA_DIR ?? path.join(os.tmpdir(), 'aos-data'));52 const p = pool;53 return {54 pool: p,55 dataDir,56 log,57 close: async () => {58 if (owns) await p.end();59 },60 };61}6263let defaultKernel: Kernel | null = null;64/** Process-wide default kernel (from env) used by the top-level contract functions. */65export function getKernel(): Kernel {66 defaultKernel ??= createKernel();67 return defaultKernel;68}69export function configureKernel(opts: KernelOptions | Kernel): Kernel {70 defaultKernel = 'close' in opts && 'log' in opts ? (opts as Kernel) : createKernel(opts as KernelOptions);71 return defaultKernel;72}73export async function closeKernel(): Promise<void> {74 const k = defaultKernel;75 defaultKernel = null;76 if (k) await k.close();77}7879export interface NewEvent {80 missionId: string;81 runId?: string | null;82 seq?: number | null;83 kind: EventKind;84 trust?: Trust;85 payload?: Record<string, unknown>;86}8788export type Row = Record<string, unknown>;89export interface Tx {90 q<T = Row>(sql: string, params?: unknown[]): Promise<T[]>;91 one<T = Row>(sql: string, params?: unknown[]): Promise<T | null>;92 /** Queue a persisted event; written at the end of the transaction (see flush). */93 emit(ev: NewEvent): void;94 /** Queue a NOTIFY on the wake channel (delivered at commit). */95 wake(msg: string): void;96 /** Write queued events now and return their ids. No row lock may be taken after this call. */97 flush(): Promise<number[]>;98}99100/**101 * Runs `fn` in one transaction. Queued events are inserted last, under a global advisory lock, so that102 * event ids are committed in increasing order (a cursor reader can never skip a late-committing lower103 * id). Lock order everywhere: run rows (by id) → wait rows → mission row → spend_days → events lock.104 * Non-kernel errors (pg, network) are wrapped in KernelStorageError.105 */106export async function withTx<T>(k: Kernel, fn: (tx: Tx) => Promise<T>): Promise<T> {107 let client: pg.PoolClient;108 try {109 client = await k.pool.connect();110 } catch (err) {111 throw new KernelStorageError(`database unavailable: ${(err as Error).message}`, { cause: err });112 }113 let broken: Error | undefined;114 const events: NewEvent[] = [];115 const wakes = new Set<string>();116 const tx: Tx = {117 q: async <R>(sql: string, params: unknown[] = []) => (await client.query(sql, params)).rows as R[],118 one: async <R>(sql: string, params: unknown[] = []) => ((await client.query(sql, params)).rows[0] as R | undefined) ?? null,119 emit: (ev) => void events.push(ev),120 wake: (msg) => void wakes.add(msg),121 flush: async () => {122 if (events.length === 0) return [];123 const batch = events.splice(0, events.length);124 await client.query('SELECT pg_advisory_xact_lock($1, $2)', [LOCK_CLASS, LOCK_EVENTS]);125 const ids: number[] = [];126 for (const ev of batch) {127 const r = await client.query(128 `INSERT INTO events (mission_id, run_id, seq, kind, trust, payload) VALUES ($1,$2,$3,$4,$5,$6::jsonb) RETURNING id`,129 [ev.missionId, ev.runId ?? null, ev.seq ?? null, ev.kind, ev.trust ?? 'system', JSON.stringify(ev.payload ?? {})],130 );131 ids.push(Number((r.rows[0] as { id: unknown }).id));132 }133 await client.query('SELECT pg_notify($1, $2)', [EVENTS_CHANNEL, JSON.stringify({ id: ids[ids.length - 1] })]);134 return ids;135 },136 };137 try {138 await client.query('BEGIN');139 const out = await fn(tx);140 await tx.flush();141 for (const w of wakes) await client.query('SELECT pg_notify($1, $2)', [WAKE_CHANNEL, w]);142 await client.query('COMMIT');143 return out;144 } catch (err) {145 try {146 await client.query('ROLLBACK');147 } catch (e) {148 broken = e as Error; // connection is unusable: destroy it instead of returning it to the pool149 }150 if (err instanceof KernelError) throw err;151 throw new KernelStorageError(`kernel storage error: ${(err as Error).message}`, { cause: err });152 } finally {153 client.release(broken);154 }155}156157/** Single statement outside a transaction, with the same error wrapping. */158export async function query<T = Row>(k: Kernel, sql: string, params: unknown[] = []): Promise<T[]> {159 try {160 return (await k.pool.query(sql, params)).rows as T[];161 } catch (err) {162 throw new KernelStorageError(`kernel storage error: ${(err as Error).message}`, { cause: err });163 }164}165166export const num = (v: unknown): number => (v === null || v === undefined ? 0 : Number(v));167export const iso = (v: unknown): string => (v instanceof Date ? v.toISOString() : new Date(String(v)).toISOString());168