From 54b39690e57517e77587a078f558bc6b67fadee6 Mon Sep 17 00:00:00 2001 From: stanley Date: Thu, 20 Aug 2026 23:10:58 +0700 Subject: [PATCH] fix(compression): bound RTK raw-output store growth and make pointer reads O(bucket) (#10659) (#10660) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Merged — locally validated together with related stanleytejakusuma PRs (typecheck:core clean, complexity/cognitive/file-size/changelog gates green, focused tests passing). Great incident writeup and clean fix. Thanks! --- .../compression/engines/rtk/configSchema.ts | 21 ++ .../services/compression/engines/rtk/index.ts | 23 +- .../compression/engines/rtk/rawOutput.ts | 268 ++++++++++++++++-- open-sse/services/compression/types.ts | 6 + .../rtk-raw-output-retention.test.ts | 111 ++++++++ 5 files changed, 408 insertions(+), 21 deletions(-) create mode 100644 tests/unit/compression/rtk-raw-output-retention.test.ts diff --git a/open-sse/services/compression/engines/rtk/configSchema.ts b/open-sse/services/compression/engines/rtk/configSchema.ts index e7699120e3..2af6385015 100644 --- a/open-sse/services/compression/engines/rtk/configSchema.ts +++ b/open-sse/services/compression/engines/rtk/configSchema.ts @@ -66,6 +66,22 @@ export const RTK_SCHEMA: EngineConfigField[] = [ { value: "always", label: "always" }, ], }, + { + key: "rawOutputMaxFiles", + type: "number", + label: "Max raw-output files (oldest purged beyond this)", + defaultValue: DEFAULT_RTK_CONFIG.rawOutputMaxFiles, + min: 1, + max: 10_000_000, + }, + { + key: "rawOutputMaxAgeDays", + type: "number", + label: "Max raw-output age (days)", + defaultValue: DEFAULT_RTK_CONFIG.rawOutputMaxAgeDays, + min: 1, + max: 3650, + }, { key: "enableRenderers", type: "boolean", @@ -113,5 +129,10 @@ export function validateRtkEngineConfig(config: Record): Engine ) { errors.push("rawOutputRetention must be never, failures, or always"); } + for (const key of ["rawOutputMaxFiles", "rawOutputMaxAgeDays"]) { + if (config[key] !== undefined && (typeof config[key] !== "number" || config[key] < 1)) { + errors.push(`${key} must be a positive number`); + } + } return { valid: errors.length === 0, errors }; } diff --git a/open-sse/services/compression/engines/rtk/index.ts b/open-sse/services/compression/engines/rtk/index.ts index 0d34980043..33b6c90b59 100644 --- a/open-sse/services/compression/engines/rtk/index.ts +++ b/open-sse/services/compression/engines/rtk/index.ts @@ -9,7 +9,11 @@ import { matchRtkFilter } from "./filterLoader.ts"; import { applyLineFilter } from "./lineFilter.ts"; import { smartTruncate } from "./smartTruncate.ts"; import { normalizeCodeLanguage, stripCode } from "./codeStripper.ts"; -import { maybePersistRtkRawOutput, type RtkRawOutputPointer } from "./rawOutput.ts"; +import { + maybePersistRtkRawOutput, + scheduleRtkRawOutputPurge, + type RtkRawOutputPointer, +} from "./rawOutput.ts"; import { applyRenderer } from "./renderers/index.ts"; import { isTextBlock } from "../../messageContent.ts"; import { adaptBodyForCompression } from "../../bodyAdapter.ts"; @@ -121,6 +125,15 @@ function mergeRtkConfig(base?: Partial, override?: Record//...`) so reads are O(bucket), + * and a bounded async purge (see purgeRtkRawOutput) caps total files/age. + */ +const RAW_OUTPUT_BUCKET_LEN = 2; +/** Legacy flat-store entries beyond this size are not synchronously scanned (freeze guard). */ +const LEGACY_FLAT_SCAN_GUARD = 100_000; + +function rawOutputDir(): string { + return path.join(dataDir(), "rtk", "raw-output"); +} + +function bucketDir(id: string): string { + return path.join(rawOutputDir(), id.slice(0, RAW_OUTPUT_BUCKET_LEN)); +} + export function maybePersistRtkRawOutput( raw: string, options: { @@ -93,8 +112,9 @@ export function maybePersistRtkRawOutput( .replace(/^_+|_+$/g, "") .slice(0, 48); const id = safeId(`${now}:${commandSlug}:${raw.length}:${redaction.text}`); - const dir = path.join(dataDir(), "rtk", "raw-output"); - const filePath = path.join(dir, `${now}-${commandSlug || "tool-output"}-${id}.log`); + const dir = bucketDir(id); + const fileName = `${now}-${commandSlug || "tool-output"}-${id}.log`; + const filePath = path.join(dir, fileName); try { fs.mkdirSync(dir, { recursive: true }); fs.writeFileSync(filePath, redaction.text); @@ -135,11 +155,33 @@ export function maybePersistRtkRawOutput( } export function readRtkRawOutput(pointerId: string): string | null { - const dir = path.join(dataDir(), "rtk", "raw-output"); + const dir = rawOutputDir(); if (!fs.existsSync(dir)) return null; - const entry = fs - .readdirSync(dir) - .find((file) => file.endsWith(".log") && file.includes(pointerId)); + + // Bucketed layout first (new writes): one tiny subdir read instead of a full-store scan. + const bucket = bucketDir(pointerId); + if (fs.existsSync(bucket)) { + const entry = fs + .readdirSync(bucket) + .find((file) => file.endsWith(".log") && file.includes(pointerId)); + if (entry) { + const fullPath = path.join(bucket, entry); + if (!fullPath.startsWith(dir)) return null; + return fs.readFileSync(fullPath, "utf8"); + } + } + + // Legacy flat layout (pre-bucket writes). Guarded: scanning a multi-million-entry flat + // store synchronously is exactly the event-loop freeze #10659 reports, so refuse once + // the flat store is pathologically large instead of stalling the gateway. + const entries = fs.readdirSync(dir); + if (entries.length > LEGACY_FLAT_SCAN_GUARD) { + console.warn( + `[rtk-raw-output] legacy flat store has ${entries.length} entries; skipping O(n) pointer scan for ${pointerId}` + ); + return null; + } + const entry = entries.find((file) => file.endsWith(".log") && file.includes(pointerId)); if (!entry) return null; const fullPath = path.join(dir, entry); if (!fullPath.startsWith(dir)) return null; @@ -156,6 +198,51 @@ function commandFromSlug(fileName: string): string { return slug.replace(/_+/g, " ").trim(); } +/** + * Collect every `.log` path in the store (legacy flat + buckets). The flat store is + * guarded so a pathological legacy directory cannot freeze the loop; bucket dirs are + * small by construction (the purge cap keeps each bucket bounded). + */ +function collectRawOutputLogFiles(dir: string): Array<{ name: string; fullPath: string }> { + const logs: Array<{ name: string; fullPath: string }> = []; + let entries: string[]; + try { + entries = fs.readdirSync(dir); + } catch { + return logs; + } + if (entries.length <= LEGACY_FLAT_SCAN_GUARD) { + for (const entry of entries) { + if (entry.endsWith(".log")) logs.push({ name: entry, fullPath: path.join(dir, entry) }); + } + } else { + console.warn( + `[rtk-raw-output] legacy flat store has ${entries.length} entries; skipping sample scan this run` + ); + } + for (const entry of entries) { + if (entry.length !== RAW_OUTPUT_BUCKET_LEN) continue; + const subPath = path.join(dir, entry); + let isDir = false; + try { + isDir = fs.statSync(subPath).isDirectory(); + } catch { + continue; + } + if (!isDir) continue; + let subEntries: string[]; + try { + subEntries = fs.readdirSync(subPath); + } catch { + continue; + } + for (const name of subEntries) { + if (name.endsWith(".log")) logs.push({ name, fullPath: path.join(subPath, name) }); + } + } + return logs; +} + /** * Read the opt-in RTK raw-output store (`DATA_DIR/rtk/raw-output/*.log`) into * `CommandSample[]` for the pure miners `discoverRepeatedNoise()` / `suggestFilter()`. @@ -166,24 +253,17 @@ function commandFromSlug(fileName: string): string { * memory. No throw: a corrupt entry is dropped, not propagated. */ export function listRtkCommandSamples(opts: { limit?: number } = {}): CommandSample[] { - const dir = path.join(dataDir(), "rtk", "raw-output"); + const dir = rawOutputDir(); if (!fs.existsSync(dir)) return []; const limit = Math.max(1, Math.floor(opts.limit ?? 500)); - let logs: string[]; - try { - logs = fs.readdirSync(dir).filter((f) => f.endsWith(".log")); - } catch { - return []; - } + const logs = collectRawOutputLogFiles(dir); // Newest first: the filename is timestamp-prefixed, so a reverse lexical sort works. - logs.sort((a, b) => (a < b ? 1 : a > b ? -1 : 0)); + logs.sort((a, b) => (a.name < b.name ? 1 : a.name > b.name ? -1 : 0)); const samples: CommandSample[] = []; - for (const fileName of logs) { + for (const { name, fullPath } of logs) { if (samples.length >= limit) break; - const fullPath = path.join(dir, fileName); - if (!fullPath.startsWith(dir)) continue; let output: string; try { output = fs.readFileSync(fullPath, "utf8"); @@ -191,7 +271,6 @@ export function listRtkCommandSamples(opts: { limit?: number } = {}): CommandSam continue; } if (output.trim().length === 0) continue; - let command = ""; try { const metaRaw = fs.readFileSync(fullPath.replace(/\.log$/, ".meta.json"), "utf8"); @@ -200,9 +279,158 @@ export function listRtkCommandSamples(opts: { limit?: number } = {}): CommandSam } catch { // No/!invalid sidecar → fall back to the filename slug below. } - if (!command) command = commandFromSlug(fileName) || "tool-output"; - + if (!command) command = commandFromSlug(name) || "tool-output"; samples.push({ command, output }); } return samples; } + +export interface RtkRawOutputPurgeOptions { + maxAgeDays?: number; + maxFiles?: number; +} + +export interface RtkRawOutputPurgeResult { + skipped: boolean; + scanned: number; + deleted: number; + errors: number; +} + +const PURGE_THROTTLE_MS = 60_000; +let lastRawOutputPurgeAt = 0; + +/** Test hook: clear the purge throttle so a test can exercise two consecutive purges. */ +export function resetRtkRawOutputPurgeThrottle(): void { + lastRawOutputPurgeAt = 0; +} + +async function mapLimit( + items: T[], + limit: number, + fn: (item: T) => Promise +): Promise { + let index = 0; + const workers = Array.from({ length: Math.min(limit, items.length) }, async () => { + while (index < items.length) { + const item = items[index++]; + await fn(item); + } + }); + await Promise.all(workers); +} + +/** + * #10659: bounded retention for the raw-output store. Enforces max age and max file count + * asynchronously (never blocks the event loop), best-effort (never throws into callers), + * and throttled to once per minute from the scheduler. + * + * The legacy flat store is skipped when it is pathologically large (guard) — scanning it + * synchronously/async with millions of entries is what froze gateways; the operator does + * a one-off cleanup and the bucketized layout keeps new growth bounded. + */ +export async function purgeRtkRawOutput( + opts: RtkRawOutputPurgeOptions = {} +): Promise { + const now = Date.now(); + if (now - lastRawOutputPurgeAt < PURGE_THROTTLE_MS) { + return { skipped: true, scanned: 0, deleted: 0, errors: 0 }; + } + lastRawOutputPurgeAt = now; + + const maxAgeDays = Math.max(1, Math.floor(opts.maxAgeDays ?? 30)); + const maxFiles = Math.max(1, Math.floor(opts.maxFiles ?? 100_000)); + const maxAgeMs = maxAgeDays * 86_400_000; + const dir = rawOutputDir(); + const result: RtkRawOutputPurgeResult = { skipped: false, scanned: 0, deleted: 0, errors: 0 }; + if (!fs.existsSync(dir)) return result; + + try { + const candidates: Array<{ file: string; meta: string | null; ts: number }> = []; + const flat = await fsp.readdir(dir); + if (flat.length > LEGACY_FLAT_SCAN_GUARD) { + console.warn( + `[rtk-raw-output] legacy flat store has ${flat.length} entries; purge skips flat scan this run (one-off manual cleanup recommended)` + ); + } else { + for (const name of flat) { + if (!name.endsWith(".log")) continue; + candidates.push({ + file: path.join(dir, name), + meta: path.join(dir, name.replace(/\.log$/, ".meta.json")), + ts: parseInt(name, 10) || 0, + }); + } + } + for (const entry of flat) { + if (entry.length !== RAW_OUTPUT_BUCKET_LEN) continue; + const subPath = path.join(dir, entry); + let isDir = false; + try { + isDir = (await fsp.stat(subPath)).isDirectory(); + } catch { + continue; + } + if (!isDir) continue; + let subEntries: string[]; + try { + subEntries = await fsp.readdir(subPath); + } catch { + continue; + } + for (const name of subEntries) { + if (!name.endsWith(".log")) continue; + candidates.push({ + file: path.join(subPath, name), + meta: path.join(subPath, name.replace(/\.log$/, ".meta.json")), + ts: parseInt(name, 10) || 0, + }); + } + } + result.scanned = candidates.length; + + const agedOut = candidates.filter((c) => c.ts > 0 && now - c.ts > maxAgeMs); + const remaining = candidates.filter((c) => !agedOut.includes(c)); + remaining.sort((a, b) => b.ts - a.ts || (a.file < b.file ? 1 : -1)); + const keep = new Set(remaining.slice(0, maxFiles).map((c) => c.file)); + const overflow = remaining.filter((c) => !keep.has(c.file)); + + await mapLimit([...agedOut, ...overflow], 32, async (c) => { + try { + await fsp.unlink(c.file); + result.deleted++; + } catch { + result.errors++; + } + if (c.meta) { + try { + await fsp.unlink(c.meta); + } catch { + // Missing/never-written sidecar is fine. + } + } + }); + + if (result.deleted > 0 || result.errors > 0) { + console.log( + `[rtk-raw-output] purge: scanned=${result.scanned} deleted=${result.deleted} errors=${result.errors} (maxFiles=${maxFiles}, maxAgeDays=${maxAgeDays})` + ); + } + } catch (err) { + console.warn("[rtk-raw-output] purge failed:", (err as Error).message); + result.errors++; + } + return result; +} + +/** + * Schedule a throttled best-effort purge off the hot path. Safe to call on every write: + * purgeRtkRawOutput itself throttles to once per minute. + */ +export function scheduleRtkRawOutputPurge(opts: RtkRawOutputPurgeOptions = {}): void { + setImmediate(() => { + void purgeRtkRawOutput(opts).catch(() => { + /* best-effort */ + }); + }); +} diff --git a/open-sse/services/compression/types.ts b/open-sse/services/compression/types.ts index f35dfac37f..70d457aa91 100644 --- a/open-sse/services/compression/types.ts +++ b/open-sse/services/compression/types.ts @@ -103,6 +103,10 @@ export interface RtkConfig { trustProjectFilters: boolean; rawOutputRetention: RtkRawOutputRetention; rawOutputMaxBytes: number; + /** #10659: cap on total raw-output files before the oldest are purged. Default: 100_000. */ + rawOutputMaxFiles?: number; + /** #10659: max age (days) of retained raw-output files. Default: 30. */ + rawOutputMaxAgeDays?: number; /** R5: enable grouping of near-equivalent consecutive lines. Default: false. */ enableGrouping?: boolean; /** R5: minimum consecutive similar-line run to trigger grouping. Default: 3. */ @@ -473,6 +477,8 @@ export const DEFAULT_RTK_CONFIG: RtkConfig = { trustProjectFilters: false, rawOutputRetention: "never", rawOutputMaxBytes: 1_048_576, + rawOutputMaxFiles: 100_000, + rawOutputMaxAgeDays: 30, enableGrouping: false, groupingThreshold: 3, stripCodeComments: false, diff --git a/tests/unit/compression/rtk-raw-output-retention.test.ts b/tests/unit/compression/rtk-raw-output-retention.test.ts new file mode 100644 index 0000000000..32349bcafb --- /dev/null +++ b/tests/unit/compression/rtk-raw-output-retention.test.ts @@ -0,0 +1,111 @@ +import { describe, it, afterEach } from "node:test"; +import assert from "node:assert/strict"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +import { + maybePersistRtkRawOutput, + purgeRtkRawOutput, + readRtkRawOutput, + resetRtkRawOutputPurgeThrottle, +} from "../../../open-sse/services/compression/engines/rtk/rawOutput.ts"; + +const originalDataDir = process.env.DATA_DIR; + +afterEach(() => { + if (originalDataDir === undefined) delete process.env.DATA_DIR; + else process.env.DATA_DIR = originalDataDir; + resetRtkRawOutputPurgeThrottle(); +}); + +function freshDataDir(): string { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-rtk-store-")); + process.env.DATA_DIR = dir; + return dir; +} + +function storeRoot(dataDir: string): string { + return path.join(dataDir, "rtk", "raw-output"); +} + +/** Write a raw-output file in the bucketized layout and return its pointer id. */ +function writeBucketFile( + dataDir: string, + ts: number, + command: string, + idHex: string, + content: string +): string { + const bucket = path.join(storeRoot(dataDir), idHex.slice(0, 2)); + fs.mkdirSync(bucket, { recursive: true }); + const slug = command.replace(/[^A-Za-z0-9_-]+/g, "_").slice(0, 48); + fs.writeFileSync(path.join(bucket, `${ts}-${slug}-${idHex}.log`), content); + return idHex; +} + +describe("RTK raw-output bounded retention (#10659)", () => { + it("writes new captures into id-prefix buckets and reads them back", () => { + const dataDir = freshDataDir(); + const text = "error: boom\nnoise\n".repeat(8); + const pointer = maybePersistRtkRawOutput(text, { retention: "always" }); + assert.ok(pointer, "pointer should be produced with retention=always"); + // Bucketed layout: pointer.path sits one level below the store root. + assert.equal(path.dirname(pointer!.path), path.join(storeRoot(dataDir), pointer!.id.slice(0, 2))); + assert.ok(fs.existsSync(pointer!.path), "bucket file exists on disk"); + assert.equal(readRtkRawOutput(pointer!.id), text, "read resolves via bucket lookup"); + }); + + it("still reads legacy flat-store files (backward compatibility)", () => { + const dataDir = freshDataDir(); + const store = storeRoot(dataDir); + fs.mkdirSync(store, { recursive: true }); + const id = "ab".padEnd(24, "0"); + fs.writeFileSync(path.join(store, `1710000000000-tool-output-${id}.log`), "legacy content"); + assert.equal(readRtkRawOutput(id), "legacy content"); + }); + + it("returns null for unknown pointer ids", () => { + freshDataDir(); + assert.equal(readRtkRawOutput("ffffffffffffffffffffffff"), null); + }); + + it("purge deletes files older than maxAgeDays and keeps recent ones", async () => { + const dataDir = freshDataDir(); + const now = Date.now(); + const oldId = writeBucketFile(dataDir, now - 40 * 86_400_000, "old", "aa".padEnd(24, "0"), "old"); + writeBucketFile(dataDir, now - 40 * 86_400_000, "old2", "ab".padEnd(24, "0"), "old2"); + const recentId = writeBucketFile(dataDir, now - 1000, "recent", "ac".padEnd(24, "0"), "recent"); + + const result = await purgeRtkRawOutput({ maxAgeDays: 30, maxFiles: 100_000 }); + assert.equal(result.skipped, false); + assert.equal(result.deleted, 2); + assert.equal(readRtkRawOutput(oldId), null, "aged-out file purged"); + assert.equal(readRtkRawOutput(recentId), "recent", "recent file kept"); + }); + + it("purge caps the store at maxFiles, keeping the newest", async () => { + const dataDir = freshDataDir(); + const now = Date.now(); + const ids: string[] = []; + for (let i = 0; i < 8; i++) { + const id = `b${i}`.padEnd(24, "b").slice(0, 24); + writeBucketFile(dataDir, now - i * 1000, `cmd${i}`, id, `content${i}`); + ids.push(id); + } + const result = await purgeRtkRawOutput({ maxAgeDays: 30, maxFiles: 5 }); + assert.equal(result.deleted, 3); + // Newest 5 (i=0..4) survive; oldest 3 (i=5..7) are purged. + assert.equal(readRtkRawOutput(ids[0]), "content0"); + assert.equal(readRtkRawOutput(ids[4]), "content4"); + assert.equal(readRtkRawOutput(ids[5]), null); + assert.equal(readRtkRawOutput(ids[7]), null); + }); + + it("retention=never writes nothing to disk", () => { + const dataDir = freshDataDir(); + const pointer = maybePersistRtkRawOutput("some output", { retention: "never" }); + assert.equal(pointer, null); + assert.equal(fs.existsSync(storeRoot(dataDir)), false); + }); +});