context-graph: dedicated finite projection heap budget (HYP_GRAPH_PROJECTION_MAX_HEAP_MB) - step 1 of #376

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<ExecuteSqlResult> }} 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<ExecuteSqlResult>} executeSql
  * @returns {Promise<GraphRow[]>}
  * @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<string>} */
   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]

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