pjandro/OmniRoute
1
0
Fork 0
mirror of https://github.com/diegosouzapw/OmniRoute.git synced 2026-09-19 05:12:16 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix(compression): fall back in-process when the compression worker fa… (#13637)

Browse Source 
* 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) <noreply@anthropic.com>
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) <noreply@anthropic.com>

* 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) <noreply@anthropic.com>
Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com>
Co-authored-by: marcs7 <marcs7@users.noreply.github.com>
This commit is contained in:
Marco
2026-09-18 18:13:25 +02:00
committed by GitHub
parent 1e8c913ca2
commit 55b6ca6573
5 changed files with 268 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
changelog.d/fixes/13637-compression-worker-silent-noop.md Normal file
View File

@@ -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))

79
open-sse/services/compression/compressionWorkerPool.ts
View File

@@ -80,9 +80,31 @@ export function resolveWorkerFile(): string {
function unchanged(body: Record<string, unknown>): 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<string, unknown>;
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<CompressionResult> {
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

45
open-sse/services/compression/strategySelector.ts
View File

@@ -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<string, unknown>,
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 (

145
tests/unit/13145-compression-worker-failure-fallback.test.ts Normal file
View File

@@ -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);
});

16
tests/unit/compression/compression-worker.test.ts
View File

@@ -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();
}