diff --git a/open-sse/services/compression/compressionWorkerPool.ts b/open-sse/services/compression/compressionWorkerPool.ts index 109940a14d..352aabd39c 100644 --- a/open-sse/services/compression/compressionWorkerPool.ts +++ b/open-sse/services/compression/compressionWorkerPool.ts @@ -1,6 +1,5 @@ import { existsSync } from "node:fs"; -import { dirname, join } from "node:path"; -import { fileURLToPath, pathToFileURL } from "node:url"; +import { dirname, join, resolve } from "node:path"; import { Worker } from "node:worker_threads"; import type { CompressionResult } from "./types.ts"; import type { StackedCompressionStep } from "./strategySelector.ts"; @@ -14,14 +13,70 @@ function positiveInteger(value: string | undefined, fallback: number): number { const parsed = Number(value); return Number.isSafeInteger(parsed) && parsed > 0 ? parsed : fallback; } -function workerUrl(): URL { - const dir = dirname(fileURLToPath(import.meta.url)); - for (const name of ["compressionWorker.js", "compressionWorker.ts"]) { - const candidate = join(dir, name); - if (existsSync(candidate)) return pathToFileURL(candidate); + +/** Relative path (from an install root) to the compression worker. */ +const WORKER_JS_REL = join("open-sse", "services", "compression", "compressionWorker.js"); +const WORKER_TS_REL = join("open-sse", "services", "compression", "compressionWorker.ts"); + +const MAX_WALK_UP = 8; + +/** + * Walk up from each anchor directory (≤ MAX_WALK_UP levels) and return the first + * ancestor that actually contains `relPath`, or null. Pure + exported for tests. + * + * This deliberately avoids `import.meta.url`/`__dirname` (both dead in the standalone + * bundle) — see the LLMLingua worker comments in llmlingua/worker.ts. + */ +export function firstAncestorWith(anchors: string[], relPath: string): string | null { + for (const anchor of anchors) { + if (!anchor) continue; + let dir = resolve(anchor); + for (let i = 0; i <= MAX_WALK_UP; i++) { + if (existsSync(join(dir, relPath))) return dir; + const parent = dirname(dir); + if (parent === dir) break; + dir = parent; + } } - return pathToFileURL(join(dir, "compressionWorker.js")); + return null; } + +/** + * Runtime install-root anchors that SURVIVE the standalone bundle: + * - `process.cwd()` — `dist/server.js` runs `process.chdir(__dirname)` → the dist root. + * - `dirname(process.argv[1])` — the entry script (server.js / bin), walked up. + */ +function runtimeAnchors(): string[] { + const anchors = [process.cwd()]; + const argv1 = process.argv[1]; + if (typeof argv1 === "string" && argv1) anchors.push(dirname(argv1)); + return anchors; +} + +/** + * Resolve the worker entry file across dev and prod WITHOUT `import.meta.url`. + * + * Prod: the worker is likely a .js file under the install root + * Dev: the same relative path resolves to the `.ts` source under the project + * root (cwd) and runs via the default Node.js loader. + * + * First existing candidate wins. Exported for tests. + */ +export function resolveWorkerFile(): string { + const anchors = runtimeAnchors(); + + // Prod first: the .js under the install root. + const jsRoot = firstAncestorWith(anchors, WORKER_JS_REL); + if (jsRoot) return join(jsRoot, WORKER_JS_REL); + + // Dev: the .ts source. + const tsRoot = firstAncestorWith(anchors, WORKER_TS_REL); + if (tsRoot) return join(tsRoot, WORKER_TS_REL); + + // Nothing found — return a cwd-relative .js path; the spawn will fail-open. + return join(process.cwd(), WORKER_JS_REL); +} + function unchanged(body: Record): CompressionResult { return { body, compressed: false, stats: null }; } @@ -80,7 +135,7 @@ export class CompressionWorkerPool { } private spawn(): PoolWorker { const slot: PoolWorker = { - worker: new Worker(workerUrl()), + worker: new Worker(resolveWorkerFile()), job: null, timeout: null, idle: null, diff --git a/tests/unit/compression/compression-worker-file-resolution.test.ts b/tests/unit/compression/compression-worker-file-resolution.test.ts new file mode 100644 index 0000000000..56cbd335ec --- /dev/null +++ b/tests/unit/compression/compression-worker-file-resolution.test.ts @@ -0,0 +1,143 @@ +import assert from "node:assert/strict"; +import { after, describe, it } from "node:test"; +import { mkdirSync, mkdtempSync, rmSync, writeFileSync, realpathSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { + firstAncestorWith, + resolveWorkerFile, +} from "../../../open-sse/services/compression/compressionWorkerPool.ts"; + +/** + * Regression tests for the runtime-anchor worker-file resolution (#12183): the + * standalone bundle kills `import.meta.url`/`__dirname`, so the pool resolves + * `compressionWorker.{js,ts}` from `process.cwd()` and `dirname(process.argv[1])` + * with a bounded walk-up — the same pattern documented in + * open-sse/services/compression/engines/llmlingua/worker.ts. + * + * All fixtures live in mkdtemp sandboxes; the real repo tree is never touched — + * the sandboxes sit under os.tmpdir(), whose ancestors do not contain an + * `open-sse/services/compression/` install root, so the walk-up cannot escape + * into the actual project. + */ + +const WORKER_JS_REL = join("open-sse", "services", "compression", "compressionWorker.js"); +const WORKER_TS_REL = join("open-sse", "services", "compression", "compressionWorker.ts"); + +const sandboxes: string[] = []; + +function makeSandbox(): string { + // realpath so assertions survive tmpdir symlinks (e.g. /tmp → /private/tmp). + const dir = realpathSync(mkdtempSync(join(tmpdir(), "omni-worker-anchors-"))); + sandboxes.push(dir); + return dir; +} + +/** Creates `/open-sse/services/compression/` with stub content. */ +function makeInstallRoot(root: string, fileName: string): void { + const dir = join(root, "open-sse", "services", "compression"); + mkdirSync(dir, { recursive: true }); + writeFileSync(join(dir, fileName), "// test stub worker\n"); +} + +/** Runs `fn` with a fake cwd + argv[1], restoring both afterwards. */ +function withRuntime(cwd: string, argv1: string, fn: () => T): T { + const originalCwd = process.cwd(); + const originalArgv1 = process.argv[1]; + process.chdir(cwd); + process.argv[1] = argv1; + try { + return fn(); + } finally { + process.argv[1] = originalArgv1; + process.chdir(originalCwd); + } +} + +after(() => { + for (const dir of sandboxes) rmSync(dir, { recursive: true, force: true }); +}); + +describe("resolveWorkerFile (runtime anchors)", () => { + it("resolves the .js worker from the cwd anchor", () => { + const installRoot = makeSandbox(); + makeInstallRoot(installRoot, "compressionWorker.js"); + const elsewhere = makeSandbox(); + const resolved = withRuntime(installRoot, join(elsewhere, "server.js"), () => + resolveWorkerFile() + ); + assert.equal(resolved, join(installRoot, WORKER_JS_REL)); + }); + + it("resolves the .js worker from dirname(argv[1]) when cwd has none", () => { + const emptyCwd = makeSandbox(); + const installRoot = makeSandbox(); + makeInstallRoot(installRoot, "compressionWorker.js"); + const resolved = withRuntime(emptyCwd, join(installRoot, "server.js"), () => + resolveWorkerFile() + ); + assert.equal(resolved, join(installRoot, WORKER_JS_REL)); + }); + + it("walks up from a nested cwd until it finds the install root", () => { + const installRoot = makeSandbox(); + makeInstallRoot(installRoot, "compressionWorker.js"); + const nested = join(installRoot, "a", "b", "c"); + mkdirSync(nested, { recursive: true }); + const elsewhere = makeSandbox(); + const resolved = withRuntime(nested, join(elsewhere, "server.js"), () => resolveWorkerFile()); + assert.equal(resolved, join(installRoot, WORKER_JS_REL)); + }); + + it("falls back to the .ts source when no .js exists (dev loader path)", () => { + const installRoot = makeSandbox(); + makeInstallRoot(installRoot, "compressionWorker.ts"); + const elsewhere = makeSandbox(); + const resolved = withRuntime(installRoot, join(elsewhere, "server.js"), () => + resolveWorkerFile() + ); + assert.equal(resolved, join(installRoot, WORKER_TS_REL)); + }); + + it("prefers a .js root on ANY anchor over a .ts root (prod-first ordering)", () => { + const tsRoot = makeSandbox(); + makeInstallRoot(tsRoot, "compressionWorker.ts"); + const jsRoot = makeSandbox(); + makeInstallRoot(jsRoot, "compressionWorker.js"); + // cwd only has the .ts source; argv[1] sits in the .js install root. + const resolved = withRuntime(tsRoot, join(jsRoot, "server.js"), () => resolveWorkerFile()); + assert.equal(resolved, join(jsRoot, WORKER_JS_REL)); + }); + + it("returns the cwd-relative .js fallback (without throwing) when nothing exists", () => { + const emptyCwd = makeSandbox(); + const emptyBin = makeSandbox(); + const resolved = withRuntime(emptyCwd, join(emptyBin, "server.js"), () => resolveWorkerFile()); + assert.equal(resolved, join(emptyCwd, WORKER_JS_REL)); + }); +}); + +describe("firstAncestorWith", () => { + it("returns the anchor itself when it already contains relPath", () => { + const installRoot = makeSandbox(); + makeInstallRoot(installRoot, "compressionWorker.js"); + assert.equal(firstAncestorWith([installRoot], WORKER_JS_REL), installRoot); + }); + + it("skips empty anchors and returns null when nothing matches", () => { + const empty = makeSandbox(); + assert.equal(firstAncestorWith(["", empty], WORKER_JS_REL), null); + assert.equal(firstAncestorWith([], WORKER_JS_REL), null); + }); + + it("finds a root up to 8 levels above the anchor, but not 9 (walk-up cap)", () => { + const installRoot = makeSandbox(); + makeInstallRoot(installRoot, "compressionWorker.js"); + const eightDeep = join(installRoot, ...Array.from({ length: 8 }, (_, i) => `d${i}`)); + mkdirSync(eightDeep, { recursive: true }); + assert.equal(firstAncestorWith([eightDeep], WORKER_JS_REL), installRoot); + const nineDeep = join(installRoot, ...Array.from({ length: 9 }, (_, i) => `d${i}`)); + mkdirSync(nineDeep, { recursive: true }); + assert.equal(firstAncestorWith([nineDeep], WORKER_JS_REL), null); + }); +});