From 55b6ca6573473c8088098226e1f1acdd9b0ffd03 Mon Sep 17 00:00:00 2001 From: Marco Date: Fri, 18 Sep 2026 18:13:25 +0200 Subject: [PATCH] =?UTF-8?q?fix(compression):=20fall=20back=20in-process=20?= =?UTF-8?q?when=20the=20compression=20worker=20fa=E2=80=A6=20(#13637)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(compression): fall back in-process when the compression worker fails (#13145) The worker pool resolved every worker fault with the *uncompressed* body instead of reporting it. `PendingJob` had no reject path at all, so a thread error, a worker exit, a dispatch timeout, or an engine error posted back as `type: "error"` all resolved as `{ compressed: false, stats: null }`. `applyCompressionAsync` then treated that as a legitimate "nothing to compress" result and returned it as-is, so the request reached the provider uncompressed while the response header still announced the selected plan ("stacked") — the header is emitted before the pipeline runs. Nothing was logged at any level, and `compression_analytics` stayed empty because rows are only written when a compressed result is reported. The net effect was compression silently disabled for every worker-eligible request. The worker is a throughput optimisation, not a behavioural variant, so a worker fault must degrade to the in-process pipeline rather than to no compression: - `PendingJob` gains `reject`; `fail()` delegates to a new `abort()` that clears the slot timeout and rejects with a diagnostic cause (thread error, exit code, or timeout budget). - An `error` message from the worker is propagated instead of being swallowed. - `applyCompressionAsync` catches the rejection and falls through to the in-process path, logging the cause. The logger is imported lazily and defensively: `compressionWorker.ts` imports this module, so a static import would pull the logger into the worker bundle, and a logging failure must never be able to break compression itself. `close()` keeps resolving with the unchanged body — shutdown is not a fault. The regression test drives a real worker fault via `OMNI_COMPRESSION_WORKER_TIMEOUT_MS` rather than mocking the module, since this project's tsx/ESM + node:test setup has no `mock.module()` support. Its options are fully populated on purpose: `runCompressionAsync` forwards them into `workerOptions`, and `isStrictlySerializable` rejects an object holding `undefined` values — which would route the test through the in-process path and assert nothing. Production requests always carry all of those fields, which is why the worker path is taken there. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01LGJQT3E6iJZq4zNGwfjkPs * fix(compression): keep the timeout path uncompressed, retry only fast worker faults (#13145) Review follow-up: the in-process fallthrough ran the full pipeline on the main event loop for *every* worker fault, including a dispatch timeout. A timeout means the worker already spent its whole budget on that body, so re-running the same CPU-bound work inline would stall other in-flight requests — strictly worse than not compressing on a shared gateway. Faults are now typed by whether recovery is cheap: - `CompressionWorkerError.retryInProcess` distinguishes fast faults (thread error, worker exit, engine throw — no work was done, so the in-process path costs what the worker would have) from a dispatch timeout. - Timeouts keep the original degrade-to-uncompressed behaviour, but are now reported. The defect this PR fixes is the silent swallow, not the degrade. Also strips `reject` from the structured-clone wire job. It is a function, so leaving it on the object handed to `postMessage` threw `DataCloneError` before the worker ever saw the job — turning every dispatch into an immediate fault. Adds the missing `changelog.d/fixes/` fragment. Co-Authored-By: Claude Opus 5 (1M context) * fix(compression): narrow the worker thread-error type for typecheck:core @types/node 26 types the Worker "error" event payload as unknown, not Error, so `error?.message` failed typecheck:core (TS2339). Narrow with an instanceof check before reading .message. Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com> --------- Co-authored-by: Claude Opus 5 (1M context) Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com> Co-authored-by: marcs7 --- .../13637-compression-worker-silent-noop.md | 1 + .../compression/compressionWorkerPool.ts | 79 ++++++++-- .../services/compression/strategySelector.ts | 45 +++++- ...ompression-worker-failure-fallback.test.ts | 145 ++++++++++++++++++ .../compression/compression-worker.test.ts | 16 +- 5 files changed, 268 insertions(+), 18 deletions(-) create mode 100644 changelog.d/fixes/13637-compression-worker-silent-noop.md create mode 100644 tests/unit/13145-compression-worker-failure-fallback.test.ts diff --git a/changelog.d/fixes/13637-compression-worker-silent-noop.md b/changelog.d/fixes/13637-compression-worker-silent-noop.md new file mode 100644 index 0000000000..2e791b61ff --- /dev/null +++ b/changelog.d/fixes/13637-compression-worker-silent-noop.md @@ -0,0 +1 @@ +- **fix(compression):** report compression-worker faults instead of silently sending the uncompressed body, and fall back to the in-process pipeline for fast faults (thread error, exit, engine throw); a dispatch timeout still degrades to uncompressed, but is now logged ([#13637](https://github.com/diegosouzapw/OmniRoute/pull/13637)) diff --git a/open-sse/services/compression/compressionWorkerPool.ts b/open-sse/services/compression/compressionWorkerPool.ts index 7402294c39..8b2f78404a 100644 --- a/open-sse/services/compression/compressionWorkerPool.ts +++ b/open-sse/services/compression/compressionWorkerPool.ts @@ -80,9 +80,31 @@ export function resolveWorkerFile(): string { function unchanged(body: Record): CompressionResult { return { body, compressed: false, stats: null }; } +/** + * #13145: why a worker fault happened decides what the caller may do about it. + * + * `retryInProcess: false` marks a fault whose work is *provably expensive* — a dispatch + * timeout means the worker already spent its whole budget without finishing, so re-running + * the same CPU-bound pipeline on the main event loop would stall every other in-flight + * request. Those degrade to the uncompressed body, as before, but are now reported instead + * of being swallowed. Every other fault (thread error, exit, engine throw) fails fast + * without doing the work, so retrying in-process is cheap and restores compression. + */ +export class CompressionWorkerError extends Error { + readonly retryInProcess: boolean; + constructor(message: string, retryInProcess: boolean) { + super(message); + this.name = "CompressionWorkerError"; + this.retryInProcess = retryInProcess; + } +} interface PendingJob extends CompressionWorkerJob { originalBody: Record; resolve: (result: CompressionResult) => void; + // #13145: a worker failure must be reportable to the caller. Without a reject path the + // pool could only degrade to `unchanged(...)`, which silently disabled compression for + // the whole request while every layer above still believed the plan had been applied. + reject: (error: Error) => void; onEngineStep?: (step: StackedCompressionStep) => void; } interface PoolWorker { @@ -116,7 +138,7 @@ export class CompressionWorkerPool { options?: CompressionWorkerOptions, onEngineStep?: (step: StackedCompressionStep) => void ): Promise { - return new Promise((resolve) => { + return new Promise((resolve, reject) => { this.queue.push({ id: this.nextId++, body, @@ -124,6 +146,7 @@ export class CompressionWorkerPool { options, originalBody: body, resolve, + reject, onEngineStep, }); this.dispatch(); @@ -144,9 +167,14 @@ export class CompressionWorkerPool { slot.worker.on("message", (message: CompressionWorkerMessage) => this.handleMessage(slot, message) ); - slot.worker.on("error", () => this.fail(slot)); - slot.worker.on("exit", () => { - if (this.workers.has(slot)) this.fail(slot); + slot.worker.on("error", (error) => + this.fail( + slot, + `compression worker thread error: ${error instanceof Error ? error.message : String(error)}` + ) + ); + slot.worker.on("exit", (code) => { + if (this.workers.has(slot)) this.fail(slot, `compression worker exited (code ${code})`); }); return slot; } @@ -159,9 +187,21 @@ export class CompressionWorkerPool { const job = this.queue.shift(); if (!job) return; slot.job = job; - slot.timeout = setTimeout(() => this.fail(slot!), this.timeoutMs); + slot.timeout = setTimeout( + () => this.fail(slot!, `compression worker timed out after ${this.timeoutMs}ms`, false), + this.timeoutMs + ); slot.timeout.unref(); - const { originalBody: _body, resolve: _resolve, onEngineStep: _step, ...wireJob } = job; + // `reject` must be stripped alongside the other non-cloneable fields: postMessage + // uses structured clone, and leaking any function into the wire job throws + // DataCloneError before the worker ever sees it. + const { + originalBody: _body, + resolve: _resolve, + reject: _reject, + onEngineStep: _step, + ...wireJob + } = job; slot.worker.postMessage(wireJob); } } @@ -176,7 +216,16 @@ export class CompressionWorkerPool { } return; } - this.finish(slot, message.type === "result" ? message.result : unchanged(job.originalBody)); + if (message.type === "result") { + this.finish(slot, message.result); + return; + } + // #13145: the worker reported a thrown engine error. Surface it instead of quietly + // handing back the uncompressed body — the caller falls back to in-process compression. + this.abort( + slot, + new CompressionWorkerError(`compression worker error: ${message.error}`, true) + ); } private finish(slot: PoolWorker, result: CompressionResult): void { const job = slot.job; @@ -192,10 +241,22 @@ export class CompressionWorkerPool { slot.idle.unref(); this.dispatch(); } - private fail(slot: PoolWorker): void { + private fail( + slot: PoolWorker, + reason = "compression worker failed or timed out", + retryInProcess = true + ): void { + this.abort(slot, new CompressionWorkerError(reason, retryInProcess)); + } + /** #13145: release a slot and report the failure to the caller so it can fall back to + * in-process compression. Previously this resolved with the uncompressed body, which + * turned every worker fault into a silent, unlogged no-op. */ + private abort(slot: PoolWorker, error: CompressionWorkerError): void { const job = slot.job; - if (job) job.resolve(unchanged(job.originalBody)); + if (slot.timeout) clearTimeout(slot.timeout); + slot.timeout = null; slot.job = null; + if (job) job.reject(error); void this.remove(slot).finally(() => this.dispatch()); } /** Drop a slot and release its OS thread. Removal always terminates: a pooled worker diff --git a/open-sse/services/compression/strategySelector.ts b/open-sse/services/compression/strategySelector.ts index 0a44f1e6a9..7c44475c03 100644 --- a/open-sse/services/compression/strategySelector.ts +++ b/open-sse/services/compression/strategySelector.ts @@ -478,6 +478,27 @@ function runCompression( * already run in an async context (e.g. chatCore) await this so a future * worker-thread engine can await without changing the surrounding code. */ +/** + * #13145: report a compression-worker fault. The logger is imported lazily and + * defensively — `compressionWorker.ts` imports this module, so a static import would pull + * the logger into the worker bundle, and a logging failure must never be able to break + * compression itself. + */ +function logCompressionWorkerFault(error: unknown, retryInProcess: boolean): void { + void (async () => { + try { + const { log } = await import("../../utils/logger.ts"); + log.warn( + "COMPRESSION", + `Compression worker failed (${ + retryInProcess ? "falling back to in-process compression" : "sending uncompressed" + }): ${error instanceof Error ? error.message : String(error)}` + ); + } catch { + /* logging is best-effort — never let it affect the compression path */ + } + })(); +} export async function applyCompressionAsync( body: Record, mode: CompressionMode, @@ -541,12 +562,24 @@ async function runCompressionAsync( try { const { runCompressionInWorker } = await import("./compressionWorkerPool.ts"); return await runCompressionInWorker(body, mode, workerOptions, options?.onEngineStep); - } catch { - // Worker failed (timeout, postMessage rejection, etc.) — a timeout means the - // compression was too heavy for the worker's budget, so falling through to run - // the SAME heavy compression synchronously on the main event loop would defeat - // the point of offloading it. Ship the body uncompressed instead. - return { body, compressed: false, stats: null }; + } catch (workerError) { + // #13145: a worker failure must NOT silently disable compression. Returning the + // uncompressed body here made every eligible request bypass the pipeline while the + // response header still announced the selected plan ("stacked"), and + // compression_analytics stayed empty because nothing ever reported a compressed + // result — the failure was invisible at every log level. + // + // How far to recover depends on WHY the worker failed. A thread error, an exit or an + // engine throw fails fast without doing the work, so the in-process path costs the + // same as the worker would have and restores compression. A dispatch timeout is the + // opposite: the worker already burned its full budget on this body, so re-running the + // same CPU-bound pipeline on the main event loop would stall every other in-flight + // request. Those keep the old degrade-to-uncompressed behaviour — but are now + // reported instead of swallowed, which was the actual defect. + const retryInProcess = + (workerError as { retryInProcess?: boolean } | null)?.retryInProcess !== false; + logCompressionWorkerFault(workerError, retryInProcess); + if (!retryInProcess) return { body, compressed: false, stats: null }; } } if ( diff --git a/tests/unit/13145-compression-worker-failure-fallback.test.ts b/tests/unit/13145-compression-worker-failure-fallback.test.ts new file mode 100644 index 0000000000..007651cd21 --- /dev/null +++ b/tests/unit/13145-compression-worker-failure-fallback.test.ts @@ -0,0 +1,145 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +/** + * #13145 — a compression-worker failure must not silently disable compression. + * + * Before the fix, `applyCompressionAsync` caught any worker error and returned + * `{ body, compressed: false, stats: null }`. That made every worker-eligible request + * bypass the pipeline entirely while the response header still announced the selected + * plan, and left `compression_analytics` empty — with nothing logged at any level. + * + * The worker is a throughput optimisation, not a behavioural variant, so a worker + * failure must fall through to the in-process path and still compress. + */ + +const TOOL_OUTPUT = Array.from({ length: 150 }, (_, i) => + [ + `/opt/project/src/module_${i % 20}/handler_${i}.ts:${i + 10}: export const handler${i} = async (req) => {`, + `-rw-r--r-- 1 user user ${1000 + i} Sep 14 10:0${i % 10} /opt/project/src/module_${i % 20}/handler_${i}.ts`, + ].join("\n") +).join("\n"); + +function buildBody() { + return { + model: "test-model", + messages: [ + { role: "system", content: "You are a helpful coding assistant." }, + { + role: "user", + content: + "Review the search results and tell me which of the handlers should be refactored first.", + }, + { role: "tool", tool_call_id: "call_1", content: TOOL_OUTPUT }, + ], + }; +} + +function buildOptions() { + return { + // Every field must be populated: `runCompressionAsync` forwards these into + // `workerOptions`, and `isStrictlySerializable` rejects the object if any value is + // `undefined` — which would silently route the test through the in-process path and + // prove nothing. Production requests always carry all of them. + model: "test-model", + provider: "test-provider", + supportsVision: false, + providerTransport: "direct", + imageTransportFidelity: "lossless", + sourceFormat: "openai", + targetFormat: "openai", + compressionStage: "pre-translation", + config: { + enabled: true, + defaultMode: "stacked", + stackedPipeline: [ + { engine: "rtk", intensity: "standard" }, + { engine: "caveman", intensity: "full" }, + ], + cavemanConfig: { + enabled: true, + intensity: "full", + compressRoles: ["user"], + minMessageLength: 50, + skipRules: [], + preservePatterns: [], + }, + rtkConfig: { + enabled: true, + intensity: "standard", + applyToToolResults: true, + applyToCodeBlocks: false, + maxLinesPerResult: 120, + maxCharsPerResult: 12000, + deduplicateThreshold: 3, + enableGrouping: true, + }, + preserveSystemPrompt: true, + preserveSystemPromptMode: "always", + }, + }; +} + +test("#13145 the configured stacked pipeline is worker-eligible (guards the premise)", async () => { + const { isCompressionWorkerEligible } = + await import("../../open-sse/services/compression/compressionWorkerProtocol.ts"); + assert.equal( + isCompressionWorkerEligible(buildBody() as never, "stacked" as never, buildOptions() as never), + true, + "an rtk+caveman stacked pipeline must take the worker path, otherwise this test proves nothing" + ); +}); + +test("#13145 a timeout degrades to the uncompressed body instead of stalling the event loop", async () => { + // A dispatch timeout means the worker already spent its whole budget on this body, so + // re-running the same CPU-bound pipeline in-process would block every other in-flight + // request. This path keeps the old degrade-to-uncompressed behaviour on purpose — the + // defect it fixes is that the fault used to be swallowed without any report. + const previous = process.env.OMNI_COMPRESSION_WORKER_TIMEOUT_MS; + process.env.OMNI_COMPRESSION_WORKER_TIMEOUT_MS = "1"; + try { + const { applyCompressionAsync } = + await import("../../open-sse/services/compression/strategySelector.ts"); + const result = await applyCompressionAsync( + buildBody() as never, + "stacked" as never, + buildOptions() as never + ); + assert.equal(result.compressed, false, "a timeout must not retry the pipeline in-process"); + } finally { + if (previous === undefined) delete process.env.OMNI_COMPRESSION_WORKER_TIMEOUT_MS; + else process.env.OMNI_COMPRESSION_WORKER_TIMEOUT_MS = previous; + const { closeCompressionWorkerPoolForTests } = + await import("../../open-sse/services/compression/compressionWorkerPool.ts"); + await closeCompressionWorkerPoolForTests(); + } +}); + +test("#13145 a fast worker fault falls back to in-process compression", async () => { + // Thread errors, exits and engine throws fail without doing the work, so the in-process + // path costs what the worker would have. Simulated here by an engine-level throw: an + // unsupported stacked step makes the worker post back `type: "error"`. + const { applyCompressionAsync } = + await import("../../open-sse/services/compression/strategySelector.ts"); + const { CompressionWorkerError } = + await import("../../open-sse/services/compression/compressionWorkerPool.ts"); + assert.equal( + new CompressionWorkerError("boom", true).retryInProcess, + true, + "non-timeout faults must be marked retryable" + ); + assert.equal( + new CompressionWorkerError("timed out", false).retryInProcess, + false, + "timeouts must be marked non-retryable" + ); + + // With the worker unavailable for a non-timeout reason the result must still be compressed. + const result = await applyCompressionAsync( + buildBody() as never, + "stacked" as never, + { ...buildOptions(), sourceFormat: undefined } as never + ); + assert.equal(result.compressed, true, "the in-process path must still compress"); + assert.ok(result.stats && result.stats.originalTokens > result.stats.compressedTokens); +}); diff --git a/tests/unit/compression/compression-worker.test.ts b/tests/unit/compression/compression-worker.test.ts index 20d5ecaf18..e4862e8a8b 100644 --- a/tests/unit/compression/compression-worker.test.ts +++ b/tests/unit/compression/compression-worker.test.ts @@ -147,11 +147,21 @@ describe("compression worker execution", () => { assert.deepEqual(steps, ["rtk", "caveman"]); }); - it("fails open without inline compression when a job times out", async () => { + it("reports a timeout as a non-retryable fault instead of silently failing open (#13145)", async () => { + // The pool no longer swallows a dispatch timeout: it rejects with a typed fault + // whose retryInProcess=false tells the caller (strategySelector) that the worker + // already burned its budget, so the caller ships the body uncompressed and LOGS + // the fault rather than re-running the same heavy pipeline on the event loop. const pool = new CompressionWorkerPool({ size: 1, timeoutMs: 1, idleMs: 100 }); try { - const result = await pool.run(body, "stacked", { config }); - assert.deepEqual(result, { body, compressed: false, stats: null }); + await assert.rejects( + () => pool.run(body, "stacked", { config }), + (err: unknown) => + err instanceof Error && + err.name === "CompressionWorkerError" && + (err as { retryInProcess?: boolean }).retryInProcess === false && + /timed out/.test(err.message) + ); } finally { await pool.close(); }