createServerHost
tcm's server host, #tcm/host (host protocol 1): what a server API library needs of the process. The call being
answered, the app's SQLite files (their tables as tcm keeps them), the lock every table call and transaction takes
on the store, transactions, the job queue with its worker and schedules, the wire codec and the schema checker.
One per process, however many libraries call it. Its own source is written to .tcm/cache/tcm-host.js, so it uses
only what comes in through deps.
import type { decode, encode } from "#codebase/tcmts/cli/codec"; import type { checkValue } from "#codebase/tcmts/core/schema"; import type { AsyncLocalStorage } from "node:async_hooks"; import type { randomBytes } from "node:crypto"; import type { DatabaseSync } from "node:sqlite"; export function createServerHost(deps: { AsyncLocalStorage: typeof AsyncLocalStorage; randomBytes: typeof randomBytes; DatabaseSync: typeof DatabaseSync; encode: typeof encode; decode: typeof decode; checkValue: typeof checkValue; }) { interface ServerRequest { method: string; url: string; headers: Headers; cookies: Record<string, string>; ip: string | undefined; signal: AbortSignal; } interface ServerCall { /** absent in a background job */ request?: ServerRequest; setCookies: string[]; secure: boolean; /** where app data lives; in memory when absent (tests) */ dataDir?: string; /** the path cookies are for unless they say: the app's base (`tcm start --base`), so apps sharing a site keep theirs apart */ cookiePath?: string; /** what React's `cache` keeps for the request (`tcm-cache.js`): shared by every call made for one request */ memo?: WeakMap<object, unknown>; } type Db = InstanceType<typeof DatabaseSync>; type Fn = (...args: unknown[]) => unknown; /** A random id people may type: base64url, never starting with `-` (a command line would read it as a flag). */ const randomId = (bytes: number): string => { let id: string; do id = deps.randomBytes(bytes).toString("base64url"); while (id.startsWith("-")); return id; }; const storage = new deps.AsyncLocalStorage<ServerCall>(); // React's cache and cacheSignal, as tcm-cache.js has them: the request being answered, if any const memos = new WeakMap<object, WeakMap<object, unknown>>(); (globalThis as { __tcmCall?: () => unknown }).__tcmCall = () => { const c = storage.getStore(); return c?.memo ? { memo: c.memo, signal: c.request?.signal ?? null } : undefined; }; /** The call being answered (a request, or a background job), if any. */ const call = (): ServerCall | undefined => storage.getStore(); let hostDataDir: string | undefined; /** Where the app's data lives now: the call's, else the worker's; in memory when neither (tests). */ const dataDir = (): string | undefined => storage.getStore()?.dataDir ?? hostDataDir; // ---- the app's data files: one connection to each, its tables as tcm keeps them ------------------------------ const SCHEMAS = { store: `PRAGMA journal_mode = WAL; PRAGMA busy_timeout = 5000; CREATE TABLE IF NOT EXISTS rows (tbl TEXT NOT NULL, id TEXT NOT NULL, data TEXT NOT NULL, schema_hash TEXT NOT NULL, updated INTEGER NOT NULL, PRIMARY KEY (tbl, id)); CREATE TABLE IF NOT EXISTS schemas (tbl TEXT NOT NULL, hash TEXT NOT NULL, schema TEXT NOT NULL, first_seen INTEGER NOT NULL, PRIMARY KEY (tbl, hash)); CREATE INDEX IF NOT EXISTS rows_by_schema ON rows (tbl, schema_hash);`, sessions: "PRAGMA journal_mode = WAL; CREATE TABLE IF NOT EXISTS sessions (id TEXT PRIMARY KEY, data TEXT NOT NULL, expires INTEGER NOT NULL)", jobs: `PRAGMA journal_mode = WAL; PRAGMA busy_timeout = 5000; CREATE TABLE IF NOT EXISTS jobs ( id TEXT PRIMARY KEY, fn_hash TEXT NOT NULL, fn_name TEXT NOT NULL, args TEXT NOT NULL, run_at INTEGER NOT NULL, attempts INTEGER NOT NULL DEFAULT 0, max_attempts INTEGER NOT NULL, status TEXT NOT NULL, lease_until INTEGER, last_error TEXT, created INTEGER NOT NULL, finished INTEGER, slot TEXT UNIQUE ); CREATE INDEX IF NOT EXISTS jobs_due ON jobs (status, run_at); CREATE TABLE IF NOT EXISTS schedules (name TEXT PRIMARY KEY, first_seen INTEGER NOT NULL);`, }; const dbs = new Map<string, Db>(); const open = (name: keyof typeof SCHEMAS, dir: string | undefined): Db => { const file = dir ? `${dir}/${name}.sqlite` : ":memory:"; const key = `${name} ${file}`; let db = dbs.get(key); if (!db) { db = new deps.DatabaseSync(file); db.exec(SCHEMAS[name]); dbs.set(key, db); } return db; }; /** * The app's SQLite file for where its data lives now: `store` (tables' `rows` and the `schemas` they were written * with) or `sessions` (`sessions`: id, data, expires). */ const database = (name: "store" | "sessions"): Db => open(name, dataDir()); // ---- the store's lock: one connection is shared by every request, so a transaction must own it ---------------- // A transaction holds a lock on the store for its whole (async) duration: table calls made inside it run in it, // and table calls from anywhere else wait until it commits or rolls back. Every table call takes the same lock // briefly, so nothing reads uncommitted rows or writes into someone else's transaction. Holding is tracked with // AsyncLocalStorage, and a hold ends when its call does: a timer started inside fires as an outsider. type Hold = { key: object; open: boolean; parent?: Hold }; const holds = new deps.AsyncLocalStorage<Hold>(); const tails = new Map<object, Promise<void>>(); const holding = (key: object) => { for (let h = holds.getStore(); h; h = h.parent) if (h.key === key && h.open) return true; return false; }; /** Run `fn` with `key` (a store, or a transaction for its nested ones) to itself; reentrant. */ const exclusive = async <R>(key: object, fn: () => R | Promise<R>): Promise<R> => { if (holding(key)) return fn(); const prev = tails.get(key) ?? Promise.resolve(); let release!: () => void; const mine = new Promise<void>((r) => (release = r)); const tail = prev.then(() => mine); tails.set(key, tail); await prev; const hold: Hold = { key, open: true, parent: holds.getStore() }; try { return await holds.run(hold, fn); } finally { hold.open = false; release(); if (tails.get(key) === tail) tails.delete(key); } }; type Txn = { db: Db; open: boolean; afterCommit: (() => void)[] }; const txns = new deps.AsyncLocalStorage<Txn>(); const openTxn = () => { const t = txns.getStore(); return t?.open ? t : undefined; }; let savepoints = 0; /** * Run `fn` as one transaction over the store: all its writes happen, or (if it throws) none do. Jobs it enqueues * are queued when it commits and dropped if it rolls back. Nested transactions are savepoints. */ const transaction = async <R>(fn: () => R | Promise<R>): Promise<R> => { const db = database("store"); const outer = openTxn(); if (outer && outer.db === db) { // a savepoint: its failure undoes only its own writes. Nested ones take turns (savepoints are a stack). return exclusive(outer, async () => { const name = `tcm_tx_${++savepoints}`; const inner: Txn = { db, open: true, afterCommit: [] }; db.exec(`SAVEPOINT ${name}`); try { const r = await txns.run(inner, fn); db.exec(`RELEASE ${name}`); outer.afterCommit.push(...inner.afterCommit); return r; } catch (e) { db.exec(`ROLLBACK TO ${name}`); db.exec(`RELEASE ${name}`); throw e; } finally { inner.open = false; } }); } return exclusive(db, async () => { const t: Txn = { db, open: true, afterCommit: [] }; db.exec("BEGIN IMMEDIATE"); let r: R; try { r = await txns.run(t, fn); db.exec("COMMIT"); } catch (e) { if (db.isTransaction) db.exec("ROLLBACK"); throw e; } finally { t.open = false; } for (const f of t.afterCommit) f(); return r; }); }; // ---- background jobs and schedules ------------------------------------------------------------------------ // Jobs live in .tcm/data/jobs.sqlite and name their function by hash, so a job runs the exact code version that // enqueued it, even across deploys. A claim is one atomic UPDATE with a lease, so several processes can work // the same queue; a scheduled slot is claimed by a unique key, so it runs once however many servers there are. const HASH_MARK = Symbol.for("tcm.hash"); interface JobRow { id: string; fn_hash: string; fn_name: string; args: string; run_at: number; attempts: number; max_attempts: number; status: string; } const memoryFns = new Map<string, Fn>(); const enqueueAt = ( runAt: number, fn: Fn, args: unknown[], opts: { attempts?: number; slot?: string; name?: string; hash?: string } = {}, ): string => { if (typeof fn !== "function") throw new TypeError("enqueue() needs a function"); const dir = dataDir(); const hash = opts.hash ?? (fn as unknown as Record<symbol, string | undefined>)[HASH_MARK]; if (dir && !hash) throw new Error( `enqueue() needs a codebase function (a top-level definition, so it can be found by hash), not an inline one`, ); const id = randomId(12); const encoded = deps.encode(args, `arguments to ${fn.name || "the job"}`); const insert = () => { const r = open("jobs", dir) .prepare( "INSERT OR IGNORE INTO jobs (id, fn_hash, fn_name, args, run_at, max_attempts, status, created, slot) VALUES (?, ?, ?, ?, ?, ?, 'queued', ?, ?)", ) .run( id, hash ?? "", opts.name ?? (fn.name || "anonymous"), encoded, runAt, opts.attempts ?? 5, Date.now(), opts.slot ?? null, ); if (r.changes && !dir) memoryFns.set(id, fn); return r.changes > 0; }; // inside a transaction, the job is part of it: queued if it commits, never if it rolls back const t = openTxn(); if (t && !opts.slot) { t.afterCommit.push(insert); return id; } return insert() ? id : ""; }; /** * Queue `fn(...args)` to run at `runAt` (ms since the epoch), retried `attempts` times (5); returns the job's id. * `fn` is a codebase function (found by its hash), `args` plain data. */ const enqueue = (runAt: number, fn: Fn, args: unknown[], opts: { attempts?: number } = {}) => enqueueAt(runAt, fn, args, { attempts: opts.attempts }); // schedules: "30s" / "15m" / "2h" / "1d" intervals, or five-field cron ("0 3 * * *", UTC) const INTERVAL = /^(\d+)(s|m|h|d)$/; const UNITS: Record<string, number> = { s: 1000, m: 60_000, h: 3_600_000, d: 86_400_000 }; const cronField = (spec: string, min: number, max: number): Set<number> => { const out = new Set<number>(); for (const part of spec.split(",")) { const [range, stepText] = part.split("/") as [string, string | undefined]; const step = stepText === undefined ? 1 : Number(stepText); const [lo, hi] = range === "*" ? [min, max] : range.includes("-") ? range.split("-").map(Number) : [Number(range), stepText === undefined ? Number(range) : max]; if (![lo, hi, step].every((n) => Number.isInteger(n)) || lo! < min || hi! > max || step < 1) throw new Error(`bad cron field '${spec}'`); for (let v = lo!; v <= hi!; v += step) out.add(v); } return out; }; const parseSchedule = (spec: string) => { const m = INTERVAL.exec(spec); if (m) return { every: Number(m[1]) * UNITS[m[2]!]! }; const f = spec.trim().split(/\s+/); if (f.length !== 5) throw new Error( `schedule '${spec}': use an interval like "15m" or five cron fields like "0 3 * * *"`, ); return { cron: [ cronField(f[0]!, 0, 59), cronField(f[1]!, 0, 23), cronField(f[2]!, 1, 31), cronField(f[3]!, 1, 12), cronField(f[4]!, 0, 6), ], }; }; /** Whether `spec` is a schedule this host runs: an interval ("15m") or five cron fields (UTC); throws if not. */ const checkSchedule = (spec: string): void => { parseSchedule(spec); }; /** The scheduled slot at or before `now`, if one is due (intervals always; cron only in the minute it names). */ const dueSlot = (spec: ReturnType<typeof parseSchedule>, now: number): number | undefined => { if (spec.every) return Math.floor(now / spec.every) * spec.every; const d = new Date(Math.floor(now / 60_000) * 60_000); const [mi, h, dom, mo, dow] = spec.cron!; return mi!.has(d.getUTCMinutes()) && h!.has(d.getUTCHours()) && dom!.has(d.getUTCDate()) && mo!.has(d.getUTCMonth() + 1) && dow!.has(d.getUTCDay()) ? d.getTime() : undefined; }; const backoff = (attempt: number) => Math.min(3600, 2 ** attempt) * 1000; const finish = (db: Db, row: JobRow, error: unknown) => { if (error === undefined) { db.prepare( "UPDATE jobs SET status = 'done', finished = ?, lease_until = NULL WHERE id = ?", ).run(Date.now(), row.id); return "done"; } const message = error instanceof Error ? (error.stack ?? error.message) : String(error); if (row.attempts >= row.max_attempts) { db.prepare( "UPDATE jobs SET status = 'failed', finished = ?, last_error = ?, lease_until = NULL WHERE id = ?", ).run(Date.now(), message, row.id); return "failed"; } db.prepare( "UPDATE jobs SET status = 'queued', run_at = ?, last_error = ?, lease_until = NULL WHERE id = ?", ).run(Date.now() + backoff(row.attempts), message, row.id); return "retrying"; }; const runJob = async (fn: Fn, row: JobRow, dir: string | undefined) => { const ctx: ServerCall = { setCookies: [], secure: false, dataDir: dir }; try { await storage.run(ctx, async () => fn(...(deps.decode(row.args) as unknown[]))); return undefined; } catch (e) { return e ?? new Error("the job threw undefined"); } }; /** For tests: run the queued jobs that are due (all of them with `{ all: true }`), inline. */ const runJobs = async ( opts: { all?: boolean } = {}, ): Promise<{ done: number; failed: number }> => { const db = open("jobs", dataDir()); let done = 0; let failed = 0; for (;;) { const row = db .prepare( "SELECT * FROM jobs WHERE status = 'queued' AND run_at <= ? ORDER BY run_at LIMIT 1", ) .get(opts.all ? Number.MAX_SAFE_INTEGER : Date.now()) as JobRow | undefined; if (!row) return { done, failed }; db.prepare("UPDATE jobs SET attempts = attempts + 1, status = 'running' WHERE id = ?").run( row.id, ); row.attempts++; const fn = memoryFns.get(row.id); const outcome = fn ? finish(db, row, await runJob(fn, row, dataDir())) : finish( db, { ...row, attempts: row.max_attempts }, new Error("no function for this job in memory"), ); if (outcome === "done") done++; else if (outcome === "failed") failed++; } }; /** For the runtime: work the queue in .tcm/data, `concurrency` jobs at a time, loading functions by hash. */ const startWorker = (opts: { dataDir: string; load: (hash: string) => Promise<Fn>; log: (line: string) => void; concurrency?: number; }) => { hostDataDir = opts.dataDir; const db = open("jobs", opts.dataDir); const claim = db.prepare( `UPDATE jobs SET status = 'running', attempts = attempts + 1, lease_until = ? WHERE id = (SELECT id FROM jobs WHERE (status = 'queued' AND run_at <= ?) OR (status = 'running' AND lease_until < ?) ORDER BY run_at LIMIT 1) RETURNING *`, ); let running = 0; let stopped = false; const tick = () => { while (!stopped && running < (opts.concurrency ?? 4)) { const now = Date.now(); const row = claim.get(now + 10 * 60_000, now, now) as JobRow | undefined; if (!row) return; running++; const started = performance.now(); (async () => { let error: unknown; try { error = await runJob(await opts.load(row.fn_hash), row, opts.dataDir); } catch (e) { error = e; } const outcome = finish(db, row, error); const ms = Math.round(performance.now() - started); opts.log( `job ${row.fn_name} ${outcome}${outcome === "done" ? "" : ` (attempt ${row.attempts}/${row.max_attempts}: ${String((error as Error)?.message ?? error).split("\n")[0]})`} ${ms}ms`, ); })().finally(() => { running--; tick(); }); } }; const timer = setInterval(tick, 250); tick(); return () => { stopped = true; clearInterval(timer); }; }; /** * For the runtime: the recurring jobs found in the app's namespace; replaces the previous set. `hash` is the * schedule definition's: its function may be inline, so a run records "the fn of schedule <hash>" (pinned all the * same). */ let scheduleTimer: ReturnType<typeof setInterval> | undefined; const setSchedules = (opts: { dataDir: string; jobs: { name: string; spec: string; fn: Fn; hash: string }[]; log?: (line: string) => void; }) => { hostDataDir = opts.dataDir; const db = open("jobs", opts.dataDir); const jobs = opts.jobs.map((j) => ({ ...j, parsed: parseSchedule(j.spec) })); for (const j of jobs) db.prepare("INSERT OR IGNORE INTO schedules (name, first_seen) VALUES (?, ?)").run( j.name, Date.now(), ); const tick = () => { const now = Date.now(); for (const j of jobs) { // one broken schedule must never take the server down try { const slot = dueSlot(j.parsed, now); const first = ( db.prepare("SELECT first_seen FROM schedules WHERE name = ?").get(j.name) as { first_seen: number; } ).first_seen; // only slots since the schedule first appeared: a new schedule doesn't fire at once, a restart catches up if (slot === undefined || slot < first) continue; enqueueAt(slot, j.fn, [], { slot: `${j.name}@${slot}`, name: j.name, hash: `${j.hash}#fn`, }); } catch (e) { opts.log?.(`schedule ${j.name}: ${(e as Error).message}`); } } }; if (scheduleTimer) clearInterval(scheduleTimer); scheduleTimer = setInterval(tick, 1000); tick(); }; // ---- calls ------------------------------------------------------------------------------------------------- const parseCookies = (header: string | null): Record<string, string> => { const out: Record<string, string> = {}; for (const part of (header ?? "").split(";")) { const i = part.indexOf("="); if (i < 0) continue; try { out[part.slice(0, i).trim()] = decodeURIComponent(part.slice(i + 1).trim()); } catch { // ignore malformed cookies } } return out; }; /** For the runtime: the Set-Cookie headers the current call has produced so far (a stream sends them up front). */ const pendingCookies = (): string[] => storage.getStore()?.setCookies ?? []; /** Run one call inside its request's context; reports the Set-Cookie headers it produced. */ const withCall = async <T>( req: { method: string; url: string; headers: Record<string, string | string[] | undefined>; ip: string | undefined; signal: AbortSignal; }, opts: { secure: boolean; dataDir?: string; cookiePath?: string }, fn: () => Promise<T>, ): Promise< | { ok: true; value: T; setCookies: string[] } | { ok: false; error: unknown; setCookies: string[] } > => { const headers = new Headers(); for (const [k, v] of Object.entries(req.headers)) if (v !== undefined) headers.set(k, Array.isArray(v) ? v.join(", ") : v); const ctx: ServerCall = { request: { method: req.method, url: req.url, headers, cookies: parseCookies(headers.get("cookie")), ip: req.ip, signal: req.signal, }, setCookies: [], secure: opts.secure, dataDir: opts.dataDir, cookiePath: opts.cookiePath, // (calls made for one request, as a page's server components are, share one: the same `req`) memo: memos.get(req) ?? (memos.set(req, new WeakMap()), memos.get(req)!), }; try { return { ok: true, value: await storage.run(ctx, fn), setCookies: ctx.setCookies }; } catch (error) { return { ok: false, error, setCookies: ctx.setCookies }; } }; return { call, dataDir, database, exclusive, transaction, enqueue, runJobs, checkSchedule, encode: deps.encode, decode: deps.decode, checkValue: deps.checkValue, randomId, withCall, pendingCookies, startWorker, setSchedules, }; }