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
18 changes: 14 additions & 4 deletions open-sse/services/compression/compressionWorkerProtocol.ts
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,11 @@ function isPlainObject(value: object): value is Record<string, unknown> {
const prototype = Object.getPrototypeOf(value);
return prototype === Object.prototype || prototype === null;
}

// `seen` tracks only the current recursion PATH (ancestors), not every node ever visited:
// add before descending, remove after returning. That way a real cycle (a node reachable
// from itself) is still rejected, but two sibling branches that happen to reference the
// SAME non-cyclic sub-object (a false positive with a globally-shared `seen` set) are not.
export function isStrictlySerializable(value: unknown, seen = new Set<object>()): boolean {
if (
value === null ||
Expand All @@ -41,11 +46,16 @@ export function isStrictlySerializable(value: unknown, seen = new Set<object>())
) {
return typeof value !== "number" || Number.isFinite(value);
}
if (typeof value !== "object" || seen.has(value)) return false;
if (typeof value !== "object") return false;
if (seen.has(value)) return false;
seen.add(value);
if (Array.isArray(value)) return value.every((entry) => isStrictlySerializable(entry, seen));
if (!isPlainObject(value)) return false;
return Object.values(value).every((entry) => isStrictlySerializable(entry, seen));
try {
if (Array.isArray(value)) return value.every((entry) => isStrictlySerializable(entry, seen));
if (!isPlainObject(value)) return false;
return Object.values(value).every((entry) => isStrictlySerializable(entry, seen));
} finally {
seen.delete(value);
}
}

const WORKER_STACK_ENGINES = new Set(["caveman", "rtk", "standard"]);
Expand Down
4 changes: 4 additions & 0 deletions open-sse/services/compression/strategySelector.ts
Original file line number Diff line number Diff line change
Expand Up @@ -542,6 +542,10 @@ async function runCompressionAsync(
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 };
}
}
Expand Down
20 changes: 20 additions & 0 deletions tests/unit/compression/compression-worker.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -96,6 +96,26 @@ describe("compression worker eligibility", () => {
cyclic.self = cyclic;
assert.equal(isStrictlySerializable(cyclic), false);
});

it("#13154: does not misread a shared (non-cyclic) sub-object referenced by two sibling branches as a cycle", () => {
// Original bug: a single `seen` set shared across the whole recursion tree (never
// backtracked) meant visiting the SAME object twice via two different, non-cyclic
// paths (e.g. two messages both pointing at the same cached template object) was
// indistinguishable from a real cycle. Path-based tracking (add before descending,
// delete after) must treat this as eligible.
const shared = { nested: true };
const sharedBody = { messages: [shared, shared] };
assert.equal(isStrictlySerializable(sharedBody), true);
assert.equal(isCompressionWorkerEligible(sharedBody, "standard", { config }), true);
});

it("still rejects a body with a genuine cycle before it ever reaches postMessage", () => {
const cyclicMessage: Record<string, unknown> = { role: "user" };
cyclicMessage.self = cyclicMessage;
const cyclicBody = { messages: [cyclicMessage] };
assert.equal(isStrictlySerializable(cyclicBody), false);
assert.equal(isCompressionWorkerEligible(cyclicBody, "standard", { config }), false);
});
});

describe("compression worker execution", () => {
Expand Down
Loading