From e6bed1231ae286698d440fdfec4d19ce3362af42 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 27 Sep 2026 20:48:43 -0500 Subject: [PATCH] fix: separate journal proof queue and execution budgets Co-Authored-By: Paperclip --- .../paperclip-runner-compatibility.md | 3 +- .../native-journal-projection-async.ts | 15 ++-- .../native-journal-projection-queue.test.ts | 86 +++++++++++++++++++ 3 files changed, 96 insertions(+), 8 deletions(-) create mode 100644 server/src/services/native-runtime/native-journal-projection-queue.test.ts diff --git a/doc/architecture/paperclip-runner-compatibility.md b/doc/architecture/paperclip-runner-compatibility.md index 2824fdb3121..f655cb8ab17 100644 --- a/doc/architecture/paperclip-runner-compatibility.md +++ b/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 diff --git a/server/src/services/native-runtime/native-journal-projection-async.ts b/server/src/services/native-runtime/native-journal-projection-async.ts index 5091e362ace..0a1713b4604 100644 --- a/server/src/services/native-runtime/native-journal-projection-async.ts +++ b/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(); }); }, diff --git a/server/src/services/native-runtime/native-journal-projection-queue.test.ts b/server/src/services/native-runtime/native-journal-projection-queue.test.ts new file mode 100644 index 00000000000..9498697dd08 --- /dev/null +++ b/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; + 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(); + } + }); +});