konnosDashboard
runtime-bootstrap.ts136 lines · main
1import { resolve } from 'node:path';
2
3import { spawn as bunSpawn } from 'bun';
4
5import { withProjectTx } from './db.js';
6import { env } from './env.js';
7import { ingestIsolateLogLine } from './log-collector.js';
8import { IsolatePoolImpl, type SpawnFn, type SpawnedChild } from './pool-manager.js';
9import { fetchProjectEnv } from './project-env-cache.js';
10
11/**
12 * Module-level singleton — constructed lazily on first `getPool()` call so
13 * importing this module (e.g. for type access) doesn't spin up timers,
14 * spawn child processes, or open the data-plane pool. The idle sweeper
15 * starts immediately after construction.
16 */
17let pool: IsolatePoolImpl | null = null;
18
19export function getPool(): IsolatePoolImpl {
20 if (pool) return pool;
21 pool = new IsolatePoolImpl({
22 spawn: bunChildSpawn,
23 runtimeStubDir: resolve(import.meta.dir, 'isolate-runtime'),
24 isolateBaseDir: env.BRIVEN_RUNTIME_TMP_DIR,
25 maxIsolates: env.BRIVEN_RUNTIME_MAX_ISOLATES,
26 maxMemoryMb: env.BRIVEN_RUNTIME_ISOLATE_MAX_MEMORY_MB,
27 invocationTimeoutMs: env.BRIVEN_RUNTIME_INVOCATION_TIMEOUT_MS,
28 idleKillMs: env.BRIVEN_RUNTIME_IDLE_KILL_MS,
29 maxInvocationsPerIsolate: env.BRIVEN_RUNTIME_MAX_INVOCATIONS_PER_ISOLATE,
30 crashLoopThreshold: env.BRIVEN_RUNTIME_CRASH_LOOP_THRESHOLD,
31 crashLoopWindowMs: env.BRIVEN_RUNTIME_CRASH_LOOP_WINDOW_MS,
32 runQueryProxy: async (projectId, _requestId, sql, params, _table) => {
33 // The host runs the query under the project's tx — exactly the
34 // same trust boundary as the inline executor. The isolate never
35 // sees raw connection state; it only sees the rows we ship back.
36 // @README-BRIVEN ADR 0001: the `pg`-backed `tx.unsafe(sql, params)`
37 // adapter (DoltGres via node-postgres) returns the rows directly —
38 // same shape callers relied on under the earlier postgres.js path.
39 return withProjectTx(projectId, async (tx) => {
40 const rows = await tx.unsafe(sql, params as never[]);
41 return rows as readonly unknown[];
42 });
43 },
44 loadProjectEnv: fetchProjectEnv,
45 onLog: (line, projectId, envValues) => {
46 // Phase 1 sink: structured stderr. Task 16 swaps in a Redis-stream
47 // publisher so `briven logs --tail` sees isolate logs alongside
48 // user logs. envValues is the project's pre-bound env values from
49 // the spawn that emitted this line — feeding it into the sanitizer
50 // closes the §5.1 redaction loop for isolate panics.
51 ingestIsolateLogLine(line, projectId, envValues);
52 },
53 denoPath: env.BRIVEN_RUNTIME_DENO_PATH,
54 });
55 pool.startIdleSweeper();
56 return pool;
57}
58
59/** Test/shutdown hook — drops the singleton so a fresh pool can be built. */
60export async function resetPool(): Promise<void> {
61 if (!pool) return;
62 await pool.shutdown();
63 pool = null;
64}
65
66/**
67 * Adapter that maps Bun's `spawn` API onto the `SpawnFn` interface the
68 * pool expects. We pin Deno's binary path from env config and wrap
69 * stdout/stderr in newline-stripped async iterators so the pool's reader
70 * loops can `await child.stdout.next()` without re-deriving framing.
71 */
72const bunChildSpawn: SpawnFn = async ({ args, env: childEnv, cwd }) => {
73 const proc = bunSpawn({
74 cmd: [env.BRIVEN_RUNTIME_DENO_PATH, ...args],
75 cwd,
76 env: childEnv,
77 stdin: 'pipe',
78 stdout: 'pipe',
79 stderr: 'pipe',
80 });
81
82 const stdoutLines = lineIterator(proc.stdout as ReadableStream<Uint8Array>);
83 const stderrLines = lineIterator(proc.stderr as ReadableStream<Uint8Array>);
84
85 const child: SpawnedChild = {
86 pid: proc.pid,
87 stdin: {
88 write: async (line: string) => {
89 const n = proc.stdin.write(line);
90 await proc.stdin.flush();
91 return typeof n === 'number' ? n > 0 : Boolean(n);
92 },
93 end: () => proc.stdin.end(),
94 },
95 stdout: {
96 next: () => stdoutLines.next().then((r) => (r.done ? null : r.value)),
97 },
98 stderr: {
99 next: () => stderrLines.next().then((r) => (r.done ? null : r.value)),
100 },
101 wait: async () => {
102 const exitCode = await proc.exited;
103 return { exitCode, signal: null };
104 },
105 kill: (signal: string) => proc.kill(signal as never),
106 };
107 return child;
108};
109
110/**
111 * Decode a Uint8Array stream into one yielded string per `\n`-terminated
112 * line. Drops the trailing newline. If the stream ends with a partial
113 * line we yield it before completing — Deno panics typically arrive that
114 * way and we'd otherwise lose them.
115 */
116async function* lineIterator(
117 stream: ReadableStream<Uint8Array>,
118): AsyncGenerator<string, void, void> {
119 const reader = stream.getReader();
120 const decoder = new TextDecoder();
121 let buf = '';
122 while (true) {
123 const { value, done } = await reader.read();
124 if (done) {
125 if (buf.length > 0) yield buf;
126 return;
127 }
128 buf += decoder.decode(value, { stream: true });
129 let nl: number;
130 while ((nl = buf.indexOf('\n')) !== -1) {
131 const line = buf.slice(0, nl);
132 buf = buf.slice(nl + 1);
133 if (line) yield line;
134 }
135 }
136}