From 14cfcb5e42c7e84642a1965b63941628533f2d1a Mon Sep 17 00:00:00 2001 From: wefio <48851810+wefio@users.noreply.github.com> Date: Thu, 29 Oct 2026 15:06:09 +0800 Subject: [PATCH] feat(onlion): trace where the invocation memory goes The 2026-10-29 incident has 220 invocations that died at the 120 s ceiling carrying 762-1014 MB, and the log cannot say what 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 a narrow band across killed requests is exactly what a retry loop plateaued by GC would look like too. Three fields are what separate those cases: - `baseline` is read when the module is 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 is read against request number rather than wall time: a leak only means anything inside one warm instance. - the breakdown separates `heapUsed` (JS objects: a manifest, a parsed body) from `external`/`arrayBuffers` (byte buffers, where a WASM linear memory that only grows shows up), and `sqlite` reports the live image count, the instance high-water mark and the WASM linear size. Fast requests are sampled on purpose - one in twenty by default, all of them with NMG_MEM_TRACE=1, none with =0 - because a trace that only ever saw hung requests cannot be compared with anything. The slow-request report and /admin/status carry the fields unconditionally, so memory can be watched while a deployment is live. The same change stops retaining a partition image longer than it must: once the bytes are copied into the WASM heap, the SQLite module owns them (FREEONCLOSE), so the JS copy is dropped while the connection is still open instead of at the end of the request. Not an optimisation yet: the in-flight peak the code can account for is about 32 MiB per partition operation, an order of magnitude below the reported figure, so there is nothing to optimise until the baseline is known. Verified: onlion npm run verify 35 PASS (new test-mem-trace), eslint and prettier clean, complexity gate clean for every file touched here. --- onlion/backend/mem-trace.mjs | 98 ++++++++++++++++++++++ onlion/backend/sqlite.mjs | 130 +++++++++++++++++++++++++----- onlion/backend/worker.mjs | 26 +++++- onlion/package.json | 5 +- onlion/scripts/test-mem-trace.mjs | 68 ++++++++++++++++ 5 files changed, 300 insertions(+), 27 deletions(-) create mode 100644 onlion/backend/mem-trace.mjs create mode 100644 onlion/scripts/test-mem-trace.mjs 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.", +);