fix(compression): bound RTK raw-output store growth and make pointer reads O(bucket) (#10659) (#10660)

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!

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<string, unknown>): 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 };
 }

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<RtkConfig>, override?: Record<string, unk
       typeof merged.rawOutputMaxBytes === "number" && Number.isFinite(merged.rawOutputMaxBytes)
         ? Math.max(1024, Math.floor(merged.rawOutputMaxBytes))
         : DEFAULT_RTK_CONFIG.rawOutputMaxBytes,
+    rawOutputMaxFiles:
+      typeof merged.rawOutputMaxFiles === "number" && Number.isFinite(merged.rawOutputMaxFiles)
+        ? Math.max(1, Math.floor(merged.rawOutputMaxFiles))
+        : DEFAULT_RTK_CONFIG.rawOutputMaxFiles,
+    rawOutputMaxAgeDays:
+      typeof merged.rawOutputMaxAgeDays === "number" &&
+      Number.isFinite(merged.rawOutputMaxAgeDays)
+        ? Math.max(1, Math.floor(merged.rawOutputMaxAgeDays))
+        : DEFAULT_RTK_CONFIG.rawOutputMaxAgeDays,
   };
 }
 
@@ -352,6 +365,14 @@ export function processRtkText(
       techniquesUsed.push("rtk-raw-output-retention");
       rulesApplied.push("rtk:raw-output-retention");
     }
+    // #10659: bounded retention — schedule a throttled async purge whenever retention is
+    // on so the store cannot grow unbounded again. Never blocks the hot path.
+    if (config.rawOutputRetention !== "never") {
+      scheduleRtkRawOutputPurge({
+        maxFiles: config.rawOutputMaxFiles,
+        maxAgeDays: config.rawOutputMaxAgeDays,
+      });
+    }
   }
   return {
     text: result,

open-sse/services/compression/engines/rtk/rawOutput.ts

@@ -1,4 +1,5 @@
 import fs from "node:fs";
+import fsp from "node:fs/promises";
 import path from "node:path";
 import os from "node:os";
 import crypto from "node:crypto";
@@ -71,6 +72,24 @@ export function isLikelyFailureOutput(value: string): boolean {
   );
 }
 
+/**
+ * #10659: the raw-output store used to grow unbounded and every pointer read did a full
+ * readdirSync over the whole store, freezing the event loop with millions of files.
+ * New writes now land in id-prefix buckets (`<store>/<id[0:2]>/...`) 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<T>(
+  items: T[],
+  limit: number,
+  fn: (item: T) => Promise<void>
+): Promise<void> {
+  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<RtkRawOutputPurgeResult> {
+  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 */
+    });
+  });
+}

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,

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