From 5eecf1a87e025295f3a49bb62fcaa4a05de644b7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gerg=C5=91=20Magyar?= Date: Mon, 8 Jun 2026 07:20:12 +0100 Subject: [PATCH] fix(parse): correct worker-pool docs drift + surface worker-side stack on crash (#2068) (#2070) --- ARCHITECTURE.md | 14 +- README.md | 4 +- gitnexus/bench/parse-throughput.md | 43 +++-- gitnexus/package-lock.json | 1 - gitnexus/package.json | 1 - gitnexus/src/cli/analyze.ts | 16 ++ gitnexus/src/core/ingestion/ast-cache.ts | 77 -------- gitnexus/src/core/ingestion/call-processor.ts | 88 +-------- .../src/core/ingestion/export-detection.ts | 3 +- .../src/core/ingestion/filesystem-walker.ts | 22 --- .../ingestion/languages/python/captures.ts | 7 +- .../pipeline-phases/orm-extraction.ts | 106 ----------- .../ingestion/pipeline-phases/parse-impl.ts | 11 -- .../core/ingestion/workers/parse-worker.ts | 18 +- .../src/core/ingestion/workers/worker-pool.ts | 38 +++- gitnexus/src/core/lbug/lbug-adapter.ts | 33 +++- gitnexus/src/core/lbug/pool-adapter.ts | 56 +++++- gitnexus/src/core/lbug/query-result-utils.ts | 31 +++ gitnexus/src/mcp/local/local-backend.ts | 15 +- gitnexus/src/types/pipeline.ts | 9 +- gitnexus/test/integration/lbug-pool.test.ts | 80 ++++++++ gitnexus/test/unit/ast-cache.test.ts | 95 ---------- .../test/unit/lbug-query-result-utils.test.ts | 45 +++++ .../test/unit/lbug-readonly-error.test.ts | 28 +++ .../test/unit/worker-pool-error-stack.test.ts | 176 ++++++++++++++++++ 25 files changed, 562 insertions(+), 455 deletions(-) delete mode 100644 gitnexus/src/core/ingestion/ast-cache.ts delete mode 100644 gitnexus/src/core/ingestion/pipeline-phases/orm-extraction.ts create mode 100644 gitnexus/src/core/lbug/query-result-utils.ts delete mode 100644 gitnexus/test/unit/ast-cache.test.ts create mode 100644 gitnexus/test/unit/lbug-query-result-utils.test.ts create mode 100644 gitnexus/test/unit/worker-pool-error-stack.test.ts diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index d90d132f6a..01013c71c4 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -15,7 +15,7 @@ Monorepo: **CLI/MCP** (`gitnexus/`) + **browser UI** (`gitnexus-web/`). ## End-to-end flow: index → graph → tools -1. **Ingestion** — `analyze.ts` → `runFullAnalysis` (`run-analyze.ts`) → `runPipelineFromRepo` (`pipeline.ts`). DAG of 12 phases builds a `KnowledgeGraph` in memory, then loads into LadybugDB under `.gitnexus/`. Repo registered in `~/.gitnexus/registry.json` for MCP discovery. +1. **Ingestion** — `analyze.ts` → `runFullAnalysis` (`run-analyze.ts`) → `runPipelineFromRepo` (`pipeline.ts`). DAG of 14 phases builds a `KnowledgeGraph` in memory, then loads into LadybugDB under `.gitnexus/`. Repo registered in `~/.gitnexus/registry.json` for MCP discovery. 2. **Persistence** — `repo-manager.ts` (paths, registry, KuzuDB cleanup). `lbug-adapter.ts` (graph load, queries, embedding batches). @@ -101,7 +101,7 @@ scan → structure → [markdown, cobol] → parse → [routes, tools, orm] | `communities` | `communities.ts` | `mro`, `pruneLocalSymbols`, `structure` | Community nodes + MEMBER_OF edges (Leiden algorithm) | | `processes` | `processes.ts` | `communities`, `routes`, `tools`, `pruneLocalSymbols`, `structure` | Process nodes + STEP_IN_PROCESS edges | -**Non-phase files in the same directory:** `parse-impl.ts`, `cross-file-impl.ts` (implementation), `wildcard-synthesis.ts` (whole-module import expansion), `orm-extraction.ts` (sequential ORM fallback), `types.ts`, `runner.ts`, `index.ts`. +**Non-phase files in the same directory:** `parse-impl.ts`, `cross-file-impl.ts` (implementation), `wildcard-synthesis.ts` (whole-module import expansion), `types.ts`, `runner.ts`, `index.ts`. ### DAG runner @@ -121,7 +121,7 @@ scan → structure → [markdown, cobol] → parse → [routes, tools, orm] - **Single graph accumulator** — all phases mutate the same `KnowledgeGraph` in `ctx`; the graph is the primary output. - **Typed phase access** — `getPhaseOutput(deps, 'name')` for type-safe upstream results. - **Binding accumulator lifecycle** — created in `parse`, disposed by `crossFile` (in `finally`). No other phase should take ownership. -- **Skippable phases** — `skipGraphPhases` omits MRO/communities/processes (faster tests); `pruneLocalSymbols` still runs (it is graph cleanup, not analysis). `skipWorkers` forces sequential parsing. +- **Skippable phases** — `skipGraphPhases` omits MRO/communities/processes (faster tests); `pruneLocalSymbols` still runs (it is graph cleanup, not analysis). `skipWorkers` is no longer a sequential escape hatch — it (like `--workers 0` / `GITNEXUS_WORKER_POOL_SIZE=0`) is rejected with an actionable error, since the worker pool is the sole parse path (§ Chunked parse-and-resolve). - **Local-symbol pruning** — `pruneLocalSymbols` removes inert block-local value symbols after scope resolution has consumed them. Opt out per-call with `PipelineOptions.keepLocalValueSymbols` or globally with the `GITNEXUS_KEEP_LOCAL_VALUE_SYMBOLS` env var. ### How to add a new phase @@ -202,7 +202,7 @@ Language-agnostic scope-resolution resolver. This is the resolution path for eve ``` Orchestrator: `runScopeResolution(input, provider)` in `scope-resolution/pipeline/run.ts`. -Pipeline phase: `scopeResolutionPhase` in `scope-resolution/pipeline/phase.ts` — iterates the registered `SCOPE_RESOLVERS`, reads per-file Trees from the parse phase's `scopeTreeCache`, disposes the cache at the end. +Pipeline phase: `scopeResolutionPhase` in `scope-resolution/pipeline/phase.ts` — iterates the registered `SCOPE_RESOLVERS` over the worker-serialized `ParsedFile`s. (Per-language `emitScopeCaptures` hooks may reuse a cached Tree via the orchestrator's `treeCache`, but in worker-pool runs that cache is empty — Trees can't cross MessageChannels — so they consume the pre-extracted `ParsedFile` instead; § Performance notes.) ### `ScopeResolver` contract @@ -251,7 +251,7 @@ CI auto-discovers the set via `tsx`. No workflow edit required. ### Performance notes -- **Cross-phase Tree cache**: parse phase writes Trees into `scopeTreeCache` (separate from the chunk-local `astCache`) ONLY for languages with `emitScopeCaptures`. Scope-resolution reads from it to skip the second parse. Cleared at end of the phase. Workers leave the cache empty — Trees can't cross MessageChannels; cache miss = fresh parse. `PROF_SCOPE_RESOLUTION=1` emits hit/miss counters and a worker-engaged warning. +- **Cross-phase Tree cache**: the orchestrator's `treeCache` (`RunScopeResolutionInput.treeCache`) lets a scope-resolution per-language hook (`emitScopeCaptures`) reuse a tree instead of re-parsing. Workers leave it empty — Trees can't cross MessageChannels — so in normal (worker-pool) runs scope-resolution does NOT rely on it: workers serialize each file's `ParsedFile` (+ capture side-channel) and stream them in, so scope-resolution consumes the pre-extracted artifact rather than re-parsing on the main thread (§ Chunked parse-and-resolve). `PROF_SCOPE_RESOLUTION=1` emits hit/miss counters and a worker-engaged warning. - **Typed relationship iteration**: heritage + MRO walk only the EXTENDS / IMPLEMENTS / HAS_METHOD edges via `iterRelationshipsByType`, not the full relationship map. - **Workspace-resolution-index**: O(1) `findOwnedMember` / `findExportedDef` / `classScopeByDefId` built once per run. - **SCC-ordered cross-file return-type propagation** (PR #1050): `propagateImportedReturnTypes` walks `indexes.sccs` in reverse-topological order (leaves first), so multi-hop alias chains like `models.User → service.user → app.user` collapse to the terminal class in a single linear pass. Within each importer, the source module's `typeBindings` is chain-followed BEFORE mirroring (so we mirror terminal types, not intermediate refs), and the importer's own `typeBindings` is chain-followed AFTER mirroring (so local `const x = importedFn()` resolves before downstream importers run). Cyclic SCCs reach a partial fixpoint within a single pass without iterating to convergence — see the `ts-circular` cross-file-binding fixture which only asserts pipeline-no-throw. PROF output (`PROF_SCOPE_RESOLUTION=1`) splits `finalize` from `propagate` so quadratic regressions in the chain-follow surface independently. @@ -314,7 +314,7 @@ Unified 3-tier algorithm (`model/resolution-context.ts`), per-language `importSe ### Chunked parse-and-resolve `parse` processes files in ~20 MB byte-budget chunks to bound memory. Per chunk: -1. Worker pool dispatches files (or sequential fallback via `skipWorkers`) +1. Worker pool dispatches files (the sole parse path — there is no sequential fallback; `skipWorkers`, `--workers 0`, and `GITNEXUS_WORKER_POOL_SIZE=0` are rejected with an actionable error) 2. Each worker: detect language → load grammar → run queries → return unified `ParseWorkerResult` 3. Synthesize wildcard bindings (`wildcard-synthesis.ts`) 4. Resolve imports @@ -324,6 +324,8 @@ Inheritance edges are emitted later, by the scope-resolution phase (`preEmitInhe Workers: `workers/worker-pool.ts`, `workers/parse-worker.ts`. +**Worker-serialized ParsedFiles (#2038).** To index very large repos (e.g. the Linux kernel) without OOM, the worker pool is the *sole* parse path and workers serialize each file's `ParsedFile` (plus its capture side-channel) in parallel, streaming them to scope-resolution through a disk-backed store. Scope-resolution consumes the pre-extracted artifact instead of re-parsing every file on the main thread — tree-sitter's native input buffers are not GC-reclaimable, so the former main-thread re-parse leaked native memory until the process died. Pool creation is lazy / cache-miss-gated, so a warm all-cache-hit run replays cached worker output without spawning a worker (hence `usedWorkerPool` can be false even when the repo has parseable files). + ### Inheritance and MRO Inheritance is captured by the `@reference.inherits` tag and emitted by the scope-resolution phase: `preEmitInheritanceEdges` resolves each base in scope, then `emitHeritageEdges` writes the `EXTENDS`/`IMPLEMENTS` edges. The phase then computes method resolution order via each `ScopeResolver`'s `buildMro` hook, feeding a `MethodDispatchIndex` used for owner-scoped lookups. Per-language strategy: diff --git a/README.md b/README.md index 28229c98c6..2df823d9ac 100644 --- a/README.md +++ b/README.md @@ -236,7 +236,7 @@ gitnexus analyze --embeddings [limit] # Enable embedding generation (slower, be gitnexus analyze --verbose # Log skipped files when parsers are unavailable gitnexus analyze --worker-timeout 60 # Increase worker idle timeout for slow parses gitnexus analyze --wal-checkpoint-threshold 67108864 # 64 MiB. Control LadybugDB WAL auto-checkpoint threshold (default: 67108864 = 64 MiB; -1 keeps Ladybug stock ~16 MiB) -gitnexus analyze --workers # Parse worker pool size (default: cores-1, capped at 16; 0 = sequential) +gitnexus analyze --workers # Parse worker pool size (>=1; default: cores-1, capped at 16, auto-sized to the repo). 0 is rejected — there is no sequential mode. gitnexus mcp # Start MCP server (stdio) — serves all indexed repos gitnexus serve # Start local HTTP server (multi-repo) for web UI connection gitnexus list # List all indexed repositories @@ -314,7 +314,7 @@ Most `analyze` knobs are also CLI flags (`--workers`, `--worker-timeout`, `--max | Variable | Default | Effect | Tune when… | | -------------------------------------- | ------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------- | -| `GITNEXUS_WORKER_POOL_SIZE` | `cores - 1`, capped at 16 | Parse worker pool size. `0` disables the pool (sequential fallback). Equivalent to `--workers `. | Constrained containers (cgroup CPU limits), CI runners with explicit quotas, or debugging a worker-only crash via `0`. | +| `GITNEXUS_WORKER_POOL_SIZE` | `cores - 1`, capped at 16 | Parse worker pool size (must be ≥ 1). Equivalent to `--workers `. The worker pool is the sole parse path — there is no sequential parser, so `0` is rejected with an actionable error (the pool self-heals via quarantine + respawn). | Constrained containers (cgroup CPU limits) or CI runners with explicit quotas. To narrow down a worker crash set `1` for a single-worker pool — not `0`. | | `GITNEXUS_PARSE_CHUNK_CONCURRENCY` | `2` | Number of chunks whose file contents may be read into memory in parallel while the pool dispatches the current chunk. Worker dispatch itself stays serial. | Repos large enough to chunk (multi-MB total source) where disk I/O is a measurable fraction of analyze wall-clock. | | `GITNEXUS_VERBOSE` | unset | When `1`, enables verbose ingestion logs (skipped-file warnings, per-chunk throughput, parse-cache stats). Equivalent to `--verbose`. | Debugging an analyze that "completed" but seems to have missed files; tuning `--workers` / chunk concurrency against observable throughput. | | `GITNEXUS_PROFILE_DEFERRED` | unset | When `1`, emits `[deferred-profile]` timing/progress logs for the post-chunk deferred resolution band (imports → heritage → buildHeritageMap → legacy call resolution). Implied by `GITNEXUS_VERBOSE`. | Diagnosing analyze stalls in "Resolving calls (all chunks)" on large Java/Kotlin repos (issue #1741) without the full verbose ingestion noise. | diff --git a/gitnexus/bench/parse-throughput.md b/gitnexus/bench/parse-throughput.md index 24a7c98b1f..4d53cd2dcf 100644 --- a/gitnexus/bench/parse-throughput.md +++ b/gitnexus/bench/parse-throughput.md @@ -70,10 +70,17 @@ this doc, run it under instrumentation: ```bash # From the gitnexus/ subdir: cd gitnexus -# Single-threaded baseline (sequential fallback): -npx vitest run test/integration/parse-impl-large-fixture.test.ts --reporter=verbose +# The worker pool is the sole parse path, so every run needs the dist worker +# (`npm run build`) and a pool size pinned via GITNEXUS_WORKER_POOL_SIZE. -# Worker-pool path (requires built dist/ — pre-built by `npm run build`): +# Single-worker-pool baseline (closest analog to the old single-threaded run — +# sequential parsing was removed, so a 1-worker pool is the floor): +npm run build && \ + GITNEXUS_WORKER_POOL_SIZE=1 \ + GITNEXUS_VERBOSE=1 \ + npx vitest run test/integration/parse-impl-large-fixture.test.ts --reporter=verbose + +# Multi-worker path: npm run build && \ GITNEXUS_WORKER_POOL_SIZE=4 \ GITNEXUS_PARSE_CHUNK_CONCURRENCY=2 \ @@ -97,22 +104,26 @@ node --inspect=0 \ ## Latest measurement > _No measurement data has been collected yet — this file is the -> methodology + harness scaffold. The single recorded data point is the -> U6 wall-clock smoke baseline below; the worker-pool rows are -> placeholders for future bench-pass output._ +> methodology + harness scaffold. The U6 smoke test confirms the +> worker-pool path stays well within its wall-clock budget, but every +> throughput/heap cell below is a `_TBD_` placeholder for a future +> bench-pass._ The U6 integration test (`gitnexus/test/integration/parse-impl-large-fixture.test.ts`) -was observed completing the synthetic fixture in **~6 seconds** under -the sequential path (`skipWorkers: true`) on the development machine, -well under the 30 s `Promise.race` wall-clock budget. That number is a -smoke baseline only — recorded here for reference, not as a regression -target. +runs the worker pool — the sole parse path now that sequential parsing +has been removed (disabling the pool on a repo with parseable files +raises a hard `WorkerPoolDisabledError`). It completes the synthetic +fixture well within the 30 s `Promise.race` wall-clock budget on the +development machine, but no worker-pool throughput/heap numbers have been +captured yet, so the rows below are all `_TBD_`. (An earlier ~6 s figure +recorded here was measured on the now-removed sequential path; it has +been dropped rather than relabelled as a worker-pool baseline, since the +two paths are not comparable.) -| Path | files/s | wall-clock | peak heap | chunks | quarantined | -| ------------------------------------------ | ------- | -------------------- | --------- | ------ | ----------- | -| Sequential (`skipWorkers: true`, U6 smoke) | _TBD_ | ~6 s _(observation)_ | _TBD_ | 17 | 0 | -| Worker pool, `--workers 4`, concurrency 2 | _TBD_ | _TBD_ | _TBD_ | _TBD_ | 0 | -| Worker pool, `--workers 1`, concurrency 1 | _TBD_ | _TBD_ | _TBD_ | _TBD_ | 0 | +| Path | files/s | wall-clock | peak heap | chunks | quarantined | +| ------------------------------------------------------------------------- | ------- | ---------- | --------- | ------ | ----------- | +| Worker pool, `--workers 1` (`GITNEXUS_WORKER_POOL_SIZE=1`), concurrency 1 | _TBD_ | _TBD_ | _TBD_ | _TBD_ | 0 | +| Worker pool, `--workers 4`, concurrency 2 | _TBD_ | _TBD_ | _TBD_ | _TBD_ | 0 | **Hardware:** _TBD — record OS, CPU, RAM, Node version, gitnexus SHA at the time of the bench-pass that populates the table above._ diff --git a/gitnexus/package-lock.json b/gitnexus/package-lock.json index 9ec0598f9a..83534ed773 100644 --- a/gitnexus/package-lock.json +++ b/gitnexus/package-lock.json @@ -26,7 +26,6 @@ "ignore": "^7.0.5", "js-yaml": "^4.1.1", "jsonc-parser": "^3.3.1", - "lru-cache": "^11.0.0", "mnemonist": "^0.40.3", "onnxruntime-node": "^1.24.0", "pandemonium": "^2.4.0", diff --git a/gitnexus/package.json b/gitnexus/package.json index 54c0b5d5f8..1dbb1ed809 100644 --- a/gitnexus/package.json +++ b/gitnexus/package.json @@ -70,7 +70,6 @@ "ignore": "^7.0.5", "js-yaml": "^4.1.1", "jsonc-parser": "^3.3.1", - "lru-cache": "^11.0.0", "mnemonist": "^0.40.3", "onnxruntime-node": "^1.24.0", "pandemonium": "^2.4.0", diff --git a/gitnexus/src/cli/analyze.ts b/gitnexus/src/cli/analyze.ts index 70bc9a566b..3f196b11d1 100644 --- a/gitnexus/src/cli/analyze.ts +++ b/gitnexus/src/cli/analyze.ts @@ -60,6 +60,22 @@ const writeFatalToStderr = (label: string, err: unknown): void => { const message = isErr ? err.message : String(err); realStderrWrite(`\n ${label}: ${message}\n`); if (isErr && err.stack) realStderrWrite(`${err.stack}\n`); + // Walk and print the `cause` chain. The phase runner wraps the underlying + // failure as `new Error("Phase 'X' failed: …", { cause })`, so the original + // error (e.g. a WorkerPoolDispatchError carrying the worker-side stack from + // #2068) is only reachable via `.cause`. Without this the user sees the + // wrapper's main-thread stack and never the real frame. `cause.stack` already + // begins with the cause's message, so we print the stack alone (not message + + // stack) to avoid repeating it. Depth-bounded so a cyclic `cause` can't loop + // (the phase runner wraps one level; the bound leaves headroom for future + // nesting); uses realStderrWrite so the redirected console.error's ANSI + // clear-line wrapping can't erase it (#1169). + const MAX_CAUSE_DEPTH = 5; + let cause: unknown = isErr ? (err as { cause?: unknown }).cause : undefined; + for (let depth = 0; depth < MAX_CAUSE_DEPTH && cause instanceof Error; depth++) { + realStderrWrite(`\n Caused by: ${cause.stack ?? cause.message}\n`); + cause = (cause as { cause?: unknown }).cause; + } }; let fatalHandlersInstalled = false; diff --git a/gitnexus/src/core/ingestion/ast-cache.ts b/gitnexus/src/core/ingestion/ast-cache.ts deleted file mode 100644 index 454c60df2e..0000000000 --- a/gitnexus/src/core/ingestion/ast-cache.ts +++ /dev/null @@ -1,77 +0,0 @@ -import { LRUCache } from 'lru-cache'; -import Parser from 'tree-sitter'; - -import { logger } from '../logger.js'; -/** - * Minimal structural shape consumers need when reading Trees back - * through a phase-dependency boundary. Declared here so phases that - * receive ASTCache via `getPhaseOutput<...>` don't hand-roll their - * own inline structural types that silently drift when ASTCache's - * contract changes. - * - * Typed as `unknown` at the Tree boundary because consumers on the - * other side of the phase-output map don't share tree-sitter's type - * graph (e.g. COBOL's standalone processor). - */ -export interface ASTCacheReader { - get(filePath: string): unknown; - clear(): void; -} - -// Define the interface for the Cache -export interface ASTCache extends ASTCacheReader { - get: (filePath: string) => Parser.Tree | undefined; - set: (filePath: string, tree: Parser.Tree) => void; - clear: () => void; - stats: () => { size: number; maxSize: number }; -} - -export const createASTCache = (maxSize: number = 50): ASTCache => { - const effectiveMax = Math.max(maxSize, 1); - // Initialize the cache with a 'dispose' handler - // This is the magic: When an item is evicted (dropped), this runs automatically. - const cache = new LRUCache({ - max: effectiveMax, - dispose: (tree) => { - try { - // NOTE: web-tree-sitter has tree.delete(); native tree-sitter - // trees are GC-managed and .delete is absent (no-op here). - // - // Single-owner invariant (load-bearing under WASM): a given - // Parser.Tree reference must live in AT MOST ONE ASTCache - // that disposes. The parse-phase chunk-local cache clears - // between chunks; the cross-phase `scopeTreeCache` (also an - // ASTCache today) holds the same Tree by reference. Under - // native tree-sitter this is benign (dispose is a no-op). - // If/when GitNexus adopts web-tree-sitter for sequential - // parsing, the cross-phase cache must either (a) skip - // writing Trees that are already owned by a disposing cache, - // or (b) use tree.copy() per entry. Failing to pick one - // will hand freed memory to scope-resolution. - (tree as unknown as { delete?: () => void }).delete?.(); - } catch (e) { - logger.warn({ e }, 'Failed to delete tree from WASM memory'); - } - }, - }); - - return { - get: (filePath: string) => { - const tree = cache.get(filePath); - return tree; // Returns undefined if not found - }, - - set: (filePath: string, tree: Parser.Tree) => { - cache.set(filePath, tree); - }, - - clear: () => { - cache.clear(); - }, - - stats: () => ({ - size: cache.size, - maxSize: effectiveMax, - }), - }; -}; diff --git a/gitnexus/src/core/ingestion/call-processor.ts b/gitnexus/src/core/ingestion/call-processor.ts index 5c8e447f6e..b5b258dcd1 100644 --- a/gitnexus/src/core/ingestion/call-processor.ts +++ b/gitnexus/src/core/ingestion/call-processor.ts @@ -9,25 +9,17 @@ * * - `processRoutesFromExtracted` — CALLS edges from framework routes * (e.g. Laravel) to their controller methods. - * - `processNextjsFetchRoutes` / `extractFetchCallsFromFiles` / - * `extractConsumerAccessedKeys` — FETCHES edges from `fetch()` calls to - * Next.js Route nodes. + * - `processNextjsFetchRoutes` / `extractConsumerAccessedKeys` — FETCHES edges + * from `fetch()` calls to Next.js Route nodes. * - `buildExportedTypeMapFromGraph` — exported symbol → return/declared type * map, consumed by the cross-file enrichment pass. */ -import Parser from 'tree-sitter'; import { KnowledgeGraph } from '../graph/types.js'; -import { ASTCache } from './ast-cache.js'; import type { SemanticModel, SymbolTableReader } from './model/index.js'; -import { isLanguageAvailable, loadParser, loadLanguage } from '../tree-sitter/parser-loader.js'; -import { getProvider } from './languages/index.js'; import { generateId } from '../../lib/utils.js'; -import { getLanguageFromFilename } from 'gitnexus-shared'; import type { SymbolDefinition } from 'gitnexus-shared'; import { yieldToEventLoop } from './utils/event-loop.js'; -import { parseSourceSafe } from '../tree-sitter/safe-parse.js'; -import { getTreeSitterBufferSize } from './constants.js'; import type { ExtractedRoute, ExtractedFetchCall } from './workers/parse-worker.js'; import { normalizeFetchURL, routeMatches } from './route-extractors/nextjs.js'; import { extractReturnTypeName } from './type-extractors/shared.js'; @@ -454,79 +446,3 @@ export const processNextjsFetchRoutes = ( } } }; - -/** - * Extract fetch() calls from source files (sequential path). - * Workers handle this via tree-sitter captures in parse-worker; this function - * provides the same extraction for the sequential fallback path. - */ -export const extractFetchCallsFromFiles = async ( - files: { path: string; content: string }[], - astCache: ASTCache, -): Promise => { - const parser = await loadParser(); - const result: ExtractedFetchCall[] = []; - - for (const file of files) { - const language = getLanguageFromFilename(file.path); - if (!language) continue; - if (!isLanguageAvailable(language)) continue; - - const provider = getProvider(language); - const queryStr = provider.treeSitterQueries; - if (!queryStr) continue; - - await loadLanguage(language, file.path); - - let tree = astCache.get(file.path); - if (!tree) { - const parseContent = provider.preprocessSource?.(file.content, file.path) ?? file.content; - try { - tree = parseSourceSafe(parser, parseContent, undefined, { - bufferSize: getTreeSitterBufferSize(parseContent), - }); - } catch { - continue; - } - astCache.set(file.path, tree); - } - - let matches; - try { - const lang = parser.getLanguage(); - const query = new Parser.Query(lang, queryStr); - matches = query.matches(tree.rootNode); - } catch { - continue; - } - - for (const match of matches) { - const captureMap: Record = {}; - match.captures.forEach((c) => (captureMap[c.name] = c.node)); - - if (captureMap['route.fetch']) { - const urlNode = captureMap['route.url'] ?? captureMap['route.template_url']; - if (urlNode) { - result.push({ - filePath: file.path, - fetchURL: urlNode.text, - lineNumber: captureMap['route.fetch'].startPosition.row, - }); - } - } else if (captureMap['http_client'] && captureMap['http_client.url']) { - const method = captureMap['http_client.method']?.text; - const url = captureMap['http_client.url'].text; - const HTTP_CLIENT_ONLY = new Set(['head', 'options', 'request', 'ajax']); - if (method && HTTP_CLIENT_ONLY.has(method) && url.startsWith('/')) { - result.push({ - filePath: file.path, - fetchURL: url, - lineNumber: captureMap['http_client'].startPosition.row, - }); - } - } - } - } - - return result; -}; diff --git a/gitnexus/src/core/ingestion/export-detection.ts b/gitnexus/src/core/ingestion/export-detection.ts index 31d0722f4e..17494e7bb3 100644 --- a/gitnexus/src/core/ingestion/export-detection.ts +++ b/gitnexus/src/core/ingestion/export-detection.ts @@ -4,7 +4,8 @@ * Determines whether a symbol (function, class, etc.) is exported/public * in its language. This is a pure function — safe for use in worker threads. * - * Shared between parse-worker.ts (worker pool) and parsing-processor.ts (sequential fallback). + * Used by the language providers during worker parsing (parse-worker.ts) — the + * sole parse path. (Sequential parsing was removed.) */ import { findSiblingChild, type SyntaxNode } from './utils/ast-helpers.js'; diff --git a/gitnexus/src/core/ingestion/filesystem-walker.ts b/gitnexus/src/core/ingestion/filesystem-walker.ts index 9ba959ea64..0af28a9586 100644 --- a/gitnexus/src/core/ingestion/filesystem-walker.ts +++ b/gitnexus/src/core/ingestion/filesystem-walker.ts @@ -6,10 +6,6 @@ import { glob } from 'glob'; import { createIgnoreFilter } from '../../config/ignore-service.js'; import { logger } from '../logger.js'; -export interface FileEntry { - path: string; - content: string; -} /** Lightweight entry — path + size from stat, no content in memory */ export interface ScannedFile { @@ -153,21 +149,3 @@ export const readFileContents = async ( return contents; }; - -/** - * Legacy API — scans and reads everything into memory. - * Used by sequential fallback path only. - */ -export const walkRepository = async ( - repoPath: string, - onProgress?: (current: number, total: number, filePath: string) => void, -): Promise => { - const scanned = await walkRepositoryPaths(repoPath, onProgress); - const contents = await readFileContents( - repoPath, - scanned.map((f) => f.path), - ); - return scanned - .filter((f) => contents.has(f.path)) - .map((f) => ({ path: f.path, content: contents.get(f.path)! })); -}; diff --git a/gitnexus/src/core/ingestion/languages/python/captures.ts b/gitnexus/src/core/ingestion/languages/python/captures.ts index 761404ebb3..756066382b 100644 --- a/gitnexus/src/core/ingestion/languages/python/captures.ts +++ b/gitnexus/src/core/ingestion/languages/python/captures.ts @@ -38,9 +38,10 @@ export function emitPythonScopeCaptures( _filePath: string, cachedTree?: unknown, ): readonly CaptureMatch[] { - // Skip the parse when the caller (parse phase's ASTCache) already - // produced a Tree for this source. Cache miss = re-parse, same as - // before. The cachedTree parameter is typed as `unknown` at the + // Skip the parse when the caller (the scope-resolution orchestrator's + // `treeCache`) already produced a Tree for this source — empty under + // worker-pool runs, so cache miss = re-parse. The cachedTree parameter + // is typed as `unknown` at the // contract layer (see `LanguageProvider.emitScopeCaptures`); cast // here at the use site. let tree = cachedTree as ReturnType['parse']> | undefined; diff --git a/gitnexus/src/core/ingestion/pipeline-phases/orm-extraction.ts b/gitnexus/src/core/ingestion/pipeline-phases/orm-extraction.ts deleted file mode 100644 index c6b46610e6..0000000000 --- a/gitnexus/src/core/ingestion/pipeline-phases/orm-extraction.ts +++ /dev/null @@ -1,106 +0,0 @@ -/** - * Inline ORM query extraction (sequential fallback path). - * - * Extracts Prisma and Supabase query calls from source content using - * regex patterns. Used by the sequential parse path when workers are - * not available — the worker path extracts ORM queries via tree-sitter - * queries instead. - * - * @module - */ - -import type { ExtractedORMQuery } from '../workers/parse-worker.js'; - -// ── Regex patterns ───────────────────────────────────────────────────────── - -/** Matches Prisma client method calls: `prisma.user.findMany(...)` */ -const PRISMA_QUERY_RE = - /\bprisma\.(\w+)\.(findMany|findFirst|findUnique|findUniqueOrThrow|findFirstOrThrow|create|createMany|update|updateMany|delete|deleteMany|upsert|count|aggregate|groupBy)\s*\(/g; - -/** Matches Supabase client method calls: `supabase.from('users').select(...)` */ -const SUPABASE_QUERY_RE = - /\bsupabase\.from\s*\(\s*['"](\w+)['"]\s*\)\s*\.(select|insert|update|delete|upsert)\s*\(/g; - -// ── Extraction function ─────────────────────────────────────────────────── - -/** - * Extract ORM query calls from file content using regex. - * - * Fast-path: skips files that don't contain `prisma.` or `supabase.from`. - * Results are appended to the `out` array (push pattern avoids allocation). - * - * @param filePath Relative path of the source file - * @param content File content string - * @param out Output array to append extracted queries to - */ -export function extractORMQueriesInline( - filePath: string, - content: string, - out: ExtractedORMQuery[], -): void { - const hasPrisma = content.includes('prisma.'); - const hasSupabase = content.includes('supabase.from'); - if (!hasPrisma && !hasSupabase) return; - - // Pre-compute line number offsets to avoid O(n²) substring+split per match - const lineOffsets = buildLineOffsets(content); - - if (hasPrisma) { - PRISMA_QUERY_RE.lastIndex = 0; - let m; - while ((m = PRISMA_QUERY_RE.exec(content)) !== null) { - const model = m[1]; - if (model.startsWith('$')) continue; - out.push({ - filePath, - orm: 'prisma', - model, - method: m[2], - lineNumber: lineNumberAtOffset(lineOffsets, m.index), - }); - } - } - - if (hasSupabase) { - SUPABASE_QUERY_RE.lastIndex = 0; - let m; - while ((m = SUPABASE_QUERY_RE.exec(content)) !== null) { - out.push({ - filePath, - orm: 'supabase', - model: m[1], - method: m[2], - lineNumber: lineNumberAtOffset(lineOffsets, m.index), - }); - } - } -} - -// ── Line offset helpers ─────────────────────────────────────────────────── - -/** Build an array of byte offsets where each newline occurs (O(n) once). */ -function buildLineOffsets(content: string): number[] { - const offsets: number[] = []; - for (let i = 0; i < content.length; i++) { - if (content[i] === '\n') offsets.push(i); - } - return offsets; -} - -/** - * Binary search for 0-based line number at a given character offset. - * - * Returns the number of newlines that occur before `offset` in the content, - * which is the 0-based line number. When `offset` is beyond the last newline, - * returns `lineOffsets.length` (i.e., the last line index). - */ -function lineNumberAtOffset(lineOffsets: number[], offset: number): number { - let lo = 0; - let hi = lineOffsets.length; - while (lo < hi) { - const mid = (lo + hi) >>> 1; - if (lineOffsets[mid] < offset) lo = mid + 1; - else hi = mid; - } - return lo; -} diff --git a/gitnexus/src/core/ingestion/pipeline-phases/parse-impl.ts b/gitnexus/src/core/ingestion/pipeline-phases/parse-impl.ts index 86bb5e5879..244f178dd5 100644 --- a/gitnexus/src/core/ingestion/pipeline-phases/parse-impl.ts +++ b/gitnexus/src/core/ingestion/pipeline-phases/parse-impl.ts @@ -43,7 +43,6 @@ import { type ExportedTypeMap, } from '../call-processor.js'; import { createSemanticModel, type MutableSemanticModel } from '../model/index.js'; -import { createASTCache } from '../ast-cache.js'; import { type PipelineProgress, getLanguageFromFilename } from 'gitnexus-shared'; import { readFileContents } from '../filesystem-walker.js'; import { isLanguageAvailable } from '../../tree-sitter/parser-loader.js'; @@ -466,14 +465,6 @@ export async function runChunkedParseAndResolve( let filesParsedSoFar = 0; - // Chunk-local tree-sitter cache, cleared between chunks — call / heritage / - // import processors read it during parse to avoid re-parsing within the same - // chunk. (The former cross-phase `scopeTreeCache` was only ever populated by - // the sequential parser, which has been removed; workers can't return native - // Trees across the MessageChannel, so scope-resolution re-parses as needed.) - const maxChunkFiles = chunks.reduce((max, c) => Math.max(max, c.length), 0); - const astCache = createASTCache(maxChunkFiles); - const exportedTypeMap: ExportedTypeMap = new Map(); const bindingAccumulator = new BindingAccumulator(); const allFetchCalls: ExtractedFetchCall[] = []; @@ -649,7 +640,6 @@ export async function runChunkedParseAndResolve( } filesParsedSoFar += chunkFiles.length; - astCache.clear(); if (verboseThroughputLog && chunkStartMs !== null) { const elapsedMs = Date.now() - chunkStartMs; @@ -943,7 +933,6 @@ export async function runChunkedParseAndResolve( // Finalize the accumulator and propagate any fixpoint-inferred exports before // `crossFile` disposes it downstream. Wrapped in try/catch so a cleanup // failure never masks a real parse error; disposal stays with `crossFile`. - astCache.clear(); try { bindingAccumulator.finalize(); const enriched = enrichExportedTypeMap(bindingAccumulator, graph, exportedTypeMap); diff --git a/gitnexus/src/core/ingestion/workers/parse-worker.ts b/gitnexus/src/core/ingestion/workers/parse-worker.ts index c5763a66d5..9a3837494e 100644 --- a/gitnexus/src/core/ingestion/workers/parse-worker.ts +++ b/gitnexus/src/core/ingestion/workers/parse-worker.ts @@ -2434,7 +2434,21 @@ parentPort!.on('message', (msg: WorkerIncomingMessage) => { return; } } catch (err) { - const message = err instanceof Error ? err.message : String(err); - parentPort!.postMessage({ type: 'error', error: message }); + // Carry the worker-side stack across the MessageChannel, not just the + // message. Without this, an unexpected worker throw (e.g. the minified + // `this.# is not a function` family) reaches the operator as a bare + // one-liner with no file:line — exactly what made #2068 undebuggable. The + // pool embeds `errorStack` into its death/circuit-breaker reason so the + // surfaced "Phase 'parse' failed" message points at the real frame (the + // stack's first line already carries the error's type + message). We send + // primitive fields (not the raw Error) so a non-cloneable `cause` payload + // can never turn the report itself into a `messageerror`. `errorStack` is + // optional on the wire, so an older pool ignores it. + const e = err instanceof Error ? err : new Error(String(err)); + parentPort!.postMessage({ + type: 'error', + error: e.message, + errorStack: e.stack, + }); } }); diff --git a/gitnexus/src/core/ingestion/workers/worker-pool.ts b/gitnexus/src/core/ingestion/workers/worker-pool.ts index 93fb6863d5..5082453a0a 100644 --- a/gitnexus/src/core/ingestion/workers/worker-pool.ts +++ b/gitnexus/src/core/ingestion/workers/worker-pool.ts @@ -310,7 +310,15 @@ type WorkerOutgoingMessage = | { type: 'progress'; filesProcessed: number } | { type: 'warning'; message: string } | { type: 'sub-batch-done' } - | { type: 'error'; error: string } + /** + * Worker-side caught error. `error` is the message; `errorStack` carries the + * worker thread's stack so the pool can embed a real file:line into its + * death / circuit-breaker reason instead of surfacing a bare one-liner (the + * #2068 diagnosability gap). `errorStack` is optional so an older worker + * build that only sends `error` still validates and degrades to message-only + * — and a newer pool reading it just gets no stack. + */ + | { type: 'error'; error: string; errorStack?: string } | { type: 'result'; data: unknown } /** * Authoritative in-flight signal: worker is about to process this file. @@ -656,6 +664,25 @@ function withStderr(worker: Worker, message: string): string { return tail ? `${message}. Worker stderr:\n${tail}` : message; } +/** + * Build a worker-death reason string that carries the worker-side stack when one + * is available (#2068). The stack is appended AFTER the `Worker N error: ` + * prefix so every prefix/substring consumer downstream — recoverAndResume → + * handleWorkerDeath → the circuit-breaker `WorkerPoolDispatchError` message, and + * the tests that regex-match those — keeps working unchanged, while the operator + * now gets the real frame instead of a bare one-liner. The stack's first line is + * normally the message itself; keeping both is harmless and the indented block + * scans cleanly in a log. The stack is capped at WORKER_STDERR_TAIL_LIMIT, + * mirroring the sibling stderr-tail bound, so a pathological error type (or a + * raised `Error.stackTraceLimit`) can't bloat the death reason. `stack` is + * `undefined` for an older worker build (or a thrown non-Error), in which case + * the reason is exactly the prior message-only form. + */ +function workerErrorReason(workerIndex: number, message: string, stack?: string): string { + const base = `Worker ${workerIndex} error: ${message}`; + return stack ? `${base}\n worker stack:\n${stack.slice(0, WORKER_STDERR_TAIL_LIMIT)}` : base; +} + /** * Wait for a freshly-spawned replacement worker to emit the * `{type:'ready'}` handshake from `parse-worker.ts` before treating its @@ -1832,7 +1859,7 @@ export const createWorkerPool = ( settled = true; cleanup(); void recoverAndResume( - `Worker ${workerIndex} error: ${msg.error}`, + workerErrorReason(workerIndex, msg.error, msg.errorStack), resolveExcludePaths(), ); } else if (msg.type === 'result') { @@ -1877,8 +1904,13 @@ export const createWorkerPool = ( if (!settled) { settled = true; cleanup(); + // The Node 'error' event fires on an UNCAUGHT worker throw (one that + // escaped the worker's own try/catch, or an async rejection). Unlike + // the `{type:'error'}` message, the event delivers a real Error whose + // `.stack` is the worker-side frame — carry it so the surfaced reason + // points at the actual failure site, not just `err.message` (#2068). void recoverAndResume( - `Worker ${workerIndex} error: ${err.message}`, + workerErrorReason(workerIndex, err.message, err.stack), resolveExcludePaths(), ); } diff --git a/gitnexus/src/core/lbug/lbug-adapter.ts b/gitnexus/src/core/lbug/lbug-adapter.ts index a41a9d2011..51f5ce860f 100644 --- a/gitnexus/src/core/lbug/lbug-adapter.ts +++ b/gitnexus/src/core/lbug/lbug-adapter.ts @@ -7,6 +7,7 @@ import path from 'path'; import os from 'os'; import crypto from 'crypto'; import lbug from '@ladybugdb/core'; +import { closeQueryResults } from './query-result-utils.js'; import { KnowledgeGraph } from '../graph/types.js'; import { NODE_TABLES, @@ -213,8 +214,18 @@ const DB_LOCK_RETRY_DELAY_MS = 500; * analyze` and either already happened or will happen on the next run. */ export const isReadOnlyDbError = (err: unknown): boolean => { - const msg = err instanceof Error ? err.message : String(err); - return /read-only database/i.test(msg); + // Walk the `cause` chain (bounded) so a wrapped read-only error — e.g. the + // pool adapter's `new Error('…read-only.', { cause: nativeReadOnlyErr })` — + // is still detected by callers that only see the wrapper (#2068 follow-up). + // The same strict regex is re-applied at each level, so a non-read-only + // chain stays false; the depth bound guards a cyclic `cause`. + let cur: unknown = err; + for (let depth = 0; depth < 5 && cur != null; depth++) { + const msg = cur instanceof Error ? cur.message : String(cur); + if (/read-only database/i.test(msg)) return true; + cur = cur instanceof Error ? (cur as { cause?: unknown }).cause : undefined; + } + return false; }; const isMissingFileError = (err: unknown): boolean => { @@ -392,12 +403,10 @@ const runWithSessionLock = async (operation: () => Promise): Promise => const normalizeCopyPath = (filePath: string): string => toNativeSafePath(filePath).replace(/\\/g, '/'); +// Single-result convenience wrapper over the shared best-effort closer +// (drainQueryResult / readQueryRows close one cursor at a time). const closeQueryResult = async (result: lbug.QueryResult): Promise => { - try { - await result.close(); - } catch { - // Best-effort cleanup only. - } + await closeQueryResults(result); }; const drainQueryResult = async ( @@ -1758,8 +1767,9 @@ export const queryImporters = async (targetFilePath: string): Promise WHERE r.type = 'IMPORTS' AND b.filePath = '${escaped}' RETURN DISTINCT a.filePath AS importer `; + let queryResult: lbug.QueryResult | lbug.QueryResult[] | undefined; try { - const queryResult = await conn.query(cypher); + queryResult = await conn.query(cypher); const result = Array.isArray(queryResult) ? queryResult[0] : queryResult; const rows = await result.getAll(); const out: string[] = []; @@ -1770,6 +1780,8 @@ export const queryImporters = async (targetFilePath: string): Promise return out; } catch { return []; + } finally { + if (queryResult) await closeQueryResults(queryResult); } }; @@ -1788,8 +1800,9 @@ export const deleteAllCommunitiesAndProcesses = async (): Promise<{ } let nodesDeleted = 0; for (const label of ['Community', 'Process']) { + let countResult: lbug.QueryResult | lbug.QueryResult[] | undefined; try { - const countResult = await conn.query(`MATCH (n:${label}) RETURN count(n) AS cnt`); + countResult = await conn.query(`MATCH (n:${label}) RETURN count(n) AS cnt`); const result = Array.isArray(countResult) ? countResult[0] : countResult; const rows = await result.getAll(); const count = Number(rows[0]?.cnt ?? rows[0]?.[0] ?? 0); @@ -1799,6 +1812,8 @@ export const deleteAllCommunitiesAndProcesses = async (): Promise<{ } } catch { // Table may not exist yet on a freshly-initialized DB — fine. + } finally { + if (countResult) await closeQueryResults(countResult); } } return { nodesDeleted }; diff --git a/gitnexus/src/core/lbug/pool-adapter.ts b/gitnexus/src/core/lbug/pool-adapter.ts index 1b868d8c1f..030688d145 100644 --- a/gitnexus/src/core/lbug/pool-adapter.ts +++ b/gitnexus/src/core/lbug/pool-adapter.ts @@ -18,6 +18,7 @@ import fs from 'fs/promises'; import lbug from '@ladybugdb/core'; import { isReadOnlyDbError, loadFTSExtension } from './lbug-adapter.js'; +import { closeQueryResults } from './query-result-utils.js'; import { createLbugDatabase, isWalCorruptionError, @@ -41,8 +42,14 @@ interface PoolEntry { available: lbug.Connection[]; /** Number of connections currently checked out */ checkedOut: number; - /** Queued waiters for when all connections are busy */ - waiters: Array<(conn: lbug.Connection) => void>; + /** Queued waiters for when all connections are busy. Each carries `resolve` + * (hand off a freed connection) and `reject` (fail fast when the pool is + * closed before a connection frees, instead of hanging until the waiter + * timeout — #2068 follow-up). */ + waiters: Array<{ + resolve: (conn: lbug.Connection) => void; + reject: (err: Error) => void; + }>; lastUsed: number; dbPath: string; /** Set to true when the pool entry is closed — checkin will close orphaned connections */ @@ -176,6 +183,20 @@ function closeOne(repoId: string): void { entry.closed = true; + // Reject any callers still queued for a connection: the pool is going away + // (re-init / teardown / LRU eviction), so they must fail fast with an + // actionable error instead of hanging until WAITER_TIMEOUT_MS and then + // surfacing a misleading "pool exhausted" (#2068 follow-up). Draining the + // queue also guarantees checkin() below finds no waiter expecting a + // connection, so a connection returned after close is simply closed. + if (entry.waiters.length > 0) { + const closedErr = new Error( + `LadybugDB connection pool closed for repo "${repoId}" (re-init/teardown); retry the query.`, + ); + for (const waiter of entry.waiters) waiter.reject(closedErr); + entry.waiters.length = 0; + } + // Close available connections — fire-and-forget with .catch() to prevent // unhandled rejections. Native close() returns Promise but can crash // the N-API destructor on macOS/Windows; deferring to process exit lets @@ -680,14 +701,20 @@ function checkout(entry: PoolEntry): Promise { // At capacity — queue the caller with a timeout. return new Promise((resolve, reject) => { - const waiter = (conn: lbug.Connection) => { - clearTimeout(timer); - resolve(conn); + const waiter = { + resolve: (conn: lbug.Connection) => { + clearTimeout(timer); + resolve(conn); + }, + reject: (err: Error) => { + clearTimeout(timer); + reject(err); + }, }; const timer = setTimeout(() => { const idx = entry.waiters.indexOf(waiter); if (idx !== -1) entry.waiters.splice(idx, 1); - reject( + waiter.reject( new Error( `Connection pool exhausted: timed out after ${WAITER_TIMEOUT_MS}ms waiting for a free connection`, ), @@ -713,7 +740,7 @@ function checkin(entry: PoolEntry, conn: lbug.Connection): void { if (entry.waiters.length > 0) { // Hand directly to the next waiter — no intermediate available state const waiter = entry.waiters.shift()!; - waiter(conn); + waiter.resolve(conn); } else { entry.checkedOut--; entry.available.push(conn); @@ -756,22 +783,33 @@ export const executeParameterized = async ( const conn = await checkout(entry); silenceStdout(); activeQueryCount++; + 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 queryResult = await withTimeout(conn.execute(stmt, params), QUERY_TIMEOUT_MS, 'Execute'); + 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)) { - throw new Error('Write operations are not allowed. The pool adapter is read-only.'); + // 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); activeQueryCount--; restoreStdout(); checkin(entry, conn); diff --git a/gitnexus/src/core/lbug/query-result-utils.ts b/gitnexus/src/core/lbug/query-result-utils.ts new file mode 100644 index 0000000000..938e28a2b3 --- /dev/null +++ b/gitnexus/src/core/lbug/query-result-utils.ts @@ -0,0 +1,31 @@ +import lbug from '@ladybugdb/core'; + +/** + * Best-effort close of one or more native `QueryResult` cursors. + * + * `result.getAll()` materializes rows into a JS array but does not release the + * native cursor — leaving it open holds native resources for the connection's + * lifetime. Both the pooled adapter (`pool-adapter.ts`) and the direct adapter + * (`lbug-adapter.ts`) must release cursors after reading; this is the single + * shared implementation so neither re-rolls the array-normalize + swallow loop + * (a #2068 follow-up de-dup). Lives in its own leaf module so the two adapters + * depend on it rather than on each other. + * + * `conn.execute()` can return either a single `QueryResult` or an array + * (multi-statement); this normalizes both. Each close is independent and + * best-effort: a failing or absent `close()` on one cursor never throws and + * never prevents the others from closing — a cleanup failure must not mask the + * query result or a real error at the call site. + */ +export async function closeQueryResults( + queryResult: lbug.QueryResult | lbug.QueryResult[], +): Promise { + const results = Array.isArray(queryResult) ? queryResult : [queryResult]; + for (const r of results) { + try { + await r?.close(); + } catch { + // Best-effort cleanup only. + } + } +} diff --git a/gitnexus/src/mcp/local/local-backend.ts b/gitnexus/src/mcp/local/local-backend.ts index 46ecf40308..77411b1078 100644 --- a/gitnexus/src/mcp/local/local-backend.ts +++ b/gitnexus/src/mcp/local/local-backend.ts @@ -177,8 +177,19 @@ function logQueryError(context: string, err: unknown): void { logger.error({ context, err: msg }, 'GitNexus query failed'); } -const isReadOnlyDbError = (err: unknown): boolean => - /read-only database/i.test(err instanceof Error ? err.message : String(err)); +const isReadOnlyDbError = (err: unknown): boolean => { + // Walk the `cause` chain (bounded) so a wrapped read-only error (e.g. the + // pool adapter's `{ cause }` wrapper) is still detected here — this is the + // copy the MCP cypher handler uses to surface its curated read-only message + // (#2068 follow-up). Mirrors lbug-adapter's isReadOnlyDbError. + let cur: unknown = err; + for (let depth = 0; depth < 5 && cur != null; depth++) { + const msg = cur instanceof Error ? cur.message : String(cur); + if (/read-only database/i.test(msg)) return true; + cur = cur instanceof Error ? (cur as { cause?: unknown }).cause : undefined; + } + return false; +}; /** * Per-query latency telemetry for production aggregation (#553). diff --git a/gitnexus/src/types/pipeline.ts b/gitnexus/src/types/pipeline.ts index becf817577..5ad08cae64 100644 --- a/gitnexus/src/types/pipeline.ts +++ b/gitnexus/src/types/pipeline.ts @@ -19,9 +19,12 @@ export interface PipelineResult { */ resolutionOutcomes: readonly ResolutionOutcome[]; /** - * True if the parse phase spawned a worker pool for this run. False means - * the sequential fallback handled every chunk. Primarily a test affordance - * so regression suites can prove which path executed. + * True if a worker pool was actually constructed for this run. The worker + * pool is the sole parse path (sequential parsing was removed). False means + * no pool was needed: either there were no parseable files, or every chunk + * was a parse-cache hit and the cached worker output was replayed without + * spawning workers (a warm all-cache-hit run, #2038). Primarily a test + * affordance so regression suites can prove the pool engaged. */ usedWorkerPool: boolean; } diff --git a/gitnexus/test/integration/lbug-pool.test.ts b/gitnexus/test/integration/lbug-pool.test.ts index 259c15c13f..561772a753 100644 --- a/gitnexus/test/integration/lbug-pool.test.ts +++ b/gitnexus/test/integration/lbug-pool.test.ts @@ -94,6 +94,86 @@ withTestLbugDB( }); }); + // ─── closeLbug rejects pending waiters (#2068 follow-up) ───────────── + // + // Before the fix, closeOne() never rejected queued waiters: a caller + // waiting for a free connection when the pool was closed (e.g. a staleness + // reinit under concurrent query load) hung for WAITER_TIMEOUT_MS (15s) and + // then surfaced a misleading "pool exhausted" error. Now they reject + // immediately with an actionable "pool closed" message. The pool caps at + // MAX_CONNS_PER_REPO (8); firing a synchronous burst larger than that queues + // the surplus as waiters, and closing synchronously (before any query + // settles) must reject every queued waiter at once. The default 5s test + // timeout also guards promptness — a regression would block ~15s and time + // out rather than reject. + describe('closeLbug waiter handling (#2068)', () => { + it('rejects queued waiters promptly with a pool-closed error on close', async () => { + await initLbug('test-repo', handle.dbPath); + + // Fire a burst larger than the 8-connection cap WITHOUT awaiting: the + // first 8 check out connections synchronously, the surplus queue as + // waiters — all before the synchronous closeLbug below runs. + const BURST = 24; + const MAX_CONNS = 8; + const inflight = Array.from({ length: BURST }, () => + executeQuery('test-repo', 'MATCH (n:Function) RETURN n.name AS name'), + ); + // Close in the same synchronous tick — no microtask has served a waiter. + const closing = closeLbug('test-repo'); + + const settled = await Promise.allSettled(inflight); + await closing; + + const reasons = settled + .filter((r): r is PromiseRejectedResult => r.status === 'rejected') + .map((r) => String(r.reason?.message ?? r.reason)); + + // The surplus (BURST - MAX_CONNS) waiters must reject with "pool closed". + const poolClosed = reasons.filter((m) => /pool closed/i.test(m)); + expect(poolClosed.length).toBeGreaterThanOrEqual(BURST - MAX_CONNS); + // And none should have hit the 15s "exhausted" waiter-timeout path. + expect(reasons.some((m) => /waiting for a free connection/i.test(m))).toBe(false); + + expect(isLbugReady('test-repo')).toBe(false); + }); + + it('settles in-flight queries and fully tears down when closed mid-flight', async () => { + // closeOne-vs-checkin interleave (F4b): with 8 connections in-flight and + // surplus callers queued, a synchronous close must (a) let every promise + // settle — no hang — and (b) fully delete the pool entry so checked-in + // connections are closed as orphans rather than handed to a rejected + // waiter. We assert the observable contract; the "orphan not handed to a + // rejected waiter" invariant is single-threaded-by-construction (closeOne + // drains waiters with no await before any checkin can run). + await initLbug('test-repo', handle.dbPath); + + const inflight = Array.from({ length: 16 }, () => + executeQuery('test-repo', 'MATCH (n:Function) RETURN n.name AS name'), + ); + const closing = closeLbug('test-repo'); + + // allSettled only resolves once EVERY query settled — proving none hangs + // (a 15s waiter-timeout regression would blow the default test timeout). + const settled = await Promise.allSettled(inflight); + await closing; + expect(settled).toHaveLength(16); + expect( + settled.some( + (r) => + r.status === 'rejected' && + /waiting for a free connection/i.test(String(r.reason?.message ?? r.reason)), + ), + ).toBe(false); + + // Pool entry fully gone — a subsequent query fails fast with the + // not-initialized error, not a hang or a stale connection. + expect(isLbugReady('test-repo')).toBe(false); + await expect(executeQuery('test-repo', 'MATCH (n) RETURN n LIMIT 1')).rejects.toThrow( + /not initialized/i, + ); + }); + }); + // ─── Parameterized queries ─────────────────────────────────────────── describe('executeParameterized', () => { diff --git a/gitnexus/test/unit/ast-cache.test.ts b/gitnexus/test/unit/ast-cache.test.ts deleted file mode 100644 index 66cc6e3c3f..0000000000 --- a/gitnexus/test/unit/ast-cache.test.ts +++ /dev/null @@ -1,95 +0,0 @@ -import { describe, it, expect, beforeEach } from 'vitest'; -import { createASTCache, type ASTCache } from '../../src/core/ingestion/ast-cache.js'; - -// Create a minimal mock tree object (mimics Parser.Tree interface) -function mockTree(id: string): any { - return { rootNode: { type: 'program', text: id }, delete: vi.fn() }; -} - -describe('ASTCache', () => { - let cache: ASTCache; - - beforeEach(() => { - cache = createASTCache(3); - }); - - describe('get / set', () => { - it('returns undefined for cache miss', () => { - expect(cache.get('nonexistent.ts')).toBeUndefined(); - }); - - it('returns cached tree on hit', () => { - const tree = mockTree('test'); - cache.set('src/index.ts', tree); - expect(cache.get('src/index.ts')).toBe(tree); - }); - - it('overwrites existing entry for same key', () => { - const tree1 = mockTree('v1'); - const tree2 = mockTree('v2'); - cache.set('src/index.ts', tree1); - cache.set('src/index.ts', tree2); - expect(cache.get('src/index.ts')).toBe(tree2); - }); - }); - - describe('LRU eviction', () => { - it('evicts least recently used when capacity exceeded', () => { - cache.set('a.ts', mockTree('a')); - cache.set('b.ts', mockTree('b')); - cache.set('c.ts', mockTree('c')); - // Cache is full (maxSize=3). Adding one more evicts 'a' - cache.set('d.ts', mockTree('d')); - expect(cache.get('a.ts')).toBeUndefined(); - expect(cache.get('b.ts')).toBeDefined(); - expect(cache.get('d.ts')).toBeDefined(); - }); - - it('accessing an entry makes it recently used', () => { - cache.set('a.ts', mockTree('a')); - cache.set('b.ts', mockTree('b')); - cache.set('c.ts', mockTree('c')); - // Touch 'a' to make it recently used - cache.get('a.ts'); - // Now 'b' is LRU - cache.set('d.ts', mockTree('d')); - expect(cache.get('a.ts')).toBeDefined(); - expect(cache.get('b.ts')).toBeUndefined(); - }); - }); - - describe('clear', () => { - it('removes all entries', () => { - cache.set('a.ts', mockTree('a')); - cache.set('b.ts', mockTree('b')); - cache.clear(); - expect(cache.get('a.ts')).toBeUndefined(); - expect(cache.get('b.ts')).toBeUndefined(); - expect(cache.stats().size).toBe(0); - }); - }); - - describe('stats', () => { - it('reports size and maxSize', () => { - expect(cache.stats()).toEqual({ size: 0, maxSize: 3 }); - cache.set('a.ts', mockTree('a')); - expect(cache.stats()).toEqual({ size: 1, maxSize: 3 }); - cache.set('b.ts', mockTree('b')); - expect(cache.stats()).toEqual({ size: 2, maxSize: 3 }); - }); - - it('uses default maxSize of 50', () => { - const defaultCache = createASTCache(); - expect(defaultCache.stats().maxSize).toBe(50); - }); - - it('clamps maxSize of 0 to 1 to prevent LRU cache error', () => { - const zeroCache = createASTCache(0); - expect(zeroCache.stats().maxSize).toBe(1); - // Should still function correctly - const tree = mockTree('test'); - zeroCache.set('a.ts', tree); - expect(zeroCache.get('a.ts')).toBe(tree); - }); - }); -}); diff --git a/gitnexus/test/unit/lbug-query-result-utils.test.ts b/gitnexus/test/unit/lbug-query-result-utils.test.ts new file mode 100644 index 0000000000..9a7ac3ad87 --- /dev/null +++ b/gitnexus/test/unit/lbug-query-result-utils.test.ts @@ -0,0 +1,45 @@ +/** + * Unit tests: closeQueryResults — the shared best-effort cursor-close helper + * (#2068 follow-up). Guards the contract the two LadybugDB adapters rely on: + * a single result OR an array of results are all closed, and a failing/absent + * `close()` never throws and never skips the rest. + */ +import { describe, it, expect, vi } from 'vitest'; +import { closeQueryResults } from '../../src/core/lbug/query-result-utils.js'; + +// Minimal QueryResult stand-in — only `close()` matters here. +function fakeResult(close: () => unknown = () => undefined) { + return { close: vi.fn(close) } as unknown as import('@ladybugdb/core').QueryResult; +} + +describe('closeQueryResults', () => { + it('closes a single QueryResult exactly once', async () => { + const r = fakeResult(); + await closeQueryResults(r); + expect(r.close as ReturnType).toHaveBeenCalledTimes(1); + }); + + it('closes EVERY element of an array (not just the first)', async () => { + const rs = [fakeResult(), fakeResult(), fakeResult()]; + await closeQueryResults(rs); + for (const r of rs) { + expect(r.close as ReturnType).toHaveBeenCalledTimes(1); + } + }); + + it('keeps closing the rest when one close() rejects (best-effort, no throw)', async () => { + const ok1 = fakeResult(); + const bad = fakeResult(() => { + throw new Error('native close failed'); + }); + const ok2 = fakeResult(() => Promise.reject(new Error('async close failed'))); + const ok3 = fakeResult(); + await expect(closeQueryResults([ok1, bad, ok2, ok3])).resolves.toBeUndefined(); + expect(ok1.close as ReturnType).toHaveBeenCalledTimes(1); + expect(ok3.close as ReturnType).toHaveBeenCalledTimes(1); + }); + + it('is a no-op on an empty array', async () => { + await expect(closeQueryResults([])).resolves.toBeUndefined(); + }); +}); diff --git a/gitnexus/test/unit/lbug-readonly-error.test.ts b/gitnexus/test/unit/lbug-readonly-error.test.ts index 4e412c323d..514ddc07d1 100644 --- a/gitnexus/test/unit/lbug-readonly-error.test.ts +++ b/gitnexus/test/unit/lbug-readonly-error.test.ts @@ -41,6 +41,34 @@ describe('isReadOnlyDbError', () => { expect(isReadOnlyDbError(undefined)).toBe(false); }); + it('detects a read-only error wrapped as a `cause` (#2068 — pool-adapter wrapper)', () => { + // The pool adapter rethrows native read-only failures as a friendly + // message with the original error preserved on `cause`. Detection must see + // through the wrapper so the MCP/HTTP handlers surface their curated message. + const wrapped = new Error('Write operations are not allowed. The pool adapter is read-only.', { + cause: new Error('Cannot execute write operations in a read-only database!'), + }); + expect(isReadOnlyDbError(wrapped)).toBe(true); + }); + + it('does NOT match a wrapper whose cause chain is unrelated', () => { + const wrapped = new Error('Query failed', { cause: new Error('Connection refused') }); + expect(isReadOnlyDbError(wrapped)).toBe(false); + }); + + it('terminates on a cyclic cause chain without hanging', () => { + const a = new Error('boom a') as Error & { cause?: unknown }; + const b = new Error('boom b') as Error & { cause?: unknown }; + a.cause = b; + b.cause = a; + expect(isReadOnlyDbError(a)).toBe(false); + }); + + it('stops cleanly on a non-Error cause', () => { + const wrapped = new Error('outer', { cause: 'just a string' }); + expect(isReadOnlyDbError(wrapped)).toBe(false); + }); + it('does NOT match unrelated errors that the ensure path must still surface', () => { // Lock contention — handled separately by isDbBusyError; must not be // silenced by the read-only filter. diff --git a/gitnexus/test/unit/worker-pool-error-stack.test.ts b/gitnexus/test/unit/worker-pool-error-stack.test.ts new file mode 100644 index 0000000000..fbd828142a --- /dev/null +++ b/gitnexus/test/unit/worker-pool-error-stack.test.ts @@ -0,0 +1,176 @@ +import { describe, expect, it, beforeEach, afterEach } from 'vitest'; +import { EventEmitter } from 'node:events'; +import path from 'node:path'; +import { pathToFileURL } from 'node:url'; +import fs from 'node:fs'; +import os from 'node:os'; + +import { + createWorkerPool, + WorkerPoolDispatchError, +} from '../../src/core/ingestion/workers/worker-pool.js'; + +/** + * #2068 regression: a worker-side crash must carry its stack across the + * MessageChannel so the surfaced "Phase 'parse' failed" error points at a real + * frame instead of a bare one-liner (the issue's `this.#q is not a function` + * reached the operator with no file:line because the worker only sent + * `err.message`). These tests assert the worker stack rides through both worker + * failure channels — the `{type:'error'}` message (a caught worker throw) and + * the Node `'error'` event (an uncaught worker throw) — into the + * `WorkerPoolDispatchError` the parse phase rejects with, and that an older + * worker build that omits the stack still degrades cleanly. + */ + +type NodeWorker = import('node:worker_threads').Worker; + +type FakeAction = + | { kind: 'error-message'; error: string; errorStack?: string } + | { kind: 'error-event'; message: string; stack: string }; + +const nextActions: FakeAction[] = []; + +/** + * Minimal worker double: emits the readiness handshake on construction, then + * runs one scripted action per dispatched sub-batch. Unlike the resilience + * suite's double, this one can emit the `{type:'error', errorStack}` MESSAGE + * (the worker's own caught-error path) in addition to the Node `'error'` event. + */ +class FakeWorker extends EventEmitter { + constructor() { + super(); + queueMicrotask(() => { + this.emit('online'); + this.emit('message', { type: 'ready' }); + }); + } + + postMessage(rawMsg: unknown): void { + const m = rawMsg as { type?: string }; + // Only a real dispatch drives an action; ignore the pool's `flush` reply. + if (!m || m.type !== 'sub-batch') return; + const action = nextActions.shift(); + if (!action) return; + queueMicrotask(() => { + if (action.kind === 'error-message') { + this.emit('message', { + type: 'error', + error: action.error, + errorStack: action.errorStack, + }); + } else { + const e = new Error(action.message); + e.stack = action.stack; + this.emit('error', e); + } + }); + } + + async terminate(): Promise { + this.emit('exit', 0); + return 0; + } +} + +let tempDir: string; +let workerUrl: URL; + +beforeEach(() => { + nextActions.length = 0; + tempDir = fs.mkdtempSync(path.join(os.tmpdir(), 'gitnexus-worker-error-stack-')); + const workerPath = path.join(tempDir, 'fake-worker.js'); + fs.writeFileSync(workerPath, '// fake'); + workerUrl = pathToFileURL(workerPath) as URL; +}); + +afterEach(() => { + try { + fs.rmSync(tempDir, { recursive: true, force: true }); + } catch { + // best-effort cleanup + } +}); + +// Trip the breaker on the very first death so the reason string is surfaced +// verbatim in the rejected WorkerPoolDispatchError. +const TRIP_ON_FIRST_DEATH = { + consecutiveFailureThreshold: 1, + maxRespawnsPerSlot: 0, +} as const; + +async function dispatchAndCatch(pool: ReturnType): Promise { + try { + await pool.dispatch<{ path: string; content: string }, unknown>([ + { path: 'src/a.ts', content: '' }, + ]); + return undefined; + } catch (e) { + return e; + } +} + +describe('worker-pool error stack propagation (#2068)', () => { + it('embeds the worker stack from a {type:error} message into the surfaced error', async () => { + const workerStack = + 'TypeError: this.#q is not a function\n' + + ' at frobnicate (/dist/core/ingestion/workers/parse-worker.js:1234:56)'; + const pool = createWorkerPool(workerUrl, 1, { + workerFactory: () => new FakeWorker() as unknown as NodeWorker, + ...TRIP_ON_FIRST_DEATH, + }); + nextActions.push({ + kind: 'error-message', + error: 'this.#q is not a function', + errorStack: workerStack, + }); + + const caught = await dispatchAndCatch(pool); + + expect(caught).toBeInstanceOf(WorkerPoolDispatchError); + const msg = (caught as Error).message; + expect(msg).toContain('this.#q is not a function'); + expect(msg).toContain('worker stack:'); + expect(msg).toContain('frobnicate (/dist/core/ingestion/workers/parse-worker.js:1234:56)'); + await pool.terminate(); + }); + + // The Node 'error' event fires on an UNCAUGHT JS throw / async rejection (which + // carries a real JS stack). A true NATIVE abort (tree-sitter SIGSEGV / OOM kill) + // instead fires the 'exit' event and is intentionally stackless — no JS frame + // exists — so it is NOT exercised here. + it('embeds the worker stack from a Node error event (uncaught throw) into the surfaced error', async () => { + const workerStack = + 'Error: uncaught worker throw\n' + + ' at processFileGroup (/dist/core/ingestion/workers/parse-worker.js:777:9)'; + const pool = createWorkerPool(workerUrl, 1, { + workerFactory: () => new FakeWorker() as unknown as NodeWorker, + ...TRIP_ON_FIRST_DEATH, + }); + nextActions.push({ kind: 'error-event', message: 'uncaught worker throw', stack: workerStack }); + + const caught = await dispatchAndCatch(pool); + + expect(caught).toBeInstanceOf(WorkerPoolDispatchError); + const msg = (caught as Error).message; + expect(msg).toContain('worker stack:'); + expect(msg).toContain('processFileGroup (/dist/core/ingestion/workers/parse-worker.js:777:9)'); + await pool.terminate(); + }); + + it('degrades to message-only when an older worker build omits errorStack', async () => { + const pool = createWorkerPool(workerUrl, 1, { + workerFactory: () => new FakeWorker() as unknown as NodeWorker, + ...TRIP_ON_FIRST_DEATH, + }); + // No errorStack — the wire field is optional for back/forward compat. + nextActions.push({ kind: 'error-message', error: 'legacy worker failure' }); + + const caught = await dispatchAndCatch(pool); + + expect(caught).toBeInstanceOf(WorkerPoolDispatchError); + const msg = (caught as Error).message; + expect(msg).toContain('legacy worker failure'); + expect(msg).not.toContain('worker stack:'); + await pool.terminate(); + }); +});