Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
- **fix(logging):** run source call-log artifact workers with native Node.js TypeScript support so idle shutdown and forced termination do not leave esbuild zombies.
2 changes: 1 addition & 1 deletion src/lib/logEnv.ts
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
import path from "path";
import { resolveDataDir } from "@/lib/dataPaths";
import { resolveDataDir } from "./dataPaths.ts";

const DEFAULT_APP_LOG_RETENTION_DAYS = 7;
const DEFAULT_CALL_LOG_RETENTION_DAYS = 7;
Expand Down
25 changes: 4 additions & 21 deletions src/lib/usage/callLogArtifactWriter.ts
Original file line number Diff line number Diff line change
Expand Up @@ -72,34 +72,20 @@ export function resolveCallLogArtifactWorker(context: WorkerResolutionContext =
const cwdJs = path.resolve(cwd, "src/lib/usage/callLogArtifactWorker.js");
if (exists(cwdJs)) return { workerFile: cwdJs, execArgv: [] };

// Native TypeScript workers avoid esbuild zombies when terminated.
const moduleTs = path.join(moduleDir, "callLogArtifactWorker.ts");
if (exists(moduleTs)) {
return { workerFile: moduleTs, execArgv: ["--import", "tsx/esm"] };
return { workerFile: moduleTs, execArgv: [] };
}

const cwdTs = path.resolve(cwd, "src/lib/usage/callLogArtifactWorker.ts");
if (exists(cwdTs)) {
return { workerFile: cwdTs, execArgv: ["--import", "tsx/esm"] };
return { workerFile: cwdTs, execArgv: [] };
}

return { workerFile: entryJs ?? cwdJs, execArgv: [] };
}

/**
* Worker options for a resolved worker file. A source checkout runs the worker through tsx, which
* lists its whole disk cache (os.tmpdir()/tsx-<uid>) with one readdirSync the first time a thread
* transforms a file. A shared dev box grows that directory to hundreds of thousands of entries, and
* the worker's first write then waited 5-7 s on the listing (0.4 s without it). The worker only
* transforms a handful of small modules, so it runs without the disk cache.
*/
export function callLogWorkerOptions(execArgv: string[]): {
execArgv: string[];
env?: NodeJS.ProcessEnv;
} {
if (!execArgv.includes("tsx/esm")) return { execArgv };
return { execArgv, env: { ...process.env, TSX_DISABLE_CACHE: "1" } };
}

function clearIdleTimer(): void {
if (!idleTimer) return;
clearTimeout(idleTimer);
Expand Down Expand Up @@ -174,10 +160,7 @@ function ensureWorker(): Worker {
const { workerFile, execArgv } = workerFileOverride ?? resolveCallLogArtifactWorker();
// Reflect.construct keeps Next/Turbopack from interpreting the runtime-selected
// worker path as a build-time glob and tracing tens of thousands of unrelated files.
const created = Reflect.construct(Worker, [
pathToFileURL(workerFile),
callLogWorkerOptions(execArgv),
]) as Worker;
const created = Reflect.construct(Worker, [pathToFileURL(workerFile), { execArgv }]) as Worker;
worker = created;
created.on("message", (reply: WorkerReply) => {
if (!active || reply.id !== active.id) return;
Expand Down
6 changes: 3 additions & 3 deletions src/lib/usage/callLogArtifacts.ts
Original file line number Diff line number Diff line change
@@ -1,14 +1,14 @@
import fs from "node:fs";
import path from "node:path";
import type { RequestPipelinePayloads } from "@omniroute/open-sse/utils/requestLogger.ts";
import { resolveWritableDataDir } from "../dataPaths";
import { getCallLogPipelineMaxSizeBytes, isChatDebugFileEnabled } from "../logEnv";
import { resolveWritableDataDir } from "../dataPaths.ts";
import { getCallLogPipelineMaxSizeBytes, isChatDebugFileEnabled } from "../logEnv.ts";
import {
CALL_LOG_SIZE_LIMIT_REASON as SIZE_LIMIT_EXCEEDED_REASON,
CALL_LOG_BODY_OMITTED_FOR_SIZE_LIMIT as OMITTED_FOR_SIZE_LIMIT,
CALL_LOG_STREAM_CHUNKS_OMITTED_FOR_SIZE_LIMIT as STREAM_CHUNKS_OMITTED_FOR_SIZE_LIMIT,
isSizeLimitOmissionMarker,
} from "@/shared/constants/callLogSizeLimitMarkers";
} from "../../shared/constants/callLogSizeLimitMarkers.ts";

const isCloud = typeof globalThis.caches === "object" && globalThis.caches !== null;
const isBuildPhase =
Expand Down
10 changes: 4 additions & 6 deletions tests/helpers/callLogSettle.ts
Original file line number Diff line number Diff line change
@@ -1,11 +1,9 @@
/**
* Bound for tests that wait on a call-log save or on its detail.
*
* The detail artifact is written by a worker thread that starts cold on its first write; a
* source checkout spawns it through tsx, and loading its import graph took 6.7–8.4 s on the
* shared devbox (14 s+ under load) against ~0.2 s on hosted CI. A call-log wait is a hang
* guard, not a latency budget (green runs return as soon as the save settles), so keep the
* bound well above that cold start. No imports on purpose: tests set DATA_DIR before they
* import the call-log modules.
* The detail artifact is written by a worker thread that starts cold on its first write.
* A call-log wait is a hang guard, not a latency budget: green runs return as soon as the
* save settles, so keep the bound generous under load. No imports on purpose: tests set
* DATA_DIR before they import the call-log modules.
*/
export const CALL_LOG_SETTLE_TIMEOUT_MS = 60_000;
182 changes: 182 additions & 0 deletions tests/unit/call-log-artifact-worker-native.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,182 @@
import test from "node:test";
import assert from "node:assert/strict";
import fs from "node:fs";
import os from "node:os";
import path from "node:path";
import { Worker } from "node:worker_threads";
import { spawnSync } from "node:child_process";
import { pathToFileURL } from "node:url";

const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-call-log-native-"));
process.env.DATA_DIR = TEST_DATA_DIR;

const { writeCallArtifactAsync, closeCallLogArtifactWriter } =
await import("../../src/lib/usage/callLogArtifactWriter.ts");

test.after(async () => {
await closeCallLogArtifactWriter(0);
fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true, maxRetries: 5, retryDelay: 100 });
});

function buildArtifact(id: string) {
return {
schemaVersion: 5 as const,
summary: {
id,
timestamp: "2026-10-28T12:34:56.789Z",
method: "POST",
path: "/v1/chat/completions",
status: 200,
model: "test-model",
requestedModel: null,
provider: "test-provider",
account: "test-account",
connectionId: null,
duration: 10,
tokens: {
in: 1,
out: 2,
cacheRead: null,
cacheWrite: null,
reasoning: null,
compressed: null,
},
requestType: "chat",
sourceFormat: "openai",
targetFormat: "openai",
apiKeyId: null,
apiKeyName: null,
comboName: null,
comboStepId: null,
comboExecutionKey: null,
},
requestBody: { messages: [{ role: "user", content: id }] },
responseBody: { content: "written by the source worker" },
error: null,
};
}

function esbuildChildren(): number[] {
if (process.platform !== "linux") return [];
const children = new Set<number>();
for (const task of fs.readdirSync("/proc/self/task")) {
try {
const pids = fs.readFileSync(`/proc/self/task/${task}/children`, "utf8").trim();
for (const pid of pids.split(/\s+/).filter(Boolean)) {
if (fs.readFileSync(`/proc/${pid}/comm`, "utf8").trim() === "esbuild") {
children.add(Number(pid));
}
}
} catch (error) {
if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error;
}
}
return [...children];
}

test("native source writes survive idle restarts and forced close without esbuild children", async (t) => {
const baseline = new Set(esbuildChildren());
const extraChildren = new Set<number>();
const workers = new Set<Worker>();
const terminations: Promise<number>[] = [];
const postMessage = Worker.prototype.postMessage;
const terminate = Worker.prototype.terminate;
t.mock.method(
Worker.prototype,
"postMessage",
function (this: Worker, ...args: Parameters<Worker["postMessage"]>) {
workers.add(this);
return postMessage.apply(this, args);
}
);
t.mock.method(Worker.prototype, "terminate", function (this: Worker) {
const terminated = terminate.call(this);
terminations.push(terminated);
return terminated;
});
t.mock.timers.enable({ apis: ["setTimeout"] });

const census = () => {
for (const pid of esbuildChildren()) {
if (!baseline.has(pid)) extraChildren.add(pid);
}
};

try {
for (let cycle = 0; cycle < 3; cycle++) {
const artifact = buildArtifact(`native-cycle-${cycle}`);
const result = await writeCallArtifactAsync(artifact);
assert.ok(result, "the real source worker must write an artifact");
const serialized = fs.readFileSync(
path.join(TEST_DATA_DIR, "call_logs", result.relPath),
"utf8"
);
assert.deepEqual(JSON.parse(serialized), artifact);
assert.equal(result.sizeBytes, Buffer.byteLength(serialized));
assert.match(result.sha256, /^[0-9a-f]{8}$/);
assert.equal(workers.size, cycle + 1, "an idle worker must be restarted");
census();

if (cycle < 2) {
t.mock.timers.tick(29_999);
assert.equal(terminations.length, cycle, "keep the existing idle timeout");
t.mock.timers.tick(1);
assert.equal(terminations.length, cycle + 1);
await Promise.all(terminations);
census();
}
}

const pending = [
writeCallArtifactAsync(buildArtifact("forced-active")),
writeCallArtifactAsync(buildArtifact("forced-queued")),
];
await closeCallLogArtifactWriter(0);
assert.deepEqual(await Promise.all(pending), [null, null]);
await Promise.all(terminations);
assert.equal(terminations.length, 3);
census();
assert.equal(await writeCallArtifactAsync(buildArtifact("after-close")), null);
assert.deepEqual(
[...extraChildren],
[],
"worker lifecycles must not spawn esbuild children/zombies"
);
} finally {
await closeCallLogArtifactWriter(0);
await Promise.all(terminations);
t.mock.timers.reset();
}
});

test("source worker does not inherit CLI preloads", () => {
const preloadFile = path.join(TEST_DATA_DIR, "main-thread-only.mjs");
fs.writeFileSync(
preloadFile,
'import { isMainThread } from "node:worker_threads";\n' +
'if (!isMainThread) throw new Error("inherited CLI preload");\n'
);
const childFile = path.join(TEST_DATA_DIR, "source-writer-child.mjs");
const writerUrl = new URL("../../src/lib/usage/callLogArtifactWriter.ts", import.meta.url).href;
fs.writeFileSync(
childFile,
[
`import { writeCallArtifactAsync, closeCallLogArtifactWriter } from ${JSON.stringify(writerUrl)};`,
`const result = await writeCallArtifactAsync(${JSON.stringify(buildArtifact("cli-preloads"))});`,
"await closeCallLogArtifactWriter();",
'if (!result) throw new Error("source write failed");',
"console.log(JSON.stringify(result));",
].join("\n")
);
const child = spawnSync(
process.execPath,
["--import", "tsx/esm", "--import", pathToFileURL(preloadFile).href, childFile],
{ encoding: "utf8", timeout: 20_000 }
);
assert.equal(child.status, 0, child.stderr);
const result = JSON.parse(child.stdout);
assert.deepEqual(
JSON.parse(fs.readFileSync(path.join(TEST_DATA_DIR, "call_logs", result.relPath), "utf8")),
buildArtifact("cli-preloads")
);
});
94 changes: 0 additions & 94 deletions tests/unit/call-log-artifact-worker-tsx-cache.test.ts

This file was deleted.

Loading
Loading