diff --git a/.github/workflows/ci-tests.yml b/.github/workflows/ci-tests.yml index f8ec70bd5..d4a98bbce 100644 --- a/.github/workflows/ci-tests.yml +++ b/.github/workflows/ci-tests.yml @@ -266,22 +266,44 @@ jobs: - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 with: persist-credentials: false + - name: Identify the first executing attempt + id: execution + shell: bash + env: + GH_TOKEN: ${{ github.token }} + CI_REPOSITORY: ${{ github.repository }} + CI_RUN_ID: ${{ github.run_id }} + CI_RUN_ATTEMPT: ${{ github.run_attempt }} + run: | + initialize=false + if [[ "$CI_RUN_ATTEMPT" == 1 ]]; then + initialize=true + elif [[ "$CI_RUN_ATTEMPT" == 2 ]]; then + # Fork approval increments run_attempt even though no job ran. + attempt="repos/$CI_REPOSITORY/actions/runs/$CI_RUN_ID/attempts/1" + conclusion=$(gh api "$attempt" --jq .conclusion) + if [[ "$conclusion" == action_required ]]; then + jobs=$(gh api "$attempt/jobs" --jq .total_count) + if [[ "$jobs" == 0 ]]; then initialize=true; fi + fi + fi + printf 'initialize=%s\n' "$initialize" >> "$GITHUB_OUTPUT" - name: Reuse this run's timing input on a retry id: previous-input - if: github.run_attempt > 1 + if: steps.execution.outputs.initialize != 'true' uses: actions/download-artifact@9000827ccba6bdab643e8b6fd33ac0654aef8333 # v8 with: name: shard-timing-input path: .vitest/timings - name: Restore last trusted timing history - if: github.run_attempt == 1 + if: steps.execution.outputs.initialize == 'true' uses: actions/cache/restore@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v5 with: path: .vitest/timings key: test-timings-v1-node22-vitest5-${{ github.run_id }}-${{ github.run_attempt }} restore-keys: test-timings-v1-node22-vitest5- - name: Prepare timing snapshot - if: github.run_attempt == 1 + if: steps.execution.outputs.initialize == 'true' shell: bash run: | mkdir -p .vitest/timings @@ -291,7 +313,7 @@ jobs: done sha256sum .vitest/timings/*.json >> "$GITHUB_STEP_SUMMARY" - name: Freeze timing input for every shard and retry - if: github.run_attempt == 1 + if: steps.execution.outputs.initialize == 'true' uses: ./.github/actions/upload-test-artifact with: name: shard-timing-input diff --git a/RUNBOOK.md b/RUNBOOK.md index ac937284a..7971d1bcc 100644 --- a/RUNBOOK.md +++ b/RUNBOOK.md @@ -195,6 +195,41 @@ Orchestrator: `.github/workflows/ci.yml`. --- +## Query memory limit + +**Symptom:** a CLI, MCP, or HTTP server process exits with code **70** and +`GitNexus stopped: process memory reached …` on stderr. + +The pooled-read guard detected memory above its process-wide budget. Native +query allocations are outside both V8's heap limit and LadybugDB's buffer pool. +The Node binding cannot interrupt them, so the guard exits the process to +release memory. Every request in that process is interrupted; restart the +CLI/server or reconnect the MCP client. A read-only query does not require a +reindex after this exit. + +To diagnose a repeat, set `GITNEXUS_QUERY_TRACE=1` in the query/server environment +and retain stderr. It includes query text but not bound parameters. Review the +output before sharing it with [issue #3696](https://github.com/abhigyanpatwari/GitNexus/issues/3696). +Set `GITNEXUS_QUERY_MAX_RSS_MB=` to change the limit; `0` disables it. The +default is max(8 GiB, 25% of available RAM), capped at 75% of available RAM, +including a reported container/runtime constraint. + +RSS is checked at query start/end and every 200 ms while reads run. On macOS, +queries lasting at least one second also sample physical footprint at most +once per second, including compressed/swapped memory. If the OS footprint +command is unavailable, monitoring falls back to RSS. Sampling is a mitigation, +not a hard OS memory limit: rapid allocation or a blocked event loop can outrun +it. Writes during `analyze` are outside this guard. + +The caller's 30-second read deadline covers prepare, execution, and row +materialization. Timing out does not cancel native work: monitoring and its +checked-out connection remain owned by that work until it settles. An impact +frontier is separately capped at 100,000 relationship rows per depth. A capped +walk stops and reports `partial: true`. Both `risk` and `riskSharedAxes` retain +HIGH/CRITICAL or report UNKNOWN instead of LOW/MEDIUM. This row cap does not bound the database's +intermediate allocations. Narrow the target or relation filters before relying +on a partial result. + ## Memory / analyze crashes Analyze re-execs Node with a **large old-space heap** when needed (`analyze.ts`). If you still OOM on huge repos, close other processes, avoid `--embeddings` for a first pass, or analyze a smaller path if supported by your workflow. diff --git a/TESTING.md b/TESTING.md index 9b2a130e4..058ddda9d 100644 --- a/TESTING.md +++ b/TESTING.md @@ -114,6 +114,10 @@ profile because native startup and coverage costs differ. The `shard-plan` job restores history once and uploads `shard-timing-input`. Every shard and report merge downloads that same snapshot. Retries reuse it, so a newly published history cannot change the partition halfway through a run. +Fork approval can increment the attempt number before any job executes. Attempt +2 initializes a snapshot only when GitHub confirms attempt 1 required approval +and ran zero jobs. Later retries always require the frozen snapshot. + If the snapshot cannot be restored, the retry fails. If it was never uploaded, deleted, or expired, trigger a fresh workflow run with a new commit; rerunning the old run must not replace its timing input. diff --git a/docs/cli-reference.md b/docs/cli-reference.md index a387b7241..64e986754 100644 --- a/docs/cli-reference.md +++ b/docs/cli-reference.md @@ -218,8 +218,17 @@ Notes: ## Environment variables +Read queries have a process-wide native-memory guard. See the +[query memory runbook](../RUNBOOK.md#query-memory-limit) for exit code 70, +sampling limits, and recovery. + Most `analyze` knobs are also CLI flags (`--workers`, `--worker-timeout`, `--max-file-size`, `--verbose`). Use the env-var form when you'd otherwise repeat the same flag every run, or when invoking GitNexus from a long-running host (MCP server, eval-server, CI shell) that already manages its own environment. CLI flags take precedence over `.gitnexusrc`, which takes precedence over env vars, which take precedence over built-in defaults. +| Query variable | Default | Effect | When to tune | +| --- | --- | --- | --- | +| `GITNEXUS_QUERY_MAX_RSS_MB` | max(8192 MiB, 25% of available RAM), capped at 75% of available RAM | Process-memory ceiling while pooled reads run; available RAM is the smaller of physical RAM and a reported runtime/container constraint. Positive MiB values override it; `0` disables the guard. Invalid, overflowing, or sub-byte values use the default. A breach exits the process with code 70. | Set in the CLI/MCP/serve environment when a legitimate read workload needs a different budget. It does not resize the database buffer pool. | +| `GITNEXUS_QUERY_TRACE` | unset | `1` writes query duration, row count, and sampled memory to stderr. Bound parameters are omitted; query text is included. | Diagnose an expensive read or a memory-guard exit. Inspect trace output before sharing it. | + | Variable | Default | Effect | Tune when… | | ----------------------------------------------- | --------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `GITNEXUS_TEST_REPORT` | unset | Write a JSON execution receipt from `scripts/run-cross-platform.ts`; accepts a basename ending in `.json`. | CI platform shards use distinct filenames so the completeness check can require every report. | diff --git a/eval/tests/test_workflow_bench.py b/eval/tests/test_workflow_bench.py index 98248b50d..8e45f079c 100644 --- a/eval/tests/test_workflow_bench.py +++ b/eval/tests/test_workflow_bench.py @@ -411,7 +411,7 @@ def test_test_artifact_retries_cannot_hide_missing_evidence(ci_test_jobs): def test_shard_plan_retry_cannot_replace_frozen_timing_input(ci_test_jobs): steps = ci_test_jobs["shard-plan"]["steps"] restore = next(step for step in steps if step.get("id") == "previous-input") - assert restore["if"] == "github.run_attempt > 1" + assert restore["if"] == "steps.execution.outputs.initialize != 'true'" assert not restore.get("continue-on-error", False) assert restore["with"]["name"] == "shard-timing-input" @@ -423,7 +423,59 @@ def test_shard_plan_retry_cannot_replace_frozen_timing_input(ci_test_jobs): "Freeze timing input for every shard and retry", ): step = next(step for step in steps if step.get("name") == name) - assert step["if"] == "github.run_attempt == 1", name + assert step["if"] == "steps.execution.outputs.initialize == 'true'", name + + +@pytest.mark.parametrize( + "attempt,conclusion,jobs,api_status,expected", + [ + (1, "", 0, 1, "true"), + (2, "action_required", 0, 0, "true"), + (2, "action_required", 1, 0, "false"), + (2, "success", 0, 0, "false"), + (2, "failure", 0, 0, "false"), + (2, "cancelled", 0, 0, "false"), + (3, "action_required", 0, 1, "false"), + (2, "", 0, 1, None), + ], +) +def test_shard_plan_initializes_only_before_any_execution( + ci_test_jobs, tmp_path, attempt, conclusion, jobs, api_status, expected +): + step = next(step for step in ci_test_jobs["shard-plan"]["steps"] if step.get("id") == "execution") + output = tmp_path / "output" + stub = """ + gh() { + if [[ "$API_STATUS" != 0 ]]; then return "$API_STATUS"; fi + case "$2" in + */jobs) printf '%s\\n' "$INITIAL_JOBS" ;; + *) printf '%s\\n' "$INITIAL_CONCLUSION" ;; + esac + } + """ + result = subprocess.run( + ["bash", "--noprofile", "--norc", "-e", "-o", "pipefail", "-c", stub + step["run"]], + env={ + **os.environ, + "GITHUB_OUTPUT": str(output), + "CI_REPOSITORY": "owner/repo", + "CI_RUN_ID": "123", + "CI_RUN_ATTEMPT": str(attempt), + "INITIAL_CONCLUSION": conclusion, + "INITIAL_JOBS": str(jobs), + "API_STATUS": str(api_status), + }, + capture_output=True, + text=True, + timeout=10, + check=False, + ) + if expected is None: + assert result.returncode != 0 + assert not output.exists() + else: + assert result.returncode == 0, result.stdout + result.stderr + assert output.read_text() == f"initialize={expected}\n" @pytest.mark.parametrize("native_result", ["success", "failure", "cancelled", "skipped", ""]) diff --git a/gitnexus/src/core/lbug/pool-adapter.ts b/gitnexus/src/core/lbug/pool-adapter.ts index a0abd9c2e..458c29918 100644 --- a/gitnexus/src/core/lbug/pool-adapter.ts +++ b/gitnexus/src/core/lbug/pool-adapter.ts @@ -20,6 +20,7 @@ import lbug from '@ladybugdb/core'; import { isReadOnlyDbError, loadFTSExtension, loadVectorExtension } from './lbug-adapter.js'; import { closeQueryResults } from './query-result-utils.js'; import { warnIfQueryTextUnbounded } from './query-batch.js'; +import { queryMemoryGuard } from './query-memory-guard.js'; import { createLbugDatabase, isWalCorruptionError, @@ -1321,37 +1322,54 @@ export const executeParameterized = async ( entry.lastUsed = Date.now(); silenceStdout(); activeQueryCount++; + const guardToken = queryMemoryGuard.begin(cypher); + let rowCount: number | undefined; let queryResult: lbug.QueryResult | lbug.QueryResult[] | undefined; - try { - const stmt = await withTimeout(conn.prepare(cypher), QUERY_TIMEOUT_MS, 'Prepare'); - if (!stmt.isSuccess()) { - const errMsg = await stmt.getErrorMessage(); - throw new Error(`Prepare failed: ${errMsg}`); + const deadline = performance.now() + QUERY_TIMEOUT_MS; + const checkDeadline = () => { + if (performance.now() >= deadline) { + throw new Error(`Query timed out after ${QUERY_TIMEOUT_MS}ms`); } - queryResult = await withTimeout(conn.execute(stmt, params), QUERY_TIMEOUT_MS, 'Execute'); - const result = Array.isArray(queryResult) ? queryResult[0] : queryResult; - const rows = await result.getAll(); - return rows; - } catch (err) { - if (isReadOnlyDbError(err)) { - // Preserve the native error as `cause` so the original frame/message is - // not lost behind the friendly read-only message (#2068 follow-up). - throw new Error('Write operations are not allowed. The pool adapter is read-only.', { - cause: err, - }); + }; + const nativeQuery = (async () => { + try { + const stmt = await conn.prepare(cypher); + checkDeadline(); + if (!stmt.isSuccess()) { + const errMsg = await stmt.getErrorMessage(); + throw new Error(`Prepare failed: ${errMsg}`); + } + queryResult = await conn.execute(stmt, params); + checkDeadline(); + const result = Array.isArray(queryResult) ? queryResult[0] : queryResult; + const rows = await result.getAll(); + rowCount = rows.length; + return rows; + } catch (err) { + if (isReadOnlyDbError(err)) { + // Preserve the native error as `cause` so the original frame/message is + // not lost behind the friendly read-only message (#2068 follow-up). + throw new Error('Write operations are not allowed. The pool adapter is read-only.', { + cause: err, + }); + } + throw err; + } finally { + // Close the native QueryResult cursor(s) before returning the connection — + // getAll() drains rows but does not release the native cursor, so without + // this the cursor leaks for the connection's lifetime (#2068 follow-up). + // Best-effort via the shared helper; never masks the query result or a real + // error. + if (queryResult) await closeQueryResults(queryResult); + queryMemoryGuard.end(guardToken, rowCount); + activeQueryCount--; + restoreStdout(); + checkin(entry, conn); } - throw err; - } finally { - // Close the native QueryResult cursor(s) before returning the connection — - // getAll() drains rows but does not release the native cursor, so without - // this the cursor leaks for the connection's lifetime (#2068 follow-up). - // Best-effort via the shared helper; never masks the query result or a real - // error. - if (queryResult) await closeQueryResults(queryResult); - activeQueryCount--; - restoreStdout(); - checkin(entry, conn); - } + })(); + // A JS timeout cannot cancel native work. Keep its guard, cursor and checked-out + // connection until that work settles, including results delivered after timeout. + return withTimeout(nativeQuery, QUERY_TIMEOUT_MS, 'Query'); }; /** diff --git a/gitnexus/src/core/lbug/query-memory-guard.ts b/gitnexus/src/core/lbug/query-memory-guard.ts new file mode 100644 index 000000000..8f34a758a --- /dev/null +++ b/gitnexus/src/core/lbug/query-memory-guard.ts @@ -0,0 +1,254 @@ +/** + * Native-memory guard for read queries (#3696). + * + * A runaway LadybugDB query allocates native heap that neither + * `bufferManagerSize` nor V8 limits bound, and the binding cannot interrupt + * it. While a query runs, process memory is sampled; past the budget, the + * running queries are named on stderr and the process exits so the host does + * not run out of memory. On macOS, RSS excludes compressed and swapped pages, + * so queries running longer than 1 s also read the physical footprint. + * + * `GITNEXUS_QUERY_MAX_RSS_MB`: limit in MiB, `0` disables. Default is + * max(8 GiB, 25% of available RAM), capped at 75% of available RAM. + * Container memory constraints count toward available RAM. + * `GITNEXUS_QUERY_TRACE=1`: one stderr line per query with duration, rows and + * memory. + */ +import { execFile } from 'node:child_process'; +import fs from 'node:fs'; +import os from 'node:os'; +import { realStderrWrite } from '../../mcp/stdio-capture.js'; + +export const QUERY_MAX_RSS_ENV = 'GITNEXUS_QUERY_MAX_RSS_MB'; +export const QUERY_TRACE_ENV = 'GITNEXUS_QUERY_TRACE'; +/** EX_SOFTWARE from sysexits.h: distinct from the CLI's generic exit 1. */ +export const QUERY_MEMORY_LIMIT_EXIT_CODE = 70; + +const MIB = 1024 * 1024; +const MIN_DEFAULT_BUDGET_BYTES = 8 * 1024 * MIB; +const DEFAULT_RAM_FRACTION = 0.25; +const SAMPLE_INTERVAL_MS = 200; +/** Footprint reads start once a query has run this long, and repeat at most this often. */ +export const FOOTPRINT_INTERVAL_MS = 1000; +const FOOTPRINT_TIMEOUT_MS = 2000; +const LABEL_MAX_CHARS = 200; + +/** Memory budget in bytes, or `null` when disabled. Invalid values use the default. */ +export function resolveQueryRssBudgetBytes( + raw: string | undefined, + totalMemBytes: number, +): number | null { + const trimmed = raw?.trim(); + if (trimmed) { + const mb = Number(trimmed); + if (mb === 0) return null; + const bytes = Math.floor(mb * MIB); + if (Number.isSafeInteger(bytes) && bytes > 0) return bytes; + } + return Math.min( + Math.floor(totalMemBytes * 0.75), + Math.max(MIN_DEFAULT_BUDGET_BYTES, Math.floor(totalMemBytes * DEFAULT_RAM_FRACTION)), + ); +} + +/** One-line, length-bounded query label. Never includes bound parameters. */ +export function queryLabel(cypher: string): string { + const flat = cypher.replace(/\s+/g, ' ').trim(); + return flat.length > LABEL_MAX_CHARS ? `${flat.slice(0, LABEL_MAX_CHARS)}…` : flat; +} + +/** Parse `phys_footprint: B` from `footprint -f bytes` output. */ +export function parsePhysFootprintBytes(output: string): number | null { + const m = /phys_footprint:\s*(\d+)\s*B\b/.exec(output); + if (!m) return null; + const n = Number(m[1]); + return Number.isSafeInteger(n) ? n : null; +} + +interface ActiveQuery { + label: string; + startedAt: number; + startBytes: number; + peakBytes: number; +} + +export interface QueryMemoryBreach { + rssBytes: number; + budgetBytes: number; + activeLabels: string[]; +} + +export interface QueryMemoryGuardDeps { + readRss: () => number; + /** Physical footprint, where the platform can report it; `null` when unavailable. */ + readFootprint?: () => Promise; + env: () => NodeJS.ProcessEnv; + totalMem: () => number; + now: () => number; + writeStderr: (line: string) => void; + onBreach: (breach: QueryMemoryBreach) => void; + /** Start the periodic sampler; returns a stop function. */ + startSampler: (tick: () => void) => () => void; +} + +export interface QueryMemoryGuard { + /** Register a query as in flight. Pass the returned token to `end`. */ + begin(cypher: string): object; + /** Unregister a query; `rowCount` is only used for the trace line. */ + end(token: object, rowCount?: number): void; + /** Take one sample now (called by the sampler). */ + sample(): void; +} + +export function formatBreachMessage(breach: QueryMemoryBreach): string { + const gb = (n: number) => (n / (1024 * MIB)).toFixed(1); + const queries = breach.activeLabels.map((l) => ` - ${l}`).join('\n'); + return ( + `GitNexus stopped: process memory reached ${gb(breach.rssBytes)} GB, over the ` + + `${gb(breach.budgetBytes)} GB limit, while running a graph query. The database ` + + `cannot cancel a running query, so the process is exiting to protect the host.\n` + + `Running queries:\n${queries}\n` + + `Set ${QUERY_MAX_RSS_ENV}= to change the limit (0 disables it). ` + + `Please report this with ${QUERY_TRACE_ENV}=1 output (see issue #3696).\n` + ); +} + +export function createQueryMemoryGuard(deps: QueryMemoryGuardDeps): QueryMemoryGuard { + const active = new Set(); + let stopSampler: (() => void) | null = null; + let budgetBytes: number | null = null; + let tracing = false; + let breached = false; + let footprintInFlight = false; + let lastFootprintAt = Number.NEGATIVE_INFINITY; + + const observe = (bytes: number): void => { + if (active.size === 0 || breached) return; + for (const q of active) if (bytes > q.peakBytes) q.peakBytes = bytes; + if (budgetBytes !== null && bytes > budgetBytes) { + breached = true; + deps.onBreach({ + rssBytes: bytes, + budgetBytes, + activeLabels: [...active].map((q) => q.label), + }); + } + }; + + const maybeReadFootprint = (): void => { + if (!deps.readFootprint || footprintInFlight) return; + const now = deps.now(); + if (now - lastFootprintAt < FOOTPRINT_INTERVAL_MS) return; + let oldest = Number.POSITIVE_INFINITY; + for (const q of active) oldest = Math.min(oldest, q.startedAt); + if (now - oldest < FOOTPRINT_INTERVAL_MS) return; + footprintInFlight = true; + lastFootprintAt = now; + const sampledQueries = [...active]; + deps.readFootprint().then( + (bytes) => { + footprintInFlight = false; + if (bytes !== null && sampledQueries.some((q) => active.has(q))) observe(bytes); + }, + () => { + footprintInFlight = false; + }, + ); + }; + + const sample = (): void => { + if (active.size === 0 || breached) return; + observe(deps.readRss()); + if (!breached) maybeReadFootprint(); + }; + + return { + begin(cypher: string): object { + if (active.size === 0) { + const env = deps.env(); + budgetBytes = resolveQueryRssBudgetBytes(env[QUERY_MAX_RSS_ENV], deps.totalMem()); + tracing = env[QUERY_TRACE_ENV] === '1'; + } + if (budgetBytes === null && !tracing) return {}; + const rss = deps.readRss(); + const q: ActiveQuery = { + label: queryLabel(cypher), + startedAt: deps.now(), + startBytes: rss, + peakBytes: rss, + }; + active.add(q); + observe(rss); + if (!stopSampler) { + stopSampler = deps.startSampler(sample); + } + return q; + }, + + end(token: object, rowCount?: number): void { + const q = token as ActiveQuery; + if (!active.has(q)) return; + const endRss = deps.readRss(); + observe(endRss); + active.delete(q); + if (tracing) { + const mb = (n: number) => Math.round(n / MIB); + deps.writeStderr( + `[query-trace] ms=${deps.now() - q.startedAt} rows=${rowCount ?? '?'} ` + + `memStartMB=${mb(q.startBytes)} memPeakMB=${mb(Math.max(q.peakBytes, endRss))} ` + + `rssEndMB=${mb(endRss)} active=${active.size} q="${q.label}"\n`, + ); + } + if (active.size === 0 && stopSampler) { + stopSampler(); + stopSampler = null; + } + }, + + sample, + }; +} + +/** `footprint` ships with macOS; once it fails, stop trying for this process. */ +let footprintUnavailable = false; +function readDarwinFootprint(): Promise { + if (footprintUnavailable) return Promise.resolve(null); + return new Promise((resolve) => { + execFile( + '/usr/bin/footprint', + ['--noCategories', '-f', 'bytes', '-p', String(process.pid)], + { timeout: FOOTPRINT_TIMEOUT_MS }, + (err, stdout) => { + const bytes = err ? null : parsePhysFootprintBytes(String(stdout)); + if (bytes === null) footprintUnavailable = true; + resolve(bytes); + }, + ); + }); +} + +/** Process-wide guard used by the read pool. */ +export const queryMemoryGuard: QueryMemoryGuard = createQueryMemoryGuard({ + readRss: () => process.memoryUsage.rss(), + ...(process.platform === 'darwin' ? { readFootprint: readDarwinFootprint } : {}), + env: () => process.env, + totalMem: () => Math.min(os.totalmem(), process.constrainedMemory() || Number.POSITIVE_INFINITY), + now: () => Date.now(), + writeStderr: (line) => { + realStderrWrite(line); + }, + onBreach: (breach) => { + // Synchronous so the message lands before exit (pipe writes are async). + try { + fs.writeSync(2, formatBreachMessage(breach)); + } catch { + /* exit regardless */ + } + process.exit(QUERY_MEMORY_LIMIT_EXIT_CODE); + }, + startSampler: (tick) => { + const timer = setInterval(tick, SAMPLE_INTERVAL_MS); + timer.unref(); + return () => clearInterval(timer); + }, +}); diff --git a/gitnexus/src/mcp/local/local-backend.ts b/gitnexus/src/mcp/local/local-backend.ts index 81b7cd029..23056ebb7 100644 --- a/gitnexus/src/mcp/local/local-backend.ts +++ b/gitnexus/src/mcp/local/local-backend.ts @@ -217,6 +217,12 @@ const VALUE_CANDIDATE_TYPES: ReadonlySet = new Set(['Const', 'Variable', */ const CANDIDATE_WINDOW = 20; +/** + * Most relationship rows `_runImpactBFS` reads per depth level (#3696). Real + * hub walks stay in the tens of thousands. + */ +export const IMPACT_FRONTIER_ROW_CAP = 100_000; + /** * Concurrent BFS walks when `query()` attaches per-process `chain`. Each walk * is already node-capped inside `_computeContextChain`; this bounds how many @@ -8513,6 +8519,7 @@ export class LocalBackend { let frontier = [symId]; const objectCallableFrontier: string[] = []; let traversalComplete = true; + let traversalCappedAtDepth: number | undefined; // Fetch one sentinel row beyond the cap so generated object bindings // degrade visibly instead of allocating an unbounded seed frontier. const OBJECT_CALLABLE_MEMBER_CAP = 5000; @@ -8733,17 +8740,29 @@ export class LocalBackend { // strongest edge makes the stamped pair fully determined by the key AND // retains the strongest evidence, the safer default for a blast-radius // tool. `sourceId` closes the order for edges that tie on both. + // + // `LIMIT` bounds one level's rows (#3696); the sentinel row past the cap + // marks the level as cut. A cut level is an arbitrary subset, so the + // walk stops there and reports partial. const query = direction === 'upstream' - ? `MATCH (caller)-[r:CodeRelation]->(n) WHERE n.id IN $frontierIds AND r.type IN $relTypes${confidenceFilter} RETURN n.id AS sourceId, caller.id AS id, caller.name AS name, labels(caller)[0] AS type, caller.filePath AS filePath, r.type AS relType, r.confidence AS confidence, r.staticGated AS staticGated` - : `MATCH (n)-[r:CodeRelation]->(callee) WHERE n.id IN $frontierIds AND r.type IN $relTypes${confidenceFilter} RETURN n.id AS sourceId, callee.id AS id, callee.name AS name, labels(callee)[0] AS type, callee.filePath AS filePath, r.type AS relType, r.confidence AS confidence, r.staticGated AS staticGated`; + ? // determinism: probe — safety ceiling, not a page; rows vary only past the cap, which marks the walk partial. + `MATCH (caller)-[r:CodeRelation]->(n) WHERE n.id IN $frontierIds AND r.type IN $relTypes${confidenceFilter} RETURN n.id AS sourceId, caller.id AS id, caller.name AS name, labels(caller)[0] AS type, caller.filePath AS filePath, r.type AS relType, r.confidence AS confidence, r.staticGated AS staticGated LIMIT ${IMPACT_FRONTIER_ROW_CAP + 1}` + : // determinism: probe — safety ceiling, not a page; rows vary only past the cap, which marks the walk partial. + `MATCH (n)-[r:CodeRelation]->(callee) WHERE n.id IN $frontierIds AND r.type IN $relTypes${confidenceFilter} RETURN n.id AS sourceId, callee.id AS id, callee.name AS name, labels(callee)[0] AS type, callee.filePath AS filePath, r.type AS relType, r.confidence AS confidence, r.staticGated AS staticGated LIMIT ${IMPACT_FRONTIER_ROW_CAP + 1}`; try { - const related = await executeParameterized(repo.lbugPath, query, { + const fetched = await executeParameterized(repo.lbugPath, query, { frontierIds: frontier, relTypes: relationTypes, ...(safeMinConfidence > 0 ? { minConfidence: safeMinConfidence } : {}), }); + const levelCapped = fetched.length > IMPACT_FRONTIER_ROW_CAP; + const related = levelCapped ? fetched.slice(0, IMPACT_FRONTIER_ROW_CAP) : fetched; + if (levelCapped) { + traversalComplete = false; + traversalCappedAtDepth = depth; + } // Validate before filtering/deduplication; a discarded corrupt edge is // still evidence that this result's counts cannot be trusted. for (const row of related) { @@ -8862,6 +8881,8 @@ export class LocalBackend { break; } + if (traversalCappedAtDepth !== undefined) break; + frontier = depth === 1 && objectCallableFrontier.length > 0 ? [...new Set([...nextFrontier, ...objectCallableFrontier])] @@ -9306,7 +9327,11 @@ export class LocalBackend { enrichmentTruncated: !skipEnrichment && MAX_CHUNKS > 0 && impacted.length > MAX_CHUNKS * CHUNK_SIZE, }); - const { risk, riskSharedAxes, riskScale } = scoreImpactRisk({ + const { + risk: scoredRisk, + riskSharedAxes: scoredSharedAxes, + riskScale, + } = scoreImpactRisk({ direction, directCount, processCount, @@ -9314,6 +9339,12 @@ export class LocalBackend { impactedCount: impacted.length, unusedAxes, }); + // Capped counts are lower bounds: keep a proven HIGH/CRITICAL, never a LOW/MEDIUM (#3442). + const traversalCapped = traversalCappedAtDepth !== undefined; + const riskAfterCap = (score: typeof scoredRisk) => + traversalCapped && (score === 'LOW' || score === 'MEDIUM') ? 'UNKNOWN' : score; + const risk = riskAfterCap(scoredRisk); + const riskSharedAxes = riskAfterCap(scoredSharedAxes); // Build per-depth counts (always included, even in summaryOnly mode) const byDepthCounts: Record = {}; @@ -9345,8 +9376,12 @@ export class LocalBackend { riskScale, ...(risk === 'UNKNOWN' ? { - riskNote: - processQueryFailed || moduleQueryFailed + riskNote: traversalCapped + ? `Risk is unresolved because depth ${traversalCappedAtDepth} returned more than ` + + `${IMPACT_FRONTIER_ROW_CAP} relationships and the walk stopped there. Observed ` + + 'counts are lower bounds; narrow the walk (relationTypes, maxDepth, minConfidence) ' + + 'before treating the change as safe.' + : processQueryFailed || moduleQueryFailed ? 'Risk is unresolved because process/module enrichment failed. Observed counts ' + 'are lower bounds; retry impact before treating the change as safe.' : unusedAxes.some((axis) => axis.reason === 'enrichment-truncated') diff --git a/gitnexus/test/integration/lbug-pool.test.ts b/gitnexus/test/integration/lbug-pool.test.ts index 063ab1dd2..0ba68a0a0 100644 --- a/gitnexus/test/integration/lbug-pool.test.ts +++ b/gitnexus/test/integration/lbug-pool.test.ts @@ -14,6 +14,7 @@ import { isLbugReady, } from '../../src/mcp/core/lbug-adapter.js'; import { withTestLbugDB } from '../helpers/test-indexed-db.js'; +import { execFile } from 'node:child_process'; const POOL_SEED_DATA = [ `CREATE (f:File {id: 'file:index.ts', name: 'index.ts', filePath: 'src/index.ts', content: ''})`, @@ -24,6 +25,35 @@ const POOL_SEED_DATA = [ CREATE (a)-[:CodeRelation {type: 'CALLS', confidence: 1.0, reason: 'direct', step: 0}]->(b)`, ]; +describe('pooled query memory guard process boundary', () => { + it('exits with a synchronous diagnostic before a read above its memory budget', async () => { + const adapterUrl = new URL('../../src/core/lbug/pool-adapter.ts', import.meta.url).href; + const source = ` + import lbug from '@ladybugdb/core'; + import { initLbugWithDb, executeParameterized } from ${JSON.stringify(adapterUrl)}; + const db = new lbug.Database(':memory:', 64 * 1024 * 1024); + await initLbugWithDb('guard-process', db); + process.env.GITNEXUS_QUERY_MAX_RSS_MB = '1'; + await executeParameterized('guard-process', 'RETURN $privateValue AS n', { privateValue: 'never-log-this-value' }); + process.exit(99); + `; + const result = await new Promise<{ code: number | string | undefined; stderr: string }>( + (resolve) => { + execFile( + process.execPath, + ['--import', 'tsx', '--input-type=module', '-e', source], + { timeout: 20_000, env: { ...process.env, GITNEXUS_QUERY_TRACE: '0' } }, + (error, _stdout, stderr) => resolve({ code: error?.code, stderr }), + ); + }, + ); + expect(result.code).toBe(70); + expect(result.stderr).toContain('GitNexus stopped: process memory reached'); + expect(result.stderr).toContain('RETURN $privateValue AS n'); + expect(result.stderr).not.toContain('never-log-this-value'); + }, 30_000); +}); + // ─── Pool lifecycle tests — test the pool adapter API directly ─────── withTestLbugDB( diff --git a/gitnexus/test/unit/calltool-dispatch.test.ts b/gitnexus/test/unit/calltool-dispatch.test.ts index ccf00f74f..33ab2faa8 100644 --- a/gitnexus/test/unit/calltool-dispatch.test.ts +++ b/gitnexus/test/unit/calltool-dispatch.test.ts @@ -93,6 +93,7 @@ vi.mock('../../src/core/group/resolve-at-member.js', () => ({ })); import { + IMPACT_FRONTIER_ROW_CAP, LocalBackend, REPO_ID_HASH_LENGTH, parseListReposPagination, @@ -1935,9 +1936,10 @@ describe('LocalBackend.callTool', () => { .filter((q: string) => q.includes('$frontierIds')); expect(frontierQueries.length).toBeGreaterThan(0); expect(frontierQueries.filter((q: string) => /ORDER BY/.test(q))).toEqual([]); - // …and equally: no LIMIT was added in its place. The traversal must see - // every neighbour edge; only the ORDERING moved. - expect(frontierQueries.filter((q: string) => /LIMIT/.test(q))).toEqual([]); + for (const query of frontierQueries) { + expect(query.match(/\bLIMIT\b/gi)).toHaveLength(1); + expect(query.endsWith(` LIMIT ${IMPACT_FRONTIER_ROW_CAP + 1}`)).toBe(true); + } }); // (a) Diamond: `delta` and `epsilon` are each reached at depth 2 from BOTH diff --git a/gitnexus/test/unit/impact-frontier-row-cap.test.ts b/gitnexus/test/unit/impact-frontier-row-cap.test.ts new file mode 100644 index 000000000..7db5abd99 --- /dev/null +++ b/gitnexus/test/unit/impact-frontier-row-cap.test.ts @@ -0,0 +1,150 @@ +/** + * #3696 — the impact BFS reads at most IMPACT_FRONTIER_ROW_CAP relationship + * rows per depth level. A capped level stops the walk, marks the result + * partial, and cannot publish a LOW/MEDIUM risk. + */ +import { describe, it, expect, vi, beforeEach } from 'vitest'; + +const executeQueryMock = vi.fn(); +const executeParameterizedMock = vi.fn(); + +vi.mock('../../src/core/lbug/pool-adapter.js', async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + initLbug: vi.fn(), + executeQuery: (...args: any[]) => executeQueryMock(...args), + executeParameterized: (...args: any[]) => executeParameterizedMock(...args), + closeLbug: vi.fn(), + isLbugReady: vi.fn().mockReturnValue(true), + }; +}); +vi.mock('../../src/mcp/core/lbug-adapter.js', async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + initLbug: vi.fn(), + executeQuery: (...args: any[]) => executeQueryMock(...args), + executeParameterized: (...args: any[]) => executeParameterizedMock(...args), + closeLbug: vi.fn(), + isLbugReady: vi.fn().mockReturnValue(true), + }; +}); + +import { LocalBackend, IMPACT_FRONTIER_ROW_CAP } from '../../src/mcp/local/local-backend.js'; + +function makeBackend() { + const backend = new LocalBackend(); + const repoHandle = { + id: 'repo1', + name: 'repo1', + repoPath: '/tmp/repo', + storagePath: '/tmp/repo/.gitnexus', + lbugPath: '/tmp/repo/.gitnexus/lbug', + indexedAt: 'now', + lastCommit: 'c', + stats: {}, + } as any; + (backend as any).repos.set(repoHandle.id, repoHandle); + (backend as any).ensureInitialized = vi.fn().mockResolvedValue(undefined); + return { backend, repoHandle }; +} + +/** + * Every frontier query returns `rowCount` CALLS rows spread over + * `distinctCallers` caller ids — duplicate edges, the shape a row-multiplying + * query produces. Returns the frontier query texts seen. + */ +function setupFrontier(rowCount: number, distinctCallers: number): string[] { + const frontierQueries: string[] = []; + executeParameterizedMock.mockImplementation(async (...args: any[]) => { + const query = typeof args[1] === 'string' ? args[1] : String(args[0] ?? ''); + if (query.includes('STEP_IN_PROCESS')) return []; + if (query.includes('MEMBER_OF')) return []; + if (query.includes('RETURN h.id AS hid')) return []; + if (query.includes('iface')) return []; + if (query.includes('r.type IN')) { + frontierQueries.push(query); + const rows: any[] = new Array(rowCount); + for (let i = 0; i < rowCount; i++) { + const c = i % distinctCallers; + rows[i] = { + sourceId: 'hub1', + id: `caller-${c}`, + name: `caller${c}`, + filePath: `src/caller-${c}.ts`, + relType: 'CALLS', + confidence: null, + }; + } + return rows; + } + return [{ id: 'hub1', name: 'HubSymbol', filePath: 'hub.ts' }]; + }); + executeQueryMock.mockImplementation(async () => []); + return frontierQueries; +} + +const impact = (backend: any, repoHandle: any, direction = 'upstream') => + backend._impactImpl(repoHandle, { target: 'HubSymbol', direction, maxDepth: 3 }); + +describe('impact: per-level frontier row cap (#3696)', () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it('asks the engine for at most one row past the cap', async () => { + const { backend, repoHandle } = makeBackend(); + const queries = setupFrontier(1, 1); + await impact(backend, repoHandle); + expect(queries.length).toBeGreaterThan(0); + for (const q of queries) expect(q).toContain(`LIMIT ${IMPACT_FRONTIER_ROW_CAP + 1}`); + }); + + it('a level exactly at the cap is complete and keeps its measured risk', async () => { + const { backend, repoHandle } = makeBackend(); + setupFrontier(IMPACT_FRONTIER_ROW_CAP, 1); + const res = await impact(backend, repoHandle); + expect(res.impactedCount).toBe(1); + expect(res.risk).toBe('LOW'); + expect(res.riskSharedAxes).toBe('LOW'); + expect(res.partial).toBeUndefined(); + }); + + it('a capped level stops the walk and cannot read as LOW', async () => { + const { backend, repoHandle } = makeBackend(); + const queries = setupFrontier(IMPACT_FRONTIER_ROW_CAP + 1, 1); + const res = await impact(backend, repoHandle); + + // No depth-2 query from an arbitrary subset of depth 1. + expect(queries).toHaveLength(1); + expect(res.impactedCount).toBe(1); + expect(res.partial).toBe(true); + expect(res.risk).toBe('UNKNOWN'); + expect(res.riskSharedAxes).toBe('UNKNOWN'); + expect(res.riskNote).toContain('depth 1'); + expect(res.riskNote).toContain(String(IMPACT_FRONTIER_ROW_CAP)); + }); + + it('a capped level keeps a CRITICAL its counts already prove', async () => { + const { backend, repoHandle } = makeBackend(); + setupFrontier(IMPACT_FRONTIER_ROW_CAP + 1, 300); + const res = await impact(backend, repoHandle); + expect(res.impactedCount).toBe(300); + expect(res.risk).toBe('CRITICAL'); + expect(res.riskSharedAxes).toBe('CRITICAL'); + expect(res.partial).toBe(true); + }); + + it('also bounds downstream traversal and reports its truncated counts', async () => { + const { backend, repoHandle } = makeBackend(); + const queries = setupFrontier(IMPACT_FRONTIER_ROW_CAP + 1, 1); + const res = await impact(backend, repoHandle, 'downstream'); + expect(queries).toHaveLength(1); + expect(queries[0]).toContain(`LIMIT ${IMPACT_FRONTIER_ROW_CAP + 1}`); + expect(res.impactedCount).toBe(1); + expect(res.partial).toBe(true); + expect(res.risk).toBe('UNKNOWN'); + expect(res.riskSharedAxes).toBe('UNKNOWN'); + }); +}); diff --git a/gitnexus/test/unit/lbug-pool-query-lifetime.test.ts b/gitnexus/test/unit/lbug-pool-query-lifetime.test.ts new file mode 100644 index 000000000..da1aece0c --- /dev/null +++ b/gitnexus/test/unit/lbug-pool-query-lifetime.test.ts @@ -0,0 +1,114 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +const { prepare, execute, begin, end } = vi.hoisted(() => ({ + prepare: vi.fn(), + execute: vi.fn(), + begin: vi.fn(() => ({})), + end: vi.fn(), +})); + +vi.mock('@ladybugdb/core', () => ({ + default: { + Connection: vi.fn(function (this: any) { + this.prepare = prepare; + this.execute = execute; + this.close = vi.fn().mockResolvedValue(undefined); + }), + }, +})); +vi.mock('../../src/core/lbug/lbug-adapter.js', () => ({ + isReadOnlyDbError: () => false, + loadFTSExtension: async () => false, + loadVectorExtension: async () => false, +})); +vi.mock('../../src/core/lbug/query-memory-guard.js', () => ({ + queryMemoryGuard: { begin, end }, +})); + +import { + closeLbug, + executeParameterized, + initLbugWithDb, +} from '../../src/core/lbug/pool-adapter.js'; + +describe('pooled query native lifetime', () => { + beforeEach(async () => { + vi.useFakeTimers(); + vi.clearAllMocks(); + prepare.mockResolvedValue({ isSuccess: () => true }); + execute.mockResolvedValue({ getAll: async () => [{ n: 1 }], close: vi.fn() }); + await initLbugWithDb('query-lifetime', {} as any, '/tmp/query-lifetime-db'); + }); + + afterEach(async () => { + await closeLbug(); + vi.useRealTimers(); + }); + + it.each(['prepare', 'execute', 'getAll'] as const)( + 'keeps monitoring after a caller times out in %s until native work settles', + async (stage) => { + let release!: (value: any) => void; + const pending = new Promise((resolve) => { + release = resolve; + }); + const close = vi.fn(); + const getAll = vi.fn(async () => [{ n: 1 }]); + const result = { getAll, close }; + if (stage === 'prepare') prepare.mockReturnValueOnce(pending); + else if (stage === 'execute') execute.mockReturnValueOnce(pending); + else execute.mockResolvedValueOnce({ getAll: () => pending, close }); + + const query = executeParameterized('query-lifetime', 'RETURN 1 AS n', {}).catch((e) => e); + await vi.advanceTimersByTimeAsync(30_001); + expect(end).not.toHaveBeenCalled(); + expect(close).not.toHaveBeenCalled(); + release( + stage === 'prepare' ? { isSuccess: () => true } : stage === 'execute' ? result : [{ n: 1 }], + ); + const error = await query; + await vi.advanceTimersByTimeAsync(0); + expect(error).toBeInstanceOf(Error); + expect(error.message).toContain('timed out'); + expect(end).toHaveBeenCalledTimes(1); + if (stage === 'prepare') expect(execute).not.toHaveBeenCalled(); + if (stage === 'execute') expect(getAll).not.toHaveBeenCalled(); + if (stage !== 'prepare') expect(close).toHaveBeenCalledTimes(1); + await expect(executeParameterized('query-lifetime', 'RETURN 1 AS n', {})).resolves.toEqual([ + { n: 1 }, + ]); + }, + ); + + it('releases monitoring and the connection after a native rejection', async () => { + execute.mockRejectedValueOnce(new Error('native failure')); + await expect(executeParameterized('query-lifetime', 'RETURN 1', {})).rejects.toThrow( + 'native failure', + ); + expect(end).toHaveBeenCalledTimes(1); + await expect(executeParameterized('query-lifetime', 'RETURN 1', {})).resolves.toEqual([ + { n: 1 }, + ]); + }); + + it('does not reuse timed-out connections while their native operations still run', async () => { + let release!: (value: any) => void; + const pending = new Promise((resolve) => { + release = resolve; + }); + execute.mockReturnValue(pending); + const first = Array.from({ length: 8 }, () => + executeParameterized('query-lifetime', 'RETURN 1', {}).catch((e) => e), + ); + await vi.advanceTimersByTimeAsync(30_001); + expect((await Promise.all(first)).every((e) => e instanceof Error)).toBe(true); + const waiting = executeParameterized('query-lifetime', 'RETURN 2', {}); + await vi.advanceTimersByTimeAsync(1); + expect(execute).toHaveBeenCalledTimes(8); + expect(end).not.toHaveBeenCalled(); + execute.mockResolvedValue({ getAll: async () => [{ n: 2 }], close: vi.fn() }); + release({ getAll: async () => [{ n: 1 }], close: vi.fn() }); + await expect(waiting).resolves.toEqual([{ n: 2 }]); + expect(end).toHaveBeenCalledTimes(9); + }); +}); diff --git a/gitnexus/test/unit/query-memory-guard.test.ts b/gitnexus/test/unit/query-memory-guard.test.ts new file mode 100644 index 000000000..332123173 --- /dev/null +++ b/gitnexus/test/unit/query-memory-guard.test.ts @@ -0,0 +1,370 @@ +/** + * #3696 — native-memory guard for read queries. Driven entirely through + * injected deps: no real timers, RSS, or process exit. + */ +import { describe, it, expect, vi } from 'vitest'; +import { + createQueryMemoryGuard, + formatBreachMessage, + queryLabel, + parsePhysFootprintBytes, + resolveQueryRssBudgetBytes, + FOOTPRINT_INTERVAL_MS, + QUERY_MAX_RSS_ENV, + QUERY_TRACE_ENV, + type QueryMemoryGuardDeps, +} from '../../src/core/lbug/query-memory-guard.js'; + +const MIB = 1024 * 1024; +const GIB = 1024 * MIB; + +function harness( + env: Record = {}, + totalMem = 64 * GIB, + readFootprint?: () => Promise, +) { + let rss = 500 * MIB; + let now = 1_000; + let tick: (() => void) | null = null; + const stop = vi.fn(() => { + tick = null; + }); + const deps: QueryMemoryGuardDeps = { + readRss: () => rss, + ...(readFootprint ? { readFootprint } : {}), + env: () => env, + totalMem: () => totalMem, + now: () => now, + writeStderr: vi.fn(), + onBreach: vi.fn(), + startSampler: vi.fn((t: () => void) => { + tick = t; + return stop; + }), + }; + const guard = createQueryMemoryGuard(deps); + return { + guard, + deps, + stop, + setRss: (mb: number) => { + rss = mb * MIB; + }, + advance: (ms: number) => { + now += ms; + }, + tick: () => tick?.(), + samplerRunning: () => tick !== null, + }; +} + +describe('resolveQueryRssBudgetBytes', () => { + it('defaults to 25% of RAM', () => { + expect(resolveQueryRssBudgetBytes(undefined, 64 * GIB)).toBe(16 * GIB); + }); + + it('defaults to 8 GiB when 25% is smaller and the host has sufficient headroom', () => { + expect(resolveQueryRssBudgetBytes(undefined, 16 * GIB)).toBe(8 * GIB); + }); + + it('honours an explicit MiB value', () => { + expect(resolveQueryRssBudgetBytes('2048', 64 * GIB)).toBe(2 * GIB); + expect(resolveQueryRssBudgetBytes(' 1.5 ', 64 * GIB)).toBe(Math.floor(1.5 * MIB)); + }); + + it('treats 0 as disabled', () => { + expect(resolveQueryRssBudgetBytes('0', 64 * GIB)).toBeNull(); + }); + + it('falls back to the default for unparseable or negative values', () => { + for (const raw of ['', ' ', 'all', '10junk', '-5', 'NaN', 'Infinity']) { + expect(resolveQueryRssBudgetBytes(raw, 64 * GIB)).toBe(16 * GIB); + } + }); + + it('rejects overflowing and sub-byte budgets', () => { + for (const raw of ['1e308', '1e-20']) { + expect(resolveQueryRssBudgetBytes(raw, 64 * GIB)).toBe(16 * GIB); + } + }); + + it('leaves headroom on small hosts and constrained runtimes', () => { + expect(resolveQueryRssBudgetBytes(undefined, 8 * GIB)).toBe(6 * GIB); + expect(resolveQueryRssBudgetBytes(undefined, 2 * GIB)).toBe(1.5 * GIB); + }); +}); + +describe('queryLabel', () => { + it('flattens whitespace and bounds length', () => { + expect(queryLabel('MATCH (n)\n WHERE n.id = $id\n RETURN n')).toBe( + 'MATCH (n) WHERE n.id = $id RETURN n', + ); + const long = queryLabel(`MATCH ${'x'.repeat(500)}`); + expect(long.length).toBe(201); + expect(long.endsWith('…')).toBe(true); + }); +}); + +describe('createQueryMemoryGuard', () => { + it('checks the budget before starting even a short query', () => { + const h = harness({ [QUERY_MAX_RSS_ENV]: '1000' }); + h.setRss(1500); + h.guard.begin('RETURN 1'); + expect(h.deps.onBreach).toHaveBeenCalledTimes(1); + }); + + it('checks memory growth when a query ends between sampler ticks', () => { + const h = harness({ [QUERY_MAX_RSS_ENV]: '1000' }); + const t = h.guard.begin('RETURN 1'); + h.setRss(1500); + h.guard.end(t, 1); + expect(h.deps.onBreach).toHaveBeenCalledTimes(1); + }); + + it('reports a breach with every running query once RSS passes the budget', () => { + const h = harness({ [QUERY_MAX_RSS_ENV]: '1000' }); + h.guard.begin('MATCH (caller)-[r]->(n) RETURN caller'); + h.guard.begin('MATCH (s)-[r]->(p:Process) RETURN p'); + expect(h.samplerRunning()).toBe(true); + + h.setRss(900); + h.tick(); + expect(h.deps.onBreach).not.toHaveBeenCalled(); + + h.setRss(1500); + h.tick(); + expect(h.deps.onBreach).toHaveBeenCalledTimes(1); + expect(h.deps.onBreach).toHaveBeenCalledWith({ + rssBytes: 1500 * MIB, + budgetBytes: 1000 * MIB, + activeLabels: [ + 'MATCH (caller)-[r]->(n) RETURN caller', + 'MATCH (s)-[r]->(p:Process) RETURN p', + ], + }); + + // Only once, even if the exit is slow to take effect. + h.tick(); + expect(h.deps.onBreach).toHaveBeenCalledTimes(1); + }); + + it('does not sample between queries', () => { + const h = harness({ [QUERY_MAX_RSS_ENV]: '1000' }); + const t = h.guard.begin('RETURN 1'); + h.guard.end(t, 1); + expect(h.stop).toHaveBeenCalledTimes(1); + expect(h.samplerRunning()).toBe(false); + + // RSS held by something else while no query runs is not this guard's call. + h.setRss(5000); + h.guard.sample(); + expect(h.deps.onBreach).not.toHaveBeenCalled(); + }); + + it('keeps sampling until the last concurrent query ends', () => { + const h = harness({ [QUERY_MAX_RSS_ENV]: '1000' }); + const a = h.guard.begin('A'); + const b = h.guard.begin('B'); + expect(h.deps.startSampler).toHaveBeenCalledTimes(1); + h.guard.end(a); + expect(h.samplerRunning()).toBe(true); + h.guard.end(b); + expect(h.samplerRunning()).toBe(false); + }); + + it('starts no sampler when disabled and not tracing', () => { + const h = harness({ [QUERY_MAX_RSS_ENV]: '0' }); + const t = h.guard.begin('RETURN 1'); + expect(h.deps.startSampler).not.toHaveBeenCalled(); + h.setRss(100_000); + h.guard.sample(); + h.guard.end(t); + expect(h.deps.onBreach).not.toHaveBeenCalled(); + expect(h.deps.writeStderr).not.toHaveBeenCalled(); + }); + + it('is on by default with no environment set', () => { + const h = harness({}, 16 * GIB); + h.guard.begin('RETURN 1'); + h.setRss(8 * 1024); + h.tick(); + expect(h.deps.onBreach).not.toHaveBeenCalled(); + h.setRss(8 * 1024 + 1); + h.tick(); + expect(h.deps.onBreach).toHaveBeenCalledTimes(1); + }); + + it('writes one trace line per query with start, peak and end RSS', () => { + const h = harness({ [QUERY_MAX_RSS_ENV]: '0', [QUERY_TRACE_ENV]: '1' }); + const t = h.guard.begin('MATCH (n)\n RETURN n'); + expect(h.samplerRunning()).toBe(true); + h.setRss(2000); + h.tick(); + h.setRss(700); + h.advance(42); + h.guard.end(t, 7); + expect(h.deps.writeStderr).toHaveBeenCalledWith( + '[query-trace] ms=42 rows=7 memStartMB=500 memPeakMB=2000 rssEndMB=700 active=0 q="MATCH (n) RETURN n"\n', + ); + }); + + it('ignores an unknown or already-ended token', () => { + const h = harness({ [QUERY_TRACE_ENV]: '1' }); + const t = h.guard.begin('RETURN 1'); + h.guard.end(t); + h.guard.end(t); + h.guard.end({}); + expect(h.deps.writeStderr).toHaveBeenCalledTimes(1); + }); +}); + +describe('createQueryMemoryGuard: physical footprint (macOS)', () => { + const flush = () => new Promise((r) => setImmediate(r)); + + it('never reads the footprint for a short query', async () => { + const readFootprint = vi.fn(async () => 50 * GIB); + const h = harness({ [QUERY_MAX_RSS_ENV]: '1000' }, 64 * GIB, readFootprint); + const t = h.guard.begin('RETURN 1'); + h.advance(FOOTPRINT_INTERVAL_MS - 1); + h.tick(); + h.guard.end(t); + await flush(); + expect(readFootprint).not.toHaveBeenCalled(); + expect(h.deps.onBreach).not.toHaveBeenCalled(); + }); + + it('breaches on the footprint while RSS stays low (compressed pages)', async () => { + const readFootprint = vi.fn(async () => 3 * GIB); + const h = harness({ [QUERY_MAX_RSS_ENV]: '1000' }, 64 * GIB, readFootprint); + h.guard.begin('MATCH (a), (b) RETURN a, b'); + h.setRss(300); + h.advance(FOOTPRINT_INTERVAL_MS); + h.tick(); + expect(readFootprint).toHaveBeenCalledTimes(1); + await flush(); + expect(h.deps.onBreach).toHaveBeenCalledWith({ + rssBytes: 3 * GIB, + budgetBytes: 1000 * MIB, + activeLabels: ['MATCH (a), (b) RETURN a, b'], + }); + }); + + it('reads at most once per interval and never overlaps reads', async () => { + let release: (v: number) => void = () => {}; + const readFootprint = vi.fn( + () => new Promise((r) => (release = r as (v: number) => void)), + ); + const h = harness({ [QUERY_MAX_RSS_ENV]: '1000' }, 64 * GIB, readFootprint); + h.guard.begin('Q'); + h.advance(FOOTPRINT_INTERVAL_MS); + h.tick(); + expect(readFootprint).toHaveBeenCalledTimes(1); + release(600 * MIB); + await flush(); + + // Interval: measured from the start of the previous read. + h.advance(FOOTPRINT_INTERVAL_MS - 1); + h.tick(); + expect(readFootprint).toHaveBeenCalledTimes(1); + h.advance(1); + h.tick(); + expect(readFootprint).toHaveBeenCalledTimes(2); + + // Overlap: a read still in flight blocks the next one, however long it takes. + h.advance(5 * FOOTPRINT_INTERVAL_MS); + h.tick(); + expect(readFootprint).toHaveBeenCalledTimes(2); + release(600 * MIB); + await flush(); + h.tick(); + expect(readFootprint).toHaveBeenCalledTimes(3); + }); + + it('ignores a footprint that resolves after the query ended', async () => { + let release: (v: number) => void = () => {}; + const readFootprint = vi.fn( + () => new Promise((r) => (release = r as (v: number) => void)), + ); + const h = harness({ [QUERY_MAX_RSS_ENV]: '1000' }, 64 * GIB, readFootprint); + const t = h.guard.begin('Q'); + h.advance(FOOTPRINT_INTERVAL_MS); + h.tick(); + h.guard.end(t); + release(50 * GIB); + await flush(); + expect(h.deps.onBreach).not.toHaveBeenCalled(); + }); + + it('does not charge a stale footprint to a later query', async () => { + let release!: (bytes: number | null) => void; + const pending = new Promise((resolve) => { + release = resolve; + }); + const h = harness({ [QUERY_MAX_RSS_ENV]: '1000' }, 64 * GIB, () => pending); + const old = h.guard.begin('old'); + h.advance(FOOTPRINT_INTERVAL_MS); + h.tick(); + h.guard.end(old); + const next = h.guard.begin('next'); + release(50 * GIB); + await flush(); + expect(h.deps.onBreach).not.toHaveBeenCalled(); + h.guard.end(next); + }); + + it('retains a footprint while a query present at sampling still runs', async () => { + let release!: (bytes: number | null) => void; + const pending = new Promise((resolve) => { + release = resolve; + }); + const h = harness({ [QUERY_MAX_RSS_ENV]: '1000' }, 64 * GIB, () => pending); + const first = h.guard.begin('first'); + const second = h.guard.begin('second'); + h.advance(FOOTPRINT_INTERVAL_MS); + h.tick(); + h.guard.end(first); + release(50 * GIB); + await flush(); + expect(h.deps.onBreach).toHaveBeenCalledTimes(1); + h.guard.end(second); + }); + + it('treats an unavailable footprint as no signal', async () => { + const readFootprint = vi.fn(async () => null); + const h = harness({ [QUERY_MAX_RSS_ENV]: '1000' }, 64 * GIB, readFootprint); + h.guard.begin('Q'); + h.advance(FOOTPRINT_INTERVAL_MS); + h.tick(); + await flush(); + expect(h.deps.onBreach).not.toHaveBeenCalled(); + }); +}); + +describe('parsePhysFootprintBytes', () => { + it('reads the byte count from `footprint -f bytes` output', () => { + const out = + 'node [123]: 64-bit Footprint: 2474344 B (16384 bytes per page)\n' + + ' phys_footprint: 2474344 B\n phys_footprint_peak: 9000000 B\n'; + expect(parsePhysFootprintBytes(out)).toBe(2474344); + }); + + it('returns null for unexpected output', () => { + expect(parsePhysFootprintBytes('')).toBeNull(); + expect(parsePhysFootprintBytes('phys_footprint: 11 MB')).toBeNull(); + }); +}); + +describe('formatBreachMessage', () => { + it('names the limit, the running queries and how to change it', () => { + const msg = formatBreachMessage({ + rssBytes: 17 * GIB, + budgetBytes: 16 * GIB, + activeLabels: ['MATCH (a) RETURN a'], + }); + expect(msg).toContain('17.0 GB'); + expect(msg).toContain('16.0 GB limit'); + expect(msg).toContain(' - MATCH (a) RETURN a'); + expect(msg).toContain(`${QUERY_MAX_RSS_ENV}=`); + expect(msg).toContain('#3696'); + }); +});