fix: separate journal proof queue and execution budgets

Co-Authored-By: Paperclip <noreply@paperclip.ing>

doc/architecture/paperclip-runner-compatibility.md

@@ -221,7 +221,8 @@ Catalog generation and production authorization are separate steps.
 The durable writer and controller accept the same 192 MiB raw journal limit.
 Identity, cleanup, warm-transition, and quiescent recovery reads scan the JSON
 off the API event loop with a fixed 48 KiB input buffer. A single worker admits
-at most 32 queued proofs and enforces a 30-second deadline including queue time.
+at most 32 queued proofs. Proofs waiting over 30 seconds are rejected as busy;
+each admitted proof gets a full, separate 30-second execution budget.
 An expired or failed worker is terminated and joined before another job starts.
 Its V8 heap is capped independently of the controller. Idle workers are unrefed
 and stopped after 15 seconds. They retain only the fields used to prove

server/src/services/native-runtime/native-journal-projection-async.ts

@@ -21,11 +21,13 @@ type Pending = {
 };
 
 /** One worker and a bounded queue cap CPU and memory independently of callers.
- * The deadline includes queue time; termination completes before another job starts. */
+ * Queue admission and execution have separate budgets. Termination completes
+ * before another job starts. */
 export function createNativeJournalProofReader(
-  options: { timeoutMs?: number; maxQueue?: number } = {},
+  options: { timeoutMs?: number; queueTimeoutMs?: number; maxQueue?: number } = {},
 ) {
   const timeoutMs = options.timeoutMs ?? 30_000;
+  const queueTimeoutMs = options.queueTimeoutMs ?? 30_000;
   const maxQueue = options.maxQueue ?? 32;
   const queue: Pending[] = [];
   let active: Pending | undefined;
@@ -107,9 +109,8 @@ export function createNativeJournalProofReader(
     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"));
+      if (pending.deadline <= performance.now()) {
+        pending.reject(new Error("native_journal_worker_busy"));
         continue;
       }
       clearTimeout(idleTimer);
@@ -119,7 +120,7 @@ export function createNativeJournalProofReader(
         worker.ref();
         timer = setTimeout(() => {
           void finish(new Error("native_journal_worker_timeout"));
-        }, remaining);
+        }, timeoutMs);
         worker.postMessage(pending.job);
       } catch {
         void finish(new Error("native_journal_worker_failed"));
@@ -132,7 +133,7 @@ export function createNativeJournalProofReader(
       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 });
+        queue.push({ job, resolve, reject, deadline: performance.now() + queueTimeoutMs });
         drain();
       });
     },

server/src/services/native-runtime/native-journal-projection-queue.test.ts

@@ -0,0 +1,86 @@
+import { afterEach, describe, expect, it, vi } from "vitest";
+
+const state = vi.hoisted(() => ({ active: 0, peak: 0, terminated: 0 }));
+vi.mock("node:perf_hooks", () => ({ performance: { now: () => Date.now() } }));
+vi.mock("node:worker_threads", async () => {
+  const { EventEmitter } = await import("node:events");
+  return {
+    Worker: class extends EventEmitter {
+      timer?: ReturnType<typeof setTimeout>;
+      constructor() {
+        super();
+        state.peak = Math.max(state.peak, ++state.active);
+      }
+      postMessage(job: { path: string }) {
+        this.timer = setTimeout(() => this.emit("message", {
+          result: { sha256: job.path, byteSize: 190 * 1024 * 1024 },
+        }), job.path === "slow" ? 110 : 60);
+      }
+      ref() {}
+      unref() {}
+      async terminate() {
+        clearTimeout(this.timer);
+        state.active--;
+        state.terminated++;
+        return 0;
+      }
+    },
+  };
+});
+import { createNativeJournalProofReader } from "./native-journal-projection-async.js";
+
+afterEach(() => vi.useRealTimers());
+describe("journal proof queue admission", () => {
+  it("rejects stale queued work as busy without starting it", async () => {
+    vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date"] });
+    const reader = createNativeJournalProofReader({ timeoutMs: 100, queueTimeoutMs: 50 });
+    const job = { kind: "scan" as const, path: "valid", maximum: 192 * 1024 * 1024 };
+    const results = Promise.allSettled([reader.read(job), reader.read(job)]);
+    try {
+      await vi.advanceTimersByTimeAsync(60);
+      expect(await results).toMatchObject([
+        { status: "fulfilled", value: { sha256: "valid" } },
+        { status: "rejected", reason: { message: "native_journal_worker_busy" } },
+      ]);
+    } finally {
+      await reader.close();
+    }
+  });
+  it("gives each admitted proof a full execution budget after waiting", async () => {
+    vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date"] });
+    const reader = createNativeJournalProofReader({ timeoutMs: 100 });
+    const job = { kind: "scan" as const, path: "valid", maximum: 192 * 1024 * 1024 };
+    // Each proof fits its budget; their aggregate duration exceeds one budget.
+    const results = Promise.allSettled([reader.read(job), reader.read(job)]);
+    try {
+      await vi.advanceTimersByTimeAsync(120);
+      expect(await results).toMatchObject([
+        { status: "fulfilled", value: { sha256: "valid" } },
+        { status: "fulfilled", value: { sha256: "valid" } },
+      ]);
+    } finally {
+      await reader.close();
+    }
+  });
+  it("terminates an actual overrun before giving the next proof its own budget", async () => {
+    vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date"] });
+    state.peak = state.active = state.terminated = 0;
+    const reader = createNativeJournalProofReader({ timeoutMs: 100 });
+    const job = { kind: "scan" as const, maximum: 192 * 1024 * 1024 };
+    const results = Promise.allSettled([
+      reader.read({ ...job, path: "slow" }),
+      reader.read({ ...job, path: "valid" }),
+    ]);
+    try {
+      await vi.advanceTimersByTimeAsync(160);
+      expect(await results).toMatchObject([
+        { status: "rejected", reason: { message: "native_journal_worker_timeout" } },
+        { status: "fulfilled", value: { sha256: "valid" } },
+      ]);
+      expect(state.peak).toBe(1);
+      expect(state.terminated).toBe(1);
+    } finally {
+      await reader.close();
+    }
+  });
+});