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 <noreply@paperclip.ing>

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" } });
+  });
+});

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<typeof readProjection>
+  | { 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<typeof setTimeout> | undefined;
+  let idleTimer: ReturnType<typeof setTimeout> | undefined;
+  let stopping: Promise<unknown> | 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<Result> {
+      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<typeof readProjection>;
+}
+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[];
+  };
+}

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 });
+  }
+});