From 5c4b8ce12f6e65ce7a382e945cd97362aba4d81a Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 27 Sep 2026 20:20:32 -0500 Subject: [PATCH] fix(runtime): isolate journal proof work in a bounded worker Use one off-thread proof reader with a bounded queue, monotonic deadlines, heap limits and joined termination. Preserve raw hashes and canonical base64 fingerprints, and prove event-loop responsiveness on a 190 MiB journal. Co-Authored-By: Paperclip --- .../native-journal-projection-async.test.ts | 149 ++++++++++++++ .../native-journal-projection-async.ts | 184 ++++++++++++++++++ .../native-journal-projection-worker.ts | 47 +++++ 3 files changed, 380 insertions(+) create mode 100644 server/src/services/native-runtime/native-journal-projection-async.test.ts create mode 100644 server/src/services/native-runtime/native-journal-projection-async.ts create mode 100644 server/src/services/native-runtime/native-journal-projection-worker.ts diff --git a/server/src/services/native-runtime/native-journal-projection-async.test.ts b/server/src/services/native-runtime/native-journal-projection-async.test.ts new file mode 100644 index 00000000000..3ebc6c48db6 --- /dev/null +++ b/server/src/services/native-runtime/native-journal-projection-async.test.ts @@ -0,0 +1,149 @@ +import { createHash } from "node:crypto"; +import { + closeSync, + mkdirSync, + mkdtempSync, + openSync, + rmSync, + writeFileSync, + writeSync, +} from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { performance } from "node:perf_hooks"; +import { afterEach, describe, expect, it } from "vitest"; +import { + createNativeJournalProofReader, + nativeStateBase64Fingerprint, + nativeStateFileIncludes, + readNativeJournalProjection, + scanNativeStateFile, +} from "./native-journal-projection-async.js"; +import { runnerdStateProvesIncompleteBootstrap } from "./native-session-executor.js"; + +const roots: string[] = []; +afterEach(() => { + for (const root of roots.splice(0)) + rmSync(root, { recursive: true, force: true }); +}); +function fixture(historyMiB = 0) { + const root = mkdtempSync(join(tmpdir(), "journal-responsive-")); + roots.push(root); + mkdirSync(join(root, "control-plane")); + const path = join(root, "control-plane/control-plane-state.json"); + const fd = openSync(path, "w"); + writeSync( + fd, + '{"schema":"paperclip.runner.durable.control-plane-state.v1","identity":{"runId":"bounded"},"connectionCount":0,"commands":[],"committedEvents":[{"envelope":{"payload":{"eventType":"item.completed","payload":{"text":"', + ); + const chunk = Buffer.alloc(1024 * 1024, 120); + for (let i = 0; i < historyMiB; i++) writeSync(fd, chunk); + writeSync(fd, '"}}}}]}'); + closeSync(fd); + return { root, path }; +} +describe("off-thread durable journal proofs", () => { + it("keeps the event loop responsive during near-limit controller admission", async () => { + const { root, path } = fixture(190); + let ticks = 0, + largestGap = 0, + last = performance.now(); + const pulse = setInterval(() => { + const now = performance.now(); + largestGap = Math.max(largestGap, now - last); + last = now; + ticks++; + }, 5); + const started = performance.now(); + try { + const [proof, bootstrap] = await Promise.all([ + readNativeJournalProjection(path), + runnerdStateProvesIncompleteBootstrap(root), + ]); + expect(proof.byteSize).toBeGreaterThan(190 * 1024 * 1024); + expect(proof.value).toMatchObject({ identity: { runId: "bounded" } }); + expect(bootstrap).toBe(false); + expect(ticks).toBeGreaterThan(0); + expect(largestGap).toBeLessThan(500); + expect(performance.now() - started).toBeLessThan(30_000); + } finally { + clearInterval(pulse); + } + }, 40_000); + it("preserves complete raw hashes and the existing base64 fingerprint", async () => { + const { root, path } = fixture(3); + const second = join(root, "other.json"), + bytes = Buffer.from('{"value":"second"}'); + writeFileSync(second, bytes); + const projection = await readNativeJournalProjection(path); + expect(projection.value).toMatchObject({ identity: { runId: "bounded" } }); + const file = await scanNativeStateFile(path, 4 * 1024 * 1024); + expect(file.sha256).toBe(projection.sha256); + expect(file.byteSize).toBe(projection.byteSize); + const fingerprint = await nativeStateBase64Fingerprint([ + { path: second, maximum: 100 }, + { path: second, maximum: 100 }, + ]); + expect(fingerprint.sha256).toBe( + createHash("sha256") + .update( + JSON.stringify([bytes.toString("base64"), bytes.toString("base64")]), + ) + .digest("hex"), + ); + expect(fingerprint.fileSha256).toEqual( + Array(2).fill(createHash("sha256").update(bytes).digest("hex")), + ); + expect( + await nativeStateFileIncludes(path, 4 * 1024 * 1024, [ + '"runId":"bounded"', + ]), + ).toBe(true); + expect( + await nativeStateFileIncludes(path, 4 * 1024 * 1024, ["absent-marker"]), + ).toBe(false); + }); + it("bounds queued work and terminates timed-out workers before closing", async () => { + const { path } = fixture(190); + const reader = createNativeJournalProofReader({ + timeoutMs: 20, + maxQueue: 1, + }); + const job = { + kind: "projection" as const, + path, + purpose: "evidence" as const, + }; + const results = Promise.allSettled([ + reader.read(job), + reader.read(job), + reader.read(job), + ]); + try { + expect(await results).toMatchObject([ + { + status: "rejected", + reason: { message: "native_journal_worker_timeout" }, + }, + { + status: "rejected", + reason: { message: "native_journal_worker_timeout" }, + }, + { + status: "rejected", + reason: { message: "native_journal_worker_busy" }, + }, + ]); + } finally { + await reader.close(); + } + await expect(reader.read(job)).rejects.toThrow( + "native_journal_worker_busy", + ); + // An independent ordinary proof still succeeds after worker termination. + const ordinary = fixture(); + expect( + (await readNativeJournalProjection(ordinary.path)).value, + ).toMatchObject({ identity: { runId: "bounded" } }); + }); +}); diff --git a/server/src/services/native-runtime/native-journal-projection-async.ts b/server/src/services/native-runtime/native-journal-projection-async.ts new file mode 100644 index 00000000000..5091e362ace --- /dev/null +++ b/server/src/services/native-runtime/native-journal-projection-async.ts @@ -0,0 +1,184 @@ +import { Worker } from "node:worker_threads"; +import { performance } from "node:perf_hooks"; +import type { readNativeJournalProjection as readProjection } from "./native-journal-projection.js"; + +type File = { path: string; maximum: number }; +export type NativeJournalProofJob = + | { kind: "projection"; path: string; purpose: "identity" | "evidence" } + | ({ kind: "scan" } & File) + | ({ kind: "includes"; needles: string[] } & File) + | { kind: "fingerprint"; files: File[] }; +type Result = + | ReturnType + | { sha256: string; byteSize: number } + | { sha256: string; fileSha256: string[] } + | boolean; +type Pending = { + job: NativeJournalProofJob; + resolve: (result: Result) => void; + reject: (error: Error) => void; + deadline: number; +}; + +/** One worker and a bounded queue cap CPU and memory independently of callers. + * The deadline includes queue time; termination completes before another job starts. */ +export function createNativeJournalProofReader( + options: { timeoutMs?: number; maxQueue?: number } = {}, +) { + const timeoutMs = options.timeoutMs ?? 30_000; + const maxQueue = options.maxQueue ?? 32; + const queue: Pending[] = []; + let active: Pending | undefined; + let worker: Worker | undefined; + let closed = false; + let timer: ReturnType | undefined; + let idleTimer: ReturnType | undefined; + let stopping: Promise | undefined; + function spawn() { + const source = new URL(import.meta.url).pathname.endsWith(".ts"); + const url = new URL( + source + ? "./native-journal-projection-worker.ts" + : "./native-journal-projection-worker.js", + import.meta.url, + ); + const resourceLimits = { + maxOldGenerationSizeMb: 128, + maxYoungGenerationSizeMb: 16, + stackSizeMb: 4, + }; + const next = source + ? new Worker( + `import(${JSON.stringify(import.meta.resolve("tsx/esm/api"))}).then(({tsImport}) => tsImport(${JSON.stringify(url.href)}, ${JSON.stringify(import.meta.url)}));`, + { eval: true, resourceLimits }, + ) + : new Worker(url, { resourceLimits }); + next.on("message", (message: { result?: Result; error?: string }) => { + if (worker !== next || !active) return; + if (message.error) void finish(new Error(message.error)); + else void finish(undefined, message.result); + }); + next.on("error", () => { + if (worker === next) + void finish(new Error("native_journal_worker_failed")); + }); + next.on("exit", () => { + if (worker === next) + void finish(new Error("native_journal_worker_exited")); + }); + return next; + } + async function stop() { + clearTimeout(idleTimer); + const previous = worker; + worker = undefined; + if (previous) { + stopping = previous.terminate(); + await stopping; + stopping = undefined; + } + } + async function finish(error?: Error, result?: Result) { + const pending = active; + if (!pending) { + if (error) { + await stop(); + drain(); + } + return; + } + active = undefined; + clearTimeout(timer); + if (error) { + // Keep the slot occupied until the old thread has actually exited. + await stop(); + pending.reject(error); + } else { + pending.resolve(result!); + worker?.unref(); + idleTimer = setTimeout(() => { + if (!active) void stop().then(drain); + }, 15_000); + idleTimer.unref(); + } + drain(); + } + function drain() { + if (closed || active || stopping) return; + while (queue.length) { + const pending = queue.shift()!; + const remaining = pending.deadline - performance.now(); + if (remaining <= 0) { + pending.reject(new Error("native_journal_worker_timeout")); + continue; + } + clearTimeout(idleTimer); + active = pending; + try { + worker ??= spawn(); + worker.ref(); + timer = setTimeout(() => { + void finish(new Error("native_journal_worker_timeout")); + }, remaining); + worker.postMessage(pending.job); + } catch { + void finish(new Error("native_journal_worker_failed")); + } + return; + } + } + return { + read(job: NativeJournalProofJob): Promise { + if (closed || queue.length >= maxQueue) + return Promise.reject(new Error("native_journal_worker_busy")); + return new Promise((resolve, reject) => { + queue.push({ job, resolve, reject, deadline: performance.now() + timeoutMs }); + drain(); + }); + }, + async close() { + closed = true; + for (const pending of queue.splice(0)) + pending.reject(new Error("native_journal_worker_closed")); + if (active) await finish(new Error("native_journal_worker_closed")); + await stop(); + await stopping; + }, + }; +} + +const reader = createNativeJournalProofReader(); +export async function readNativeJournalProjection( + path: string, + purpose: "identity" | "evidence" = "evidence", +) { + return (await reader.read({ + kind: "projection", + path, + purpose, + })) as ReturnType; +} +export async function scanNativeStateFile(path: string, maximum: number) { + return (await reader.read({ kind: "scan", path, maximum })) as { + sha256: string; + byteSize: number; + }; +} +export async function nativeStateFileIncludes( + path: string, + maximum: number, + needles: string[], +) { + return (await reader.read({ + kind: "includes", + path, + maximum, + needles, + })) as boolean; +} +export async function nativeStateBase64Fingerprint(files: File[]) { + return (await reader.read({ kind: "fingerprint", files })) as { + sha256: string; + fileSha256: string[]; + }; +} diff --git a/server/src/services/native-runtime/native-journal-projection-worker.ts b/server/src/services/native-runtime/native-journal-projection-worker.ts new file mode 100644 index 00000000000..d201535174b --- /dev/null +++ b/server/src/services/native-runtime/native-journal-projection-worker.ts @@ -0,0 +1,47 @@ +import { parentPort } from "node:worker_threads"; +import { createHash } from "node:crypto"; +import { + readNativeJournalProjection, + scanNativeStateFile, +} from "./native-journal-projection.js"; +import type { NativeJournalProofJob } from "./native-journal-projection-async.js"; + +if (!parentPort) throw new Error("native_journal_worker_requires_parent"); +parentPort.on("message", (job: NativeJournalProofJob) => { + try { + let result: unknown; + if (job.kind === "projection") + result = readNativeJournalProjection(job.path, job.purpose); + else if (job.kind === "scan") + result = scanNativeStateFile(job.path, job.maximum); + else if (job.kind === "includes") { + let tail = "", + found = false; + const overlap = Math.max(0, ...job.needles.map((value) => value.length)); + scanNativeStateFile(job.path, job.maximum, (chunk) => { + const value = tail + chunk.toString("utf8"); + if (job.needles.some((needle) => value.includes(needle))) found = true; + tail = value.slice(-overlap); + }); + result = found; + } else { + const digest = createHash("sha256").update("["); + const fileSha256 = job.files.map((file, index) => { + digest.update(index ? ',"' : '"'); + const proof = scanNativeStateFile(file.path, file.maximum, (chunk) => + digest.update(chunk.toString("base64")), + ); + digest.update('"'); + return proof.sha256; + }); + result = { sha256: digest.update("]").digest("hex"), fileSha256 }; + } + parentPort!.postMessage({ result }); + } catch (error) { + const message = + error instanceof Error && /^native_[a-z_]+$/.test(error.message) + ? error.message + : "native_journal_worker_read_failed"; + parentPort!.postMessage({ error: message }); + } +});