From ad33b24f941386036485aa5bb312e8a87b422f87 Mon Sep 17 00:00:00 2001 From: neutral-loop Date: Wed, 22 Jul 2026 23:06:37 +0000 Subject: [PATCH] context-graph: dedicated finite projection heap budget (step 1 of #376) Graph projection ran its three executeQuerySql scans (the shared per-contract scan, the raw-SQL rule scans, and the dedup id read) without a maxHeapBytes override, so they inherited the 1 GiB user-query heap budget added in #295 (LLP 0097). The shared scan fully materializes the source table before the rule loop sees a row, so on the ~1.66M-row prod table it trips that budget every sweep, logs graph_projection.scope_failed, and freezes the fleet graph. Step 1 (immediate unblock): resolve a DEDICATED projection budget from a new knob HYP_GRAPH_PROJECTION_MAX_HEAP_MB (default 3 GiB) in one helper, resolveProjectionMaxHeapBytes, and pass it as maxHeapBytes at all three scan sites. Never 0: a blank, zero, non-positive, or non-numeric knob falls back to the default rather than disabling executeQuerySql's watchdog (0 risks a daemon OOM crash-loop, strictly worse than a stale graph). This leaves the 1 GiB user-query guard untouched and preserves the fail-clean property: when the table outgrows the dedicated budget, projection refuses cleanly instead of crashing the daemon. The durable incremental + chunked projection (step 2) is deferred to #377. Adds a regression test asserting the finite knob-derived budget reaches every scan site (default 3 GiB, override propagates, unset != 0), captured through an injectable query-exec seam. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../context-graph/src/project.js | 61 +++++-- .../context-graph-projection-budget.test.js | 152 ++++++++++++++++++ 2 files changed, 203 insertions(+), 10 deletions(-) create mode 100644 test/plugins/context-graph-projection-budget.test.js diff --git a/hypaware-core/plugins-workspace/context-graph/src/project.js b/hypaware-core/plugins-workspace/context-graph/src/project.js index 98436f27..074cdfe6 100644 --- a/hypaware-core/plugins-workspace/context-graph/src/project.js +++ b/hypaware-core/plugins-workspace/context-graph/src/project.js @@ -14,9 +14,39 @@ import { /** * @import { HypAwareV2Config, QueryRegistry } from '../../../../hypaware-plugin-kernel-types.js' * @import { ExtendedQueryStorageService } from '../../../../src/core/cache/types.js' + * @import { ExecuteSqlOptions, ExecuteSqlResult } from '../../../../src/core/query/types.js' * @import { Contract, ContractRule, GraphRow, RulePredicate } from './types.js' */ +/** The dedicated projection heap budget when the knob is unset: 3 GiB. */ +const GRAPH_PROJECTION_DEFAULT_MAX_HEAP_BYTES = 3072 * 1024 * 1024 + +/** + * The graph projection's DEDICATED execution heap budget in bytes, from + * HYP_GRAPH_PROJECTION_MAX_HEAP_MB (default 3 GiB). The T0 shared scan + * materializes the whole source table into memory before the rule loop sees a + * row, so it needs headroom the 1 GiB user-query default (LLP 0056, added in + * #295) rightly denies interactive traffic. This is a dedicated budget, not an + * exemption: passing a finite bound keeps the fail-clean property (over-budget + * projection refuses with a typed error the scheduler logs as + * `graph_projection.scope_failed`) while leaving the user-query guard untouched. + * Never returns 0: 0 disables executeQuerySql's watchdog entirely, risking a + * daemon OOM crash-loop that is strictly worse than a stale graph, so a blank, + * zero, non-positive, or non-numeric knob falls back to the default rather than + * removing the guard. Mirrors resolveHeapBudgetBytes' blank-var handling. + * + * @returns {number} + * @ref LLP 0097 [constrained-by]: reuses the maxHeapBytes option and inherits its fail-clean refusal; deliberately does NOT use the "0 disables" escape the doc exposes + */ +export function resolveProjectionMaxHeapBytes() { + const raw = process.env.HYP_GRAPH_PROJECTION_MAX_HEAP_MB?.trim() + if (raw) { + const mb = Number(raw) + if (Number.isFinite(mb) && mb > 0) return mb * 1024 * 1024 + } + return GRAPH_PROJECTION_DEFAULT_MAX_HEAP_BYTES +} + /** * Run the T0 deterministic projection: read each registered source contract's * rules, materialize node/edge rows (deterministic ids + inline provenance), @@ -26,11 +56,16 @@ import { * structurally converge for free. Idempotent: a second run with no new source * data writes zero rows. * - * @param {{ query: QueryRegistry, storage: ExtendedQueryStorageService, contracts: Contract[], config?: HypAwareV2Config, dryRun?: boolean }} args + * @param {{ query: QueryRegistry, storage: ExtendedQueryStorageService, contracts: Contract[], config?: HypAwareV2Config, dryRun?: boolean, __executeSql?: (args: ExecuteSqlOptions) => Promise }} args * @returns {Promise<{ nodes: number, edges: number, nodesWritten: number, edgesWritten: number }>} * @ref LLP 0023#contract-contribution [implements]: the engine runs every registered contract; adding a source is contributing one */ -export async function projectGraph({ query, storage, contracts, config, dryRun = false }) { +export async function projectGraph({ query, storage, contracts, config, dryRun = false, __executeSql = executeQuerySql }) { + // A dedicated finite budget for every projection scan: the shared scan below + // fully materializes the source table, so it needs more than the 1 GiB + // user-query default without stripping the guard (never 0). Resolved once and + // applied at all three scan sites (shared scan, raw-SQL rules, dedup read). + const maxHeapBytes = resolveProjectionMaxHeapBytes() return withSpan( 'graph.project', { @@ -96,13 +131,14 @@ export async function projectGraph({ query, storage, contracts, config, dryRun = // enforced when the graph datasets are READ, via their // localOnlyContentColumns declaration (datasets.js). // @ref LLP 0105#surfaces [constrained-by]: the filter governs read surfaces; derived-cache builds keep full fidelity and the derived dataset carries its own declaration - const result = await executeQuerySql({ + const result = await __executeSql({ query: `SELECT ${[...columns].sort().join(', ')} FROM ${contract.sourceDataset}`, registry: query, storage, config, refresh: 'always', includeLocalOnly: true, + maxHeapBytes, }) sourceRows += result.rows.length scanCount += 1 @@ -125,14 +161,15 @@ export async function projectGraph({ query, storage, contracts, config, dryRun = else bySql.set(sql, [rule]) } for (const [sql, rules] of bySql) { - // Same cache-to-cache bypass as the shared scan above. - const result = await executeQuerySql({ + // Same cache-to-cache bypass and dedicated budget as the shared scan. + const result = await __executeSql({ query: sql, registry: query, storage, config, refresh: 'always', includeLocalOnly: true, + maxHeapBytes, }) sourceRows += result.rows.length scanCount += 1 @@ -155,8 +192,8 @@ export async function projectGraph({ query, storage, contracts, config, dryRun = return { nodes: nodeRows.length, edges: edgeRows.length, nodesWritten: 0, edgesWritten: 0 } } - const freshNodes = await dedupExisting(nodeRows, 'node_id', NODE_DATASET, query, storage, config) - const freshEdges = await dedupExisting(edgeRows, 'edge_id', EDGE_DATASET, query, storage, config) + const freshNodes = await dedupExisting(nodeRows, 'node_id', NODE_DATASET, query, storage, config, maxHeapBytes, __executeSql) + const freshEdges = await dedupExisting(edgeRows, 'edge_id', EDGE_DATASET, query, storage, config, maxHeapBytes, __executeSql) if (freshNodes.length > 0) { await storage.appendRows(graphTablePath(storage, NODE_DATASET), [...NODE_COLUMNS], freshNodes) @@ -190,24 +227,28 @@ export async function projectGraph({ query, storage, contracts, config, dryRun = * @param {QueryRegistry} query * @param {ExtendedQueryStorageService} storage * @param {HypAwareV2Config | undefined} config + * @param {number} maxHeapBytes + * @param {(args: ExecuteSqlOptions) => Promise} executeSql * @returns {Promise} * @ref LLP 0023#pre-write-dedup [implements]: only a missing dataset is benign; real failures abort instead of duplicating */ -async function dedupExisting(rows, idCol, dataset, query, storage, config) { +async function dedupExisting(rows, idCol, dataset, query, storage, config, maxHeapBytes, executeSql) { if (rows.length === 0) return rows /** @type {Set} */ const seen = new Set() try { // Dedup must see EVERY committed id (ids are content-addressed hashes, // not content): a visibility-filtered id set would re-append rows the - // filter hid and corrupt idempotency. Cache-internal, so bypass. - const res = await executeQuerySql({ + // filter hid and corrupt idempotency. Cache-internal, so bypass. Same + // dedicated projection budget as the source scans (never the 1 GiB default). + const res = await executeSql({ query: `SELECT ${idCol} FROM ${dataset}`, registry: query, storage, config, refresh: 'always', includeLocalOnly: true, + maxHeapBytes, }) for (const r of res.rows) { const v = r[idCol] diff --git a/test/plugins/context-graph-projection-budget.test.js b/test/plugins/context-graph-projection-budget.test.js new file mode 100644 index 00000000..5ace63bb --- /dev/null +++ b/test/plugins/context-graph-projection-budget.test.js @@ -0,0 +1,152 @@ +// @ts-check + +import assert from 'node:assert/strict' +import test from 'node:test' + +import { + projectGraph, + resolveProjectionMaxHeapBytes, +} from '../../hypaware-core/plugins-workspace/context-graph/src/project.js' + +// Regression for issue #376 step 1: graph projection must pass its OWN finite +// heap budget (HYP_GRAPH_PROJECTION_MAX_HEAP_MB, default 3 GiB) to every +// executeQuerySql scan, instead of inheriting the 1 GiB user-query default +// (#295). The shared scan fully materializes the source table, so on prod it +// trips that default and freezes the graph. A dedicated FINITE budget (never 0) +// unblocks it while preserving the fail-clean property and leaving the +// user-query guard untouched. + +const KNOB = 'HYP_GRAPH_PROJECTION_MAX_HEAP_MB' +const DEFAULT_BYTES = 3072 * 1024 * 1024 + +/** @returns {import('../../hypaware-core/plugins-workspace/context-graph/src/types.js').Contract} */ +function probeContract() { + return { + name: 'budget-probe', + plugin: '@test/budget', + sourceDataset: 'src_table', + projector: 'test.t0', + projectorVersion: 1, + rules: [ + // Declarative rule -> exercises the shared per-contract scan site. + { + kind: 'node', + type: 'Thing', + columns: ['id'], + toRow: (r) => ({ node_id: `n-${r.id}`, first_seen: '2026-01-01T00:00:00.000Z' }), + }, + // Raw-SQL rule -> exercises the raw-SQL rule scan site. + { + kind: 'edge', + type: 'rel', + sql: 'SELECT a, b FROM src_table', + toRow: (r) => ({ edge_id: `e-${r.a}-${r.b}`, first_seen: '2026-01-01T00:00:00.000Z' }), + }, + ], + } +} + +/** + * Run projectGraph with a recording SQL seam and a no-op storage. Returns the + * captured { query, maxHeapBytes } for every executeQuerySql call: the shared + * scan, the raw-SQL rule scan, and both dedup id reads. + * + * @returns {Promise<{ query: string, maxHeapBytes: number | undefined }[]>} + */ +async function captureBudgets() { + /** @type {{ query: string, maxHeapBytes: number | undefined }[]} */ + const calls = [] + /** @type {any} */ + const fakeExec = async (args) => { + calls.push({ query: args.query, maxHeapBytes: args.maxHeapBytes }) + if (args.query === 'SELECT id FROM src_table') return { rows: [{ id: '1' }] } + if (args.query === 'SELECT a, b FROM src_table') return { rows: [{ a: 'x', b: 'y' }] } + // Dedup reads: empty committed set, so freshNodes/freshEdges are non-empty + // and the append (a no-op storage) fires. + return { rows: [] } + } + /** @type {any} */ + const storage = { + cacheTablePath: (dataset) => `/fake/${dataset}`, + appendRows: async () => {}, + } + await projectGraph({ + query: /** @type {any} */ ({}), + storage, + contracts: [probeContract()], + __executeSql: fakeExec, + }) + return calls +} + +test('projectGraph passes the default 3 GiB projection budget at all three scan sites when the knob is unset', async () => { + const prev = process.env[KNOB] + delete process.env[KNOB] + try { + const calls = await captureBudgets() + + const shared = calls.find((c) => c.query === 'SELECT id FROM src_table') + const rawSql = calls.find((c) => c.query === 'SELECT a, b FROM src_table') + const dedupNode = calls.find((c) => c.query === 'SELECT node_id FROM node') + const dedupEdge = calls.find((c) => c.query === 'SELECT edge_id FROM edge') + + assert.ok(shared, 'the shared per-contract scan ran') + assert.ok(rawSql, 'the raw-SQL rule scan ran') + assert.ok(dedupNode, 'the node dedup id read ran') + assert.ok(dedupEdge, 'the edge dedup id read ran') + + for (const c of [shared, rawSql, dedupNode, dedupEdge]) { + assert.equal( + c.maxHeapBytes, + DEFAULT_BYTES, + `${c.query}: must carry the dedicated 3 GiB projection budget, not inherit the 1 GiB user-query default` + ) + // The whole point of a DEDICATED budget over an exemption: never 0. + assert.notEqual(c.maxHeapBytes, 0, `${c.query}: budget must never be 0 (0 disables the OOM guard)`) + assert.ok(Number.isFinite(c.maxHeapBytes), `${c.query}: budget must be finite`) + } + } finally { + if (prev === undefined) delete process.env[KNOB] + else process.env[KNOB] = prev + } +}) + +test('HYP_GRAPH_PROJECTION_MAX_HEAP_MB overrides the projection budget at every scan site', async () => { + const prev = process.env[KNOB] + process.env[KNOB] = '512' + try { + const calls = await captureBudgets() + assert.ok(calls.length >= 4, 'all scan sites ran') + for (const c of calls) { + assert.equal(c.maxHeapBytes, 512 * 1024 * 1024, `${c.query}: knob override reaches the scan`) + } + } finally { + if (prev === undefined) delete process.env[KNOB] + else process.env[KNOB] = prev + } +}) + +test('resolveProjectionMaxHeapBytes defaults to 3 GiB and never returns 0', () => { + const prev = process.env[KNOB] + try { + // Unset -> default. + delete process.env[KNOB] + assert.equal(resolveProjectionMaxHeapBytes(), DEFAULT_BYTES) + + // A positive knob converts MB -> bytes. + process.env[KNOB] = '2048' + assert.equal(resolveProjectionMaxHeapBytes(), 2048 * 1024 * 1024) + + // The dangerous inputs must NEVER remove the guard: blank, zero, negative, + // and non-numeric all fall back to the finite default, never 0. + for (const bad of ['', ' ', '0', '-1', 'abc']) { + process.env[KNOB] = bad + const resolved = resolveProjectionMaxHeapBytes() + assert.equal(resolved, DEFAULT_BYTES, `knob=${JSON.stringify(bad)} falls back to the default`) + assert.notEqual(resolved, 0, `knob=${JSON.stringify(bad)} must never disable the guard`) + } + } finally { + if (prev === undefined) delete process.env[KNOB] + else process.env[KNOB] = prev + } +})