konnosDashboard
pool-manager.ts952 lines · main
1import { newId } from '@briven/shared';
2
3import {
4 cleanupIsolate,
5 materializeIsolate,
6 sweepOrphans,
7} from './bundle-materializer.js';
8import { sanitizeErrorMessage } from './error-sanitizer.js';
9import type { LogLine, RuntimeErrorCode } from './isolate-runtime/types.js';
10import { incCounter, observeHistogram } from './metrics.js';
11import { formatDenyNet } from './network-filter.js';
12import type { Bundle, InvokeRequest, InvokeResult } from './types.js';
13
14export type IsolateState = 'spawning' | 'ready' | 'in_flight' | 'retiring' | 'dead';
15
16export interface IsolateEntry {
17 isolateId: string;
18 projectId: string;
19 deploymentId: string;
20 state: IsolateState;
21 pid: number;
22 invocationCount: number;
23 /** ms epoch — last time the isolate finished work or accepted an invoke. */
24 lastActivityAt: number;
25 tmpDir: string;
26 /** ms epoch — when the isolate process was spawned. */
27 createdAt: number;
28 /** Per-(projectId, deploymentId) consecutive crashes; cleared on success. */
29 consecutiveCrashes: number;
30 /** Per-spawn env values used for §5.1 redaction in log/error sanitizers. */
31 envValues: readonly string[];
32}
33
34export type StateEvent =
35 | { kind: 'spawn_ready' }
36 | { kind: 'invoke_start' }
37 | { kind: 'invoke_complete' }
38 | { kind: 'retire' }
39 | { kind: 'crash' }
40 | { kind: 'exit' };
41
42/**
43 * Pure state-transition function. Returns a new entry; does not mutate.
44 *
45 * spawning --spawn_ready--> ready
46 * ready --invoke_start--> in_flight
47 * in_flight --invoke_complete--> ready
48 * * --retire--> retiring
49 * * --crash--> dead
50 * * --exit--> dead
51 *
52 * Invalid transitions are silently no-op (state unchanged) — callers should
53 * not depend on this for safety; check state first.
54 */
55export function transitionState(entry: IsolateEntry, evt: StateEvent): IsolateEntry {
56 let next: IsolateState = entry.state;
57 switch (evt.kind) {
58 case 'spawn_ready':
59 if (entry.state === 'spawning') next = 'ready';
60 break;
61 case 'invoke_start':
62 if (entry.state === 'ready') next = 'in_flight';
63 break;
64 case 'invoke_complete':
65 if (entry.state === 'in_flight') next = 'ready';
66 break;
67 case 'retire':
68 next = 'retiring';
69 break;
70 case 'crash':
71 next = 'dead';
72 break;
73 case 'exit':
74 next = 'dead';
75 break;
76 }
77 return { ...entry, state: next };
78}
79
80export type KillReason =
81 | 'idle'
82 | 'max_invocations'
83 | 'crash'
84 | 'deploy_invalidation'
85 | 'host_cap_evict';
86
87/**
88 * Decide whether a ready isolate should be killed based on activity and
89 * invocation count. Returns null if the isolate is fine, or a reason
90 * string for the metric label.
91 *
92 * Triggers checked here (CLAUDE.md §7.3):
93 * - idle > idleKillMs → 'idle'
94 * - invocationCount ≥ maxInvocations → 'max_invocations'
95 *
96 * Other kill triggers (crash, deploy_invalidation, host_cap_evict) are
97 * decided elsewhere and surfaced through the same KillReason union for
98 * metrics consistency.
99 */
100export function computeKillReason(
101 entry: IsolateEntry,
102 config: { idleKillMs: number; maxInvocations: number },
103 now: number = Date.now(),
104): KillReason | null {
105 if (entry.state !== 'ready') return null;
106 if (entry.invocationCount >= config.maxInvocations) return 'max_invocations';
107 if (now - entry.lastActivityAt >= config.idleKillMs) return 'idle';
108 return null;
109}
110
111/**
112 * The full PoolManager interface. Implementation arrives in Task 10+.
113 * The stub class for `PoolManager` here gives downstream tasks something
114 * to import as a type.
115 */
116export interface PoolManager {
117 invoke(bundle: Bundle, request: InvokeRequest): Promise<InvokeResult>;
118 shutdown(): Promise<void>;
119 startIdleSweeper(): void;
120 describeForMetrics(): {
121 isolatesByState: Record<IsolateState, number>;
122 poolSize: number;
123 };
124}
125
126export interface SpawnedChild {
127 pid: number;
128 stdin: {
129 write: (line: string) => boolean | Promise<boolean | number>;
130 end: () => void;
131 };
132 /** Yields newline-stripped stdout lines; null on EOF. */
133 stdout: { next: () => Promise<string | null> };
134 /** Yields newline-stripped stderr lines; null on EOF. */
135 stderr: { next: () => Promise<string | null> };
136 wait: () => Promise<{ exitCode: number; signal: number | null }>;
137 kill: (signal: string) => void;
138}
139
140export interface SpawnFn {
141 (opts: {
142 args: string[];
143 env: Record<string, string>;
144 cwd: string;
145 }): Promise<SpawnedChild>;
146}
147
148export interface IsolatePoolConfig {
149 spawn: SpawnFn;
150 runtimeStubDir: string;
151 isolateBaseDir: string;
152 maxIsolates: number;
153 maxMemoryMb: number;
154 invocationTimeoutMs: number;
155 idleKillMs: number;
156 maxInvocationsPerIsolate: number;
157 crashLoopThreshold: number;
158 crashLoopWindowMs: number;
159 runQueryProxy: (
160 projectId: string,
161 requestId: string,
162 sql: string,
163 params: readonly unknown[],
164 table: string,
165 ) => Promise<readonly unknown[]>;
166 onLog: (line: LogLine, projectId: string, envValues: readonly string[]) => void;
167 loadProjectEnv: (projectId: string) => Promise<Record<string, string>>;
168 denoPath?: string;
169}
170
171interface CrashHistory {
172 /** ms timestamps of recent crashes */
173 crashes: number[];
174}
175
176type ResultEnvelope =
177 | { type: 'result'; requestId: string; value: unknown; durationMs: number }
178 | { type: 'error'; requestId: string; code: string; message: string; durationMs: number };
179
180/**
181 * Error codes that mean the underlying isolate is dead or its IO channel is
182 * unusable. When a result envelope carries one of these, the entry must NOT
183 * be transitioned back to ready — the next invoke needs a cold start.
184 */
185const CRASH_CODES: ReadonlySet<RuntimeErrorCode> = new Set<RuntimeErrorCode>([
186 'invocation_timeout',
187 'isolate_crashed',
188 'isolate_spawn_timeout',
189 'isolate_protocol_error',
190]);
191
192export class IsolatePoolImpl implements PoolManager {
193 private readonly map = new Map<string, IsolateEntry>();
194 private readonly children = new Map<string, SpawnedChild>();
195 private readonly crashHistory = new Map<string, CrashHistory>();
196 private readonly readyWaiters = new Map<string, Array<() => void>>();
197 private readonly resultResolvers = new Map<string, (msg: ResultEnvelope) => void>();
198 private readonly touchedTablesByRequest = new Map<string, Set<string>>();
199 /** Track which isolateIds have a stdout reader running. */
200 private readonly stdoutReaders = new Set<string>();
201 /**
202 * Per-projectId in-flight cold-start promise. Two concurrent invokes for
203 * the same projectId await the same spawn promise rather than each
204 * spawning a new isolate. Per CLAUDE.md §5.2: at most one live isolate
205 * per project.
206 */
207 private readonly pendingSpawns = new Map<string, Promise<IsolateEntry>>();
208 private shutdownInProgress = false;
209 private idleSweeperHandle: ReturnType<typeof setInterval> | null = null;
210
211 constructor(private readonly config: IsolatePoolConfig) {}
212
213 async invoke(bundle: Bundle, request: InvokeRequest): Promise<InvokeResult> {
214 if (this.shutdownInProgress) {
215 return errResult('host_overloaded', 'runtime shutting down');
216 }
217 if (!bundle.functionNames.includes(request.functionName)) {
218 return errResult('function_not_found', `function '${request.functionName}' not found`);
219 }
220
221 const breakerKey = `${request.projectId}:${request.deploymentId}`;
222 if (this.isCrashLoopBroken(breakerKey)) {
223 return errResult(
224 'deployment_unhealthy',
225 'deployment marked unhealthy after repeated crashes',
226 );
227 }
228
229 let entry = this.map.get(request.projectId);
230
231 // Deploy invalidation: deployment changed → retire and respawn.
232 if (entry && entry.deploymentId !== request.deploymentId) {
233 await this.retireAndAwait(entry, 'deploy_invalidation');
234 entry = undefined;
235 }
236
237 // Single-flight queueing: if an entry exists but is in-flight, wait
238 // for it to return to ready before piggybacking the next invoke.
239 if (entry && entry.state === 'in_flight') {
240 await this.waitForReady(request.projectId);
241 entry = this.map.get(request.projectId);
242 if (!entry || entry.deploymentId !== request.deploymentId) {
243 // Mid-wait the entry was retired or deploy-invalidated; fall
244 // back through the full flow.
245 return this.invoke(bundle, request);
246 }
247 }
248
249 if (!entry) {
250 // C1 fix: gate concurrent cold-starts on a per-project promise so
251 // two parallel invokes don't both spawn. Per CLAUDE.md §5.2.
252 let spawnPromise = this.pendingSpawns.get(request.projectId);
253 const isInitiator = !spawnPromise;
254 if (!spawnPromise) {
255 // Host-cap eviction: if at maxIsolates, retire the oldest idle
256 // entry before spawning a fresh one. If no idle entry exists the
257 // invoke falls through and runs (we let downstream metrics/alerts
258 // catch sustained over-cap rather than fail hard here).
259 if (this.map.size >= this.config.maxIsolates) {
260 await this.evictOldestIdle();
261 }
262 spawnPromise = this.spawnIsolate(bundle).finally(() => {
263 this.pendingSpawns.delete(request.projectId);
264 });
265 this.pendingSpawns.set(request.projectId, spawnPromise);
266 }
267 try {
268 entry = await spawnPromise;
269 } catch (err) {
270 // `spawnIsolate` already emitted the `timeout` counter; everything
271 // else falls into `spawn_error` (materialization failure, exec
272 // not-found, OOM during ready handshake, …).
273 const tagged = err as Error & { spawnOutcome?: string };
274 if (tagged.spawnOutcome !== 'timeout') {
275 incCounter('briven_runtime_isolate_spawns_total', { outcome: 'spawn_error' });
276 }
277 this.recordCrash(breakerKey);
278 return errResult(
279 'isolate_spawn_timeout',
280 err instanceof Error ? err.message : 'spawn failed',
281 );
282 }
283 // Only the spawn-initiating caller writes the entry + starts the
284 // readers; concurrent awaiters observe the entry already in the map.
285 if (isInitiator && !this.map.has(request.projectId)) {
286 this.map.set(request.projectId, entry);
287 this.startStdoutReader(entry);
288 this.startStderrReader(entry);
289 this.watchForExit(entry);
290 }
291 }
292
293 return this.runInvoke(entry, request);
294 }
295
296 private async spawnIsolate(bundle: Bundle): Promise<IsolateEntry> {
297 const spawnStart = performance.now();
298 const isolateId = newId('iso');
299 const mat = await materializeIsolate(isolateId, bundle, {
300 isolateBaseDir: this.config.isolateBaseDir,
301 runtimeStubDir: this.config.runtimeStubDir,
302 });
303
304 const userEnv = await this.config.loadProjectEnv(bundle.projectId);
305 const allowEnvKeys = Object.keys(userEnv).join(',');
306 // Bind env values now so log/error sanitizers can redact them per §5.1.
307 const envValues: readonly string[] = Object.values(userEnv).filter(
308 (v): v is string => typeof v === 'string',
309 );
310
311 const args: string[] = [
312 'run',
313 '--no-prompt',
314 '--no-remote',
315 '--allow-net',
316 `--deny-net=${formatDenyNet()}`,
317 `--allow-read=${mat.tmpDir}`,
318 `--allow-write=${mat.tmpDir}`,
319 `--v8-flags=--max-old-space-size=${this.config.maxMemoryMb}`,
320 `--import-map=${mat.importMapPath}`,
321 ];
322 if (allowEnvKeys) args.splice(args.indexOf('--allow-net'), 0, `--allow-env=${allowEnvKeys}`);
323 args.push(mat.entryPath);
324
325 const env: Record<string, string> = {
326 ...(process.env as Record<string, string>),
327 ...userEnv,
328 };
329
330 const child = await this.config.spawn({ args, env, cwd: mat.tmpDir });
331
332 // Wait for ready handshake — 500ms timeout (twice the cold-start budget).
333 const readyLine = await Promise.race([
334 child.stdout.next(),
335 new Promise<null>((resolve) => setTimeout(() => resolve(null), 500)),
336 ]);
337 if (!readyLine) {
338 // Drain a few stderr lines so the host log surfaces Deno's actual error
339 // (e.g. flag-parse failure) rather than a generic timeout.
340 const stderrLines: string[] = [];
341 for (let i = 0; i < 5; i++) {
342 const line = await Promise.race([
343 child.stderr.next(),
344 new Promise<null>((resolve) => setTimeout(() => resolve(null), 50)),
345 ]);
346 if (line === null) break;
347 stderrLines.push(line);
348 }
349 child.kill('SIGTERM');
350 await cleanupIsolate(mat.tmpDir);
351 incCounter('briven_runtime_isolate_spawns_total', { outcome: 'timeout' });
352 // Tag the error so the caller in `invoke()` can attribute it to
353 // 'timeout' rather than the generic 'spawn_error' bucket.
354 const stderrTail = stderrLines.length > 0 ? `; stderr: ${stderrLines.join(' | ')}` : '';
355 const err = new Error(`isolate did not emit ready within 500ms${stderrTail}`);
356 (err as Error & { spawnOutcome?: string }).spawnOutcome = 'timeout';
357 throw err;
358 }
359 let parsed: { type?: string; deploymentId?: string };
360 try {
361 parsed = JSON.parse(readyLine);
362 } catch {
363 child.kill('SIGKILL');
364 await cleanupIsolate(mat.tmpDir);
365 throw new Error('isolate ready handshake malformed');
366 }
367 if (parsed.type !== 'ready' || parsed.deploymentId !== bundle.deploymentId) {
368 child.kill('SIGKILL');
369 await cleanupIsolate(mat.tmpDir);
370 throw new Error(
371 `isolate ready handshake mismatch: got ${JSON.stringify(parsed)}`,
372 );
373 }
374 const entry: IsolateEntry = {
375 isolateId,
376 projectId: bundle.projectId,
377 deploymentId: bundle.deploymentId,
378 state: 'ready',
379 pid: child.pid,
380 invocationCount: 0,
381 lastActivityAt: Date.now(),
382 tmpDir: mat.tmpDir,
383 createdAt: Date.now(),
384 consecutiveCrashes: 0,
385 envValues,
386 };
387 this.children.set(isolateId, child);
388 incCounter('briven_runtime_isolate_spawns_total', { outcome: 'success' });
389 observeHistogram(
390 'briven_runtime_cold_start_ms',
391 Math.round(performance.now() - spawnStart),
392 );
393 return entry;
394 }
395
396 /**
397 * Reads stdout lines from the child in a loop. Dispatches:
398 * - `query` → forward to `runQueryProxy`, send `query_result` back.
399 * - `result`/`error`→ resolve the matching `resultResolvers` entry.
400 *
401 * On JSON parse failure we SIGKILL the child (the protocol is broken)
402 * and stop reading. The reader is started after a successful spawn
403 * and runs until the child closes stdout.
404 */
405 private startStdoutReader(entry: IsolateEntry): void {
406 if (this.stdoutReaders.has(entry.isolateId)) return;
407 this.stdoutReaders.add(entry.isolateId);
408 const child = this.children.get(entry.isolateId);
409 if (!child) return;
410
411 const projectId = entry.projectId;
412 const isolateId = entry.isolateId;
413
414 void (async () => {
415 try {
416 while (true) {
417 const line = await child.stdout.next();
418 if (line === null) return;
419 if (!line.trim()) continue;
420 let msg: { type?: string; [k: string]: unknown };
421 try {
422 msg = JSON.parse(line);
423 } catch {
424 // Protocol break — drain any pending resolvers so callers don't
425 // wait the full invocation timeout for a result that will never
426 // arrive once we kill the child below.
427 this.drainResolversWith(
428 'isolate_protocol_error',
429 'malformed output from isolate',
430 );
431 child.kill('SIGKILL');
432 return;
433 }
434 if (msg.type === 'query') {
435 const requestId = msg.requestId as string;
436 const qid = msg.qid as string;
437 const sql = msg.sql as string;
438 const params = (msg.params as readonly unknown[]) ?? [];
439 const table = msg.table as string;
440 const tables = this.touchedTablesByRequest.get(requestId);
441 if (tables) tables.add(table);
442 let writePayload: string;
443 try {
444 const rows = await this.config.runQueryProxy(
445 projectId,
446 requestId,
447 sql,
448 params,
449 table,
450 );
451 writePayload = `${JSON.stringify({ type: 'query_result', requestId, qid, rows })}\n`;
452 } catch (err) {
453 const message = err instanceof Error ? err.message : String(err);
454 writePayload = `${JSON.stringify({
455 type: 'query_result',
456 requestId,
457 qid,
458 error: { code: 'function_threw', message },
459 })}\n`;
460 }
461 try {
462 await child.stdin.write(writePayload);
463 } catch (err) {
464 // Stdin closed mid-query — child is dead. Drain pending
465 // resolvers so callers see the truthful error code rather
466 // than waiting for a 30s timeout.
467 const message = err instanceof Error ? err.message : 'isolate stdin closed';
468 this.drainResolversWith('isolate_crashed', message);
469 try {
470 child.kill('SIGKILL');
471 } catch {
472 /* ignore */
473 }
474 return;
475 }
476 } else if (msg.type === 'result' || msg.type === 'error') {
477 const requestId = msg.requestId as string;
478 const resolver = this.resultResolvers.get(requestId);
479 if (resolver) {
480 this.resultResolvers.delete(requestId);
481 resolver(msg as unknown as ResultEnvelope);
482 }
483 }
484 // Anything else (logs, unknown types) → ignore at this layer.
485 }
486 } finally {
487 this.stdoutReaders.delete(isolateId);
488 }
489 })();
490 }
491
492 /**
493 * Reads stderr lines from the child in a loop. Each line is parsed as
494 * JSON and, if it shapes as a `LogLine` envelope, routed to
495 * `config.onLog`. Anything that doesn't parse — or doesn't match the
496 * envelope — is treated as a raw stderr panic (Deno crash trace,
497 * unhandled rejection trail, etc.) and surfaced as an error-level
498 * log so ops still sees it.
499 *
500 * The reader exits when stderr returns null (EOF). It does not kill
501 * the child on parse errors — the protocol is stdout-only; stderr is
502 * advisory.
503 */
504 private startStderrReader(entry: IsolateEntry): void {
505 const child = this.children.get(entry.isolateId);
506 if (!child) return;
507 const projectId = entry.projectId;
508 const envValues = entry.envValues;
509 void (async () => {
510 while (true) {
511 const line = await child.stderr.next();
512 if (line === null) return;
513 if (!line.trim()) continue;
514 try {
515 const parsed = JSON.parse(line) as { type?: string };
516 if (parsed && parsed.type === 'log') {
517 this.config.onLog(parsed as LogLine, projectId, envValues);
518 continue;
519 }
520 } catch {
521 /* fallthrough to raw-line capture */
522 }
523 this.config.onLog(
524 {
525 type: 'log',
526 requestId: null,
527 level: 'error',
528 msg: line,
529 ts: Date.now(),
530 },
531 projectId,
532 envValues,
533 );
534 }
535 })();
536 }
537
538 /**
539 * Find the entry in `ready` state with the oldest `lastActivityAt`
540 * and retire it. Used when a cold-start would push us past
541 * `maxIsolates`. If no ready entry exists (every isolate is busy),
542 * returns without doing anything — the caller proceeds with the
543 * spawn anyway and pool size briefly exceeds the cap.
544 */
545 private async evictOldestIdle(): Promise<void> {
546 let oldest: IsolateEntry | null = null;
547 for (const e of this.map.values()) {
548 if (e.state !== 'ready') continue;
549 if (!oldest || e.lastActivityAt < oldest.lastActivityAt) oldest = e;
550 }
551 if (!oldest) return;
552 await this.retireAndAwait(oldest, 'host_cap_evict');
553 }
554
555 private async runInvoke(entry: IsolateEntry, request: InvokeRequest): Promise<InvokeResult> {
556 const child = this.children.get(entry.isolateId);
557 if (!child) {
558 const r = errResult('isolate_crashed', 'isolate child handle missing');
559 this.emitInvocationMetrics(r);
560 return r;
561 }
562
563 // 1. Transition to in_flight.
564 let updated = transitionState(entry, { kind: 'invoke_start' });
565 this.map.set(request.projectId, updated);
566
567 // 2. Track touched tables for this requestId.
568 this.touchedTablesByRequest.set(request.requestId, new Set<string>());
569
570 // 3+4. Promise + timeout.
571 const started = performance.now();
572 let timeoutHandle: ReturnType<typeof setTimeout> | null = null;
573 const envelope = await new Promise<ResultEnvelope>((resolve) => {
574 this.resultResolvers.set(request.requestId, resolve);
575 timeoutHandle = setTimeout(() => {
576 if (!this.resultResolvers.has(request.requestId)) return;
577 this.resultResolvers.delete(request.requestId);
578 try {
579 child.kill('SIGKILL');
580 } catch {
581 /* ignore */
582 }
583 resolve({
584 type: 'error',
585 requestId: request.requestId,
586 code: 'invocation_timeout',
587 message: `invocation exceeded ${this.config.invocationTimeoutMs}ms`,
588 durationMs: Math.round(performance.now() - started),
589 });
590 }, this.config.invocationTimeoutMs);
591
592 // 5. Write invoke frame.
593 const frame: { type: 'invoke' } & Omit<InvokeRequest, 'projectId' | 'deploymentId' | 'env'> = {
594 type: 'invoke',
595 requestId: request.requestId,
596 functionName: request.functionName,
597 args: request.args,
598 auth: request.auth,
599 };
600 void Promise.resolve(child.stdin.write(`${JSON.stringify(frame)}\n`)).catch(() => {
601 if (!this.resultResolvers.has(request.requestId)) return;
602 this.resultResolvers.delete(request.requestId);
603 resolve({
604 type: 'error',
605 requestId: request.requestId,
606 code: 'isolate_crashed',
607 message: 'failed to write invoke frame to isolate stdin',
608 durationMs: Math.round(performance.now() - started),
609 });
610 });
611 });
612 if (timeoutHandle) clearTimeout(timeoutHandle);
613
614 // 7. Build the InvokeResult.
615 const touched = this.touchedTablesByRequest.get(request.requestId) ?? new Set<string>();
616 this.touchedTablesByRequest.delete(request.requestId);
617 const touchedTables = [...touched];
618
619 let result: InvokeResult;
620 if (envelope.type === 'result') {
621 result = {
622 ok: true,
623 value: envelope.value,
624 durationMs: envelope.durationMs,
625 touchedTables,
626 };
627 } else {
628 // §5.1: every customer-visible error message that flows through the
629 // host passes through the sanitizer with the isolate's bound env
630 // values so a panic echoing a secret can't leak it.
631 result = {
632 ok: false,
633 code: envelope.code as RuntimeErrorCode,
634 message: sanitizeErrorMessage(envelope.message, entry.envValues),
635 durationMs: envelope.durationMs,
636 touchedTables,
637 };
638 }
639
640 // 8. Transition back to ready — but only if the isolate is still
641 // healthy. Crash-coded errors mean the child is dead (or its IO
642 // channel is unusable); putting that entry back into ready state
643 // would trick the next invoke into writing to a dead pipe.
644 const after = this.map.get(request.projectId);
645 const isCrashCoded = !result.ok && CRASH_CODES.has(result.code);
646 if (after && after.state === 'in_flight') {
647 if (isCrashCoded) {
648 // Don't put a dead isolate back into ready state. Mark it gone
649 // and remove it from the map so the next invoke cold-starts.
650 this.map.delete(request.projectId);
651 this.children.delete(after.isolateId);
652 // Drain ready waiters so any queued invoke gets a chance to spawn fresh.
653 const waiters = this.readyWaiters.get(request.projectId) ?? [];
654 this.readyWaiters.delete(request.projectId);
655 for (const w of waiters) w();
656 // Best-effort tmp dir cleanup; don't await on the hot path.
657 void cleanupIsolate(after.tmpDir).catch(() => {});
658 } else {
659 updated = transitionState(after, { kind: 'invoke_complete' });
660 updated = {
661 ...updated,
662 invocationCount: updated.invocationCount + 1,
663 lastActivityAt: Date.now(),
664 };
665 this.map.set(request.projectId, updated);
666
667 // Drain ready waiters for this project.
668 const waiters = this.readyWaiters.get(request.projectId);
669 if (waiters && waiters.length > 0) {
670 this.readyWaiters.delete(request.projectId);
671 for (const w of waiters) w();
672 }
673
674 // Max-invocations retire.
675 if (updated.invocationCount >= this.config.maxInvocationsPerIsolate) {
676 // Fire-and-forget — the next invoke for this projectId will spawn fresh.
677 void this.retireAndAwait(updated, 'max_invocations');
678 }
679 }
680 }
681
682 // 11. Crash bookkeeping.
683 const breakerKey = `${request.projectId}:${request.deploymentId}`;
684 if (
685 !result.ok &&
686 (result.code === 'isolate_crashed' || result.code === 'isolate_spawn_timeout')
687 ) {
688 this.recordCrash(breakerKey);
689 } else if (result.ok) {
690 this.crashHistory.delete(breakerKey);
691 }
692
693 this.emitInvocationMetrics(result);
694 return result;
695 }
696
697 /**
698 * Single emit point for `briven_runtime_invocations_total` and
699 * `briven_runtime_invocation_duration_ms`. Centralized so every
700 * `runInvoke` exit goes through the same pair, regardless of whether
701 * the failure was ours (timeout, crash) or the user's (function_threw).
702 */
703 private emitInvocationMetrics(result: InvokeResult): void {
704 const code = result.ok ? 'success' : result.code;
705 incCounter('briven_runtime_invocations_total', { code });
706 observeHistogram('briven_runtime_invocation_duration_ms', result.durationMs);
707 }
708
709 /**
710 * Reject every pending in-flight invocation with the given error code +
711 * message. Used whenever the protocol breaks (malformed stdout, stdin
712 * write failure, shutdown) — without this, callers wait the full
713 * `invocationTimeoutMs` for a result that will never arrive.
714 */
715 private drainResolversWith(code: RuntimeErrorCode, message: string): void {
716 for (const [reqId, resolve] of this.resultResolvers) {
717 resolve({
718 type: 'error',
719 requestId: reqId,
720 code,
721 message,
722 durationMs: 0,
723 });
724 }
725 this.resultResolvers.clear();
726 }
727
728 private recordCrash(key: string): void {
729 const now = Date.now();
730 const hist = this.crashHistory.get(key) ?? { crashes: [] };
731 const wasBroken = hist.crashes.length >= this.config.crashLoopThreshold;
732 hist.crashes.push(now);
733 // Trim entries outside the window so the array doesn't grow unbounded.
734 const cutoff = now - this.config.crashLoopWindowMs;
735 hist.crashes = hist.crashes.filter((t) => t >= cutoff);
736 this.crashHistory.set(key, hist);
737 const isBroken = hist.crashes.length >= this.config.crashLoopThreshold;
738 // Edge-trigger: only emit on the (threshold-1)→threshold transition,
739 // not every subsequent crash inside the same broken window. The
740 // history naturally resets once `crashLoopWindowMs` passes, after
741 // which the next crash that re-trips the breaker fires this again.
742 if (!wasBroken && isBroken) {
743 incCounter('briven_runtime_crash_loop_breaks_total');
744 }
745 }
746
747 /**
748 * Returns true when the (projectId, deploymentId) key has crashed at
749 * least `crashLoopThreshold` times within `crashLoopWindowMs`. Trims
750 * stale entries on read so deployments self-heal once the window
751 * passes without further crashes.
752 */
753 private isCrashLoopBroken(key: string): boolean {
754 const hist = this.crashHistory.get(key);
755 if (!hist) return false;
756 const now = Date.now();
757 hist.crashes = hist.crashes.filter((t) => now - t < this.config.crashLoopWindowMs);
758 if (hist.crashes.length === 0) {
759 this.crashHistory.delete(key);
760 return false;
761 }
762 return hist.crashes.length >= this.config.crashLoopThreshold;
763 }
764
765 /**
766 * SIGCHLD-equivalent: awaits the child process exit and reconciles
767 * pool state. If the entry is still the live one for its project,
768 * transitions to crash/exit (both → dead), drains pending resolvers,
769 * and removes the entry from the maps. Tmp dir cleanup is fire-and-
770 * forget — failure to unlink shouldn't block the next spawn.
771 */
772 private watchForExit(entry: IsolateEntry): void {
773 const child = this.children.get(entry.isolateId);
774 if (!child) return;
775 void child.wait().then(({ exitCode, signal }) => {
776 const current = this.map.get(entry.projectId);
777 if (!current || current.isolateId !== entry.isolateId) return;
778 const updated = transitionState(
779 current,
780 signal != null || exitCode !== 0 ? { kind: 'crash' } : { kind: 'exit' },
781 );
782 this.map.set(entry.projectId, updated);
783 if (updated.state === 'dead') {
784 // `watchForExit` is the only path that reaches the `dead` state
785 // outside `retireAndAwait`, so we emit the kill counter here
786 // rather than threading a reason through every transition.
787 // Both crash (non-zero exit / signal) and clean exit count as
788 // 'crash' for ops purposes — a clean exit from a long-lived
789 // isolate is unexpected and worth alerting on.
790 incCounter('briven_runtime_isolate_kills_total', { reason: 'crash' });
791 this.children.delete(entry.isolateId);
792 this.map.delete(entry.projectId);
793 this.drainResolversWith(
794 'isolate_crashed',
795 `isolate exited (code=${exitCode}, signal=${signal})`,
796 );
797 void cleanupIsolate(entry.tmpDir);
798 }
799 });
800 }
801
802 /**
803 * @internal
804 * Test-only — runs one tick of the idle sweeper synchronously. Mirrors
805 * the body of the interval started by `startIdleSweeper`. Production
806 * code uses the timed interval; tests use this hook to deterministically
807 * exercise the idle-kill path without waiting for the 30s tick.
808 */
809 public async triggerIdleCheck(): Promise<void> {
810 if (this.shutdownInProgress) return;
811 const now = Date.now();
812 for (const entry of [...this.map.values()]) {
813 const reason = computeKillReason(
814 entry,
815 {
816 idleKillMs: this.config.idleKillMs,
817 maxInvocations: this.config.maxInvocationsPerIsolate,
818 },
819 now,
820 );
821 if (reason) await this.retireAndAwait(entry, reason);
822 }
823 }
824
825 /**
826 * Schedule a 30s sweep that retires any ready entry whose lastActivity
827 * exceeded `idleKillMs` or whose invocationCount hit
828 * `maxInvocationsPerIsolate`. Idempotent — calling twice is a no-op.
829 * Cleared in `shutdown()`.
830 */
831 startIdleSweeper(): void {
832 if (this.idleSweeperHandle) return;
833 this.idleSweeperHandle = setInterval(() => {
834 if (this.shutdownInProgress) return;
835 const now = Date.now();
836 for (const entry of [...this.map.values()]) {
837 const reason = computeKillReason(
838 entry,
839 {
840 idleKillMs: this.config.idleKillMs,
841 maxInvocations: this.config.maxInvocationsPerIsolate,
842 },
843 now,
844 );
845 if (reason) void this.retireAndAwait(entry, reason);
846 }
847 }, 30_000);
848 }
849
850 /**
851 * Single-flight queue: returns a promise that resolves once the
852 * in-flight invoke for `projectId` finishes and the entry is back
853 * to ready (or the entry is removed).
854 */
855 private waitForReady(projectId: string): Promise<void> {
856 return new Promise<void>((resolve) => {
857 const list = this.readyWaiters.get(projectId) ?? [];
858 list.push(resolve);
859 this.readyWaiters.set(projectId, list);
860 });
861 }
862
863 /**
864 * Graceful retire: send shutdown frame, escalate to SIGTERM at 5s and
865 * SIGKILL at 7s, then clean up the tmp dir. Removes the entry from
866 * the pool map so the next invoke spawns fresh.
867 *
868 * `reason` is recorded as a label on `briven_runtime_isolate_kills_total`
869 * — pass the trigger (idle sweeper, max-invocations, deploy switch,
870 * host-cap eviction). Crash kills go through `watchForExit` and emit
871 * separately, since that path doesn't reach this method.
872 */
873 private async retireAndAwait(
874 entry: IsolateEntry,
875 reason: KillReason = 'idle',
876 ): Promise<void> {
877 incCounter('briven_runtime_isolate_kills_total', { reason });
878 const retired = transitionState(entry, { kind: 'retire' });
879 this.map.set(entry.projectId, retired);
880 const child = this.children.get(entry.isolateId);
881 if (child) {
882 try {
883 await child.stdin.write('{"type":"shutdown"}\n');
884 } catch {
885 /* ignore */
886 }
887 const sigtermHandle = setTimeout(() => {
888 try {
889 child.kill('SIGTERM');
890 } catch {
891 /* ignore */
892 }
893 }, 5000);
894 const sigkillHandle = setTimeout(() => {
895 try {
896 child.kill('SIGKILL');
897 } catch {
898 /* ignore */
899 }
900 }, 7000);
901 // Don't keep the event loop alive on these timers in tests.
902 if (typeof sigtermHandle === 'object' && sigtermHandle && 'unref' in sigtermHandle) {
903 (sigtermHandle as { unref: () => void }).unref();
904 }
905 if (typeof sigkillHandle === 'object' && sigkillHandle && 'unref' in sigkillHandle) {
906 (sigkillHandle as { unref: () => void }).unref();
907 }
908 }
909 await cleanupIsolate(entry.tmpDir);
910 // Only remove if still pointing at this entry (a re-spawn may have
911 // already replaced it under our feet).
912 const current = this.map.get(entry.projectId);
913 if (current && current.isolateId === entry.isolateId) {
914 this.map.delete(entry.projectId);
915 }
916 this.children.delete(entry.isolateId);
917 }
918
919 describeForMetrics() {
920 const isolatesByState: Record<IsolateState, number> = {
921 spawning: 0,
922 ready: 0,
923 in_flight: 0,
924 retiring: 0,
925 dead: 0,
926 };
927 for (const e of this.map.values()) isolatesByState[e.state]++;
928 return { isolatesByState, poolSize: this.map.size };
929 }
930
931 async shutdown(): Promise<void> {
932 this.shutdownInProgress = true;
933 if (this.idleSweeperHandle) {
934 clearInterval(this.idleSweeperHandle);
935 this.idleSweeperHandle = null;
936 }
937 // Drain in-flight resolvers FIRST so callers don't hang waiting for
938 // results that will never arrive once we kill the children below.
939 this.drainResolversWith('isolate_crashed', 'runtime shutting down');
940 for (const child of this.children.values()) {
941 try { await child.stdin.write('{"type":"shutdown"}\n'); } catch { /* ignore */ }
942 setTimeout(() => child.kill('SIGTERM'), 5000);
943 setTimeout(() => child.kill('SIGKILL'), 7000);
944 }
945 this.crashHistory.clear();
946 await sweepOrphans(this.config.isolateBaseDir, new Set());
947 }
948}
949
950function errResult(code: RuntimeErrorCode, message: string): InvokeResult {
951 return { ok: false, code, message, durationMs: 0 };
952}