diff --git a/onlion/backend/mem-trace.mjs b/onlion/backend/mem-trace.mjs new file mode 100644 index 00000000..c86df762 --- /dev/null +++ b/onlion/backend/mem-trace.mjs @@ -0,0 +1,98 @@ +/* eslint-disable no-console -- the trace is log output by definition */ +/* Memory self-reporting. + * + * The 2026-10-29 incident has 220 invocations that died at the 120 s ceiling carrying + * 762-1014 MB, and nothing in the log says which component that was. Every one of those + * rows is a *killed* request, so the sample is conditioned on the outcome: it cannot + * separate a container baseline from a heap that grew inside a warm instance, and it + * cannot tell how much of the number the request accumulated before it was killed. + * + * These lines are what separates them, at the cost of one `process.memoryUsage()` per + * sampled request: + * + * - `baseline` is taken when this module is first evaluated, before SQLite exists. If it + * is already large, the platform figure includes the container and the V8 reservation, + * and every allocation we could optimise is noise against it. + * - `seq` counts requests in this instance, so a trend can be read against request number + * instead of wall time. A leak only means something inside one warm instance. + * - the breakdown (`heapUsed` vs `external`/`arrayBuffers`) says whether the growth is JS + * objects - a manifest or a JSON body - or byte buffers, which is where a WASM linear + * memory that only grows would show up. + * + * Sampling: `NMG_MEM_TRACE=0` disables, `=1` logs every request, unset logs one in + * `SAMPLE_EVERY`. Fast requests are sampled on purpose: a sample that only contains + * hung requests cannot be compared with anything. + */ +const MB = (bytes) => Math.round((bytes ?? 0) / 10485.76) / 100; +/** Captured at module evaluation: the earliest point our code runs, before SQLite. */ +const baseline = (() => { + const m = process.memoryUsage(); + return { rss: MB(m.rss), heapTotal: MB(m.heapTotal), uptime_s: Math.round(process.uptime()) }; +})(); +const SAMPLE_EVERY = 20; +let seq = 0; + +const mode = () => (process.env.NMG_MEM_TRACE ?? "").trim(); +const disabled = () => mode() === "0"; +const every = () => (mode() === "1" ? 1 : SAMPLE_EVERY); +/** `NMG_MEM_TRACE=1` samples every request, which `n % 1` would never select. */ +const due = (n) => { + const step = every(); + return step <= 1 || n % step === 1; +}; + +function memUsage() { + const m = process.memoryUsage(); + return { + rss: MB(m.rss), + heapUsed: MB(m.heapUsed), + heapTotal: MB(m.heapTotal), + external: MB(m.external), + arrayBuffers: MB(m.arrayBuffers), + }; +} + +/** Fields for a report that is emitted anyway (the slow-request line): never sampled, + * so a slow request always carries its memory and its instance position. */ +export function memFields(extra = {}) { + return { + mem: memUsage(), + instance: { seq, uptime_s: Math.round(process.uptime()), baseline_rss: baseline.rss }, + ...extra, + }; +} + +/** Allocates this request's sequence number and logs the entry sample when it is due. */ +export function memRequestStarted(route, extra = {}) { + seq += 1; + if (disabled() || !due(seq)) return seq; + console.log( + JSON.stringify({ + event: "nmg_mem_sample", + phase: "enter", + at: new Date().toISOString(), + route, + ...memFields(extra), + }), + ); + return seq; +} + +/** Logs one more sample for the current request when the policy allows it. */ +export function memLog(phase, route, extra = {}) { + if (disabled() || !due(seq)) return; + console.log( + JSON.stringify({ + event: "nmg_mem_sample", + phase, + at: new Date().toISOString(), + route, + ...memFields(extra), + }), + ); +} + +/** Exposed for tests: the baseline captured at module evaluation. */ +export function baselineMemory() { + return { ...baseline }; +} diff --git a/onlion/backend/sqlite.mjs b/onlion/backend/sqlite.mjs index ef6e0519..8e122907 100644 --- a/onlion/backend/sqlite.mjs +++ b/onlion/backend/sqlite.mjs @@ -1,53 +1,139 @@ -import { AsyncLocalStorage } from 'node:async_hooks'; -import { basename } from 'node:path'; +import { AsyncLocalStorage } from "node:async_hooks"; +import { basename } from "node:path"; const scope = new AsyncLocalStorage(); let sqlite; let active = false; -export function configure(value) { sqlite = value; } +/** High-water marks for one instance, exposed to the memory trace: what the largest + * simultaneous set of live SQLite images was. A single request holds one partition at a + * time, so a reading above that says an image outlived its part. */ +const highWater = { databases: 0, imageBytes: 0 }; +function noteImage() { + const store = scope.getStore(); + const databases = store ? store.opened.size : 0; + const imageBytes = store + ? [...store.opened.values()].reduce((sum, db) => sum + db.imageBytes, 0) + : 0; + highWater.databases = Math.max(highWater.databases, databases); + highWater.imageBytes = Math.max(highWater.imageBytes, imageBytes); +} +/** What the memory trace reports: the live images in this request, the instance high-water + * mark, and the WASM linear memory size (which only ever grows). */ +export function sqliteStats() { + const store = scope.getStore(); + const opened = store ? [...store.opened.values()] : []; + const imageBytes = opened.reduce((sum, db) => sum + db.imageBytes, 0); + const wasmBytes = (() => { + try { + return sqlite ? sqlite.wasm.heap8u().byteLength : null; + } catch { + return null; + } + })(); + return { + live_databases: opened.length, + live_image_mb: Math.round(imageBytes / 10485.76) / 100, + high_water_databases: highWater.databases, + high_water_image_mb: Math.round(highWater.imageBytes / 10485.76) / 100, + wasm_linear_mb: wasmBytes === null ? null : Math.round(wasmBytes / 10485.76) / 100, + }; +} +export function configure(value) { + sqlite = value; +} export function withDatabases(images, operation) { - if (active) throw Object.assign(new Error('SQLite session is busy; retry'), {status:429,code:'sqlite_busy'}); + if (active) + throw Object.assign(new Error("SQLite session is busy; retry"), { + status: 429, + code: "sqlite_busy", + }); active = true; return scope.run({ images, opened: new Map() }, async () => { - try { return await operation(scope.getStore()); } - finally { - try { for (const db of scope.getStore().opened.values()) db.close(); } - finally { active = false; } + try { + return await operation(scope.getStore()); + } finally { + try { + for (const db of scope.getStore().opened.values()) db.close(); + } finally { + active = false; + } } }); } export class DatabaseSync { constructor(path) { const state = scope.getStore(); - if (!state || !sqlite) throw new Error('SQLite request context missing'); + if (!state || !sqlite) throw new Error("SQLite request context missing"); if (state.opened.has(path)) return state.opened.get(path); this.path = path; - this.db = new sqlite.oo1.DB(':memory:'); + this.db = new sqlite.oo1.DB(":memory:"); + this.imageBytes = 0; const image = state.images.get(basename(path)); if (image) { const p = sqlite.capi.sqlite3_malloc64(BigInt(image.byteLength)); - if (!p) throw new Error('SQLite memory allocation failed'); + if (!p) throw new Error("SQLite memory allocation failed"); sqlite.wasm.heap8u().set(image, p); - const rc = sqlite.capi.sqlite3_deserialize(this.db.pointer, 'main', p, - BigInt(image.byteLength), BigInt(image.byteLength), 3); - if (rc) { sqlite.capi.sqlite3_free(p); this.db.close(); throw new Error('SQLite snapshot restore failed: ' + rc); } + const rc = sqlite.capi.sqlite3_deserialize( + this.db.pointer, + "main", + p, + BigInt(image.byteLength), + BigInt(image.byteLength), + 3, + ); + if (rc) { + sqlite.capi.sqlite3_free(p); + this.db.close(); + throw new Error("SQLite snapshot restore failed: " + rc); + } + this.imageBytes = image.byteLength; + // The WASM copy owns the bytes now (FREEONCLOSE), so the JS image can be collected + // while the connection is still open instead of after the whole request. + state.images.delete(basename(path)); } // Bound page growth before export, so one user cannot exhaust the isolate. - this.db.exec('PRAGMA max_page_count=4096; PRAGMA cache_size=-2048;'); + this.db.exec("PRAGMA max_page_count=4096; PRAGMA cache_size=-2048;"); state.opened.set(path, this); + noteImage(); + } + exec(sql) { + this.db.exec(sql); } - exec(sql) { this.db.exec(sql); } prepare(sql) { const db = this.db; function statement(args, operation) { const st = db.prepare(sql); - try { if (args.length) st.bind(args.map(x => Buffer.isBuffer(x) ? new Uint8Array(x) : x)); return operation(st); } - finally { st.finalize(); } + try { + if (args.length) st.bind(args.map((x) => (Buffer.isBuffer(x) ? new Uint8Array(x) : x))); + return operation(st); + } finally { + st.finalize(); + } } return { - all(...args) { return statement(args, st => { const rows=[]; while(st.step()) rows.push(st.get({})); return rows; }); }, - get(...args) { return statement(args, st => st.step() ? st.get({}) : undefined); }, - run(...args) { return statement(args, st => { while(st.step()) { /* run() discards row data */ } return { changes: sqlite.capi.sqlite3_changes(db.pointer), lastInsertRowid: sqlite.capi.sqlite3_last_insert_rowid(db.pointer) }; }); } + all(...args) { + return statement(args, (st) => { + const rows = []; + while (st.step()) rows.push(st.get({})); + return rows; + }); + }, + get(...args) { + return statement(args, (st) => (st.step() ? st.get({}) : undefined)); + }, + run(...args) { + return statement(args, (st) => { + while (st.step()) { + /* run() discards row data */ + } + return { + changes: sqlite.capi.sqlite3_changes(db.pointer), + lastInsertRowid: sqlite.capi.sqlite3_last_insert_rowid(db.pointer), + }; + }); + }, }; } - close() { if (this.db.pointer) this.db.close(); } + close() { + if (this.db.pointer) this.db.close(); + } } diff --git a/onlion/backend/worker.mjs b/onlion/backend/worker.mjs index c7272f3a..964d720d 100644 --- a/onlion/backend/worker.mjs +++ b/onlion/backend/worker.mjs @@ -26,7 +26,8 @@ import { } from "./diagnostics.mjs"; import init from "../sqlite-wasm/worker-loader.mjs"; import compiledSqlite from "../sqlite-wasm/sqlite3.wasm"; -import { configure, DatabaseSync, withDatabases } from "./sqlite.mjs"; +import { configure, DatabaseSync, withDatabases, sqliteStats } from "./sqlite.mjs"; +import { memRequestStarted, memLog, memFields } from "./mem-trace.mjs"; import { AmlMemoryService, AML_SEARCH_PROFILE, CORE_SOURCE } from "./memory-service.ts"; import { createEmbeddingClientFromEnv } from "../../src/core/embedding-provider.ts"; import { @@ -174,6 +175,9 @@ async function sqliteReady() { if (!ready) { ready = instance; configure(ready); + // Once per instance: what loading the SQLite module costs, before any request data is + // touched. `seq` here is the request that paid for the load. + memLog("sqlite_ready", null, { sqlite: sqliteStats() }); } return ready; } @@ -1129,6 +1133,10 @@ function watchSlowRequest(env, context, started) { elapsed_ms: Date.now() - started, add_mode: env.NMG_ADD_MODE ?? null, queue: queueStatus(), + // A killed invocation carries none of this, so the report has to be written while + // the request is still alive: memory, how many requests this instance has served, + // and how many SQLite images it has held at once. + ...memFields({ sqlite: sqliteStats() }), }), ); }, afterMs); @@ -1678,6 +1686,9 @@ async function adminStatusResponse(env) { checks_ms: { sqlite: sqlite.ms, r2: r2.ms }, load, queue: queueState, + // On demand, per instance: the same fields the slow-request report carries, so memory + // can be watched while a deployment is live instead of only after an incident. + memory: memFields({ sqlite: sqliteStats() }), failures: { scope: "shared_best_effort", retention_hours: 24, maximum: 20, entries: failures }, limitations: [ "Basic probes do not guarantee model availability.", @@ -1870,7 +1881,9 @@ export default { const context = { trace_id: traceId, route, env, execution }; return withRequest(context, () => { const track = statisticsEnabled(); - const stopSlowWatch = watchSlowRequest(env, context, Date.now()); + const startedAt = Date.now(); + context.seq = memRequestStarted(route); + const stopSlowWatch = watchSlowRequest(env, context, startedAt); if (track) requestAccepted(); const received = track ? performance.now() : 0; let begun; @@ -1925,7 +1938,14 @@ export default { return response; }) : job; - return tracked.finally(stopSlowWatch); + return tracked.finally(() => { + stopSlowWatch(); + memLog("exit", route, { + seq: context.seq, + elapsed_ms: Date.now() - startedAt, + sqlite: sqliteStats(), + }); + }); }); }, }; diff --git a/onlion/package.json b/onlion/package.json index 954bc1b3..322dda94 100644 --- a/onlion/package.json +++ b/onlion/package.json @@ -18,14 +18,15 @@ "test:worker-media": "node scripts/test-multimodal-worker.mjs", "test:cos": "node scripts/test-cos-bucket.mjs", "test:edgeone": "node scripts/test-edgeone-worker.mjs", - "verify": "npm run build && npm run check && npm run test:temporal && npm run test:media && npm run test:backend && npm run test:append-log && npm run test:manifest-pending && npm run test:add-log && npm run test:page-delta && npm run test:extractive && npm run test:worker-media", + "verify": "npm run build && npm run check && npm run test:temporal && npm run test:media && npm run test:backend && npm run test:append-log && npm run test:manifest-pending && npm run test:add-log && npm run test:page-delta && npm run test:extractive && npm run test:mem && npm run test:worker-media", "generate:callbacks": "node scripts/generate-callbacks.mjs", "smoke:edgeone": "node scripts/smoke-edgeone.mjs", "test:add-log": "node scripts/test-add-log-mode.mjs", "test:append-log": "node scripts/test-append-log.mjs", "test:manifest-pending": "node scripts/test-manifest-pending.mjs", "test:page-delta": "node scripts/test-page-delta.mjs", - "test:extractive": "node --experimental-transform-types scripts/test-extractive-summaries.mjs" + "test:extractive": "node --experimental-transform-types scripts/test-extractive-summaries.mjs", + "test:mem": "node scripts/test-mem-trace.mjs" }, "devDependencies": { "@types/node": "24.12.4", diff --git a/onlion/scripts/test-mem-trace.mjs b/onlion/scripts/test-mem-trace.mjs new file mode 100644 index 00000000..2f549703 --- /dev/null +++ b/onlion/scripts/test-mem-trace.mjs @@ -0,0 +1,68 @@ +/* eslint-disable no-console -- deliberate test output */ +/* The memory trace exists to answer one question the incident log cannot: whether the + * reported 762-1014 MB is the container baseline or something the request accumulated. It + * is only useful if it samples fast requests too, so the assertions here check the sampling + * policy and the report shape rather than any particular number. */ +import assert from "node:assert/strict"; +import { memFields, memLog, memRequestStarted, baselineMemory } from "../backend/mem-trace.mjs"; +import { sqliteStats } from "../backend/sqlite.mjs"; + +const lines = []; +const realLog = console.log; +console.log = (line) => lines.push(line); +function parse(line) { + assert.equal(typeof line, "string", "the trace writes one JSON line per sample"); + return JSON.parse(line); +} + +try { + // Off means off: an operator disabling the trace must not pay for it. + process.env.NMG_MEM_TRACE = "0"; + memRequestStarted("add"); + memLog("exit", "add", {}); + assert.equal(lines.length, 0, "NMG_MEM_TRACE=0 must emit nothing"); + + // Every request, with the baseline captured before SQLite was loaded. + process.env.NMG_MEM_TRACE = "1"; + memRequestStarted("search"); + const enter = parse(lines.at(-1)); + assert.equal(enter.event, "nmg_mem_sample"); + assert.equal(enter.phase, "enter"); + assert.equal(enter.route, "search"); + assert.ok(enter.instance.seq >= 1, "the sequence number counts requests in this instance"); + assert.equal( + enter.instance.baseline_rss, + baselineMemory().rss, + "the baseline is the reading taken when the module loaded", + ); + assert.ok(enter.instance.baseline_rss > 0); + for (const key of ["rss", "heapUsed", "heapTotal", "external", "arrayBuffers"]) + assert.equal(typeof enter.mem[key], "number", `mem.${key} must be reported`); + + // A second request in the same instance advances the counter: a leak only means anything + // per instance, so the sequence number is what a trend is read against. + const before = parse(lines.at(-1)).instance.seq; + memRequestStarted("add"); + assert.equal(parse(lines.at(-1)).instance.seq, before + 1, "each request advances by one"); + + // The slow-request line is emitted anyway, so its fields must never be sampled away. + const fields = memFields({ sqlite: sqliteStats() }); + for (const key of ["mem", "instance", "sqlite"]) + assert.ok(key in fields, `memFields must always carry ${key}`); + assert.equal(typeof fields.sqlite.high_water_image_mb, "number"); + + // Default policy samples, and samples the *first* request: a trace that only ever saw hung + // requests could not be compared with anything. + delete process.env.NMG_MEM_TRACE; + lines.length = 0; + // The counter is per instance and already advanced, so the window has to be measured + // rather than assumed: twenty consecutive requests contain exactly one sample. + for (let index = 0; index < 20; index += 1) memRequestStarted("add"); + assert.equal(lines.length, 1, "the default policy logs one request in twenty"); + assert.equal(parse(lines[0]).phase, "enter"); +} finally { + console.log = realLog; +} +console.log( + "PASS: the memory trace samples fast requests, reports the instance baseline, and stays silent when disabled.", +);