From b1f3bd25c3b09f7cf5b0ec4b52de77b0bc86067d Mon Sep 17 00:00:00 2001 From: Nii Yeboah Date: Tue, 30 Jun 2026 12:12:54 -0400 Subject: [PATCH 01/11] feat(mount-io): add bounded-concurrency semaphore --- apps/server/src/shared/mount-io/semaphore.ts | 29 ++++++++++++ apps/server/test/mount-io/semaphore.test.ts | 47 ++++++++++++++++++++ 2 files changed, 76 insertions(+) create mode 100644 apps/server/src/shared/mount-io/semaphore.ts create mode 100644 apps/server/test/mount-io/semaphore.test.ts diff --git a/apps/server/src/shared/mount-io/semaphore.ts b/apps/server/src/shared/mount-io/semaphore.ts new file mode 100644 index 00000000..b63cd671 --- /dev/null +++ b/apps/server/src/shared/mount-io/semaphore.ts @@ -0,0 +1,29 @@ +export class Semaphore { + private permits: number; + private waiters: Array<() => void> = []; + + constructor(max: number) { + this.permits = max; + } + + async acquire(): Promise { + if (this.permits > 0) { + this.permits--; + return; + } + await new Promise((resolve) => this.waiters.push(resolve)); + } + + release(): void { + const next = this.waiters.shift(); + if (next) { + next(); + } else { + this.permits++; + } + } + + get available(): number { + return this.permits; + } +} diff --git a/apps/server/test/mount-io/semaphore.test.ts b/apps/server/test/mount-io/semaphore.test.ts new file mode 100644 index 00000000..bbf24a98 --- /dev/null +++ b/apps/server/test/mount-io/semaphore.test.ts @@ -0,0 +1,47 @@ +import { describe, expect, it } from "vitest"; + +import { Semaphore } from "../../src/shared/mount-io/semaphore.js"; + +describe("Semaphore", () => { + it("grants permits immediately while capacity remains", async () => { + const sem = new Semaphore(2); + await sem.acquire(); + await sem.acquire(); + expect(sem.available).toBe(0); + }); + + it("queues acquisitions beyond capacity until release", async () => { + const sem = new Semaphore(1); + await sem.acquire(); + + let third = false; + const pending = sem.acquire().then(() => { + third = true; + }); + + await Promise.resolve(); + expect(third).toBe(false); + + sem.release(); + await pending; + expect(third).toBe(true); + }); + + it("never lets more than max run at once", async () => { + const sem = new Semaphore(2); + let active = 0; + let peak = 0; + + const task = async () => { + await sem.acquire(); + active++; + peak = Math.max(peak, active); + await Promise.resolve(); + active--; + sem.release(); + }; + + await Promise.all(Array.from({ length: 10 }, task)); + expect(peak).toBeLessThanOrEqual(2); + }); +}); From aa63a3e412dc49b206dd436ac49d0e0050de652b Mon Sep 17 00:00:00 2001 From: Nii Yeboah Date: Tue, 30 Jun 2026 12:14:47 -0400 Subject: [PATCH 02/11] feat(mount-io): add circuit breaker --- .../src/shared/mount-io/circuit-breaker.ts | 43 ++++++++++++ .../test/mount-io/circuit-breaker.test.ts | 68 +++++++++++++++++++ 2 files changed, 111 insertions(+) create mode 100644 apps/server/src/shared/mount-io/circuit-breaker.ts create mode 100644 apps/server/test/mount-io/circuit-breaker.test.ts diff --git a/apps/server/src/shared/mount-io/circuit-breaker.ts b/apps/server/src/shared/mount-io/circuit-breaker.ts new file mode 100644 index 00000000..0bad18c1 --- /dev/null +++ b/apps/server/src/shared/mount-io/circuit-breaker.ts @@ -0,0 +1,43 @@ +export type BreakerState = "closed" | "open" | "half-open"; + +export class CircuitBreaker { + private failures = 0; + private openedAt: number | null = null; + + constructor( + private readonly threshold: number, + private readonly cooldownMs: number, + private readonly now: () => number = Date.now, + ) {} + + state(): BreakerState { + if (this.openedAt === null) return "closed"; + if (this.now() - this.openedAt >= this.cooldownMs) return "half-open"; + return "open"; + } + + canProceed(): boolean { + return this.state() !== "open"; + } + + recordSuccess(): void { + this.failures = 0; + this.openedAt = null; + } + + recordFailure(): void { + if (this.state() === "half-open") { + this.openedAt = this.now(); + return; + } + this.failures++; + if (this.failures >= this.threshold) { + this.openedAt = this.now(); + } + } + + reset(): void { + this.failures = 0; + this.openedAt = null; + } +} diff --git a/apps/server/test/mount-io/circuit-breaker.test.ts b/apps/server/test/mount-io/circuit-breaker.test.ts new file mode 100644 index 00000000..18c39372 --- /dev/null +++ b/apps/server/test/mount-io/circuit-breaker.test.ts @@ -0,0 +1,68 @@ +import { describe, expect, it } from "vitest"; + +import { CircuitBreaker } from "../../src/shared/mount-io/circuit-breaker.js"; + +const clock = (start = 0) => { + let t = start; + const now = () => t; + return { now, advance: (ms: number) => (t += ms) }; +}; + +describe("CircuitBreaker", () => { + it("stays closed below the failure threshold", () => { + const b = new CircuitBreaker(3, 1000, () => 0); + b.recordFailure(); + b.recordFailure(); + expect(b.state()).toBe("closed"); + expect(b.canProceed()).toBe(true); + }); + + it("opens at the failure threshold and blocks", () => { + const b = new CircuitBreaker(3, 1000, () => 0); + b.recordFailure(); + b.recordFailure(); + b.recordFailure(); + expect(b.state()).toBe("open"); + expect(b.canProceed()).toBe(false); + }); + + it("moves to half-open after the cooldown elapses", () => { + const c = clock(); + const b = new CircuitBreaker(1, 1000, c.now); + b.recordFailure(); + expect(b.state()).toBe("open"); + c.advance(1000); + expect(b.state()).toBe("half-open"); + expect(b.canProceed()).toBe(true); + }); + + it("re-opens when the half-open probe fails", () => { + const c = clock(); + const b = new CircuitBreaker(1, 1000, c.now); + b.recordFailure(); + c.advance(1000); + expect(b.state()).toBe("half-open"); + b.recordFailure(); + expect(b.state()).toBe("open"); + }); + + it("closes when the half-open probe succeeds", () => { + const c = clock(); + const b = new CircuitBreaker(1, 1000, c.now); + b.recordFailure(); + c.advance(1000); + b.recordSuccess(); + expect(b.state()).toBe("closed"); + expect(b.canProceed()).toBe(true); + }); + + it("resets failure count on success", () => { + const b = new CircuitBreaker(3, 1000, () => 0); + b.recordFailure(); + b.recordFailure(); + b.recordSuccess(); + b.recordFailure(); + b.recordFailure(); + expect(b.state()).toBe("closed"); + }); +}); From 9bb0b6be955e75f2fb8a650189a8be1e748652fd Mon Sep 17 00:00:00 2001 From: Nii Yeboah Date: Tue, 30 Jun 2026 12:23:41 -0400 Subject: [PATCH 03/11] feat(mount-io): add MountIO guard (timeout + breaker + concurrency) --- apps/server/src/shared/mount-io/index.ts | 23 +++++ apps/server/src/shared/mount-io/mount-io.ts | 85 +++++++++++++++++ apps/server/test/mount-io/mount-io.test.ts | 100 ++++++++++++++++++++ 3 files changed, 208 insertions(+) create mode 100644 apps/server/src/shared/mount-io/index.ts create mode 100644 apps/server/src/shared/mount-io/mount-io.ts create mode 100644 apps/server/test/mount-io/mount-io.test.ts diff --git a/apps/server/src/shared/mount-io/index.ts b/apps/server/src/shared/mount-io/index.ts new file mode 100644 index 00000000..f3394fa9 --- /dev/null +++ b/apps/server/src/shared/mount-io/index.ts @@ -0,0 +1,23 @@ +import { MountIO } from "./mount-io.js"; + +export { Semaphore } from "./semaphore.js"; +export { CircuitBreaker } from "./circuit-breaker.js"; +export type { BreakerState } from "./circuit-breaker.js"; +export { + MountIO, + MountStallError, + MountUnavailableError, +} from "./mount-io.js"; +export type { MountIOConfig } from "./mount-io.js"; + +const num = (raw: string | undefined, fallback: number): number => { + const n = raw ? Number(raw) : NaN; + return Number.isFinite(n) && n > 0 ? n : fallback; +}; + +export const mountIO = new MountIO({ + timeoutMs: num(process.env.DISPATCH_MOUNT_IO_TIMEOUT_MS, 5_000), + maxConcurrency: num(process.env.DISPATCH_MOUNT_IO_CONCURRENCY, 2), + breakerThreshold: num(process.env.DISPATCH_MOUNT_IO_BREAKER_THRESHOLD, 3), + breakerCooldownMs: num(process.env.DISPATCH_MOUNT_IO_BREAKER_COOLDOWN_MS, 60_000), +}); diff --git a/apps/server/src/shared/mount-io/mount-io.ts b/apps/server/src/shared/mount-io/mount-io.ts new file mode 100644 index 00000000..c4e64148 --- /dev/null +++ b/apps/server/src/shared/mount-io/mount-io.ts @@ -0,0 +1,85 @@ +import { CircuitBreaker } from "./circuit-breaker.js"; +import { Semaphore } from "./semaphore.js"; + +export class MountStallError extends Error { + constructor(label: string, timeoutMs: number) { + super(`mount op "${label}" exceeded ${timeoutMs}ms`); + this.name = "MountStallError"; + } +} + +export class MountUnavailableError extends Error { + constructor(label: string) { + super(`mount unavailable (circuit open), skipped "${label}"`); + this.name = "MountUnavailableError"; + } +} + +export type MountIOConfig = { + timeoutMs: number; + maxConcurrency: number; + breakerThreshold: number; + breakerCooldownMs: number; + now?: () => number; +}; + +export class MountIO { + private readonly sem: Semaphore; + private readonly breaker: CircuitBreaker; + private readonly timeoutMs: number; + + constructor(cfg: MountIOConfig) { + this.sem = new Semaphore(cfg.maxConcurrency); + this.breaker = new CircuitBreaker( + cfg.breakerThreshold, + cfg.breakerCooldownMs, + cfg.now, + ); + this.timeoutMs = cfg.timeoutMs; + } + + available(): boolean { + return this.breaker.canProceed(); + } + + async run( + label: string, + fn: (signal: AbortSignal) => Promise, + ): Promise { + if (!this.breaker.canProceed()) { + throw new MountUnavailableError(label); + } + + await this.sem.acquire(); + const ac = new AbortController(); + let timer: ReturnType | null = null; + + try { + const work = fn(ac.signal); + work.catch(() => {}); + + const timeout = new Promise((_, reject) => { + timer = setTimeout(() => { + ac.abort(new MountStallError(label, this.timeoutMs)); + reject(new MountStallError(label, this.timeoutMs)); + }, this.timeoutMs); + }); + + const result = await Promise.race([work, timeout]); + this.breaker.recordSuccess(); + return result; + } catch (err) { + if (err instanceof MountStallError) { + this.breaker.recordFailure(); + } + throw err; + } finally { + if (timer) clearTimeout(timer); + this.sem.release(); + } + } + + reset(): void { + this.breaker.reset(); + } +} diff --git a/apps/server/test/mount-io/mount-io.test.ts b/apps/server/test/mount-io/mount-io.test.ts new file mode 100644 index 00000000..c28490d8 --- /dev/null +++ b/apps/server/test/mount-io/mount-io.test.ts @@ -0,0 +1,100 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import { + MountIO, + MountStallError, + MountUnavailableError, +} from "../../src/shared/mount-io/mount-io.js"; + +const cfg = { + timeoutMs: 5000, + maxConcurrency: 2, + breakerThreshold: 2, + breakerCooldownMs: 60_000, +}; + +describe("MountIO", () => { + beforeEach(() => vi.useFakeTimers()); + afterEach(() => vi.useRealTimers()); + + it("returns the result of a fast op", async () => { + const io = new MountIO(cfg); + await expect(io.run("ok", async () => 42)).resolves.toBe(42); + expect(io.available()).toBe(true); + }); + + it("times out a hanging op and aborts its signal", async () => { + const io = new MountIO(cfg); + let aborted = false; + const p = io.run("hang", (signal) => { + signal.addEventListener("abort", () => (aborted = true)); + return new Promise(() => {}); + }); + const assertion = expect(p).rejects.toBeInstanceOf(MountStallError); + await vi.advanceTimersByTimeAsync(5001); + await assertion; + expect(aborted).toBe(true); + }); + + it("opens the breaker after threshold timeouts, then skips fn", async () => { + const io = new MountIO(cfg); + const hang = () => new Promise(() => {}); + + for (let i = 0; i < 2; i++) { + const p = io.run(`hang-${i}`, hang); + const a = expect(p).rejects.toBeInstanceOf(MountStallError); + await vi.advanceTimersByTimeAsync(5001); + await a; + } + + expect(io.available()).toBe(false); + let called = false; + await expect( + io.run("blocked", async () => { + called = true; + return 1; + }), + ).rejects.toBeInstanceOf(MountUnavailableError); + expect(called).toBe(false); + }); + + it("never runs more than maxConcurrency fns at once", async () => { + const io = new MountIO({ ...cfg, maxConcurrency: 2 }); + let active = 0; + let peak = 0; + const release: Array<() => void> = []; + + const starts = Array.from({ length: 5 }, (_, i) => + io.run(`c-${i}`, () => { + active++; + peak = Math.max(peak, active); + return new Promise((resolve) => { + release.push(() => { + active--; + resolve(i); + }); + }); + }), + ); + + await vi.advanceTimersByTimeAsync(0); + expect(peak).toBeLessThanOrEqual(2); + release.forEach((r) => r()); + await Promise.all(starts); + }); + + it("does not emit an unhandled rejection when fn rejects after timeout", async () => { + const io = new MountIO(cfg); + const p = io.run( + "late-reject", + () => + new Promise((_, reject) => + setTimeout(() => reject(new Error("late")), 10_000), + ), + ); + const a = expect(p).rejects.toBeInstanceOf(MountStallError); + await vi.advanceTimersByTimeAsync(5001); + await a; + await vi.advanceTimersByTimeAsync(6000); + }); +}); From f1e4f0b5e4312ee289a0929583e549dbded25d0b Mon Sep 17 00:00:00 2001 From: Nii Yeboah Date: Tue, 30 Jun 2026 12:26:08 -0400 Subject: [PATCH 04/11] test(mount-io): fix deadlocking concurrency test --- apps/server/test/mount-io/mount-io.test.ts | 21 +++++++-------------- 1 file changed, 7 insertions(+), 14 deletions(-) diff --git a/apps/server/test/mount-io/mount-io.test.ts b/apps/server/test/mount-io/mount-io.test.ts index c28490d8..b3689e14 100644 --- a/apps/server/test/mount-io/mount-io.test.ts +++ b/apps/server/test/mount-io/mount-io.test.ts @@ -62,25 +62,18 @@ describe("MountIO", () => { const io = new MountIO({ ...cfg, maxConcurrency: 2 }); let active = 0; let peak = 0; - const release: Array<() => void> = []; - const starts = Array.from({ length: 5 }, (_, i) => - io.run(`c-${i}`, () => { + const task = (i: number) => + io.run(`c-${i}`, async () => { active++; peak = Math.max(peak, active); - return new Promise((resolve) => { - release.push(() => { - active--; - resolve(i); - }); - }); - }), - ); + await Promise.resolve(); + active--; + return i; + }); - await vi.advanceTimersByTimeAsync(0); + await Promise.all(Array.from({ length: 10 }, (_, i) => task(i))); expect(peak).toBeLessThanOrEqual(2); - release.forEach((r) => r()); - await Promise.all(starts); }); it("does not emit an unhandled rejection when fn rejects after timeout", async () => { From d329312cf8f981a447729cb23bec2b8430ae2d38 Mon Sep 17 00:00:00 2001 From: Nii Yeboah Date: Tue, 30 Jun 2026 12:28:44 -0400 Subject: [PATCH 05/11] fix(mount-io): use a single MountStallError per timeout --- apps/server/src/shared/mount-io/mount-io.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/apps/server/src/shared/mount-io/mount-io.ts b/apps/server/src/shared/mount-io/mount-io.ts index c4e64148..0274f3e8 100644 --- a/apps/server/src/shared/mount-io/mount-io.ts +++ b/apps/server/src/shared/mount-io/mount-io.ts @@ -60,8 +60,9 @@ export class MountIO { const timeout = new Promise((_, reject) => { timer = setTimeout(() => { - ac.abort(new MountStallError(label, this.timeoutMs)); - reject(new MountStallError(label, this.timeoutMs)); + const err = new MountStallError(label, this.timeoutMs); + ac.abort(err); + reject(err); }, this.timeoutMs); }); From c6505f73c3762eecb1fbde575e3bed823bf1d26a Mon Sep 17 00:00:00 2001 From: Nii Yeboah Date: Tue, 30 Jun 2026 12:33:54 -0400 Subject: [PATCH 06/11] fix(token-harvester): bound Claude session reads via mountIO --- apps/server/src/agents/token-harvester.ts | 101 ++++++++++-------- .../server/test/token-harvester-stall.test.ts | 47 ++++++++ 2 files changed, 102 insertions(+), 46 deletions(-) create mode 100644 apps/server/test/token-harvester-stall.test.ts diff --git a/apps/server/src/agents/token-harvester.ts b/apps/server/src/agents/token-harvester.ts index ef55d385..0ce13e63 100644 --- a/apps/server/src/agents/token-harvester.ts +++ b/apps/server/src/agents/token-harvester.ts @@ -6,6 +6,7 @@ import { createInterface } from "node:readline"; import type { Pool } from "pg"; +import { mountIO } from "../shared/mount-io/index.js"; import type { AgentRecord } from "./manager.js"; type ModelTokenTotals = { @@ -65,7 +66,9 @@ export function cwdToClaudeProjectDir(cwd: string): string { export async function discoverSessionFiles(dir: string): Promise { let entries: string[]; try { - entries = await readdir(dir); + entries = await mountIO.run("readdir:claude-projects", (signal) => + readdir(dir, { signal }), + ); } catch { return []; } @@ -75,60 +78,62 @@ export async function discoverSessionFiles(dir: string): Promise { } async function parseClaudeSessionTokenUsage( - filePath: string + filePath: string, ): Promise { const sessionId = path.basename(filePath, ".jsonl"); - const totals = new Map(); - let sessionStart: string | null = null; - let sessionEnd: string | null = null; - - const rl = createInterface({ - input: createReadStream(filePath, { encoding: "utf-8" }), - crlfDelay: Infinity, - }); + return mountIO.run(`read:claude:${sessionId}`, async (signal) => { + const totals = new Map(); + let sessionStart: string | null = null; + let sessionEnd: string | null = null; + + const rl = createInterface({ + input: createReadStream(filePath, { encoding: "utf-8", signal }), + crlfDelay: Infinity, + }); + + for await (const line of rl) { + if (!line.trim()) continue; + let entry: Record; + try { + entry = JSON.parse(line); + } catch { + continue; + } - for await (const line of rl) { - if (!line.trim()) continue; - let entry: Record; - try { - entry = JSON.parse(line); - } catch { - continue; - } + if (entry.type !== "assistant") continue; + const message = entry.message as Record | undefined; + if (!message?.usage) continue; - if (entry.type !== "assistant") continue; - const message = entry.message as Record | undefined; - if (!message?.usage) continue; + const usage = message.usage as Record; + const model = (message.model as string) ?? "unknown"; + const ts = entry.timestamp as string | undefined; - const usage = message.usage as Record; - const model = (message.model as string) ?? "unknown"; - const ts = entry.timestamp as string | undefined; + if (ts) { + if (!sessionStart) sessionStart = ts; + sessionEnd = ts; + } - if (ts) { - if (!sessionStart) sessionStart = ts; - sessionEnd = ts; - } + let bucket = totals.get(model); + if (!bucket) { + bucket = { + inputTokens: 0, + cacheCreationTokens: 0, + cacheReadTokens: 0, + outputTokens: 0, + messageCount: 0, + }; + totals.set(model, bucket); + } - let bucket = totals.get(model); - if (!bucket) { - bucket = { - inputTokens: 0, - cacheCreationTokens: 0, - cacheReadTokens: 0, - outputTokens: 0, - messageCount: 0, - }; - totals.set(model, bucket); + bucket.inputTokens += usage.input_tokens ?? 0; + bucket.cacheCreationTokens += usage.cache_creation_input_tokens ?? 0; + bucket.cacheReadTokens += usage.cache_read_input_tokens ?? 0; + bucket.outputTokens += usage.output_tokens ?? 0; + bucket.messageCount += 1; } - bucket.inputTokens += usage.input_tokens ?? 0; - bucket.cacheCreationTokens += usage.cache_creation_input_tokens ?? 0; - bucket.cacheReadTokens += usage.cache_read_input_tokens ?? 0; - bucket.outputTokens += usage.output_tokens ?? 0; - bucket.messageCount += 1; - } - - return { sessionId, totals, sessionStart, sessionEnd }; + return { sessionId, totals, sessionStart, sessionEnd }; + }); } async function harvestClaudeTokenUsage( @@ -151,6 +156,10 @@ async function harvestClaudeTokenUsage( } for (const file of files) { + if (!mountIO.available()) { + logger?.warn({ projectDir }, "mount unavailable, aborting Claude harvest"); + break; + } try { const summary = await parseClaudeSessionTokenUsage(file); diff --git a/apps/server/test/token-harvester-stall.test.ts b/apps/server/test/token-harvester-stall.test.ts new file mode 100644 index 00000000..a4e6f182 --- /dev/null +++ b/apps/server/test/token-harvester-stall.test.ts @@ -0,0 +1,47 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +vi.mock("node:fs/promises", async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, readdir: vi.fn() }; +}); + +import { readdir } from "node:fs/promises"; + +import { discoverSessionFiles } from "../src/agents/token-harvester.js"; +import { mountIO } from "../src/shared/mount-io/index.js"; + +describe("discoverSessionFiles under a stalled mount", () => { + beforeEach(() => { + vi.useFakeTimers(); + mountIO.reset(); + }); + afterEach(() => { + vi.useRealTimers(); + vi.mocked(readdir).mockReset(); + mountIO.reset(); + }); + + it("returns [] instead of hanging when readdir never resolves", async () => { + vi.mocked(readdir).mockReturnValue( + new Promise(() => {}) as unknown as ReturnType, + ); + + const p = discoverSessionFiles("/mnt/claude/projects/whatever"); + const assertion = expect(p).resolves.toEqual([]); + await vi.advanceTimersByTimeAsync(5001); + await assertion; + }); + + it("returns parsed file list when readdir resolves normally", async () => { + vi.mocked(readdir).mockResolvedValue([ + "a.jsonl", + "b.txt", + "c.jsonl", + ] as unknown as Awaited>); + + await expect(discoverSessionFiles("/p")).resolves.toEqual([ + "/p/a.jsonl", + "/p/c.jsonl", + ]); + }); +}); From 2d8aebc3d76201ff28367d479fdd8b4835db7343 Mon Sep 17 00:00:00 2001 From: Nii Yeboah Date: Tue, 30 Jun 2026 12:37:24 -0400 Subject: [PATCH 07/11] fix(token-harvester): bound Codex rollout reads via mountIO --- apps/server/src/agents/token-harvester.ts | 106 +++++++++--------- .../server/test/token-harvester-stall.test.ts | 27 ++++- 2 files changed, 81 insertions(+), 52 deletions(-) diff --git a/apps/server/src/agents/token-harvester.ts b/apps/server/src/agents/token-harvester.ts index 0ce13e63..afc58535 100644 --- a/apps/server/src/agents/token-harvester.ts +++ b/apps/server/src/agents/token-harvester.ts @@ -210,9 +210,9 @@ async function discoverCodexRolloutFiles(): Promise { async function walk(dir: string): Promise { let entries: import("node:fs").Dirent[]; try { - entries = (await readdir(dir, { - withFileTypes: true, - })) as import("node:fs").Dirent[]; + entries = await mountIO.run("readdir:codex-sessions", (signal) => + readdir(dir, { withFileTypes: true, signal }), + ); } catch { return; } @@ -230,6 +230,10 @@ async function discoverCodexRolloutFiles(): Promise { return files; } +export function discoverCodexRolloutFilesForTest(): Promise { + return discoverCodexRolloutFiles(); +} + /** * Parse a Codex rollout JSONL in a single pass: check for the agent tag * in the first 20 lines, then extract model and cumulative token usage. @@ -237,66 +241,62 @@ async function discoverCodexRolloutFiles(): Promise { */ async function parseCodexRolloutForAgent( filePath: string, - agentId: string + agentId: string, ): Promise<{ usage: CodexTokenUsage; model: string; sessionStart: string | null; sessionEnd: string | null; } | null> { - let matched = false; - let lastUsage: CodexTokenUsage | null = null; - let model = "unknown"; - let sessionStart: string | null = null; - let sessionEnd: string | null = null; - let linesRead = 0; - - const rl = createInterface({ - input: createReadStream(filePath, { encoding: "utf-8" }), - crlfDelay: Infinity, - }); - - for await (const line of rl) { - linesRead++; - - // Check for agent tag in the first 20 lines - if (!matched) { - if (linesRead > 20) break; - const tagMatch = DISPATCH_TAG_RE.exec(line); - if (tagMatch && tagMatch[1] === agentId) matched = true; - continue; - } - - if (!line.trim()) continue; - let entry: Record; - try { - entry = JSON.parse(line); - } catch { - continue; - } - - const ts = entry.timestamp as string | undefined; - if (ts && !sessionStart) sessionStart = ts; - if (ts) sessionEnd = ts; + return mountIO.run(`read:codex:${path.basename(filePath)}`, async (signal) => { + let matched = false; + let lastUsage: CodexTokenUsage | null = null; + let model = "unknown"; + let sessionStart: string | null = null; + let sessionEnd: string | null = null; + let linesRead = 0; - if (entry.type === "turn_context") { - const payload = entry.payload as Record | undefined; - if (payload?.model) model = payload.model as string; - } + const rl = createInterface({ + input: createReadStream(filePath, { encoding: "utf-8", signal }), + crlfDelay: Infinity, + }); - if (entry.type === "event_msg") { - const payload = entry.payload as Record | undefined; - if (payload?.type === "token_count") { - const info = payload.info as Record | undefined; - if (info?.total_token_usage) { - lastUsage = info.total_token_usage as CodexTokenUsage; + for await (const line of rl) { + linesRead++; + if (!matched) { + if (linesRead > 20) break; + const tagMatch = DISPATCH_TAG_RE.exec(line); + if (tagMatch && tagMatch[1] === agentId) matched = true; + continue; + } + if (!line.trim()) continue; + let entry: Record; + try { + entry = JSON.parse(line); + } catch { + continue; + } + const ts = entry.timestamp as string | undefined; + if (ts && !sessionStart) sessionStart = ts; + if (ts) sessionEnd = ts; + if (entry.type === "turn_context") { + const payload = entry.payload as Record | undefined; + if (payload?.model) model = payload.model as string; + } + if (entry.type === "event_msg") { + const payload = entry.payload as Record | undefined; + if (payload?.type === "token_count") { + const info = payload.info as Record | undefined; + if (info?.total_token_usage) { + lastUsage = info.total_token_usage as CodexTokenUsage; + } } } } - } - if (!matched || !lastUsage) return null; - return { usage: lastUsage, model, sessionStart, sessionEnd }; + if (!matched || !lastUsage) return null; + return { usage: lastUsage, model, sessionStart, sessionEnd }; + }); } async function harvestCodexTokenUsage( @@ -308,6 +308,10 @@ async function harvestCodexTokenUsage( if (rolloutFiles.length === 0) return; for (const file of rolloutFiles) { + if (!mountIO.available()) { + logger?.warn({}, "mount unavailable, aborting Codex harvest"); + break; + } try { const result = await parseCodexRolloutForAgent(file, agent.id); if (!result) continue; diff --git a/apps/server/test/token-harvester-stall.test.ts b/apps/server/test/token-harvester-stall.test.ts index a4e6f182..13d449b5 100644 --- a/apps/server/test/token-harvester-stall.test.ts +++ b/apps/server/test/token-harvester-stall.test.ts @@ -7,7 +7,10 @@ vi.mock("node:fs/promises", async (importOriginal) => { import { readdir } from "node:fs/promises"; -import { discoverSessionFiles } from "../src/agents/token-harvester.js"; +import { + discoverSessionFiles, + discoverCodexRolloutFilesForTest, +} from "../src/agents/token-harvester.js"; import { mountIO } from "../src/shared/mount-io/index.js"; describe("discoverSessionFiles under a stalled mount", () => { @@ -45,3 +48,25 @@ describe("discoverSessionFiles under a stalled mount", () => { ]); }); }); + +describe("codex rollout discovery under a stalled mount", () => { + beforeEach(() => { + vi.useFakeTimers(); + mountIO.reset(); + }); + afterEach(() => { + vi.useRealTimers(); + vi.mocked(readdir).mockReset(); + mountIO.reset(); + }); + + it("returns [] instead of hanging when the top-level readdir never resolves", async () => { + vi.mocked(readdir).mockReturnValue( + new Promise(() => {}) as unknown as ReturnType, + ); + const p = discoverCodexRolloutFilesForTest(); + const assertion = expect(p).resolves.toEqual([]); + await vi.advanceTimersByTimeAsync(5001); + await assertion; + }); +}); From 06d9da270fb8a0965867d6c9678f72995e24b022 Mon Sep 17 00:00:00 2001 From: Nii Yeboah Date: Tue, 30 Jun 2026 12:49:05 -0400 Subject: [PATCH 08/11] fix(token-harvester): drop unsupported readdir signal (bound via MountIO timeout) --- apps/server/src/agents/token-harvester.ts | 112 ++++++++++++---------- 1 file changed, 60 insertions(+), 52 deletions(-) diff --git a/apps/server/src/agents/token-harvester.ts b/apps/server/src/agents/token-harvester.ts index afc58535..ca1b9f7a 100644 --- a/apps/server/src/agents/token-harvester.ts +++ b/apps/server/src/agents/token-harvester.ts @@ -66,8 +66,9 @@ export function cwdToClaudeProjectDir(cwd: string): string { export async function discoverSessionFiles(dir: string): Promise { let entries: string[]; try { - entries = await mountIO.run("readdir:claude-projects", (signal) => - readdir(dir, { signal }), + // @types/node@24 readdir has no {signal} overload; bounded by MountIO timeout only + entries = await mountIO.run("readdir:claude-projects", (_signal) => + readdir(dir) ); } catch { return []; @@ -78,7 +79,7 @@ export async function discoverSessionFiles(dir: string): Promise { } async function parseClaudeSessionTokenUsage( - filePath: string, + filePath: string ): Promise { const sessionId = path.basename(filePath, ".jsonl"); return mountIO.run(`read:claude:${sessionId}`, async (signal) => { @@ -157,7 +158,10 @@ async function harvestClaudeTokenUsage( for (const file of files) { if (!mountIO.available()) { - logger?.warn({ projectDir }, "mount unavailable, aborting Claude harvest"); + logger?.warn( + { projectDir }, + "mount unavailable, aborting Claude harvest" + ); break; } try { @@ -210,8 +214,9 @@ async function discoverCodexRolloutFiles(): Promise { async function walk(dir: string): Promise { let entries: import("node:fs").Dirent[]; try { - entries = await mountIO.run("readdir:codex-sessions", (signal) => - readdir(dir, { withFileTypes: true, signal }), + // @types/node@24 readdir has no {signal} overload; bounded by MountIO timeout only + entries = await mountIO.run("readdir:codex-sessions", (_signal) => + readdir(dir, { withFileTypes: true }) ); } catch { return; @@ -241,62 +246,65 @@ export function discoverCodexRolloutFilesForTest(): Promise { */ async function parseCodexRolloutForAgent( filePath: string, - agentId: string, + agentId: string ): Promise<{ usage: CodexTokenUsage; model: string; sessionStart: string | null; sessionEnd: string | null; } | null> { - return mountIO.run(`read:codex:${path.basename(filePath)}`, async (signal) => { - let matched = false; - let lastUsage: CodexTokenUsage | null = null; - let model = "unknown"; - let sessionStart: string | null = null; - let sessionEnd: string | null = null; - let linesRead = 0; - - const rl = createInterface({ - input: createReadStream(filePath, { encoding: "utf-8", signal }), - crlfDelay: Infinity, - }); - - for await (const line of rl) { - linesRead++; - if (!matched) { - if (linesRead > 20) break; - const tagMatch = DISPATCH_TAG_RE.exec(line); - if (tagMatch && tagMatch[1] === agentId) matched = true; - continue; - } - if (!line.trim()) continue; - let entry: Record; - try { - entry = JSON.parse(line); - } catch { - continue; - } - const ts = entry.timestamp as string | undefined; - if (ts && !sessionStart) sessionStart = ts; - if (ts) sessionEnd = ts; - if (entry.type === "turn_context") { - const payload = entry.payload as Record | undefined; - if (payload?.model) model = payload.model as string; - } - if (entry.type === "event_msg") { - const payload = entry.payload as Record | undefined; - if (payload?.type === "token_count") { - const info = payload.info as Record | undefined; - if (info?.total_token_usage) { - lastUsage = info.total_token_usage as CodexTokenUsage; + return mountIO.run( + `read:codex:${path.basename(filePath)}`, + async (signal) => { + let matched = false; + let lastUsage: CodexTokenUsage | null = null; + let model = "unknown"; + let sessionStart: string | null = null; + let sessionEnd: string | null = null; + let linesRead = 0; + + const rl = createInterface({ + input: createReadStream(filePath, { encoding: "utf-8", signal }), + crlfDelay: Infinity, + }); + + for await (const line of rl) { + linesRead++; + if (!matched) { + if (linesRead > 20) break; + const tagMatch = DISPATCH_TAG_RE.exec(line); + if (tagMatch && tagMatch[1] === agentId) matched = true; + continue; + } + if (!line.trim()) continue; + let entry: Record; + try { + entry = JSON.parse(line); + } catch { + continue; + } + const ts = entry.timestamp as string | undefined; + if (ts && !sessionStart) sessionStart = ts; + if (ts) sessionEnd = ts; + if (entry.type === "turn_context") { + const payload = entry.payload as Record | undefined; + if (payload?.model) model = payload.model as string; + } + if (entry.type === "event_msg") { + const payload = entry.payload as Record | undefined; + if (payload?.type === "token_count") { + const info = payload.info as Record | undefined; + if (info?.total_token_usage) { + lastUsage = info.total_token_usage as CodexTokenUsage; + } } } } - } - if (!matched || !lastUsage) return null; - return { usage: lastUsage, model, sessionStart, sessionEnd }; - }); + if (!matched || !lastUsage) return null; + return { usage: lastUsage, model, sessionStart, sessionEnd }; + } + ); } async function harvestCodexTokenUsage( From ec6ca7d80e13abc97a8623a3739793dbe2a1b864 Mon Sep 17 00:00:00 2001 From: Nii Yeboah Date: Tue, 30 Jun 2026 12:49:08 -0400 Subject: [PATCH 09/11] docs(mount-io): document DISPATCH_MOUNT_IO_* env knobs --- .env.example | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/.env.example b/.env.example index fb3c4da9..863a4ad0 100644 --- a/.env.example +++ b/.env.example @@ -15,3 +15,9 @@ MEDIA_ROOT=~/.dispatch/media # TLS (optional — both must be set to enable HTTPS) # TLS_CERT=~/.dispatch/tls/cert.pem # TLS_KEY=~/.dispatch/tls/key.pem + +# gcsfuse mount-io guard (token harvester). Defaults shown. +# DISPATCH_MOUNT_IO_TIMEOUT_MS=5000 +# DISPATCH_MOUNT_IO_CONCURRENCY=2 +# DISPATCH_MOUNT_IO_BREAKER_THRESHOLD=3 +# DISPATCH_MOUNT_IO_BREAKER_COOLDOWN_MS=60000 From 9067c87cf3eba22e6e07ded67824247ab4612a55 Mon Sep 17 00:00:00 2001 From: Nii Yeboah Date: Tue, 30 Jun 2026 12:50:37 -0400 Subject: [PATCH 10/11] style(mount-io): apply prettier (trailing commas, wrapping) --- apps/server/src/shared/mount-io/circuit-breaker.ts | 2 +- apps/server/src/shared/mount-io/index.ts | 11 +++++------ apps/server/src/shared/mount-io/mount-io.ts | 4 ++-- apps/server/test/mount-io/mount-io.test.ts | 6 +++--- apps/server/test/token-harvester-stall.test.ts | 4 ++-- 5 files changed, 13 insertions(+), 14 deletions(-) diff --git a/apps/server/src/shared/mount-io/circuit-breaker.ts b/apps/server/src/shared/mount-io/circuit-breaker.ts index 0bad18c1..648e0727 100644 --- a/apps/server/src/shared/mount-io/circuit-breaker.ts +++ b/apps/server/src/shared/mount-io/circuit-breaker.ts @@ -7,7 +7,7 @@ export class CircuitBreaker { constructor( private readonly threshold: number, private readonly cooldownMs: number, - private readonly now: () => number = Date.now, + private readonly now: () => number = Date.now ) {} state(): BreakerState { diff --git a/apps/server/src/shared/mount-io/index.ts b/apps/server/src/shared/mount-io/index.ts index f3394fa9..e3ca0f7d 100644 --- a/apps/server/src/shared/mount-io/index.ts +++ b/apps/server/src/shared/mount-io/index.ts @@ -3,11 +3,7 @@ import { MountIO } from "./mount-io.js"; export { Semaphore } from "./semaphore.js"; export { CircuitBreaker } from "./circuit-breaker.js"; export type { BreakerState } from "./circuit-breaker.js"; -export { - MountIO, - MountStallError, - MountUnavailableError, -} from "./mount-io.js"; +export { MountIO, MountStallError, MountUnavailableError } from "./mount-io.js"; export type { MountIOConfig } from "./mount-io.js"; const num = (raw: string | undefined, fallback: number): number => { @@ -19,5 +15,8 @@ export const mountIO = new MountIO({ timeoutMs: num(process.env.DISPATCH_MOUNT_IO_TIMEOUT_MS, 5_000), maxConcurrency: num(process.env.DISPATCH_MOUNT_IO_CONCURRENCY, 2), breakerThreshold: num(process.env.DISPATCH_MOUNT_IO_BREAKER_THRESHOLD, 3), - breakerCooldownMs: num(process.env.DISPATCH_MOUNT_IO_BREAKER_COOLDOWN_MS, 60_000), + breakerCooldownMs: num( + process.env.DISPATCH_MOUNT_IO_BREAKER_COOLDOWN_MS, + 60_000 + ), }); diff --git a/apps/server/src/shared/mount-io/mount-io.ts b/apps/server/src/shared/mount-io/mount-io.ts index 0274f3e8..c27d816e 100644 --- a/apps/server/src/shared/mount-io/mount-io.ts +++ b/apps/server/src/shared/mount-io/mount-io.ts @@ -33,7 +33,7 @@ export class MountIO { this.breaker = new CircuitBreaker( cfg.breakerThreshold, cfg.breakerCooldownMs, - cfg.now, + cfg.now ); this.timeoutMs = cfg.timeoutMs; } @@ -44,7 +44,7 @@ export class MountIO { async run( label: string, - fn: (signal: AbortSignal) => Promise, + fn: (signal: AbortSignal) => Promise ): Promise { if (!this.breaker.canProceed()) { throw new MountUnavailableError(label); diff --git a/apps/server/test/mount-io/mount-io.test.ts b/apps/server/test/mount-io/mount-io.test.ts index b3689e14..a103eb69 100644 --- a/apps/server/test/mount-io/mount-io.test.ts +++ b/apps/server/test/mount-io/mount-io.test.ts @@ -53,7 +53,7 @@ describe("MountIO", () => { io.run("blocked", async () => { called = true; return 1; - }), + }) ).rejects.toBeInstanceOf(MountUnavailableError); expect(called).toBe(false); }); @@ -82,8 +82,8 @@ describe("MountIO", () => { "late-reject", () => new Promise((_, reject) => - setTimeout(() => reject(new Error("late")), 10_000), - ), + setTimeout(() => reject(new Error("late")), 10_000) + ) ); const a = expect(p).rejects.toBeInstanceOf(MountStallError); await vi.advanceTimersByTimeAsync(5001); diff --git a/apps/server/test/token-harvester-stall.test.ts b/apps/server/test/token-harvester-stall.test.ts index 13d449b5..2abd3497 100644 --- a/apps/server/test/token-harvester-stall.test.ts +++ b/apps/server/test/token-harvester-stall.test.ts @@ -26,7 +26,7 @@ describe("discoverSessionFiles under a stalled mount", () => { it("returns [] instead of hanging when readdir never resolves", async () => { vi.mocked(readdir).mockReturnValue( - new Promise(() => {}) as unknown as ReturnType, + new Promise(() => {}) as unknown as ReturnType ); const p = discoverSessionFiles("/mnt/claude/projects/whatever"); @@ -62,7 +62,7 @@ describe("codex rollout discovery under a stalled mount", () => { it("returns [] instead of hanging when the top-level readdir never resolves", async () => { vi.mocked(readdir).mockReturnValue( - new Promise(() => {}) as unknown as ReturnType, + new Promise(() => {}) as unknown as ReturnType ); const p = discoverCodexRolloutFilesForTest(); const assertion = expect(p).resolves.toEqual([]); From 2d3879c756df5f99d058a40ba5b6948c955bc855 Mon Sep 17 00:00:00 2001 From: Nii Yeboah Date: Tue, 30 Jun 2026 13:23:08 -0400 Subject: [PATCH 11/11] =?UTF-8?q?fix(mount-io):=20harden=20guard=20per=20p?= =?UTF-8?q?ersona=20review=20=E2=80=94=20idle=20timeout,=20threadpool,=20o?= =?UTF-8?q?bservability?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses findings from backend, infra, architecture, and release-readiness persona reviews of the MountIO guard: - Streaming reads now use an idle (no-progress) deadline: run() hands fn a heartbeat() that re-arms the timer, and both JSONL parse loops call it per line. A slow-but-progressing large file over a healthy mount no longer trips the process-wide breaker — only a genuine no-data stall does. readdir keeps the whole-op deadline (single-shot, can't heartbeat). - Raise default DISPATCH_MOUNT_IO_TIMEOUT_MS 5s -> 10s; document tuning. - Set UV_THREADPOOL_SIZE=16 in the launchd wrapper + .env.example so an uncancellable readdir/read syscall wedged on a stalled gcsfuse mount cannot starve fs/dns/crypto work server-wide. Documented that production's Bun runtime may ignore the knob (own threadpool, not libuv) and that the breaker is the cross-runtime backstop. - Log mount-io breaker open/recover transitions via the app logger (mountIO.setLogger in server startup) for operator visibility. - Document run()'s contract: signal MUST be honored; timeout abandons rather than cancels; available()/semaphore track JS ops, not in-flight kernel work. - Document the global-breaker single-mount assumption and the MountStallError-only breaker policy; comment the load-bearing work.catch. - Replace the discoverCodexRolloutFilesForTest pass-through with a direct exported discoverCodexRolloutFiles (testing-only comment). Co-Authored-By: Claude Opus 4.8 (1M context) --- .env.example | 15 ++- apps/server/src/agents/token-harvester.ts | 16 +-- apps/server/src/server.ts | 4 + apps/server/src/shared/mount-io/index.ts | 17 ++- apps/server/src/shared/mount-io/mount-io.ts | 111 ++++++++++++++++-- .../server/test/token-harvester-stall.test.ts | 8 +- bin/dispatch-launchd-wrapper | 11 ++ 7 files changed, 160 insertions(+), 22 deletions(-) diff --git a/.env.example b/.env.example index 863a4ad0..2095690a 100644 --- a/.env.example +++ b/.env.example @@ -17,7 +17,20 @@ MEDIA_ROOT=~/.dispatch/media # TLS_KEY=~/.dispatch/tls/key.pem # gcsfuse mount-io guard (token harvester). Defaults shown. -# DISPATCH_MOUNT_IO_TIMEOUT_MS=5000 +# TIMEOUT_MS is an idle (no-progress) deadline for streaming reads and a +# whole-op deadline for readdir. Keep it generous enough to exceed worst-case +# large-file / cold-cache latency on your mount — a too-tight value makes a +# healthy-but-slow mount trip the breaker and pause all harvesting for the +# cooldown. Raise it on high-latency mounts. +# DISPATCH_MOUNT_IO_TIMEOUT_MS=10000 # DISPATCH_MOUNT_IO_CONCURRENCY=2 # DISPATCH_MOUNT_IO_BREAKER_THRESHOLD=3 # DISPATCH_MOUNT_IO_BREAKER_COOLDOWN_MS=60000 + +# I/O thread-pool size. A stalled gcsfuse mount leaves uncancellable read/readdir +# syscalls blocked on pool threads; enlarging the pool keeps a few wedged +# harvester threads from starving other fs/dns/crypto work. Set before launch +# (the launchd wrapper defaults this to 16). NOTE: the production Bun runtime +# uses its own threadpool, not libuv, and may ignore this — it mainly hardens +# Node dev/test; the mount-io breaker is the cross-runtime backstop. +# UV_THREADPOOL_SIZE=16 diff --git a/apps/server/src/agents/token-harvester.ts b/apps/server/src/agents/token-harvester.ts index ca1b9f7a..9c8222f1 100644 --- a/apps/server/src/agents/token-harvester.ts +++ b/apps/server/src/agents/token-harvester.ts @@ -82,7 +82,7 @@ async function parseClaudeSessionTokenUsage( filePath: string ): Promise { const sessionId = path.basename(filePath, ".jsonl"); - return mountIO.run(`read:claude:${sessionId}`, async (signal) => { + return mountIO.run(`read:claude:${sessionId}`, async (signal, heartbeat) => { const totals = new Map(); let sessionStart: string | null = null; let sessionEnd: string | null = null; @@ -93,6 +93,9 @@ async function parseClaudeSessionTokenUsage( }); for await (const line of rl) { + // Progress made — re-arm the idle deadline so a slow-but-streaming read + // is not mistaken for a stall. + heartbeat(); if (!line.trim()) continue; let entry: Record; try { @@ -208,7 +211,8 @@ type CodexTokenUsage = { * Recursively discover all rollout JSONL files under ~/.codex/sessions/. * Files are organized as YYYY/MM/DD/rollout-*.jsonl. */ -async function discoverCodexRolloutFiles(): Promise { +/** Exported for testing (stall behavior); no production caller besides harvest. */ +export async function discoverCodexRolloutFiles(): Promise { const files: string[] = []; async function walk(dir: string): Promise { @@ -235,10 +239,6 @@ async function discoverCodexRolloutFiles(): Promise { return files; } -export function discoverCodexRolloutFilesForTest(): Promise { - return discoverCodexRolloutFiles(); -} - /** * Parse a Codex rollout JSONL in a single pass: check for the agent tag * in the first 20 lines, then extract model and cumulative token usage. @@ -255,7 +255,7 @@ async function parseCodexRolloutForAgent( } | null> { return mountIO.run( `read:codex:${path.basename(filePath)}`, - async (signal) => { + async (signal, heartbeat) => { let matched = false; let lastUsage: CodexTokenUsage | null = null; let model = "unknown"; @@ -269,6 +269,8 @@ async function parseCodexRolloutForAgent( }); for await (const line of rl) { + // Progress made — re-arm the idle deadline (see Claude parser above). + heartbeat(); linesRead++; if (!matched) { if (linesRead > 20) break; diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index e65b37bb..533cd490 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -36,6 +36,7 @@ import { deleteSetting, getSetting, setSetting } from "./db/settings.js"; import { runCommand } from "./shared/lib/run-command.js"; import { shouldSkipAutomaticMacPathProbe } from "./shared/mac-path-privacy.js"; import { mimeType, resolveMediaDir } from "./shared/media.js"; +import { mountIO } from "./shared/mount-io/index.js"; import { handleMcpRequest } from "./shared/mcp/server.js"; import { readReleaseStore, writeReleaseStore } from "./release-store.js"; import { @@ -151,6 +152,9 @@ const app = Fastify({ ...(config.tls && { https: { cert: config.tls.cert, key: config.tls.key } }), }); const pool = createPool(config); +// Route mount-io breaker open/recover transitions to the app logger so +// operators get clear timestamps for a stalled gcsfuse mount. +mountIO.setLogger(app.log); const agentManager = new AgentManager(pool, app.log, config); const focusTracker = new FocusTracker(); const slackNotifier = new SlackNotifier(pool, app.log); diff --git a/apps/server/src/shared/mount-io/index.ts b/apps/server/src/shared/mount-io/index.ts index e3ca0f7d..e6e44867 100644 --- a/apps/server/src/shared/mount-io/index.ts +++ b/apps/server/src/shared/mount-io/index.ts @@ -11,8 +11,23 @@ const num = (raw: string | undefined, fallback: number): number => { return Number.isFinite(n) && n > 0 ? n : fallback; }; +// Process-wide singleton. The breaker and semaphore are global (not keyed by +// label), which assumes every caller reads the SAME mount — true today: both +// token-harvester sources (~/.claude/projects and ~/.codex/sessions) live under +// the home dir on the same gcsfuse mount, so a shared breaker is desirable (it +// avoids hammering a stalled mount from multiple sources). If a future caller +// reads a *different* mount root, key the breaker/semaphore by mount root (the +// `label` prefix passed to run() is the natural seam) so a stall in one mount +// does not suppress I/O against a healthy one. +// +// Default timeout is deliberately lenient (10s): harvesting is a periodic +// background task, and a healthy-but-slow mount (cold metadata cache, large +// readdir, transient GCS latency) can momentarily exceed a tight client +// timeout. A lenient timeout trades a little stall-detection latency for far +// fewer false breaker trips. Operators on high-latency mounts can raise it +// further via DISPATCH_MOUNT_IO_TIMEOUT_MS. export const mountIO = new MountIO({ - timeoutMs: num(process.env.DISPATCH_MOUNT_IO_TIMEOUT_MS, 5_000), + timeoutMs: num(process.env.DISPATCH_MOUNT_IO_TIMEOUT_MS, 10_000), maxConcurrency: num(process.env.DISPATCH_MOUNT_IO_CONCURRENCY, 2), breakerThreshold: num(process.env.DISPATCH_MOUNT_IO_BREAKER_THRESHOLD, 3), breakerCooldownMs: num( diff --git a/apps/server/src/shared/mount-io/mount-io.ts b/apps/server/src/shared/mount-io/mount-io.ts index c27d816e..73ee763d 100644 --- a/apps/server/src/shared/mount-io/mount-io.ts +++ b/apps/server/src/shared/mount-io/mount-io.ts @@ -1,4 +1,4 @@ -import { CircuitBreaker } from "./circuit-breaker.js"; +import { CircuitBreaker, type BreakerState } from "./circuit-breaker.js"; import { Semaphore } from "./semaphore.js"; export class MountStallError extends Error { @@ -15,18 +15,27 @@ export class MountUnavailableError extends Error { } } +/** Minimal structural logger (satisfied by pino / FastifyBaseLogger). */ +export type MountIOLogger = { + warn: (obj: Record, msg: string) => void; + info: (obj: Record, msg: string) => void; +}; + export type MountIOConfig = { timeoutMs: number; maxConcurrency: number; breakerThreshold: number; breakerCooldownMs: number; now?: () => number; + logger?: MountIOLogger; }; export class MountIO { private readonly sem: Semaphore; private readonly breaker: CircuitBreaker; private readonly timeoutMs: number; + private readonly breakerCooldownMs: number; + private logger?: MountIOLogger; constructor(cfg: MountIOConfig) { this.sem = new Semaphore(cfg.maxConcurrency); @@ -36,15 +45,59 @@ export class MountIO { cfg.now ); this.timeoutMs = cfg.timeoutMs; + this.breakerCooldownMs = cfg.breakerCooldownMs; + this.logger = cfg.logger; + } + + /** + * Attach a logger after construction (the process-wide singleton is built at + * import time, before the app logger exists). + */ + setLogger(logger: MountIOLogger): void { + this.logger = logger; } available(): boolean { return this.breaker.canProceed(); } + /** + * Run a single filesystem operation against the (possibly stalled) mount, + * guarded by a circuit breaker, a bounded-concurrency semaphore, and a + * timeout. + * + * IMPORTANT — the timeout *abandons*, it does not *cancel*. When the timeout + * fires we abort the signal, release the semaphore permit, and reject the + * caller with a `MountStallError`. But the underlying syscall keeps running + * on its libuv/Bun I/O thread until the kernel/gcsfuse unblocks it (which on + * a hard mount wedge may be never). Consequences callers must understand: + * + * 1. `fn` MUST honor the supplied `AbortSignal` (e.g. pass it to + * `createReadStream({ signal })`). A `fn` that ignores the signal turns + * every timeout into a *leaked* background op holding an I/O-pool + * thread. With enough leaks the pool is exhausted and ALL threadpool + * work server-wide (fs, dns, crypto, zlib) stalls — the exact failure + * this guard exists to prevent. Where the underlying API genuinely + * cannot take a signal (e.g. `fs.readdir`), the op is bounded only by + * the caller-side timeout and the residual leak risk is real — keep + * `maxConcurrency` and `breakerThreshold` low and raise + * `UV_THREADPOOL_SIZE` for the server process so a few wedged threads + * cannot starve the pool. + * 2. `available()` and the semaphore track JS-tracked ops, not in-flight + * kernel work. A released permit lets a new op dispatch even though the + * abandoned syscall's thread is still stuck. + * + * `fn` receives a `heartbeat()` callback. The timeout starts as a + * whole-operation deadline; each `heartbeat()` call re-arms it, turning it + * into an *idle* (no-progress) deadline. Streaming readers should call + * `heartbeat()` per chunk/line so a slow-but-progressing read over a healthy + * mount is not mistaken for a stall — only a genuine no-data stall trips the + * breaker. Single-shot ops (e.g. readdir) simply never heartbeat and keep + * the whole-operation deadline. + */ async run( label: string, - fn: (signal: AbortSignal) => Promise + fn: (signal: AbortSignal, heartbeat: () => void) => Promise ): Promise { if (!this.breaker.canProceed()) { throw new MountUnavailableError(label); @@ -53,33 +106,73 @@ export class MountIO { await this.sem.acquire(); const ac = new AbortController(); let timer: ReturnType | null = null; + let settled = false; try { - const work = fn(ac.signal); - work.catch(() => {}); - + let rejectTimeout!: (err: Error) => void; const timeout = new Promise((_, reject) => { + rejectTimeout = reject; + }); + + // (Re)arm the deadline. Called once up front (whole-op deadline) and + // again on every heartbeat (turning it into an idle deadline). + const arm = (): void => { + if (settled) return; + if (timer) clearTimeout(timer); timer = setTimeout(() => { const err = new MountStallError(label, this.timeoutMs); ac.abort(err); - reject(err); + rejectTimeout(err); }, this.timeoutMs); - }); + }; + arm(); + + const work = fn(ac.signal, arm); + // Load-bearing: when the timeout wins the race, `work` has no other + // rejection handler and rejects later (once the abandoned op settles via + // abort). Without this no-op catch that becomes an unhandledRejection, + // which under Node's default policy can terminate the process. Do not + // remove. + work.catch(() => {}); const result = await Promise.race([work, timeout]); - this.breaker.recordSuccess(); + this.recordSuccess(); return result; } catch (err) { + // Only a genuine stall trips the breaker. Per-file errors (ENOENT, parse + // failures, permission errors) are intentionally neutral — a single bad + // or absent session file must never open the process-wide circuit and + // suppress harvesting for every agent. Do not broaden this to all errors. if (err instanceof MountStallError) { - this.breaker.recordFailure(); + this.recordFailure(); } throw err; } finally { + settled = true; if (timer) clearTimeout(timer); this.sem.release(); } } + private recordSuccess(): void { + const prev = this.breaker.state(); + this.breaker.recordSuccess(); + if (prev !== "closed") { + this.logger?.info({}, "mount-io: mount recovered, harvesting resumed"); + } + } + + private recordFailure(): void { + const prev: BreakerState = this.breaker.state(); + this.breaker.recordFailure(); + if (prev !== "open" && this.breaker.state() === "open") { + this.logger?.warn( + { cooldownMs: this.breakerCooldownMs }, + "mount-io: breaker opened after mount stall, pausing harvest" + ); + } + } + reset(): void { this.breaker.reset(); } diff --git a/apps/server/test/token-harvester-stall.test.ts b/apps/server/test/token-harvester-stall.test.ts index 2abd3497..f98cb7d6 100644 --- a/apps/server/test/token-harvester-stall.test.ts +++ b/apps/server/test/token-harvester-stall.test.ts @@ -9,7 +9,7 @@ import { readdir } from "node:fs/promises"; import { discoverSessionFiles, - discoverCodexRolloutFilesForTest, + discoverCodexRolloutFiles, } from "../src/agents/token-harvester.js"; import { mountIO } from "../src/shared/mount-io/index.js"; @@ -31,7 +31,7 @@ describe("discoverSessionFiles under a stalled mount", () => { const p = discoverSessionFiles("/mnt/claude/projects/whatever"); const assertion = expect(p).resolves.toEqual([]); - await vi.advanceTimersByTimeAsync(5001); + await vi.advanceTimersByTimeAsync(10_001); await assertion; }); @@ -64,9 +64,9 @@ describe("codex rollout discovery under a stalled mount", () => { vi.mocked(readdir).mockReturnValue( new Promise(() => {}) as unknown as ReturnType ); - const p = discoverCodexRolloutFilesForTest(); + const p = discoverCodexRolloutFiles(); const assertion = expect(p).resolves.toEqual([]); - await vi.advanceTimersByTimeAsync(5001); + await vi.advanceTimersByTimeAsync(10_001); await assertion; }); }); diff --git a/bin/dispatch-launchd-wrapper b/bin/dispatch-launchd-wrapper index d116a963..11bd6da1 100755 --- a/bin/dispatch-launchd-wrapper +++ b/bin/dispatch-launchd-wrapper @@ -11,6 +11,17 @@ export TERM="${TERM:-xterm-256color}" +# Enlarge the I/O thread pool. The token harvester reads agent session files +# from a gcsfuse mount; on a mount stall an in-flight readdir/read syscall +# cannot be cancelled and stays blocked on an I/O-pool thread until the kernel +# unblocks it. The default libuv pool is small (4), so a few wedged harvester +# threads could otherwise starve all other fs/dns/crypto work server-wide. +# NOTE: production runs the Bun binary, which uses its own I/O threadpool (not +# libuv) and may not honor this knob — it hardens Node-based dev/test paths, +# and the mount-io breaker's low concurrency/threshold is the primary backstop +# on the wedged-syscall leak regardless of runtime. Override before launch. +export UV_THREADPOOL_SIZE="${UV_THREADPOOL_SIZE:-16}" + ROOT_DIR="$(cd "$(dirname "$0")/.." && pwd)" OS_NAME="$(uname -s)" ARCH_NAME="$(uname -m)"