From 359b9b520b462f1ba7009ea2f548846171b0d568 Mon Sep 17 00:00:00 2001 From: "Bob.Hou" Date: Fri, 11 Sep 2026 19:45:16 -0400 Subject: [PATCH] fix(db): self-heal under sustained memory pressure, VACUUM gate, WAL housekeeping (#13011) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A 36-minute outage with a full host-level write-up is the best kind of bug report. The WAL at 2.98 GB being the high-water mark a VACUUM leaves behind — rewriting a 3 GB database after deleting ~125 rows, with auto-checkpoint never shrinking the file and TRUNCATE only every 6 hours — is exactly the kind of thing that is invisible from inside the process, where the V8 heap read 300 MB while RSS was dominated by 1.9 GB of glibc main-heap. Reconciled against the tip after the batch landed: the `ENVIRONMENT.md` table conflicted with #13035's `OMNIROUTE_SQLJS_WASM_PATH` row and both sides were kept. --- Validated in one consolidated worktree cut from `release/v3.8.51`, boarded with the other 19 PRs of this batch. Two in-batch conflicts, both additive and resolved by keeping each side: the `ENVIRONMENT.md` table (#13035 + #13011) and the `chatHelpers.ts` import block (#12975 on the tip + #13017). - `typecheck:core` clean; `check:dashboard-typecheck` OK (206 pre-existing, within baseline); `check:changelog-integrity` OK; `check:docs-counts` migrations ✓ - complexity 2816 / baseline 3218 and cognitive-complexity 1271 / baseline 1437 — both under baseline - 531 of 532 focused assertions green across the batch's 46 test files - `check-file-size` rebaselined for the batch's real growth (annotation `_rebaseline_2026_09_11_mergebatch_v3851_houminxi`, landed on #13038), attributed per PR The single red is **not this batch**: `tests/unit/combo/quota-weighted-strategy.test.ts` → "A/B isolation: 7 hard-empty + 2 at 0.5% + 1 at 40%, floor=1" asserts an order between two connections of identical weight and flakes on the pure tip too — 2 failures in 4 runs at `origin/release/v3.8.51` with nothing from this batch applied. ⚠️ base-red inherited: #12732 — `Docs Gates`, `Merge integrity`, `No new ESLint warnings`, `Unit Tests fast-path` and `Fast Quality Gates` reproduce on the pure tip (provider count 356 vs the 358 the modules define, SKILL.md drift, and `open-sse/utils/stream.ts` at 3115 > frozen 3098, untouched here). Thanks @HouMinXi — the live evidence on these (X500 logs, `storage.sqlite` state, real `/v1/models` probes, the 36-minute outage write-up) is what let a 20-PR batch be reviewed as a unit. --- .../fixes/13011-memory-pressure-self-heal.md | 1 + docs/reference/ENVIRONMENT.md | 5 + open-sse/utils/resourcePressure.ts | 139 ++++++++ src/lib/db/cleanup.ts | 73 ++++- src/lib/db/walMaintenance.ts | 107 ++++++- tests/unit/db-cleanup-vacuum-gate.test.ts | 101 ++++++ tests/unit/db-wal-passive-scheduler.test.ts | 88 ++++++ .../resource-pressure-self-restart.test.ts | 296 ++++++++++++++++++ 8 files changed, 793 insertions(+), 17 deletions(-) create mode 100644 changelog.d/fixes/13011-memory-pressure-self-heal.md create mode 100644 tests/unit/db-cleanup-vacuum-gate.test.ts create mode 100644 tests/unit/db-wal-passive-scheduler.test.ts create mode 100644 tests/unit/resource-pressure-self-restart.test.ts diff --git a/changelog.d/fixes/13011-memory-pressure-self-heal.md b/changelog.d/fixes/13011-memory-pressure-self-heal.md new file mode 100644 index 0000000000..00300f6750 --- /dev/null +++ b/changelog.d/fixes/13011-memory-pressure-self-heal.md @@ -0,0 +1 @@ +- **fix(db):** add an opt-in self-restart circuit for sustained critical memory pressure, gate post-cleanup VACUUM behind a minimum freed-rows threshold, and checkpoint the SQLite WAL every 5 minutes with a size guard that escalates to TRUNCATE, so a growing WAL can no longer stall the event loop into a full outage. diff --git a/docs/reference/ENVIRONMENT.md b/docs/reference/ENVIRONMENT.md index 9a66d935f9..0a96beadac 100644 --- a/docs/reference/ENVIRONMENT.md +++ b/docs/reference/ENVIRONMENT.md @@ -102,6 +102,11 @@ OmniRoute uses **SQLite** (via `better-sqlite3`) for all persistence. These vari | `OMNIROUTE_API_KEY_BASE64` | _(unset)_ | `src/lib/db/encryption.ts` | **Legacy alias** (Base64-encoded form) accepted as a fallback. Decoded automatically before use. | | `OMNIROUTE_DB_HEALTHCHECK_INTERVAL_MS` | _(unset)_ | `src/lib/db/core.ts` | Override the periodic SQLite healthcheck interval (ms). When unset, defaults are derived from `NODE_ENV`. | | `OMNIROUTE_WAL_TRUNCATE_INTERVAL_MS` | `21600000` (6h) | `src/lib/db/walMaintenance.ts` | Override the periodic `wal_checkpoint(TRUNCATE)` interval (ms). Auto-checkpoint never shrinks the WAL file itself, and a long-running server never closes its DB. `0` disables. | +| `OMNIROUTE_WAL_PASSIVE_INTERVAL_MS` | `300000` (5m) | `src/lib/db/walMaintenance.ts` | Override the frequent `wal_checkpoint(PASSIVE)` interval (ms). Keeps pending WAL frames small so the periodic TRUNCATE never copies a multi-GB backlog on the main thread. `0` disables. | +| `OMNIROUTE_WAL_GUARD_MAX_MB` | `256` | `src/lib/db/walMaintenance.ts` | When a PASSIVE tick finds the WAL file above this size, escalate to `wal_checkpoint(TRUNCATE)` immediately instead of waiting for the slow tick. | +| `OMNIROUTE_VACUUM_MIN_DELETED_ROWS` | `1000` | `src/lib/db/cleanup.ts` | Post-cleanup VACUUM only runs when the cleanup deleted at least this many rows. `0` means always VACUUM when a cleanup freed any rows; `1` effectively disables the gate. VACUUM rewrites the entire database (multi-GB WAL + I/O burst on large DBs), so tiny cleanups skip it; the Storage page's scheduled VACUUM (default weekly, #4437) and manual VACUUM still reclaim space. | +| `OMNIROUTE_PRESSURE_SELF_RESTART` | `false` | `open-sse/utils/resourcePressure.ts` | Set to `1`/`true`/`yes`/`on` to exit the process after critical resource pressure is sustained for `OMNIROUTE_PRESSURE_SELF_RESTART_AFTER_MS`, letting a supervisor (systemd `Restart=always`, Docker restart policy) bring back a clean process instead of serving 503s indefinitely. | +| `OMNIROUTE_PRESSURE_SELF_RESTART_AFTER_MS` | `120000` (2m) | `open-sse/utils/resourcePressure.ts` | How long critical pressure must persist before the self-restart exit fires. | | `OMNIROUTE_SQLJS_WASM_PATH` | _(auto-detect)_ | `src/lib/db/adapters/sqljsAdapter.ts` | Explicit path (absolute or relative to cwd) to `sql-wasm.wasm` when using the `sql.js` WASM fallback adapter. Auto-detected via package dependencies and candidate layouts when unset. | | `OMNIROUTE_SKIP_DB_HEALTHCHECK` | `0` | `src/lib/db/core.ts`, `src/lib/db/healthCheck.ts` | Set to `1` to skip the DB healthcheck entirely on startup. Useful for short-lived tasks and integration tests. | | `OMNIROUTE_FORCE_DB_HEALTHCHECK` | `0` | `src/lib/db/core.ts` | Set to `1` to force the DB healthcheck loop on, even when it would normally be skipped (e.g., short-lived tasks). | diff --git a/open-sse/utils/resourcePressure.ts b/open-sse/utils/resourcePressure.ts index f3ba7f77d9..57c956182c 100644 --- a/open-sse/utils/resourcePressure.ts +++ b/open-sse/utils/resourcePressure.ts @@ -40,8 +40,82 @@ export type ResourcePressureRuntimeOptions = { maxStaleMs?: number; retryAfterMs?: number; samplerDeps?: SampleResourceSignalsDeps; + selfRestart?: { + enabled?: boolean; + afterMs?: number; + exitCode?: number; + exitFn?: (code: number) => void; + }; }; +type ResolvedSelfRestart = { + enabled: boolean; + afterMs: number; + exitCode: number; + exitFn: (code: number) => void; +}; + +const SELF_RESTART_DEFAULT_AFTER_MS = 120_000; + +function envFlagEnabled(raw: string | undefined): boolean { + return raw != null && /^(1|true|yes|on)$/i.test(raw.trim()); +} + +function resolveSelfRestartOptions( + option: ResourcePressureRuntimeOptions["selfRestart"] +): ResolvedSelfRestart { + const enabled = option?.enabled ?? envFlagEnabled(process.env.OMNIROUTE_PRESSURE_SELF_RESTART); + const rawAfter = process.env.OMNIROUTE_PRESSURE_SELF_RESTART_AFTER_MS; + const envAfter = + rawAfter != null && rawAfter.trim().length > 0 && Number.isFinite(Number(rawAfter)) + ? Number(rawAfter) + : undefined; + const afterMs = requireDuration( + "selfRestart.afterMs", + option?.afterMs ?? envAfter ?? SELF_RESTART_DEFAULT_AFTER_MS + ); + const exitCode = option?.exitCode ?? 1; + if (!Number.isInteger(exitCode) || exitCode < 1 || exitCode > 255) { + throw new RangeError("selfRestart.exitCode must be an integer between 1 and 255"); + } + return { + enabled, + afterMs, + exitCode, + exitFn: option?.exitFn ?? ((code) => process.exit(code)), + }; +} + +/** + * One structured line when the tracker first enters critical. The 2026-09-07 + * P0 (cgroup working set pinned at the 5 GiB cap for 36 minutes, then a full + * HTTP stall) reached us with zero diagnostic context beyond the shed reason, + * so the first transition now dumps the numbers an operator needs to tell a + * real leak from a mistuned guard. + */ +function logCriticalTransitionDiagnostics( + reason: PressureReason, + signals: ResourceSignals | null +): void { + const usage = process.memoryUsage(); + const cgroup = signals?.cgroup; + console.warn( + `[resourcePressure] entered critical (reason=${reason}) ` + + formatPressureDetail({ + heapUsedMb: Math.round(usage.heapUsed / MB), + heapTotalMb: Math.round(usage.heapTotal / MB), + rssMb: Math.round(usage.rss / MB), + externalMb: Math.round(usage.external / MB), + arrayBuffersMb: Math.round(usage.arrayBuffers / MB), + cgroupCurrentMb: cgroup?.currentBytes != null ? Math.round(cgroup.currentBytes / MB) : null, + cgroupFileMb: cgroup?.fileBytes != null ? Math.round(cgroup.fileBytes / MB) : null, + cgroupMaxMb: cgroup?.maxBytes != null ? Math.round(cgroup.maxBytes / MB) : null, + psiSomeAvg10: signals?.psi?.someAvg10 ?? null, + psiFullAvg10: signals?.psi?.fullAvg10 ?? null, + }) + ); +} + export type ResourcePressureRuntime = { check: () => ResourcePressureGuardResult | null; getObservation: () => ResourcePressureObservation; @@ -174,6 +248,7 @@ export function createResourcePressureRuntime( handle.unref(); }); const tracker = createResourcePressureTracker(thresholds); + const selfRestart = resolveSelfRestartOptions(options.selfRestart); let lastSignals: ResourceSignals | null = null; let state = emptyState(); @@ -182,6 +257,48 @@ export function createResourcePressureRuntime( let scheduled = false; let inFlight: Promise | null = null; let disposed = false; + let criticalSinceMs: number | null = null; + let selfRestartFired = false; + + const observeSelfRestart = (settledAtMs: number): void => { + if (state.severity !== "critical") { + criticalSinceMs = null; + return; + } + if (criticalSinceMs === null) { + criticalSinceMs = settledAtMs; + logCriticalTransitionDiagnostics(state.reason, lastSignals); + return; + } + if ( + !selfRestart.enabled || + selfRestartFired || + settledAtMs - criticalSinceMs < selfRestart.afterMs + ) { + return; + } + // Sustained critical means the process can no longer serve reliably (the + // 2026-09-07 outage: 36 minutes of global 503s, then a fully stalled event + // loop until an operator restarted the container by hand). Exiting lets the + // supervisor (systemd Restart=always) bring back a clean process in seconds + // instead of leaving every caller wedged until human intervention. + console.error( + `[resourcePressure] critical pressure sustained for ${settledAtMs - criticalSinceMs}ms ` + + `(>= ${selfRestart.afterMs}ms); exiting with code ${selfRestart.exitCode} so the supervisor restarts a clean process` + ); + try { + selfRestart.exitFn(selfRestart.exitCode); + // Only reached when a custom exitFn returns (tests); process.exit never does. + selfRestartFired = true; + } catch (error: unknown) { + // A throwing exitFn must not brick the circuit: reset so the next sustained + // critical window retries, and log loudly since the pre-exit line above + // already claimed the process was leaving. + criticalSinceMs = null; + const message = error instanceof Error ? error.message : String(error); + console.error(`[resourcePressure] self-restart exit failed, circuit re-armed: ${message}`); + } + }; const refresh = (): void => { if (disposed || inFlight) return; @@ -193,6 +310,7 @@ export function createResourcePressureRuntime( const settledAtMs = nowMs(); lastSignals = signals; state = tracker.observe(signals); + observeSelfRestart(settledAtMs); lastRefreshAtMs = settledAtMs; nextRefreshAtMs = settledAtMs + staleAfterMs; }) @@ -210,6 +328,23 @@ export function createResourcePressureRuntime( schedule(refresh); }; + // The self-restart circuit measures *sustained* critical time, so it must not + // depend on incoming requests to advance: during an outage clients back off and + // check() may not be called for long stretches. An unref'd driver re-arms the + // refresh whenever the circuit is armed. A fully stalled event loop still can't + // be unwedged from inside the process — that case belongs to the supervisor's + // own watchdog, not to this circuit. + let selfRestartDriver: NodeJS.Timeout | null = null; + if (selfRestart.enabled) { + const driverIntervalMs = Math.max(1_000, Math.min(staleAfterMs, 10_000)); + selfRestartDriver = setInterval(() => { + if (disposed) return; + nextRefreshAtMs = Math.min(nextRefreshAtMs, nowMs()); + scheduleRefresh(); + }, driverIntervalMs); + selfRestartDriver.unref?.(); + } + return { check() { let heapUsedMb = 0; @@ -253,6 +388,10 @@ export function createResourcePressureRuntime( dispose() { disposed = true; scheduled = false; + if (selfRestartDriver) { + clearInterval(selfRestartDriver); + selfRestartDriver = null; + } }, }; } diff --git a/src/lib/db/cleanup.ts b/src/lib/db/cleanup.ts index 01e84a7964..5ab85942a3 100644 --- a/src/lib/db/cleanup.ts +++ b/src/lib/db/cleanup.ts @@ -888,6 +888,59 @@ export async function cleanupProxyLogs(): Promise { const CLEANUP_INTERVAL_MS = 6 * 60 * 60 * 1000; // 6 hours let _cleanupSchedulerTimer: ReturnType | null = null; +const VACUUM_MIN_DELETED_ROWS_DEFAULT = 1000; + +export function getVacuumMinDeletedRows(): number { + const raw = process.env.OMNIROUTE_VACUUM_MIN_DELETED_ROWS; + if (typeof raw === "string" && raw.trim().length > 0) { + const parsed = Number(raw); + // 0 is valid and means "always VACUUM after a cleanup that freed any rows". + if (Number.isFinite(parsed) && parsed >= 0) return Math.floor(parsed); + } + return VACUUM_MIN_DELETED_ROWS_DEFAULT; +} + +/** + * VACUUM rewrites the entire database file (a multi-GB DB produces a + * multi-GB WAL and a matching page-cache/I/O burst on the host). Running it + * after a cleanup that only freed a handful of rows buys no space and pays + * the full rewrite cost, so tiny cleanups skip it; the scheduled VACUUM + * (#4437) and large cleanups still reclaim space. + */ +export function shouldVacuumAfterCleanup( + totalDeleted: number, + minRows: number = getVacuumMinDeletedRows() +): boolean { + return totalDeleted > 0 && totalDeleted >= minRows; +} + +/** + * Runs the post-cleanup VACUUM only when the cleanup freed enough rows to + * justify a full-database rewrite. Returns true when VACUUM ran. + */ +export async function vacuumAfterCleanup( + totalDeleted: number, + exec: (sql: string) => void, + log: (message: string) => void = (m) => console.log(m), + logError: (message: string, error: unknown) => void = (m, e) => console.error(m, e) +): Promise { + if (totalDeleted <= 0) return false; + const minRows = getVacuumMinDeletedRows(); + if (!shouldVacuumAfterCleanup(totalDeleted, minRows)) { + log(`[Cleanup] Freed ${totalDeleted} rows; skipping VACUUM (below ${minRows}-row threshold).`); + return false; + } + log(`[Cleanup] Running VACUUM to reclaim ${totalDeleted} freed rows...`); + try { + exec("VACUUM"); + log("[Cleanup] VACUUM completed after cleanup."); + return true; + } catch (vacErr) { + logError("[Cleanup] VACUUM after cleanup failed:", vacErr); + return false; + } +} + /** * Start the background cleanup scheduler. Runs cleanup on startup * and then every 6 hours. Runs VACUUM after deletes to reclaim disk space. @@ -906,14 +959,8 @@ export function startCleanupScheduler(): void { const proxyResult = await cleanupProxyLogs(); const totalDeleted = result.totalDeleted + proxyResult.deleted; if (totalDeleted > 0) { - console.log(`[Cleanup] Startup cleanup freed ${totalDeleted} rows. Running VACUUM...`); - try { - const db = getDbInstance(); - db.exec("VACUUM"); - console.log("[Cleanup] VACUUM completed after startup cleanup."); - } catch (vacErr) { - console.error("[Cleanup] VACUUM after cleanup failed:", vacErr); - } + console.log(`[Cleanup] Startup cleanup freed ${totalDeleted} rows.`); + await vacuumAfterCleanup(totalDeleted, (sql) => getDbInstance().exec(sql)); } } catch (err) { console.error("[Cleanup] Startup cleanup failed:", err); @@ -927,14 +974,8 @@ export function startCleanupScheduler(): void { const proxyResult = await cleanupProxyLogs(); const totalDeleted = result.totalDeleted + proxyResult.deleted; if (totalDeleted > 0) { - console.log(`[Cleanup] Periodic cleanup freed ${totalDeleted} rows. Running VACUUM...`); - try { - const db = getDbInstance(); - db.exec("VACUUM"); - console.log("[Cleanup] VACUUM completed after periodic cleanup."); - } catch (vacErr) { - console.error("[Cleanup] VACUUM after cleanup failed:", vacErr); - } + console.log(`[Cleanup] Periodic cleanup freed ${totalDeleted} rows.`); + await vacuumAfterCleanup(totalDeleted, (sql) => getDbInstance().exec(sql)); } } catch (err) { console.error("[Cleanup] Periodic cleanup failed:", err); diff --git a/src/lib/db/walMaintenance.ts b/src/lib/db/walMaintenance.ts index 8dfad87bad..d884720a27 100644 --- a/src/lib/db/walMaintenance.ts +++ b/src/lib/db/walMaintenance.ts @@ -1,3 +1,4 @@ +import fs from "fs"; import { isAutomatedTestProcess } from "@/shared/utils/testProcess"; import { isNextBuildPhase } from "../buildPhase"; import type { SqliteAdapter } from "./adapters/types"; @@ -36,9 +37,12 @@ export interface WalMaintenanceState { const isCloud = typeof globalThis.caches === "object" && globalThis.caches !== null; const DEFAULT_WAL_TRUNCATE_INTERVAL_MS = 6 * 60 * 60 * 1000; +const DEFAULT_WAL_PASSIVE_INTERVAL_MS = 5 * 60 * 1000; +const DEFAULT_WAL_GUARD_MAX_BYTES = 256 * 1024 * 1024; const RETRY_DELAY_MS = 60_000; let walTimer: NodeJS.Timeout | null = null; +let walPassiveTimer: NodeJS.Timeout | null = null; let retryTimer: NodeJS.Timeout | null = null; let ticks = 0; let busyStreak = 0; @@ -133,6 +137,41 @@ export function getWalMaintenanceIntervalMs(env: NodeJS.ProcessEnv = process.env return DEFAULT_WAL_TRUNCATE_INTERVAL_MS; } +export function getWalPassiveIntervalMs(env: NodeJS.ProcessEnv = process.env): number { + const rawValue = env.OMNIROUTE_WAL_PASSIVE_INTERVAL_MS; + if (typeof rawValue === "string" && rawValue.trim().length > 0) { + const parsed = Number(rawValue); + if (Number.isFinite(parsed) && parsed >= 0) { + return parsed; + } + } + return DEFAULT_WAL_PASSIVE_INTERVAL_MS; +} + +export function getWalGuardMaxBytes(env: NodeJS.ProcessEnv = process.env): number { + const rawValue = env.OMNIROUTE_WAL_GUARD_MAX_MB; + if (typeof rawValue === "string" && rawValue.trim().length > 0) { + const parsed = Number(rawValue); + if (Number.isFinite(parsed) && parsed >= 1) { + return Math.floor(parsed) * 1024 * 1024; + } + } + return DEFAULT_WAL_GUARD_MAX_BYTES; +} + +function getWalFileSizeBytes(sqliteFile: string | null): number | null { + if (!sqliteFile) return null; + try { + return fs.statSync(`${sqliteFile}-wal`).size; + } catch { + return null; + } +} + +function formatWalMb(bytes: number | null): string { + return bytes == null ? "null" : String(Math.round(bytes / (1024 * 1024))); +} + export function logCheckpointOutcome( outcome: WalCheckpointOutcome, mode: WalCheckpointMode, @@ -178,6 +217,57 @@ function schedulePassiveRetry(db: SqliteAdapter): void { retryTimer.unref?.(); } +function startWalPassiveScheduler( + db: SqliteAdapter, + sqliteFile: string | null, + env: NodeJS.ProcessEnv +): void { + if (walPassiveTimer) { + clearInterval(walPassiveTimer); + walPassiveTimer = null; + } + if (sqliteFile === null || isCloud || isNextBuildPhase() || isAutomatedTestProcess()) return; + const intervalMs = getWalPassiveIntervalMs(env); + if (intervalMs <= 0) return; + walPassiveTimer = setInterval(() => { + try { + if (!db.open) return; + const walBeforeBytes = getWalFileSizeBytes(sqliteFile); + const stats = runCheckpointNow(db, "PASSIVE", { + sqliteFile, + isCloud, + isBuildPhase: isNextBuildPhase(), + }); + if (stats.skipped) return; + if (stats.busy || (stats.checkpointedFrames ?? 0) > 0) { + console.log( + `[DB] WAL passive checkpoint (busy=${stats.busy ? 1 : 0} logFrames=${stats.logFrames} ` + + `checkpointedFrames=${stats.checkpointedFrames} walMb=${formatWalMb(walBeforeBytes)})` + ); + } + const guardMaxBytes = getWalGuardMaxBytes(env); + if (walBeforeBytes != null && walBeforeBytes > guardMaxBytes) { + const startedAtMs = Date.now(); + const truncateStats = runCheckpointNow(db, "TRUNCATE", { + sqliteFile, + isCloud, + isBuildPhase: isNextBuildPhase(), + }); + console.log( + `[DB] WAL above guard (${formatWalMb(walBeforeBytes)}MB > ${Math.floor(guardMaxBytes / (1024 * 1024))}MB); ` + + `ran TRUNCATE in ${Date.now() - startedAtMs}ms ` + + `(walMbAfter=${formatWalMb(getWalFileSizeBytes(sqliteFile))} busy=${truncateStats.busy ? 1 : 0} ` + + `checkpointedFrames=${truncateStats.checkpointedFrames})` + ); + } + } catch (error: unknown) { + const message = error instanceof Error ? error.message : String(error); + console.warn("[DB] WAL passive checkpoint failed:", message); + } + }, intervalMs); + walPassiveTimer.unref?.(); +} + export function startWalMaintenance( db: SqliteAdapter, sqliteFile: string | null, @@ -186,10 +276,15 @@ export function startWalMaintenance( stopWalMaintenance(); if (sqliteFile === null || isCloud || isNextBuildPhase() || isAutomatedTestProcess()) return; const intervalMs = getWalMaintenanceIntervalMs(env); - if (intervalMs <= 0) return; + if (intervalMs <= 0) { + startWalPassiveScheduler(db, sqliteFile, env); + return; + } walTimer = setInterval(() => { try { if (!db.open) return; + const walBeforeBytes = getWalFileSizeBytes(sqliteFile); + const startedAtMs = Date.now(); const outcome = runCheckpointNow(db, "TRUNCATE"); if (outcome.skipped) return; ticks++; @@ -199,6 +294,11 @@ export function startWalMaintenance( schedulePassiveRetry(db); } else if (outcome.ok) { recordOk(); + console.log( + `[DB] Periodic SQLite WAL checkpoint completed (TRUNCATE) in ${Date.now() - startedAtMs}ms ` + + `(walMbBefore=${formatWalMb(walBeforeBytes)} walMbAfter=${formatWalMb(getWalFileSizeBytes(sqliteFile))} ` + + `busy=${outcome.busy ? 1 : 0} logFrames=${outcome.logFrames} checkpointedFrames=${outcome.checkpointedFrames})` + ); } else { logCheckpointOutcome(outcome, "TRUNCATE", busyStreak); } @@ -207,6 +307,7 @@ export function startWalMaintenance( } }, intervalMs); walTimer.unref?.(); + startWalPassiveScheduler(db, sqliteFile, env); } export function stopWalMaintenance(): void { @@ -214,6 +315,10 @@ export function stopWalMaintenance(): void { clearInterval(walTimer); walTimer = null; } + if (walPassiveTimer) { + clearInterval(walPassiveTimer); + walPassiveTimer = null; + } if (retryTimer) { clearTimeout(retryTimer); retryTimer = null; diff --git a/tests/unit/db-cleanup-vacuum-gate.test.ts b/tests/unit/db-cleanup-vacuum-gate.test.ts new file mode 100644 index 0000000000..81f6c9e798 --- /dev/null +++ b/tests/unit/db-cleanup-vacuum-gate.test.ts @@ -0,0 +1,101 @@ +/** + * Post-cleanup VACUUM gate. + * + * VACUUM rewrites the entire database file: on a ~3GB DB that is a 3GB WAL plus a + * full page-cache/I/O burst on the host. The cleanup scheduler used to run it after + * every non-empty cleanup — including startup cleanups that freed ~100 rows — which + * is pure churn. The gate skips VACUUM unless the cleanup freed enough rows to + * justify a full rewrite (OMNIROUTE_VACUUM_MIN_DELETED_ROWS, default 1000). + */ +import assert from "node:assert/strict"; +import { describe, it } from "node:test"; +import { + getVacuumMinDeletedRows, + shouldVacuumAfterCleanup, + vacuumAfterCleanup, +} from "../../src/lib/db/cleanup.ts"; + +describe("shouldVacuumAfterCleanup", () => { + it("skips VACUUM when nothing was deleted", () => { + assert.equal(shouldVacuumAfterCleanup(0, 1000), false); + }); + + it("skips VACUUM for cleanups below the threshold", () => { + assert.equal(shouldVacuumAfterCleanup(1, 1000), false); + assert.equal(shouldVacuumAfterCleanup(999, 1000), false); + }); + + it("runs VACUUM at and above the threshold", () => { + assert.equal(shouldVacuumAfterCleanup(1000, 1000), true); + assert.equal(shouldVacuumAfterCleanup(5000, 1000), true); + }); + + it("treats a 0 threshold as always-vacuum", () => { + const saved = process.env.OMNIROUTE_VACUUM_MIN_DELETED_ROWS; + process.env.OMNIROUTE_VACUUM_MIN_DELETED_ROWS = "0"; + try { + assert.equal(getVacuumMinDeletedRows(), 0); + assert.equal(shouldVacuumAfterCleanup(1, 0), true); + } finally { + if (saved === undefined) delete process.env.OMNIROUTE_VACUUM_MIN_DELETED_ROWS; + else process.env.OMNIROUTE_VACUUM_MIN_DELETED_ROWS = saved; + } + }); + + it("defaults to 1000 rows and honors the env override", () => { + const saved = process.env.OMNIROUTE_VACUUM_MIN_DELETED_ROWS; + try { + delete process.env.OMNIROUTE_VACUUM_MIN_DELETED_ROWS; + assert.equal(getVacuumMinDeletedRows(), 1000); + process.env.OMNIROUTE_VACUUM_MIN_DELETED_ROWS = "50"; + assert.equal(getVacuumMinDeletedRows(), 50); + assert.equal(shouldVacuumAfterCleanup(60), true); + process.env.OMNIROUTE_VACUUM_MIN_DELETED_ROWS = "not-a-number"; + assert.equal(getVacuumMinDeletedRows(), 1000); + } finally { + if (saved === undefined) delete process.env.OMNIROUTE_VACUUM_MIN_DELETED_ROWS; + else process.env.OMNIROUTE_VACUUM_MIN_DELETED_ROWS = saved; + } + }); +}); + +describe("vacuumAfterCleanup", () => { + it("does not exec VACUUM below the threshold but says why", async () => { + const execed: string[] = []; + const logs: string[] = []; + const ran = await vacuumAfterCleanup( + 125, + (sql) => execed.push(sql), + (m) => logs.push(m) + ); + assert.equal(ran, false); + assert.deepEqual(execed, []); + assert.ok(logs.some((line) => line.includes("skipping VACUUM"))); + }); + + it("execs VACUUM once when enough rows were freed", async () => { + const execed: string[] = []; + const ran = await vacuumAfterCleanup( + 1000, + (sql) => execed.push(sql), + () => {}, + () => {} + ); + assert.equal(ran, true); + assert.deepEqual(execed, ["VACUUM"]); + }); + + it("swallows VACUUM failures into an error log like the old inline path", async () => { + const errLogs: string[] = []; + const ran = await vacuumAfterCleanup( + 5000, + () => { + throw new Error("disk full"); + }, + () => {}, + (m) => errLogs.push(m) + ); + assert.equal(ran, false); + assert.ok(errLogs.some((line) => line.includes("VACUUM after cleanup failed"))); + }); +}); diff --git a/tests/unit/db-wal-passive-scheduler.test.ts b/tests/unit/db-wal-passive-scheduler.test.ts new file mode 100644 index 0000000000..bac4d1c7ab --- /dev/null +++ b/tests/unit/db-wal-passive-scheduler.test.ts @@ -0,0 +1,88 @@ +/** + * WAL passive-checkpoint scheduler + size guard, and TRUNCATE telemetry. + * After #12853 the scheduler lives in walMaintenance.ts; this reads the wiring. + */ +import test from "node:test"; +import assert from "node:assert/strict"; +import fs from "node:fs"; +import path from "node:path"; + +function readSource(relativePath: string): string { + return fs.readFileSync(path.join(process.cwd(), relativePath), "utf8"); +} + +const WAL_PATH = "src/lib/db/walMaintenance.ts"; +const CORE_PATH = "src/lib/db/core.ts"; + +function fnBody(source: string, name: string, span = 2200): string { + const start = source.indexOf(`function ${name}`); + assert.notEqual(start, -1, `${name} must exist`); + return source.slice(start, start + span); +} + +test("a frequent PASSIVE checkpoint scheduler boots alongside the truncate scheduler", () => { + const source = readSource(WAL_PATH); + const bootIdx = source.indexOf("function startWalMaintenance"); + assert.notEqual(bootIdx, -1); + const window = source.slice(bootIdx, bootIdx + 2200); + assert.match( + window, + /startWalPassiveScheduler\(/, + "startWalMaintenance() must start the PASSIVE scheduler next to the TRUNCATE scheduler" + ); + const core = readSource(CORE_PATH); + assert.match(core, /startWalMaintenance\(db, SQLITE_FILE\)/); +}); + +test("the passive scheduler runs wal_checkpoint(PASSIVE) and escalates to TRUNCATE over the size guard", () => { + const source = readSource(WAL_PATH); + const body = fnBody(source, "startWalPassiveScheduler", 2600); + assert.match(body, /runCheckpointNow\(db, "PASSIVE"/); + assert.match( + body, + /runCheckpointNow\(db, "TRUNCATE"/, + "when the WAL file exceeds the guard, escalate to TRUNCATE immediately instead of waiting for the 6h tick" + ); +}); + +test("the passive scheduler self-gates like the other DB schedulers", () => { + const body = fnBody(readSource(WAL_PATH), "startWalPassiveScheduler", 400); + assert.match(body, /isCloud \|\| isNextBuildPhase\(\) \|\| isAutomatedTestProcess\(\)/); +}); + +test("both WAL schedulers are cleared on close", () => { + const stop = fnBody(readSource(WAL_PATH), "stopWalMaintenance", 500); + assert.match(stop, /walTimer/); + assert.match(stop, /walPassiveTimer/); + const close = fnBody(readSource(CORE_PATH), "closeDbInstance", 400); + assert.match(close, /stopWalMaintenance\(\)/); +}); + +test("checkpoint results keep busy/frames counters so a starved checkpoint is visible", () => { + const body = fnBody(readSource(WAL_PATH), "runCheckpointNow", 900); + assert.match(body, /busy:/, "busy=1 (checkpoint blocked by readers) must not be swallowed"); + assert.match(body, /checkpointedFrames:/); +}); + +test("the TRUNCATE tick logs duration and WAL sizes for post-mortem diagnosis", () => { + const body = fnBody(readSource(WAL_PATH), "startWalMaintenance", 2200); + assert.match(body, /walMbBefore=/); + assert.match(body, /busy=/); +}); + +test("the WAL size guard rejects sub-1MB values that would floor to a 0-byte guard", () => { + const body = fnBody(readSource(WAL_PATH), "getWalGuardMaxBytes", 500); + assert.match( + body, + /parsed >= 1/, + "OMNIROUTE_WAL_GUARD_MAX_MB=0.5 would Math.floor to 0 bytes and escalate to TRUNCATE on every tick" + ); +}); + +test("the new env vars are documented", () => { + const docs = readSource("docs/reference/ENVIRONMENT.md"); + assert.match(docs, /OMNIROUTE_WAL_PASSIVE_INTERVAL_MS/); + assert.match(docs, /OMNIROUTE_WAL_GUARD_MAX_MB/); + assert.match(docs, /OMNIROUTE_VACUUM_MIN_DELETED_ROWS/); + assert.match(docs, /OMNIROUTE_PRESSURE_SELF_RESTART/); +}); diff --git a/tests/unit/resource-pressure-self-restart.test.ts b/tests/unit/resource-pressure-self-restart.test.ts new file mode 100644 index 0000000000..210673fd0e --- /dev/null +++ b/tests/unit/resource-pressure-self-restart.test.ts @@ -0,0 +1,296 @@ +/** + * Self-heal circuit for sustained critical resource pressure. + * + * The 2026-09-07 outage: the container's cgroup working set sat at the 5 GiB cap + * for 36 minutes (every request 503), then the event loop stalled completely until + * an operator restarted the container by hand. With OMNIROUTE_PRESSURE_SELF_RESTART + * enabled the runtime exits on sustained critical pressure so the supervisor + * (systemd Restart=always) brings back a clean process in seconds. + */ +import assert from "node:assert/strict"; +import { describe, it } from "node:test"; +import { + createResourcePressureRuntime, + type ResourcePressureRuntime, +} from "../../open-sse/utils/resourcePressure.ts"; +import type { ResourceSignals } from "../../open-sse/utils/resourcePressurePolicy.ts"; + +const MiB = 1024 ** 2; + +function criticalSignals(observedAtMs: number): ResourceSignals { + return { + observedAtMs, + v8: { heapUsedBytes: 100 * MiB, heapLimitBytes: 1_000 * MiB }, + process: { + rssBytes: 200 * MiB, + externalBytes: 10 * MiB, + arrayBuffersBytes: MiB, + availableBytes: null, + constrainedBytes: null, + }, + // workingset (current - file) = 950MB of a 1000MB cgroup cap -> critical ratio 0.95 + cgroup: { + currentBytes: 950 * MiB, + maxBytes: 1_000 * MiB, + highBytes: null, + fileBytes: 0, + events: null, + }, + psi: null, + }; +} + +function normalSignals(observedAtMs: number): ResourceSignals { + const signals = criticalSignals(observedAtMs); + return { + ...signals, + cgroup: { ...signals.cgroup, currentBytes: 100 * MiB }, + }; +} + +const fastThresholds = { + highRatio: 0.8, + criticalRatio: 0.9, + recoveryRatio: 0.7, + highPsiAvg10: 20, + criticalPsiAvg10: 40, + recoveryPsiAvg10: 10, + sustainedSamplesHigh: 2, + sustainedSamplesCritical: 2, + recoverySamples: 2, +}; + +function makeHarness(options: { + selfRestart?: { enabled?: boolean; afterMs?: number; exitCode?: number }; +}) { + let clock = 0; + let scheduledFn: (() => void) | null = null; + const exitCalls: number[] = []; + const warnings: string[] = []; + const errors: string[] = []; + const origWarn = console.warn; + const origError = console.error; + console.warn = (msg: unknown) => warnings.push(String(msg)); + console.error = (msg: unknown) => errors.push(String(msg)); + + const runtime = createResourcePressureRuntime({ + thresholds: fastThresholds, + immediateHeapUsedMb: () => 0, + nowMs: () => clock, + schedule: (fn) => { + scheduledFn = fn; + }, + staleAfterMs: 0, + maxStaleMs: 60 * 60 * 1000, + retryAfterMs: 1000, + sample: async () => harness.signals(clock), + selfRestart: { + enabled: options.selfRestart?.enabled, + afterMs: options.selfRestart?.afterMs, + exitCode: options.selfRestart?.exitCode, + exitFn: (code) => { + exitCalls.push(code); + }, + }, + }); + + const harness = { + runtime, + exitCalls, + warnings, + errors, + signals: criticalSignals as (ms: number) => ResourceSignals, + async tick(advanceMs: number) { + clock += advanceMs; + runtime.check(); + assert.ok(scheduledFn, "check() must schedule a refresh"); + const fn = scheduledFn; + scheduledFn = null; + fn(); + await runtime.whenRefreshSettled(); + }, + restore() { + console.warn = origWarn; + console.error = origError; + runtime.dispose(); + }, + }; + return harness; +} + +describe("resource pressure self-restart circuit", () => { + it("exits once critical pressure has been sustained for afterMs", async () => { + const h = makeHarness({ selfRestart: { enabled: true, afterMs: 60_000 } }); + try { + await h.tick(1_000); // elevated streak 1, still below critical sample count + await h.tick(1_000); // streak 2 -> critical, criticalSince = 2000 + assert.deepEqual(h.exitCalls, []); + assert.ok( + h.warnings.some((line) => line.includes("entered critical")), + "the first critical transition must log diagnostics" + ); + await h.tick(30_000); // critical for 30s < 60s afterMs + assert.deepEqual(h.exitCalls, []); + await h.tick(31_000); // critical for 61s >= 60s -> self-restart + assert.deepEqual(h.exitCalls, [1]); + assert.ok( + h.errors.some((line) => line.includes("exiting with code 1")), + "the self-restart must log the exit reason for post-mortem diagnosis" + ); + await h.tick(120_000); // never fires twice + assert.deepEqual(h.exitCalls, [1]); + } finally { + h.restore(); + } + }); + + it("stays quiet when pressure recovers before afterMs", async () => { + const h = makeHarness({ selfRestart: { enabled: true, afterMs: 60_000 } }); + try { + await h.tick(1_000); + await h.tick(1_000); // critical since 2000 + h.signals = normalSignals; + await h.tick(10_000); // recovery streak 1 + await h.tick(10_000); // recovery streak 2 -> normal, circuit resets + h.signals = criticalSignals; + await h.tick(10_000); // elevated again + await h.tick(10_000); // critical again, fresh criticalSince + await h.tick(30_000); // 30s < 60s + assert.deepEqual(h.exitCalls, []); + } finally { + h.restore(); + } + }); + + it("never exits when the circuit is disabled (the default)", async () => { + const saved = process.env.OMNIROUTE_PRESSURE_SELF_RESTART; + delete process.env.OMNIROUTE_PRESSURE_SELF_RESTART; + const h = makeHarness({ selfRestart: { afterMs: 1_000 } }); + try { + for (let i = 0; i < 10; i += 1) { + await h.tick(60_000); // critical far past afterMs + } + assert.deepEqual(h.exitCalls, []); + } finally { + h.restore(); + if (saved !== undefined) process.env.OMNIROUTE_PRESSURE_SELF_RESTART = saved; + } + }); + + it("honors the env switch and custom afterMs", async () => { + const savedFlag = process.env.OMNIROUTE_PRESSURE_SELF_RESTART; + const savedAfter = process.env.OMNIROUTE_PRESSURE_SELF_RESTART_AFTER_MS; + process.env.OMNIROUTE_PRESSURE_SELF_RESTART = "1"; + process.env.OMNIROUTE_PRESSURE_SELF_RESTART_AFTER_MS = "5000"; + const h = makeHarness({}); + try { + await h.tick(1_000); + await h.tick(1_000); // critical since 2000 + await h.tick(4_000); // 4s < 5s env afterMs + assert.deepEqual(h.exitCalls, []); + await h.tick(2_000); // 6s >= 5s + assert.deepEqual(h.exitCalls, [1]); + } finally { + h.restore(); + if (savedFlag === undefined) delete process.env.OMNIROUTE_PRESSURE_SELF_RESTART; + else process.env.OMNIROUTE_PRESSURE_SELF_RESTART = savedFlag; + if (savedAfter === undefined) delete process.env.OMNIROUTE_PRESSURE_SELF_RESTART_AFTER_MS; + else process.env.OMNIROUTE_PRESSURE_SELF_RESTART_AFTER_MS = savedAfter; + } + }); + + it("re-arms when exitFn throws instead of bricking the circuit", async () => { + let clock = 0; + let scheduledFn: (() => void) | null = null; + const exitCalls: number[] = []; + const errors: string[] = []; + const origError = console.error; + console.error = (msg: unknown) => errors.push(String(msg)); + let shouldThrow = true; + const runtime = createResourcePressureRuntime({ + thresholds: fastThresholds, + immediateHeapUsedMb: () => 0, + nowMs: () => clock, + schedule: (fn) => { + scheduledFn = fn; + }, + staleAfterMs: 0, + maxStaleMs: 60 * 60 * 1000, + sample: async () => criticalSignals(clock), + selfRestart: { + enabled: true, + afterMs: 10_000, + exitFn: (code) => { + if (shouldThrow) throw new Error("exit wedged"); + exitCalls.push(code); + }, + }, + }); + const tick = async (advanceMs: number) => { + clock += advanceMs; + runtime.check(); + const fn = scheduledFn; + scheduledFn = null; + assert.ok(fn, "check() must schedule a refresh"); + fn(); + await runtime.whenRefreshSettled(); + }; + try { + await tick(1_000); // streak 1 + await tick(1_000); // critical since t=2000 + await tick(10_000); // sustained 10s >= afterMs -> exitFn throws -> re-arm + assert.deepEqual(exitCalls, []); + assert.ok( + errors.some((line) => line.includes("self-restart exit failed")), + "a throwing exitFn must be logged, not swallowed" + ); + // circuit re-armed: new critical window starts on the next critical sample + shouldThrow = false; + await tick(1_000); // window restarts at t=13000 + await tick(9_000); // 9s < 10s, no fire yet + assert.deepEqual(exitCalls, []); + await tick(2_000); // 11s >= 10s -> retry fires + assert.deepEqual(exitCalls, [1]); + } finally { + console.error = origError; + runtime.dispose(); + } + }); + + it("advances the circuit without any incoming requests via the self-restart driver", async () => { + // Real clock, real timers: no check() calls at all. The driver must re-arm + // refresh on its own so a traffic-less outage still exits. + const exitCalls: number[] = []; + const runtime = createResourcePressureRuntime({ + thresholds: fastThresholds, + immediateHeapUsedMb: () => 0, + staleAfterMs: 1_000, + maxStaleMs: 60_000, + sample: async () => criticalSignals(Date.now()), + selfRestart: { enabled: true, afterMs: 3_000, exitFn: (code) => exitCalls.push(code) }, + }); + try { + const deadline = Date.now() + 15_000; + while (exitCalls.length === 0 && Date.now() < deadline) { + await new Promise((resolve) => setTimeout(resolve, 250)); + } + assert.deepEqual(exitCalls, [1], "driver must fire the exit without any check() traffic"); + } finally { + runtime.dispose(); + } + }, 20_000); + + it("dispose() stops the driver so a disposed runtime never exits", async () => { + const exitCalls: number[] = []; + const runtime = createResourcePressureRuntime({ + thresholds: fastThresholds, + immediateHeapUsedMb: () => 0, + staleAfterMs: 1_000, + sample: async () => criticalSignals(Date.now()), + selfRestart: { enabled: true, afterMs: 1_000, exitFn: (code) => exitCalls.push(code) }, + }); + runtime.dispose(); + await new Promise((resolve) => setTimeout(resolve, 2_500)); + assert.deepEqual(exitCalls, []); + }); +});