Compare commits

...
5 changed files with 157 additions and 146 deletions

No files matched your search

+5 -32
View File
@@ -3,7 +3,6 @@ import { performance } from "node:perf_hooks";
import { Worker } from "node:worker_threads";
import type { MiddlewareHandler } from "hono";
export const SLOW_OPERATION_BARRIER_TIMEOUT_MS = 250;
export const SLOW_OPERATION_THRESHOLD_MS = 250;
export const SLOW_OPERATION_ROTATION_BYTES = 1024 * 1024;
export const SLOW_OPERATION_ROTATION_COUNT = 3;
@@ -89,7 +88,6 @@ interface SlowOpRecorderOptions {
maxBytes?: number;
rotationCount?: number;
slowThresholdMs?: number;
barrierTimeoutMs?: number;
}
interface Span {
@@ -141,13 +139,7 @@ const WORKER_SOURCE = String.raw`
ok = false;
error = caught instanceof Error ? caught.message : String(caught);
}
if (message.signal) {
const state = new Int32Array(message.signal);
Atomics.store(state, 0, ok ? 1 : 2);
Atomics.notify(state, 0);
} else if (message.id) {
parentPort.postMessage({ id: message.id, ok, error });
}
if (message.id) parentPort.postMessage({ id: message.id, ok, error });
});
`;
@@ -157,7 +149,6 @@ export class SlowOpRecorder implements SlowOperationInstrumentation {
private readonly maxBytes: number;
private readonly rotationCount: number;
private readonly slowThresholdMs: number;
private readonly barrierTimeoutMs: number;
private readonly pending = new Map<string, { resolve: () => void; reject: (error: unknown) => void }>();
private degraded: SlowOperationSnapshot = { healthy: true };
private degradedHandler?: (snapshot: Required<Pick<SlowOperationSnapshot, "reason" | "site">>) => void;
@@ -181,7 +172,6 @@ export class SlowOpRecorder implements SlowOperationInstrumentation {
this.maxBytes = options.maxBytes ?? SLOW_OPERATION_ROTATION_BYTES;
this.rotationCount = options.rotationCount ?? SLOW_OPERATION_ROTATION_COUNT;
this.slowThresholdMs = options.slowThresholdMs ?? SLOW_OPERATION_THRESHOLD_MS;
this.barrierTimeoutMs = options.barrierTimeoutMs ?? SLOW_OPERATION_BARRIER_TIMEOUT_MS;
this.worker = new Worker(WORKER_SOURCE, { eval: true, execArgv: [] });
this.worker.unref();
this.worker.on("message", (message: { id?: string; ok?: boolean }) => {
@@ -220,27 +210,10 @@ export class SlowOpRecorder implements SlowOperationInstrumentation {
beginSyncSpan(site: string): Span {
const span: Span = { spanId: randomUUID(), site, startedAt: performance.now() };
const signal = new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT);
const state = new Int32Array(signal);
try {
this.worker.postMessage({
type: "append",
signal,
logPath: this.logPath,
maxBytes: this.maxBytes,
rotationCount: this.rotationCount,
record: { v: 1, ts: new Date().toISOString(), spanId: span.spanId, phase: "begin", site },
});
} catch {
// Synchronous postMessage failure means the Worker is gone: degrade
// through the terminal transition but still return the span so the
// wrapped operation runs to completion (its value/error stays exact).
this.handleTerminalFailure("recorder_post_failed", site);
return span;
}
const wait = Atomics.wait(state, 0, 0, this.barrierTimeoutMs);
if (wait === "timed-out") this.markDegraded("begin_barrier_timeout", site);
else if (Atomics.load(state, 0) !== 1) this.markDegraded("begin_barrier_failed", site);
// Queue before the operation without waiting on worker startup or fsync. The
// worker can write even while the main thread is wedged; only flush/close
// acknowledge durability. A crash before the worker appends can lose begin.
this.appendAsync({ v: 1, ts: new Date().toISOString(), spanId: span.spanId, phase: "begin", site });
return span;
}
@@ -1,6 +1,6 @@
import { spawnSync } from "node:child_process";
import { execFile } from "node:child_process";
import type { TmuxAdapter } from "../adapters/tmux.js";
import { runSyncSite } from "./sync-site-wrap.js";
import { runAsyncSite } from "./sync-site-wrap.js";
export interface TmuxOptionDefaultsDeps {
tmuxAdapter: TmuxAdapter;
@@ -17,11 +17,11 @@ export interface TmuxOptionDefaultsDeps {
*/
platform?: NodeJS.Platform;
/**
* OPR.0.4.6.02 S1 — cheap `command -v <bin>` probe for the linux
* copy-command fallback chain. Defaults to a real shell probe; injectable
* OPR.0.4.6.02 S1 — asynchronous `command -v <bin>` probe for the linux
* copy-command fallback chain. Defaults to a bounded asynchronous shell probe; injectable
* for deterministic tests.
*/
hasCommand?: (bin: string) => boolean;
hasCommand?: (bin: string) => boolean | Promise<boolean>;
}
/**
@@ -44,7 +44,7 @@ export class TmuxOptionDefaultsApplier {
private tmuxAdapter: TmuxAdapter;
private readTmuxOptionDefaults: () => { statusBar: boolean };
private platform: NodeJS.Platform;
private hasCommand: (bin: string) => boolean;
private hasCommand: (bin: string) => boolean | Promise<boolean>;
private serverDefaultsAsserted = false;
constructor(deps: TmuxOptionDefaultsDeps) {
@@ -120,7 +120,14 @@ export class TmuxOptionDefaultsApplier {
warnings.push(`tmux "set-clipboard" server option not set: ${clip.message}`);
}
const copyCommand = resolveCopyCommand(this.platform, this.hasCommand);
let copyCommand: string | null;
try {
copyCommand = await resolveCopyCommand(this.platform, this.hasCommand);
} catch {
this.serverDefaultsAsserted = false;
warnings.push("Clipboard lookup interrupted (250 ms limit); copy-command left unchanged (OSC 52 if unset). Will retry on the next fresh session.");
return;
}
if (copyCommand !== null) {
const cc = await this.tmuxAdapter.setServerOption("copy-command", copyCommand);
if (!cc.ok) {
@@ -141,14 +148,14 @@ export class TmuxOptionDefaultsApplier {
* Pure: platform + a `command -v` probe in, the command string (or null)
* out. One table, no side effects — the applier decides whether to write.
*/
export function resolveCopyCommand(
export async function resolveCopyCommand(
platform: NodeJS.Platform,
hasCommand: (bin: string) => boolean,
): string | null {
hasCommand: (bin: string) => boolean | Promise<boolean>,
): Promise<string | null> {
if (platform === "darwin") return "pbcopy";
if (platform === "linux") {
if (hasCommand("wl-copy")) return "wl-copy";
if (hasCommand("xclip")) return "xclip -selection clipboard -i";
if (await hasCommand("wl-copy")) return "wl-copy";
if (await hasCommand("xclip")) return "xclip -selection clipboard -i";
return null;
}
return null;
@@ -156,16 +163,18 @@ export function resolveCopyCommand(
/**
* OPR.0.4.6.02 S1 — default `command -v <bin>` probe (POSIX shell builtin).
* Returns true when the binary resolves on PATH. Cheap + best-effort; any
* failure (spawn error, non-shell env) reads as absent.
* Returns true when the binary resolves on PATH. A killed probe rejects so
* the applier can warn and retry; other failures read as absent.
*/
function defaultHasCommand(bin: string): boolean {
async function defaultHasCommand(bin: string): Promise<boolean> {
try {
const r = runSyncSite("tmux_options.command_v", () =>
spawnSync(`command -v ${bin}`, { shell: true, stdio: "ignore" })
);
return r.status === 0;
} catch {
return await runAsyncSite("tmux_options.command_v", () => new Promise<boolean>((resolve, reject) => {
execFile("/bin/sh", ["-c", 'command -v "$1"', "--", bin], {
timeout: 250, killSignal: "SIGKILL", maxBuffer: 1024,
}, error => error?.killed ? reject(error) : resolve(!error));
}));
} catch (error) {
if ((error as { killed?: boolean } | null)?.killed) throw error;
return false;
}
}
+49 -83
View File
@@ -5,7 +5,7 @@ import os from "node:os";
import path from "node:path";
import { pathToFileURL } from "node:url";
import type { Worker } from "node:worker_threads";
import { afterEach, describe, expect, it } from "vitest";
import { afterEach, describe, expect, it, vi } from "vitest";
import { Hono } from "hono";
import { createDaemon } from "../src/startup.js";
import { createSlowOpRequestMiddleware } from "../src/domain/slow-op-recorder.js";
@@ -67,12 +67,37 @@ afterEach(() => {
});
describe("SlowOpRecorder locked instrumentation contract", () => {
it("returns promptly while begin acknowledgements are delayed", async () => {
const mod = await loadRecorderModule();
if (!mod) throw new Error("recorder module missing");
const logPath = path.join(tempDir(), "pending.jsonl");
const recorder = new mod.SlowOpRecorder({ logPath });
const worker: Worker = (recorder as any).worker;
const post = worker.postMessage;
const held: unknown[] = [];
worker.postMessage = message => {
if (message?.record?.phase === "begin") held.push(message);
else post.call(worker, message);
};
try {
const start = performance.now();
const values = [1, 2, 3].map(value => recorder.runSync("test.delayed.begin", () => value));
const elapsedMs = performance.now() - start;
expect(values).toEqual([1, 2, 3]);
expect(held).toHaveLength(3);
expect(elapsedMs).toBeLessThan(100);
} finally {
worker.postMessage = post;
for (const message of held) post.call(worker, message);
await recorder.close();
}
});
it("pins the absolute boundary, rotation, filename, and secret-safe record contract", async () => {
const mod = await loadRecorderModule();
expect(mod, "slow-op-recorder production module is missing").not.toBeNull();
if (!mod) return;
expect(mod.SLOW_OPERATION_BARRIER_TIMEOUT_MS).toBe(250);
expect(mod.SLOW_OPERATION_THRESHOLD_MS).toBe(250);
expect(mod.SLOW_OPERATION_ROTATION_BYTES).toBe(1024 * 1024);
expect(mod.SLOW_OPERATION_ROTATION_COUNT).toBe(3);
@@ -85,9 +110,6 @@ describe("SlowOpRecorder locked instrumentation contract", () => {
process.env.OPENRIG_TEST_SECRET = secret;
try {
const value = recorder.runSync("test.sync.boundary", () => {
const records = fs.readFileSync(logPath, "utf8");
expect(records).toContain('"phase":"begin"');
expect(records).toContain('"site":"test.sync.boundary"');
return 41;
});
expect(value).toBe(41);
@@ -107,88 +129,34 @@ describe("SlowOpRecorder locked instrumentation contract", () => {
}
});
it("leaves a durable open span after the begin barrier and exposes fail-visible degradation", async () => {
it("flushes a queued open span and exposes acknowledged write failure in health", async () => {
const mod = await loadRecorderModule();
expect(mod, "slow-op-recorder production module is missing").not.toBeNull();
if (!mod) return;
if (!mod) throw new Error("recorder module missing");
const dir = tempDir();
const logPath = path.join(dir, "slow-operations.jsonl");
const logPath = path.join(dir, "open.jsonl");
const recorder = new mod.SlowOpRecorder({ logPath });
const span = recorder.beginSyncSpan("test.sync.wedge");
expect(span).toBeTruthy();
const records = await readRecords(logPath);
expect(records).toHaveLength(1);
expect(records[0]).toMatchObject({ phase: "begin", site: "test.sync.wedge" });
expect(recorder.snapshot()).toMatchObject({ healthy: true });
recorder.beginSyncSpan("test.sync.wedge");
await recorder.flush();
expect(await readRecords(logPath)).toMatchObject([{ phase: "begin", site: "test.sync.wedge" }]);
await recorder.close();
const degraded = new mod.SlowOpRecorder({
logPath: path.join(dir, "degraded.jsonl"),
barrierTimeoutMs: 0,
});
const degraded = new mod.SlowOpRecorder({ logPath: dir });
const oldNoKernel = process.env.OPENRIG_NO_KERNEL;
process.env.OPENRIG_NO_KERNEL = "1";
const daemon = await createTestDaemon({
dbPath: ":memory:",
tmuxExec: async () => "",
cmuxExec: async () => "",
slowOpRecorder: degraded,
} as never);
const daemon = await createTestDaemon({ dbPath: ":memory:", tmuxExec: async () => "", cmuxExec: async () => "", slowOpRecorder: degraded } as never);
try {
const worker: Worker = (degraded as any).worker;
const postMessage = worker.postMessage;
let intercepted = 0;
let withheldSignal: SharedArrayBuffer | undefined;
try {
worker.postMessage = (message, transferList) => {
if (message?.type === "append"
&& message.record?.phase === "begin"
&& message.record?.site === "test.sync.degraded"
&& message.signal instanceof SharedArrayBuffer) {
intercepted++;
const { signal, ...withoutSignal } = message;
withheldSignal = signal;
// Append through the real worker, but withhold this acknowledgement.
return postMessage.call(worker, withoutSignal, transferList);
}
return postMessage.call(worker, message, transferList);
};
expect(degraded.runSync("test.sync.degraded", () => "continued")).toBe("continued");
} finally {
worker.postMessage = postMessage;
}
expect(intercepted).toBe(1);
expect(worker.postMessage).toBe(postMessage);
expect(withheldSignal).toBeInstanceOf(SharedArrayBuffer);
expect(new Int32Array(withheldSignal!)[0]).toBe(0);
expect(degraded.snapshot()).toMatchObject({
healthy: false,
reason: "begin_barrier_timeout",
site: "test.sync.degraded",
});
const healthResponse = await daemon.app.request("/healthz");
expect(await healthResponse.json()).toMatchObject({
status: "ok",
slowOperations: {
healthy: false,
reason: "begin_barrier_timeout",
site: "test.sync.degraded",
},
});
expect(degraded.runSync("test.sync.degraded", () => "continued")).toBe("continued");
await expect(degraded.flush()).rejects.toBeInstanceOf(mod.SlowOpRecorderWriteError);
expect(degraded.snapshot()).toMatchObject({ healthy: false, reason: "recorder_write_failed", site: "recorder.worker" });
const response = await daemon.app.request("/healthz");
expect(await response.json()).toMatchObject({ status: "ok", slowOperations: { healthy: false, reason: "recorder_write_failed" } });
const observations = daemon.deps.streamStore!.list();
expect(observations).toHaveLength(1);
expect(observations[0]!.body).toContain("begin_barrier_timeout");
expect(observations[0]!.body).toContain("test.sync.degraded");
await degraded.flush();
const degradedRecords = await readRecords(path.join(dir, "degraded.jsonl"));
expect(degradedRecords.filter((record) => record.phase === "begin"
&& record.site === "test.sync.degraded")).toHaveLength(1);
expect(observations[0]!.body).toContain("recorder_write_failed");
} finally {
daemon.eventLoopMonitor.stop();
daemon.db.close();
await degraded.close();
await degraded.close().catch(() => {});
if (oldNoKernel === undefined) delete process.env.OPENRIG_NO_KERNEL;
else process.env.OPENRIG_NO_KERNEL = oldNoKernel;
}
@@ -211,20 +179,17 @@ describe("SlowOpRecorder locked instrumentation contract", () => {
});
expect(value).toBe("continued");
expect(ran).toBe(true);
expect(recorder.snapshot()).toEqual({
healthy: false,
reason: "begin_barrier_failed",
site: "test.sync.observer_failure",
});
await expect(recorder.flush()).rejects.toBeInstanceOf(mod.SlowOpRecorderWriteError);
expect(recorder.snapshot()).toEqual({ healthy: false, reason: "recorder_write_failed", site: "recorder.worker" });
} finally {
// This recorder's log path is intentionally unwritable (to force the
// begin-barrier failure above), so close() now correctly rejects on the
// asynchronously acknowledged write failure above), so close() now correctly rejects on the
// latched lost-write durability — tolerate it in teardown.
await recorder.close().catch(() => {});
}
});
it("durably records an open span before a main-thread sync wedge is killed", async () => {
it("writes a queued begin while the main thread is wedged, before a later kill", async () => {
const dir = tempDir();
const logPath = path.join(dir, "crash.jsonl");
fs.writeFileSync(logPath, "", { mode: 0o600 });
@@ -278,6 +243,7 @@ describe("SlowOpRecorder locked instrumentation contract", () => {
});
});
expect(callbackEvidence).toEqual({ isMainThread: true, threadId: 0 });
await vi.waitFor(async () => expect(await readRecords(logPath)).toMatchObject([{ phase: "begin", site: "test.sync.crash" }]));
child.kill("SIGKILL");
await once(child, "exit");
@@ -351,7 +317,7 @@ describe("SlowOpRecorder locked instrumentation contract", () => {
expect(JSON.stringify(requests)).not.toContain("must-not-appear");
});
it("wraps all nine accepted sync calls without moving them off the main thread", () => {
it("keeps the accepted slow-operation site labels", () => {
const srcRoot = path.resolve(import.meta.dirname, "../src");
for (const [relativePath, sites] of expectedSites) {
const source = fs.readFileSync(path.join(srcRoot, relativePath), "utf8");
@@ -496,7 +462,7 @@ describe("SlowOpRecorder locked instrumentation contract", () => {
expect((await app.request("/definitely-missing")).status).toBe(404);
});
it("drives the real tmux command-v spawnSync site through a hermetic daemon", async () => {
it("drives the asynchronous tmux command-v site through a hermetic daemon", async () => {
const mod = await loadRecorderModule();
expect(mod, "slow-op-recorder production module is missing").not.toBeNull();
if (!mod) return;
@@ -0,0 +1,45 @@
import { beforeEach, describe, expect, it, vi } from "vitest";
import type { TmuxAdapter } from "../src/adapters/tmux.js";
import { TmuxOptionDefaultsApplier } from "../src/domain/tmux-option-defaults.js";
const execute = vi.hoisted(() => vi.fn());
vi.mock("node:child_process", () => ({ execFile: execute }));
beforeEach(() => { execute.mockReset(); });
describe("clipboard discovery deadline", () => {
it.each([null, "custom-copy"])("warns and retries a killed default probe, preserving %j", async initial => {
let copyCommand = initial;
const setServerOption = vi.fn(async (name: string, value: string) => {
if (name === "copy-command") copyCommand = value;
return { ok: true };
});
const adapter = { setSessionOption: async () => ({ ok: true }), setServerOption } as unknown as TmuxAdapter;
execute.mockImplementationOnce((_file, _args, options, callback) => {
expect(options).toMatchObject({ timeout: 250, killSignal: "SIGKILL" });
callback(Object.assign(new Error("probe deadline"), { killed: true }));
}).mockImplementationOnce((_file, _args, _options, callback) => callback(null));
const applier = new TmuxOptionDefaultsApplier({ tmuxAdapter: adapter, platform: "linux" });
const warnings = await applier.applyToFreshSession("first@test");
expect(warnings).toHaveLength(1);
expect(warnings[0]).toMatch(/clipboard.*250 ms/i);
expect(warnings[0]).toMatch(/retry/i);
expect(copyCommand).toBe(initial);
expect(execute).toHaveBeenCalledTimes(1);
expect(await applier.applyToFreshSession("next@test")).toEqual([]);
expect(copyCommand).toBe("wl-copy");
expect(execute).toHaveBeenCalledTimes(2);
expect(await applier.applyToFreshSession("later@test")).toEqual([]);
expect(execute).toHaveBeenCalledTimes(2);
});
it("memoizes confirmed absence without a deadline warning", async () => {
execute.mockImplementation((_file, _args, _options, callback) => callback(Object.assign(new Error("absent"), { code: 1 })));
const setServerOption = vi.fn(async () => ({ ok: true }));
const adapter = { setSessionOption: async () => ({ ok: true }), setServerOption } as unknown as TmuxAdapter;
const applier = new TmuxOptionDefaultsApplier({ tmuxAdapter: adapter, platform: "linux" });
expect(await applier.applyToFreshSession("first@test")).toEqual([]);
expect(await applier.applyToFreshSession("next@test")).toEqual([]);
expect(execute).toHaveBeenCalledTimes(2);
expect(setServerOption).not.toHaveBeenCalledWith("copy-command", expect.anything());
});
});
@@ -23,6 +23,24 @@ function mockAdapter(overrides?: {
}
describe("TmuxOptionDefaultsApplier", () => {
it("awaits an asynchronous clipboard lookup without holding unrelated work", async () => {
const { adapter, setServerOption } = mockAdapter();
let release!: (present: boolean) => void;
const pending = new Promise<boolean>(resolve => { release = resolve; });
const probe = vi.fn((bin: string) => bin === "wl-copy" ? pending : Promise.resolve(true));
let finished = false;
const applier = new TmuxOptionDefaultsApplier({ tmuxAdapter: adapter, platform: "linux", hasCommand: probe });
const applied = applier.applyToFreshSession("async-clipboard@test").then(value => { finished = true; return value; });
try {
await new Promise(resolve => setImmediate(resolve));
expect(finished).toBe(false);
expect(setServerOption).not.toHaveBeenCalledWith("copy-command", expect.anything());
} finally { release(false); }
expect(await applied).toEqual([]);
expect(probe.mock.calls.map(([bin]) => bin)).toEqual(["wl-copy", "xclip"]);
expect(setServerOption).toHaveBeenCalledWith("copy-command", "xclip -selection clipboard -i");
});
it("sets mouse on + status off (default) on the given session via SESSION scope only", async () => {
const { adapter, setSessionOption, setServerOption } = mockAdapter();
// default reader (omitted) → statusBar false.
@@ -109,20 +127,20 @@ describe("TmuxOptionDefaultsApplier", () => {
});
});
describe("resolveCopyCommand (pure per-platform table)", () => {
it("darwin → pbcopy", () => {
expect(resolveCopyCommand("darwin", () => false)).toBe("pbcopy");
describe("resolveCopyCommand (pure per-platform table)", async () => {
it("darwin → pbcopy", async () => {
expect(await resolveCopyCommand("darwin", () => false)).toBe("pbcopy");
});
it("linux → wl-copy when present", () => {
expect(resolveCopyCommand("linux", (b) => b === "wl-copy")).toBe("wl-copy");
it("linux → wl-copy when present", async () => {
expect(await resolveCopyCommand("linux", (b) => b === "wl-copy")).toBe("wl-copy");
});
it("linux → xclip when wl-copy absent but xclip present", () => {
expect(resolveCopyCommand("linux", (b) => b === "xclip")).toBe("xclip -selection clipboard -i");
it("linux → xclip when wl-copy absent but xclip present", async () => {
expect(await resolveCopyCommand("linux", (b) => b === "xclip")).toBe("xclip -selection clipboard -i");
});
it("linux → null when neither present (OSC 52 fallback)", () => {
expect(resolveCopyCommand("linux", () => false)).toBeNull();
it("linux → null when neither present (OSC 52 fallback)", async () => {
expect(await resolveCopyCommand("linux", () => false)).toBeNull();
});
it("other platforms → null", () => {
expect(resolveCopyCommand("win32", () => true)).toBeNull();
it("other platforms → null", async () => {
expect(await resolveCopyCommand("win32", () => true)).toBeNull();
});
});