mirror of
https://github.com/diegosouzapw/OmniRoute.git
synced 2026-08-08 00:02:20 +03:00
fix(compression): bound session-dedup suffix-block scan to prevent OOM (#8438)
Validated in post-merge-train sweep (boards clean on release/v3.8.50 tip)
This commit is contained in:
committed by
GitHub
parent
1e55fbd20b
commit
7f36b192f0
@@ -42,6 +42,7 @@ x-common: &common
|
|||||||
- LIVE_WS_HOST=${LIVE_WS_HOST:-0.0.0.0}
|
- LIVE_WS_HOST=${LIVE_WS_HOST:-0.0.0.0}
|
||||||
- LIVE_WS_ALLOWED_ORIGINS=${LIVE_WS_ALLOWED_ORIGINS:-http://localhost:20128,http://127.0.0.1:20128}
|
- LIVE_WS_ALLOWED_ORIGINS=${LIVE_WS_ALLOWED_ORIGINS:-http://localhost:20128,http://127.0.0.1:20128}
|
||||||
- REDIS_URL=${REDIS_URL:-redis://redis:6379}
|
- REDIS_URL=${REDIS_URL:-redis://redis:6379}
|
||||||
|
- NODE_OPTIONS=--max-old-space-size=2048
|
||||||
volumes:
|
volumes:
|
||||||
- ./data:/app/data
|
- ./data:/app/data
|
||||||
healthcheck:
|
healthcheck:
|
||||||
|
|||||||
@@ -201,16 +201,43 @@ function calculateSpecificityMatch(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Pool-wide maxima used to normalize cost/latency/stability factors. These are
|
||||||
|
* identical for every candidate in a given pool, so callers scoring many
|
||||||
|
* candidates against the same pool should compute this ONCE via
|
||||||
|
* computePoolMaxima() and pass it to calculateFactors — recomputing it inside
|
||||||
|
* a per-candidate loop turns an O(n) scoring pass into O(n^2) (#OOM incident:
|
||||||
|
* a zero-config "auto" combo with no explicit candidatePool can expand the
|
||||||
|
* pool to 1000s of provider/model targets, at which point the repeated
|
||||||
|
* `pool.map()` + spread here dominates heap churn and can OOM the process).
|
||||||
|
*/
|
||||||
|
export interface PoolMaxima {
|
||||||
|
maxCost: number;
|
||||||
|
maxLatency: number;
|
||||||
|
maxStdDev: number;
|
||||||
|
}
|
||||||
|
|
||||||
|
export function computePoolMaxima(pool: ProviderCandidate[]): PoolMaxima {
|
||||||
|
let maxCost = 0.001;
|
||||||
|
let maxLatency = 1;
|
||||||
|
let maxStdDev = 0.001;
|
||||||
|
for (const p of pool) {
|
||||||
|
if (p.costPer1MTokens > maxCost) maxCost = p.costPer1MTokens;
|
||||||
|
if (p.p95LatencyMs > maxLatency) maxLatency = p.p95LatencyMs;
|
||||||
|
if (p.latencyStdDev > maxStdDev) maxStdDev = p.latencyStdDev;
|
||||||
|
}
|
||||||
|
return { maxCost, maxLatency, maxStdDev };
|
||||||
|
}
|
||||||
|
|
||||||
export function calculateFactors(
|
export function calculateFactors(
|
||||||
candidate: ProviderCandidate,
|
candidate: ProviderCandidate,
|
||||||
pool: ProviderCandidate[],
|
pool: ProviderCandidate[],
|
||||||
taskType: string,
|
taskType: string,
|
||||||
getTaskFitness: (model: string, taskType: string) => number,
|
getTaskFitness: (model: string, taskType: string) => number,
|
||||||
manifestHint?: RoutingHint | null
|
manifestHint?: RoutingHint | null,
|
||||||
|
precomputedMaxima?: PoolMaxima
|
||||||
): ScoringFactors {
|
): ScoringFactors {
|
||||||
const maxCost = Math.max(...pool.map((p) => p.costPer1MTokens), 0.001);
|
const { maxCost, maxLatency, maxStdDev } = precomputedMaxima ?? computePoolMaxima(pool);
|
||||||
const maxLatency = Math.max(...pool.map((p) => p.p95LatencyMs), 1);
|
|
||||||
const maxStdDev = Math.max(...pool.map((p) => p.latencyStdDev), 0.001);
|
|
||||||
|
|
||||||
// Every factor is contractually [0,1]. clamp01 guards against bad telemetry
|
// Every factor is contractually [0,1]. clamp01 guards against bad telemetry
|
||||||
// (negative quota / cost / latency, NaN, out-of-range candidate-supplied
|
// (negative quota / cost / latency, NaN, out-of-range candidate-supplied
|
||||||
@@ -245,9 +272,17 @@ export function scorePool(
|
|||||||
getTaskFitness: (model: string, taskType: string) => number = () => 0.5,
|
getTaskFitness: (model: string, taskType: string) => number = () => 0.5,
|
||||||
manifestHint?: RoutingHint | null
|
manifestHint?: RoutingHint | null
|
||||||
): ScoredProvider[] {
|
): ScoredProvider[] {
|
||||||
|
const poolMaxima = computePoolMaxima(pool);
|
||||||
return pool
|
return pool
|
||||||
.map((candidate) => {
|
.map((candidate) => {
|
||||||
const factors = calculateFactors(candidate, pool, taskType, getTaskFitness, manifestHint);
|
const factors = calculateFactors(
|
||||||
|
candidate,
|
||||||
|
pool,
|
||||||
|
taskType,
|
||||||
|
getTaskFitness,
|
||||||
|
manifestHint,
|
||||||
|
poolMaxima
|
||||||
|
);
|
||||||
return {
|
return {
|
||||||
provider: candidate.provider,
|
provider: candidate.provider,
|
||||||
model: candidate.model,
|
model: candidate.model,
|
||||||
|
|||||||
@@ -33,6 +33,7 @@ import { getTaskFitness } from "../autoCombo/taskFitness.ts";
|
|||||||
import {
|
import {
|
||||||
calculateFactors,
|
calculateFactors,
|
||||||
calculateScore,
|
calculateScore,
|
||||||
|
computePoolMaxima,
|
||||||
type ProviderCandidate,
|
type ProviderCandidate,
|
||||||
type ScoringWeights,
|
type ScoringWeights,
|
||||||
} from "../autoCombo/scoring.ts";
|
} from "../autoCombo/scoring.ts";
|
||||||
@@ -351,6 +352,11 @@ export function scoreAutoTargets(
|
|||||||
) {
|
) {
|
||||||
const targetByExecutionKey = new Map(targets.map((target) => [target.executionKey, target]));
|
const targetByExecutionKey = new Map(targets.map((target) => [target.executionKey, target]));
|
||||||
const activeCandidates = candidates.filter((candidate) => candidate.quotaCutoffBlocked !== true);
|
const activeCandidates = candidates.filter((candidate) => candidate.quotaCutoffBlocked !== true);
|
||||||
|
// Computed once per scoring pass, not per candidate — see computePoolMaxima's
|
||||||
|
// doc comment (scoring.ts) for the O(n^2) OOM this avoids on large auto-combo
|
||||||
|
// candidate pools (#OOM incident, zero-config auto combo expanding to 1000s
|
||||||
|
// of provider/model targets).
|
||||||
|
const poolMaxima = computePoolMaxima(activeCandidates as unknown as ProviderCandidate[]);
|
||||||
|
|
||||||
return activeCandidates
|
return activeCandidates
|
||||||
.map((candidate) => {
|
.map((candidate) => {
|
||||||
@@ -373,10 +379,11 @@ export function scoreAutoTargets(
|
|||||||
};
|
};
|
||||||
const factors = calculateFactors(
|
const factors = calculateFactors(
|
||||||
candidate as ProviderCandidate,
|
candidate as ProviderCandidate,
|
||||||
activeCandidates,
|
activeCandidates as unknown as ProviderCandidate[],
|
||||||
taskType ?? "general",
|
taskType ?? "general",
|
||||||
getTaskFitness,
|
getTaskFitness,
|
||||||
manifestHint ?? undefined
|
manifestHint ?? undefined,
|
||||||
|
poolMaxima
|
||||||
);
|
);
|
||||||
let score = calculateScore(factors, weights);
|
let score = calculateScore(factors, weights);
|
||||||
// B17: Quota Share soft-policy deprioritization
|
// B17: Quota Share soft-policy deprioritization
|
||||||
@@ -447,11 +454,16 @@ export async function expandAutoComboCandidatePool(
|
|||||||
.filter((p): p is string => typeof p === "string" && p.length > 0)
|
.filter((p): p is string => typeof p === "string" && p.length > 0)
|
||||||
),
|
),
|
||||||
];
|
];
|
||||||
|
// Pre-build a Set of already-present modelStr values so candidate-pool
|
||||||
|
// expansion doesn't turn into O(n^2) per provider. See #OOM incident
|
||||||
|
// (zero-config auto combo expanding to 1000s of provider/model targets).
|
||||||
|
const seenModelStrs = new Set(eligibleTargets.map((t) => t.modelStr));
|
||||||
for (const providerId of providerIds) {
|
for (const providerId of providerIds) {
|
||||||
const providerModels = getProviderModels(providerId);
|
const providerModels = getProviderModels(providerId);
|
||||||
for (const model of providerModels) {
|
for (const model of providerModels) {
|
||||||
const modelStr = `${providerId}/${model.id}`;
|
const modelStr = `${providerId}/${model.id}`;
|
||||||
if (!eligibleTargets.some((t) => t.modelStr === modelStr)) {
|
if (!seenModelStrs.has(modelStr)) {
|
||||||
|
seenModelStrs.add(modelStr);
|
||||||
eligibleTargets.push({
|
eligibleTargets.push({
|
||||||
kind: "model",
|
kind: "model",
|
||||||
stepId: modelStr,
|
stepId: modelStr,
|
||||||
|
|||||||
@@ -47,14 +47,19 @@ const DEFAULT_MIN_BLOCK_CHARS = 80;
|
|||||||
/** Minimum number of lines a block must span to be a dedup candidate. */
|
/** Minimum number of lines a block must span to be a dedup candidate. */
|
||||||
const MIN_BLOCK_LINES = 3;
|
const MIN_BLOCK_LINES = 3;
|
||||||
/**
|
/**
|
||||||
* Request-wide ceiling for the suffix strings materialized by the exact pass.
|
* O(n²) guard for {@link findSuffixBlocks} (OOM incident): a single message with
|
||||||
* 32 MiB keeps ordinary sessions byte-identical while preventing line-rich inputs
|
* thousands of lines otherwise generates one full-length suffix string PER line,
|
||||||
* from retaining a quadratic graph of suffix copies.
|
* all retained at once. A real agent conversation embedding a large
|
||||||
|
* line-numbered file view (e.g. a tool result pasting a multi-thousand-line
|
||||||
|
* file back into the chat) drove ~1.7GB of live suffix strings and OOM-killed
|
||||||
|
* the 2GB heap (heap snapshot confirmed 6801 `{ block }` objects). These bound
|
||||||
|
* both the number of suffix starts scanned
|
||||||
|
* and the total bytes of retained blocks, so memory is O(budget) instead of O(n²).
|
||||||
|
* Dedup is best-effort — skipping the tail only forgoes some compression, never
|
||||||
|
* changes output correctness.
|
||||||
*/
|
*/
|
||||||
const MAX_SUFFIX_WORK_CHARS = 32 * 1024 * 1024;
|
const MAX_SUFFIX_STARTS = 2000;
|
||||||
const SUFFIX_WORK_BUDGET_WARNING = "session-dedup: skipped (suffix work budget exceeded)";
|
const MAX_TOTAL_BLOCK_BYTES = 8 * 1024 * 1024;
|
||||||
|
|
||||||
type SuffixWorkBudget = { remaining: number };
|
|
||||||
|
|
||||||
// ─── hash helper (SHA-256 prefix, collision-resistant) ───────────────────────
|
// ─── hash helper (SHA-256 prefix, collision-resistant) ───────────────────────
|
||||||
|
|
||||||
@@ -67,24 +72,6 @@ function hashBlock(text: string): string {
|
|||||||
|
|
||||||
// ─── suffix-block extraction ──────────────────────────────────────────────────
|
// ─── suffix-block extraction ──────────────────────────────────────────────────
|
||||||
|
|
||||||
/**
|
|
||||||
* Reserves the characters that findSuffixBlocks() would materialize for one text.
|
|
||||||
* The scan observes line starts without splitting or constructing any suffix strings.
|
|
||||||
*/
|
|
||||||
function reserveSuffixWork(text: string, passCount: number, budget: SuffixWorkBudget): boolean {
|
|
||||||
let start = 0;
|
|
||||||
while (start <= text.length) {
|
|
||||||
const suffixChars = (text.length - start) * passCount;
|
|
||||||
if (suffixChars > budget.remaining) return false;
|
|
||||||
budget.remaining -= suffixChars;
|
|
||||||
|
|
||||||
const nextNewline = text.indexOf("\n", start);
|
|
||||||
if (nextNewline === -1) break;
|
|
||||||
start = nextNewline + 1;
|
|
||||||
}
|
|
||||||
return true;
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* For each starting line position, emit the suffix block `lines[start..end]`
|
* For each starting line position, emit the suffix block `lines[start..end]`
|
||||||
* (i.e. from `start` to the end of the line array). This ensures that any
|
* (i.e. from `start` to the end of the line array). This ensures that any
|
||||||
@@ -102,12 +89,19 @@ function findSuffixBlocks(
|
|||||||
const seen = new Set<string>();
|
const seen = new Set<string>();
|
||||||
const results: Array<{ block: string; startLine: number }> = [];
|
const results: Array<{ block: string; startLine: number }> = [];
|
||||||
|
|
||||||
for (let start = 0; start < n; start++) {
|
// O(n²) guard (#OOM): cap the number of suffix starts and the total retained
|
||||||
|
// block bytes so a huge message can't materialize thousands of full-length
|
||||||
|
// suffix strings at once. See MAX_SUFFIX_STARTS / MAX_TOTAL_BLOCK_BYTES.
|
||||||
|
const maxStarts = Math.min(n, MAX_SUFFIX_STARTS);
|
||||||
|
let totalBlockBytes = 0;
|
||||||
|
for (let start = 0; start < maxStarts; start++) {
|
||||||
const block = lines.slice(start).join("\n");
|
const block = lines.slice(start).join("\n");
|
||||||
const blockLines = n - start;
|
const blockLines = n - start;
|
||||||
if (blockLines >= MIN_BLOCK_LINES && block.length >= minBlockChars && !seen.has(block)) {
|
if (blockLines >= MIN_BLOCK_LINES && block.length >= minBlockChars && !seen.has(block)) {
|
||||||
seen.add(block);
|
seen.add(block);
|
||||||
results.push({ block, startLine: start });
|
results.push({ block, startLine: start });
|
||||||
|
totalBlockBytes += block.length;
|
||||||
|
if (totalBlockBytes >= MAX_TOTAL_BLOCK_BYTES) break;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return results;
|
return results;
|
||||||
@@ -145,9 +139,7 @@ function dedupeWithinMessage(
|
|||||||
|
|
||||||
for (const { block } of sortedBlocks) {
|
for (const { block } of sortedBlocks) {
|
||||||
// Only dedup blocks that appear 2+ times in the text.
|
// Only dedup blocks that appear 2+ times in the text.
|
||||||
const occurrences = (
|
const occurrences = (result.match(new RegExp(block.replace(/[.*+?^${}()|[\]\\]/g, "\\$&"), "g")) || []).length;
|
||||||
result.match(new RegExp(block.replace(/[.*+?^${}()|[\]\\]/g, "\\$&"), "g")) || []
|
|
||||||
).length;
|
|
||||||
if (occurrences < 2) continue;
|
if (occurrences < 2) continue;
|
||||||
|
|
||||||
const sha = hashBlock(block);
|
const sha = hashBlock(block);
|
||||||
@@ -269,7 +261,7 @@ type MessageLike = {
|
|||||||
function processMessages(
|
function processMessages(
|
||||||
messages: MessageLike[],
|
messages: MessageLike[],
|
||||||
minBlockChars: number
|
minBlockChars: number
|
||||||
): { messages: MessageLike[]; dedupCount: number; suffixWorkBudgetExceeded: boolean } {
|
): { messages: MessageLike[]; dedupCount: number } {
|
||||||
// Collect (msgIdx, text) for non-system string-content messages.
|
// Collect (msgIdx, text) for non-system string-content messages.
|
||||||
// For multipart, index each text part separately.
|
// For multipart, index each text part separately.
|
||||||
const msgTexts: Array<{ msgIdx: number; text: string }> = [];
|
const msgTexts: Array<{ msgIdx: number; text: string }> = [];
|
||||||
@@ -291,24 +283,13 @@ function processMessages(
|
|||||||
}
|
}
|
||||||
|
|
||||||
if (msgTexts.length === 0) {
|
if (msgTexts.length === 0) {
|
||||||
return { messages, dedupCount: 0, suffixWorkBudgetExceeded: false };
|
return { messages, dedupCount: 0 };
|
||||||
}
|
|
||||||
|
|
||||||
// Single-message exact dedup enumerates suffixes once; cross-message dedup does so
|
|
||||||
// in both passes. Reserve the request-wide work up front so no quadratic suffix graph
|
|
||||||
// is partially materialized before the engine decides to fail open.
|
|
||||||
const suffixWorkBudget: SuffixWorkBudget = { remaining: MAX_SUFFIX_WORK_CHARS };
|
|
||||||
const passCount = msgTexts.length === 1 ? 1 : 2;
|
|
||||||
for (const { text } of msgTexts) {
|
|
||||||
if (!reserveSuffixWork(text, passCount, suffixWorkBudget)) {
|
|
||||||
return { messages, dedupCount: 0, suffixWorkBudgetExceeded: true };
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
const { deduped, dedupCount } = dedupMessageTexts(msgTexts, minBlockChars);
|
const { deduped, dedupCount } = dedupMessageTexts(msgTexts, minBlockChars);
|
||||||
|
|
||||||
if (dedupCount === 0) {
|
if (dedupCount === 0) {
|
||||||
return { messages, dedupCount: 0, suffixWorkBudgetExceeded: false };
|
return { messages, dedupCount: 0 };
|
||||||
}
|
}
|
||||||
|
|
||||||
const result = messages.map((msg, i) => {
|
const result = messages.map((msg, i) => {
|
||||||
@@ -337,7 +318,7 @@ function processMessages(
|
|||||||
return { ...msg };
|
return { ...msg };
|
||||||
});
|
});
|
||||||
|
|
||||||
return { messages: result, dedupCount, suffixWorkBudgetExceeded: false };
|
return { messages: result, dedupCount };
|
||||||
}
|
}
|
||||||
|
|
||||||
// ─── schema & validation ──────────────────────────────────────────────────────
|
// ─── schema & validation ──────────────────────────────────────────────────────
|
||||||
@@ -383,8 +364,7 @@ function validateSessionDedupConfig(config: Record<string, unknown>): EngineVali
|
|||||||
const f = config["fuzzy"];
|
const f = config["fuzzy"];
|
||||||
if (typeof f === "object" && f !== null) {
|
if (typeof f === "object" && f !== null) {
|
||||||
const fe = (f as Record<string, unknown>)["enabled"];
|
const fe = (f as Record<string, unknown>)["enabled"];
|
||||||
if (fe !== undefined && typeof fe !== "boolean")
|
if (fe !== undefined && typeof fe !== "boolean") errors.push("fuzzy.enabled must be a boolean");
|
||||||
errors.push("fuzzy.enabled must be a boolean");
|
|
||||||
} else if (typeof f !== "boolean") {
|
} else if (typeof f !== "boolean") {
|
||||||
errors.push("fuzzy must be an object { enabled } or a boolean");
|
errors.push("fuzzy must be an object { enabled } or a boolean");
|
||||||
}
|
}
|
||||||
@@ -435,18 +415,10 @@ export const sessionDedupEngine: CompressionEngine = {
|
|||||||
}
|
}
|
||||||
|
|
||||||
const start = performance.now();
|
const start = performance.now();
|
||||||
const {
|
const { messages: exactMessages, dedupCount } = processMessages(
|
||||||
messages: exactMessages,
|
messages as MessageLike[],
|
||||||
dedupCount,
|
minBlockChars
|
||||||
suffixWorkBudgetExceeded,
|
);
|
||||||
} = processMessages(messages as MessageLike[], minBlockChars);
|
|
||||||
|
|
||||||
if (suffixWorkBudgetExceeded) {
|
|
||||||
const durationMs = Math.round(performance.now() - start);
|
|
||||||
const stats = createCompressionStats(body, body, "stacked", [], undefined, durationMs);
|
|
||||||
stats.validationWarnings = [SUFFIX_WORK_BUDGET_WARNING];
|
|
||||||
return { body, compressed: false, stats };
|
|
||||||
}
|
|
||||||
|
|
||||||
const { messages: finalMessages, fuzzyCount } = runFuzzyPass(
|
const { messages: finalMessages, fuzzyCount } = runFuzzyPass(
|
||||||
exactMessages,
|
exactMessages,
|
||||||
|
|||||||
@@ -102,6 +102,44 @@ describe("session-dedup engine", () => {
|
|||||||
assert.equal(messages[2].content, "bye");
|
assert.equal(messages[2].content, "bye");
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("O(n²) guard: a huge multi-thousand-line message stays bounded in time and memory", () => {
|
||||||
|
// Regression for the OOM incident: an agent conversation embedded a large
|
||||||
|
// line-numbered file view (a tool result pasting a multi-thousand-line file
|
||||||
|
// back into the chat) repeated across messages. findSuffixBlocks generated
|
||||||
|
// one full-length suffix string per line, all retained at once (~1.7GB of
|
||||||
|
// live strings), OOM-killing the heap.
|
||||||
|
// The MAX_SUFFIX_STARTS / MAX_TOTAL_BLOCK_BYTES guards must keep this bounded.
|
||||||
|
const hugeLines: string[] = [];
|
||||||
|
for (let i = 0; i < 6000; i++) {
|
||||||
|
// ~80 chars/line so each suffix is large — the pathological shape.
|
||||||
|
hugeLines.push(`${i}: "config_snapshot": { "value": ${i}, "pad": "xxxxxxxxxxxxxx" }`);
|
||||||
|
}
|
||||||
|
const hugeContent = hugeLines.join("\n");
|
||||||
|
const body = makeBody([
|
||||||
|
{ role: "user", content: `Read file result:\n${hugeContent}` },
|
||||||
|
{ role: "assistant", content: "analysing" },
|
||||||
|
{ role: "user", content: `Read file again:\n${hugeContent}` },
|
||||||
|
]);
|
||||||
|
|
||||||
|
const before = process.memoryUsage().heapUsed;
|
||||||
|
const t0 = Date.now();
|
||||||
|
const result = sessionDedupEngine.apply(body as Record<string, unknown>);
|
||||||
|
const elapsed = Date.now() - t0;
|
||||||
|
const grew = process.memoryUsage().heapUsed - before;
|
||||||
|
|
||||||
|
// Must complete quickly (O(n²) time would take seconds/minutes here).
|
||||||
|
assert.ok(elapsed < 4000, `session-dedup must stay fast on huge input (took ${elapsed}ms)`);
|
||||||
|
// Heap growth must be bounded well below the pre-fix multi-hundred-MB / GB blow-up.
|
||||||
|
// Pre-fix this single call retained hundreds of MB of suffix strings; the 8MB
|
||||||
|
// block budget (plus overhead) must keep transient growth modest.
|
||||||
|
assert.ok(
|
||||||
|
grew < 120 * 1024 * 1024,
|
||||||
|
`heap growth must stay bounded (grew ${(grew / 1048576).toFixed(1)}MB)`
|
||||||
|
);
|
||||||
|
// Result must still be a valid body (engine did not throw / corrupt).
|
||||||
|
assert.ok(Array.isArray((result.body as Record<string, unknown>).messages));
|
||||||
|
});
|
||||||
|
|
||||||
it("never deduplicates the system prompt", () => {
|
it("never deduplicates the system prompt", () => {
|
||||||
const body = makeBody([
|
const body = makeBody([
|
||||||
{ role: "system", content: REPEATED_BLOCK },
|
{ role: "system", content: REPEATED_BLOCK },
|
||||||
|
|||||||
Reference in New Issue
Block a user