From 9718e1247aec56745a32fbc19ae83f392f656f02 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gerg=C5=91=20Magyar?= Date: Sun, 30 Aug 2026 09:31:47 +0100 Subject: [PATCH] fix(parse): stabilize parse-cache chunks and cheapen ParsedFile loads (#3093) * fix(parse): stabilize parse-cache chunks and cheapen ParsedFile loads Hash-bucket membership so worker count and add/delete no longer reshuffle cache keys; GC and path sidecars keep small-shard scope-resolution from full-store JSON and empty GCs. Co-authored-by: Cursor * fix(review): apply review findings Drop the unused pool argument from cache-budget resolution, reuse path compare helpers, and copy durable sidecars via full shard paths. Co-authored-by: Cursor * fix(store): copy durable path sidecars via full shard paths Keep restore destinations relative to the run store even when sidecar names are derived from absolute json paths. Co-authored-by: Cursor * fix(store): fail closed on truncated ParsedFile path sidecars Skip JSON only when a sidecar is complete (NUL-free, trailing newline). Truncated listings without a NUL were able to omit wanted paths. Co-authored-by: Cursor * style: apply prettier to ParsedFile store and tests Match the PR autofix formatter so CI quality does not flag wrap-only diffs. Co-authored-by: Cursor * fix(store): tighten ParsedFile path sidecars from review Skip sidecar writes when a path contains CR/LF, and assert the skip path does not open non-intersecting JSON shards. Co-authored-by: Cursor * fix(store): yield on sidecar skips and assert restore copies listing bytes Skipped shards now count toward the 128-shard event-loop yield, and restore tests check sidecar contents rather than existence only. Co-authored-by: Cursor * fix(store): treat path sidecars as best-effort after a JSON shard write A sidecar ENOSPC/EACCES must not fail persist; load already falls back to the JSON shard when the listing is missing. Co-authored-by: Cursor * fix(store): drop stale path sidecars when a shard is no longer listing-safe Rewriting a shard with a newline-bearing path must unlink the old listing so load does not skip the JSON payload. Co-authored-by: Cursor * fix(parse): keep worker-integration tests aligned with hash buckets Quarantine cache-skip asserts the poison pack hash, clone-skip keeps poison and survivors in one bucket, and restore unlinks a stale dest sidecar when the durable source has none. Co-authored-by: Cursor * fix(parse): address review follow-ups for cache packs and sidecars Record SCHEMA_BUMP 80, pin pack locality and sidecar load/restore tests, and keep sidecar I/O best-effort with shared ENOENT handling. Co-authored-by: Cursor * fix(store): drop stale path sidecars after a failed listing write A leftover .paths file after ENOSPC (or similar) made load skip the new JSON shard. Hash expected packs with the same env budget production uses. Co-authored-by: Cursor * fix(store): drop path sidecars before overwriting parsed-file JSON Load trusts a leftover .paths listing, so rewriting a shard must unlink that listing first. Otherwise an interrupted sidecar refresh can hide newly written files. Co-authored-by: Cursor * chore(autofix): apply prettier + eslint fixes via /autofix command * fix(store): fail closed on truncated or CR path sidecars Count-prefix listings so a newline-terminated partial sidecar cannot skip the JSON shard, and reject CR instead of stripping it. Co-authored-by: Cursor * fix(ci): expect single-file watch refresh telemetry The production analyze --watch e2e was still pinned to the old pack-cascade "8 re-parsed" line, so shard 1/3 timed out after a correct 1-file refresh. Co-authored-by: Cursor * fix(ci): expect one reparsed file on a non-bean incremental touch Pack-cascade leftover: the drift-skip test still required 7 reparsed files after logger.ts-only edits. Cheap ParsedFile loads now reparse just that file. Co-authored-by: Cursor --------- Co-authored-by: Gergo Magyar Co-authored-by: Cursor Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> --- README.md | 2 +- .../ingestion/pipeline-phases/parse-impl.ts | 99 ++----- gitnexus/src/storage/parse-cache.ts | 81 ++++- gitnexus/src/storage/parsedfile-store.ts | 279 +++++++++++++++--- gitnexus/test/integration/cli-e2e.test.ts | 9 +- .../integration/parse-impl-clone-skip.test.ts | 29 +- .../integration/parse-impl-env-reads.test.ts | 39 ++- .../parse-impl-quarantine-cache-skip.test.ts | 72 +++-- .../unit/incremental-orchestration.test.ts | 2 +- .../test/unit/incremental-parse-cache.test.ts | 64 +++- gitnexus/test/unit/parsedfile-store.test.ts | 268 ++++++++++++++++- 11 files changed, 777 insertions(+), 167 deletions(-) diff --git a/README.md b/README.md index 173cd1d42..405d85201 100644 --- a/README.md +++ b/README.md @@ -546,7 +546,7 @@ Most `analyze` knobs are also CLI flags (`--workers`, `--worker-timeout`, `--max | `GITNEXUS_WORKER_CONSECUTIVE_FAILURE_THRESHOLD` | `max(3, poolSize)` | Per-slot consecutive deaths before the pool's circuit breaker trips. After tripping, every subsequent dispatch rejects until a fresh pool is created. | Hosts where a SIGSEGV-prone native grammar should trip the breaker sooner; CI runners that should fail loudly. | | `GITNEXUS_WORKER_SHUTDOWN_DRAIN_MS` | `30000` | Max wait at pool shutdown for a retired worker still inside native code. The worker is terminated at its next JS-safe point instead of mid-native-call (which aborts the whole process with `Napi::Error`, #2432); on expiry it is left running, unref'd, and terminated when it surfaces. | Shutdown latency matters more than draining a wedged worker (lower), or a legitimately-slow native grammar needs longer to surface (raise). | | `GITNEXUS_CPP_CAPTURE_BUDGET_MS` | `20000` | Per-file wall-clock budget for C++ capture extraction. On breach the file keeps the captures accumulated so far and logs a warning — the worker returns to JS instead of stalling in native-heavy loops (#2432). `0` expires immediately. | Pathological generated C++ that still exceeds the budget after the indexed lookups; raise for completeness, lower to fail-fast. | -| `GITNEXUS_CHUNK_BYTE_BUDGET` | `2097152` (2 MB) | Chunk boundary used for cache-key composition and dispatch. Smaller = finer-grained cache hits but more dispatch overhead. | Tuning incremental-analyze cache behavior on monorepos. | +| `GITNEXUS_CHUNK_BYTE_BUDGET` | `2097152` (2 MB) | Per-bucket byte budget for parse-cache packing. Files are grouped by `(language, hash(path) mod 128)`; packs inside a bucket are cut at this limit. Smaller = finer-grained invalidation and more dispatch. Default is always 2 MiB and no longer scales with worker count. | Tuning incremental-analyze cache invalidation on monorepos without changing `--workers`. | | `GITNEXUS_NO_GITIGNORE` | unset | When set, skips `.gitignore` parsing. `.gitnexusignore` is still honored. | Indexing a repo whose `.gitignore` excludes files you actually want indexed (e.g., generated code committed for cross-repo lookup). | | `GITNEXUS_SKIP_OPTIONAL_GRAMMARS` | unset | When `=1` strictly, skips the vendored grammar materialize for `tree-sitter-dart`, `tree-sitter-proto`, `tree-sitter-swift`, and `tree-sitter-kotlin` at install time (and the Dart/Proto source builds). Those four won't be parsed; the install still succeeds. | Installing on a host without a C++ toolchain or where the vendored prebuilds don't match; willing to skip Dart/Proto/Swift/Kotlin parsing. | | `GITNEXUS_MCP_READ_ONLY` | unset | Set to `1` to expose only proven single-repository read tools and resources; `0` disables the policy and any other value fails startup. | The MCP server runs in an environment where graph mutation, raw Cypher, and cross-repository group routing must be unavailable. | diff --git a/gitnexus/src/core/ingestion/pipeline-phases/parse-impl.ts b/gitnexus/src/core/ingestion/pipeline-phases/parse-impl.ts index 157e29027..b6ca75ea0 100644 --- a/gitnexus/src/core/ingestion/pipeline-phases/parse-impl.ts +++ b/gitnexus/src/core/ingestion/pipeline-phases/parse-impl.ts @@ -2,7 +2,7 @@ * Parse implementation — chunked parse + resolve loop. * * This is the core parsing engine of the ingestion pipeline. It reads - * source files in byte-budget chunks (~20MB each), parses via the worker + * source files in stable hash-bucket packs (~2MB each by default), parses via the worker * pool (the sole parse path — there is no sequential fallback), and emits * route CALLS edges. Import, * call, and inheritance resolution are owned by the scope-resolution @@ -27,6 +27,7 @@ import { loadParseCacheChunk, persistParseCacheChunk, PARSE_CACHE_VERSION, + packParseCacheChunks, } from '../../../storage/parse-cache.js'; import { clearParsedFileStore, @@ -192,39 +193,19 @@ export function heapPressureRemedy(heapLimitBytes: number): string { ); } -/** Max bytes of source content to load per parse chunk. +/** Max bytes of source content to load per parse cache pack. * - * Memory bound for the worker pool dispatch + a granularity knob for - * the parse cache. A single file change invalidates only its enclosing - * chunk, so smaller budgets → finer-grained invalidation. - * - * Override via GITNEXUS_CHUNK_BYTE_BUDGET (bytes) — the default of 2MB - * gives a useful invalidation floor (~1/N chunks on a multi-MB repo) - * while keeping worker dispatch overhead under 5% on cold runs. - */ -/** - * Built-in chunk byte budget when neither `PipelineOptions.chunkByteBudget` - * nor `GITNEXUS_CHUNK_BYTE_BUDGET` is set. Tuned to give a useful - * cache-invalidation floor (~1/N chunks on a multi-MB repo) while keeping - * worker dispatch overhead under 5% on cold runs. Resolution happens at - * call time inside `runChunkedParseAndResolve` (U14 from PR #1693 review) - * — previously this was a module-load IIFE, which froze the env value at - * import time and meant per-call option threading silently no-op'd. + * Granularity knob for the parse cache: a single file change invalidates only + * its enclosing pack. Override via GITNEXUS_CHUNK_BYTE_BUDGET. Resolution + * happens at call time (U14 from PR #1693) — not at module load. */ const DEFAULT_CHUNK_BYTE_BUDGET = 2 * 1024 * 1024; /** - * Per-worker share of a chunk's byte budget when auto-scaling (#worker-idle). - * - * A chunk is a single `WorkerPool.dispatch` unit; the pool fans a chunk's files - * into sub-batch jobs and assigns them to idle workers (`wakeIdleSlots`). When - * the chunk budget (2 MB) was far below the 8 MB sub-batch cap, every chunk - * produced exactly ONE job → ONE busy worker while the other N-1 sat idle. To - * keep all workers fed, the auto chunk budget now scales as - * `poolSize × CHUNK_BYTES_PER_WORKER`, so each dispatch carries enough work to - * fan across the whole pool. Sequential / explicit-budget runs are unaffected. + * Byte unit for auto pool sizing (one worker per this much source). Same + * magnitude as the default cache pack, but not a membership input (#3088). */ -const CHUNK_BYTES_PER_WORKER = 2 * 1024 * 1024; +const CHUNK_BYTES_PER_WORKER = DEFAULT_CHUNK_BYTE_BUDGET; /** * Target jobs-per-worker per dispatch. More jobs than workers gives the pool's @@ -236,14 +217,12 @@ const TARGET_JOBS_PER_WORKER = 3; /** Floor for a derived sub-batch so jobs don't shrink to per-file IPC churn. */ const MIN_SUB_BATCH_BYTES = 256 * 1024; -function resolveChunkByteBudget(options?: PipelineOptions, effectivePoolSize = 1): number { +function resolveChunkByteBudget(options?: PipelineOptions): number { const opt = options?.chunkByteBudget; if (typeof opt === 'number' && Number.isFinite(opt) && opt > 0) return opt; const env = Number(process.env.GITNEXUS_CHUNK_BYTE_BUDGET); if (Number.isFinite(env) && env > 0) return env; - // Auto: size each chunk so a dispatch can fan across the whole pool. A - // single-worker (tiny-repo) run keeps the original 2 MB invalidation floor. - return Math.max(DEFAULT_CHUNK_BYTE_BUDGET, effectivePoolSize * CHUNK_BYTES_PER_WORKER); + return DEFAULT_CHUNK_BYTE_BUDGET; } // ── Main parse + resolve function ────────────────────────────────────────── @@ -524,15 +503,6 @@ export async function runChunkedParseAndResolve( 0, ); - // Sort parseableScanned alphabetically for stable chunk membership - // across runs (Finding 4). Without this, filesystem-scan order can - // shift between runs (notably on macOS APFS where directory entry - // order can change after modifications) — different files in the - // same chunk → different chunk hash → cache miss even when no file - // content changed. The cache also becomes platform-specific: a - // Linux-built cache misses on macOS for the same repo. - parseableScanned.sort((a, b) => (a.path < b.path ? -1 : a.path > b.path ? 1 : 0)); - const totalParseable = parseableScanned.length; const totalBytes = parseableScanned.reduce((sum, f) => sum + f.size, 0); @@ -579,25 +549,25 @@ export async function runChunkedParseAndResolve( // runs. Resolving in the function body restores per-call configurability // and matches the pattern used by resolveAutoPoolSize and the U1 // parseChunkConcurrency resolver. - // Effective worker count, computed up-front so the chunk budget can scale to - // keep the whole pool busy (#worker-idle). The pool is ALWAYS used (sequential - // parsing was removed; the disabled channels threw above). Size it to the - // work: an explicit `--workers ` pins the size; otherwise the cores-based - // auto size is capped by the repo's worth of work (~one worker per - // CHUNK_BYTES_PER_WORKER of source) so a tiny repo spawns ~1 worker instead of - // a full pool, replacing the job the deleted small-repo threshold used to do. - // KTD-3 of the remove-sequential plan; the cap formula is intentionally coarse - // (tuning deferred). + // Effective worker count: explicit `--workers ` pins it; otherwise + // cores-based auto size is capped by source bytes / CHUNK_BYTES_PER_WORKER + // so a tiny repo does not spawn a full idle pool. Cache pack membership + // is independent of this number (#3088). const explicitPoolSize = options?.workerPoolSize; const workProportionalCap = Math.max(1, Math.ceil(totalBytes / CHUNK_BYTES_PER_WORKER)); const effectivePoolSize = explicitPoolSize && explicitPoolSize > 0 ? explicitPoolSize : Math.min(resolveAutoPoolSize(), workProportionalCap); - const chunkByteBudget = resolveChunkByteBudget(options, effectivePoolSize); - // Sub-batch size so each chunk fans into ~`TARGET_JOBS_PER_WORKER` jobs per - // worker, giving the pool's idle-slot assignment room to load-balance. An - // explicit `GITNEXUS_WORKER_SUB_BATCH_MAX_BYTES` operator override wins. + // Cache packs: stable (language, hash(path) mod 128) buckets, then the + // per-call byte budget inside each bucket (#3088). Pool size is used only + // for worker count and sub-batch fan-out, not membership. + const chunkByteBudget = resolveChunkByteBudget(options); + // Sub-batch size so a 2 MiB pack fans into ~TARGET_JOBS_PER_WORKER jobs + // per worker, floored at MIN_SUB_BATCH_BYTES (256 KiB) so an 8-worker + // pool still gets ~8 jobs from one pack instead of one idle-heavy job + // (#worker-idle). Do not derive this from pool×2 MiB while dispatching a + // 2 MiB pack. An explicit GITNEXUS_WORKER_SUB_BATCH_MAX_BYTES wins. const subBatchEnv = Number(process.env.GITNEXUS_WORKER_SUB_BATCH_MAX_BYTES); const dispatchSubBatchMaxBytes = Number.isFinite(subBatchEnv) && subBatchEnv > 0 @@ -622,19 +592,14 @@ export async function runChunkedParseAndResolve( ); } - const chunks: string[][] = []; - let currentChunk: string[] = []; - let currentBytes = 0; - for (const file of parseableScanned) { - if (currentChunk.length > 0 && currentBytes + file.size > chunkByteBudget) { - chunks.push(currentChunk); - currentChunk = []; - currentBytes = 0; - } - currentChunk.push(file.path); - currentBytes += file.size; - } - if (currentChunk.length > 0) chunks.push(currentChunk); + const chunks: string[][] = packParseCacheChunks( + parseableScanned.map((file) => ({ + path: file.path, + size: file.size, + language: getLanguageFromFilename(file.path) ?? 'unknown', + })), + chunkByteBudget, + ); const numChunks = chunks.length; diff --git a/gitnexus/src/storage/parse-cache.ts b/gitnexus/src/storage/parse-cache.ts index bcbb4c189..c99f4d87e 100644 --- a/gitnexus/src/storage/parse-cache.ts +++ b/gitnexus/src/storage/parse-cache.ts @@ -6,10 +6,12 @@ * does is skip the tree-sitter worker dispatch when a chunk's contents * haven't changed since the last run. * - * Granularity: chunk-level. The parse phase chunks files into ~20MB byte - * budgets. The cache key is `sha256(joined(filePath:contentHash for each - * file in the chunk, sorted))`. A change to a single file invalidates only - * that file's chunk — typically 1 of ~50 chunks on a 1000-file repo. + * Granularity: chunk-level. Files are assigned to a stable + * `(language, hash(path) mod 128)` bucket, then packed to a 2 MiB (or + * operator) byte budget *inside* that bucket. Membership does not depend + * on worker count. The cache key is `sha256(joined(filePath:contentHash + * for each file in the chunk, sorted))`. A content edit invalidates only + * that file's pack; add/delete/rename only the affected bucket. * * Why not per-file: * - Workers process sub-batches and emit aggregated `ParseWorkerResult`s. @@ -27,6 +29,7 @@ import { createRequire } from 'module'; import fs from 'fs/promises'; import path from 'path'; import { fileURLToPath } from 'url'; +import { compareCodeUnits } from '../lib/utils.js'; import type { ParseWorkerResult } from '../core/ingestion/workers/parse-worker.js'; /** @@ -632,6 +635,16 @@ import type { ParseWorkerResult } from '../core/ingestion/workers/parse-worker.j // therefore takes 79, the next free value above origin/main and every open PR // found by the contents-API scan at their exact head SHAs. // +// 79 -> 80 for #3088: parse-cache membership is `(language, sha256(path) mod +// 128)` then the byte budget *inside* that bucket. Worker count is no longer a +// membership input, so a warm v79 cache keyed sequential scan-order packs (and +// on multi-worker hosts, pool×2 MiB mega-chunks) must miss. Sidecar-era +// ParsedFile stores (#3086/#3087) share PARSE_CACHE_VERSION, so both stores +// invalidate in lockstep. origin/main at allocation is 79; open PRs that still +// touch gitnexus/src/storage/parse-cache.ts claim 78 (#3060), 71 (#2840), and +// 2 (#1616) — none claim 80. RE-CHECK AGAINST origin/main AND OPEN PRs +// IMMEDIATELY BEFORE MERGING. +// // WHY THIS IS STILL A HAND-PICKED NUMBER, when `SCHEMA_FINGERPRINT` next door // is a derived sha256 that cannot collide. The derivation exists and already // runs: `resolveAnalyzerRunnerIdentity` computes `build.digest` over the @@ -650,7 +663,7 @@ import type { ParseWorkerResult } from '../core/ingestion/workers/parse-worker.j // `route-extractors/` and `workers/` module content — would close the missing- // bump axis without invalidating on unrelated churn, and is the real follow-up. // RE-CHECK AGAINST origin/main AND OPEN PRs IMMEDIATELY BEFORE MERGING. -const SCHEMA_BUMP = 79; +const SCHEMA_BUMP = 80; const GITNEXUS_PKG_VERSION = (() => { try { // package.json sits at gitnexus/package.json — two levels up from @@ -676,6 +689,58 @@ const GITNEXUS_PKG_VERSION = (() => { })(); export const PARSE_CACHE_VERSION = `${SCHEMA_BUMP}+${GITNEXUS_PKG_VERSION}`; +/** SHA-256 hex of a string or buffer (paths for bucket ids, contents for cache keys). */ +const sha256Hex = (input: Buffer | string): string => + createHash('sha256') + .update(typeof input === 'string' ? Buffer.from(input) : input) + .digest('hex'); + +/** Stable parse-cache bucket count (#3088). Changing this requires SCHEMA_BUMP. */ +export const PARSE_CACHE_BUCKET_COUNT = 128; + +/** Bucket id for cache membership: `sha256(path) mod N` without IEEE-754 truncation. */ +export const parseCacheBucketId = (filePath: string): number => + Number(BigInt(`0x${sha256Hex(filePath)}`) % BigInt(PARSE_CACHE_BUCKET_COUNT)); + +export type ParseCachePackFile = { path: string; size: number; language: string }; + +/** + * Pack files into parse-cache chunks: group by (language, bucket id), sort + * paths inside the group, then cut at `byteBudget`. Bucket visit order is + * the lexicographic order of `${language}\\0${bucketId}` keys (deterministic, + * independent of scan order and worker count). + */ +export const packParseCacheChunks = ( + files: readonly ParseCachePackFile[], + byteBudget: number, +): string[][] => { + const buckets = new Map(); + for (const file of files) { + const key = `${file.language}\0${parseCacheBucketId(file.path)}`; + const list = buckets.get(key); + if (list) list.push(file); + else buckets.set(key, [file]); + } + const chunks: string[][] = []; + for (const key of [...buckets.keys()].sort()) { + const group = buckets.get(key)!; + group.sort((a, b) => compareCodeUnits(a.path, b.path)); + let current: string[] = []; + let bytes = 0; + for (const file of group) { + if (current.length > 0 && bytes + file.size > byteBudget) { + chunks.push(current); + current = []; + bytes = 0; + } + current.push(file.path); + bytes += file.size; + } + if (current.length > 0) chunks.push(current); + } + return chunks; +}; + const LEGACY_CACHE_FILENAME = 'parse-cache.json'; const CACHE_DIRNAME = 'parse-cache'; const CACHE_INDEX_FILENAME = 'index.json'; @@ -720,12 +785,6 @@ export interface ParseCache { onDiskKeys?: Set; } -/** SHA-256 hex of a single string or buffer. */ -const sha256Hex = (input: Buffer | string): string => - createHash('sha256') - .update(typeof input === 'string' ? Buffer.from(input) : input) - .digest('hex'); - /** Stable hash of a single file's contents — used by callers to compose a chunk hash. */ export const fileContentHash = (content: Buffer | string): string => sha256Hex(content); diff --git a/gitnexus/src/storage/parsedfile-store.ts b/gitnexus/src/storage/parsedfile-store.ts index c9a6e0482..1f34ff083 100644 --- a/gitnexus/src/storage/parsedfile-store.ts +++ b/gitnexus/src/storage/parsedfile-store.ts @@ -48,7 +48,7 @@ * file changes its chunk hash, which misses BOTH stores and re-dispatches. */ -import { promises as fs, mkdirSync, writeFileSync } from 'node:fs'; +import { promises as fs, mkdirSync, writeFileSync, unlinkSync } from 'node:fs'; import path from 'node:path'; import v8 from 'node:v8'; import vm from 'node:vm'; @@ -180,6 +180,140 @@ const serializeParsedFileShard = (parsedFiles: readonly ParsedFile[]): string | const shardPath = (storagePath: string, shardId: string): string => path.join(getParsedFileStoreDir(storagePath), `${shardId}.json`); +/** Sidecar listing `filePath`s in a shard; not matched by `endsWith('.json')`. */ +const shardPathsSidecarPath = (jsonPath: string): string => `${jsonPath}.paths`; + +const LOAD_YIELD_EVERY_SHARDS = 128; + +/** + * Test seam for #3086. Production always calls {@link forceGc}; unit tests + * replace `run` to count cadence without requiring `--expose-gc`. + */ +export const parsedFileLoadGc = { + run: forceGc, + /** Raw UTF-8 JSON shard bytes between GCs (#3086). Tests may lower this. */ + byteBudget: 128 * 1024 * 1024, +}; + +const encodeShardPathsSidecar = (parsedFiles: readonly ParsedFile[]): string => { + const paths = parsedFiles.map((pf) => pf.filePath); + return `${paths.length}\n${paths.length === 0 ? '' : `${paths.join('\n')}\n`}`; +}; + +/** + * Parse a counted NDJSON path listing. Returns `null` when the sidecar must + * not be trusted to skip the JSON shard: missing trailing newline, CR/NUL, + * a truncated listing that still ends on a complete line, or a count that + * does not match the remaining lines. + */ +const parseShardPathsSidecar = (sidecarRaw: string): string[] | null => { + if (sidecarRaw.includes('\0') || sidecarRaw.includes('\r') || !sidecarRaw.endsWith('\n')) { + return null; + } + const nl = sidecarRaw.indexOf('\n'); + if (nl < 0) return null; + const countToken = sidecarRaw.slice(0, nl); + if (!/^[0-9]+$/.test(countToken)) return null; + const count = Number(countToken); + const body = sidecarRaw.slice(nl + 1); + const listed = body === '' ? [] : body.slice(0, -1).split('\n'); + if (listed.length !== count) return null; + return listed; +}; + +/** NDJSON sidecars cannot encode paths that themselves contain CR/LF/NUL. */ +const shardPathsSidecarSafe = (parsedFiles: readonly ParsedFile[]): boolean => + parsedFiles.every((pf) => !/[\r\n\0]/.test(pf.filePath)); + +const isEnoent = (err: unknown): boolean => (err as NodeJS.ErrnoException).code === 'ENOENT'; + +const warnSidecarIo = (err: unknown, jsonPath: string, msg: string): void => { + logger.warn({ err, jsonPath }, msg); +}; + +const ignoreMissingSidecarUnlink = (err: unknown, jsonPath: string): void => { + if (isEnoent(err)) return; + warnSidecarIo( + err, + jsonPath, + 'parsedfile-store: failed to drop path sidecar; JSON remains authoritative', + ); +}; + +/** Drop a leftover listing before publishing JSON so load cannot skip new paths. */ +const dropPathSidecar = async (jsonPath: string): Promise => { + try { + await fs.unlink(shardPathsSidecarPath(jsonPath)); + } catch (err) { + ignoreMissingSidecarUnlink(err, jsonPath); + } +}; + +const dropPathSidecarSync = (jsonPath: string): void => { + try { + unlinkSync(shardPathsSidecarPath(jsonPath)); + } catch (err) { + ignoreMissingSidecarUnlink(err, jsonPath); + } +}; + +const writeShardPathsSidecar = async ( + jsonPath: string, + parsedFiles: readonly ParsedFile[], +): Promise => { + if (!shardPathsSidecarSafe(parsedFiles)) { + try { + await fs.unlink(shardPathsSidecarPath(jsonPath)); + } catch (err) { + ignoreMissingSidecarUnlink(err, jsonPath); + } + return; + } + try { + await fs.writeFile( + shardPathsSidecarPath(jsonPath), + encodeShardPathsSidecar(parsedFiles), + 'utf-8', + ); + } catch (err) { + warnSidecarIo( + err, + jsonPath, + 'parsedfile-store: path sidecar write failed; JSON shard remains authoritative', + ); + try { + await fs.unlink(shardPathsSidecarPath(jsonPath)); + } catch (unlinkErr) { + ignoreMissingSidecarUnlink(unlinkErr, jsonPath); + } + } +}; + +const writeShardPathsSidecarSync = (jsonPath: string, parsedFiles: readonly ParsedFile[]): void => { + if (!shardPathsSidecarSafe(parsedFiles)) { + try { + unlinkSync(shardPathsSidecarPath(jsonPath)); + } catch (err) { + ignoreMissingSidecarUnlink(err, jsonPath); + } + return; + } + try { + writeFileSync(shardPathsSidecarPath(jsonPath), encodeShardPathsSidecar(parsedFiles), 'utf-8'); + } catch (err) { + warnSidecarIo( + err, + jsonPath, + 'parsedfile-store: path sidecar write failed; JSON shard remains authoritative', + ); + try { + unlinkSync(shardPathsSidecarPath(jsonPath)); + } catch (unlinkErr) { + ignoreMissingSidecarUnlink(unlinkErr, jsonPath); + } + } +}; + /** * Write one parse chunk's `ParsedFile[]` to the store as a single shard (async). * No-op for an empty chunk. `shardId` must be unique within a run. Used by the @@ -194,7 +328,10 @@ export const persistParsedFileChunk = async ( const payload = serializeParsedFileShard(parsedFiles); if (payload === null) return; await fs.mkdir(getParsedFileStoreDir(storagePath), { recursive: true }); - await fs.writeFile(shardPath(storagePath, shardId), payload, 'utf-8'); + const dest = shardPath(storagePath, shardId); + await dropPathSidecar(dest); + await fs.writeFile(dest, payload, 'utf-8'); + await writeShardPathsSidecar(dest, parsedFiles); }; // Per-process set of store dirs we've already `mkdir`ed, so the sync worker @@ -223,7 +360,10 @@ export const persistParsedFileShardSync = ( mkdirSync(dir, { recursive: true }); createdStoreDirs.add(dir); } - writeFileSync(shardPath(storagePath, shardId), payload, 'utf-8'); + const dest = shardPath(storagePath, shardId); + dropPathSidecarSync(dest); + writeFileSync(dest, payload, 'utf-8'); + writeShardPathsSidecarSync(dest, parsedFiles); }; /** @@ -255,54 +395,90 @@ export const loadParsedFilesForPaths = async ( let filesWithDroppedSites = 0; let droppedChains = 0; let rejectedFiles = 0; + let bytesSinceGc = 0; + let shardsSinceYield = 0; + const maybeYieldAndGc = async (forceByteGc: boolean): Promise => { + if (forceByteGc) { + parsedFileLoadGc.run(); + bytesSinceGc = 0; + shardsSinceYield = 0; + await new Promise((resolve) => setImmediate(resolve)); + return; + } + shardsSinceYield++; + if (shardsSinceYield >= LOAD_YIELD_EVERY_SHARDS) { + shardsSinceYield = 0; + await new Promise((resolve) => setImmediate(resolve)); + } + }; for (let i = 0; i < shards.length; i++) { + const jsonName = shards[i]; + const jsonFull = path.join(dir, jsonName); + try { + const sidecarRaw = await fs.readFile(shardPathsSidecarPath(jsonFull), 'utf-8'); + // Fail closed: complete writers emit `\n` plus one path per line + // and a trailing newline, never CR. Stripping CR (or accepting a + // newline-terminated prefix) would let a truncated listing skip JSON. + const listed = parseShardPathsSidecar(sidecarRaw); + if (listed === null) { + throw new Error('corrupt sidecar'); + } + if (listed.length > 0 && !listed.some((p) => wantPaths.has(p))) { + await maybeYieldAndGc(false); + continue; + } + } catch { + // Missing or unreadable sidecar → read the shard (pre-sidecar stores). + } // Per-shard def pool: a SymbolDefinition's three serialized copies live within // a single shard (one ParsedFile), so the dedup is shard-local. A cross-shard // pool would retain defs of files NOT in `wantPaths` (loaded-but-discarded // shards), reintroducing the leak; per-shard drops them with the shard. const defPool = new Map(); const reviver = makeInterningReviver(pool, defPool); - let parsed: ParsedFile[]; + let raw: string; + try { + raw = await fs.readFile(jsonFull, 'utf-8'); + } catch { + continue; // skip a missing shard; missing files fall back to fresh extract + } + bytesSinceGc += Buffer.byteLength(raw, 'utf8'); + const crossedBudget = bytesSinceGc >= parsedFileLoadGc.byteBudget; + let parsed: ParsedFile[] | undefined; try { - const raw = await fs.readFile(path.join(dir, shards[i]), 'utf-8'); parsed = JSON.parse(raw, reviver) as ParsedFile[]; } catch { - continue; // skip a corrupt shard; missing files fall back to fresh extract + parsed = undefined; } - if (!Array.isArray(parsed)) continue; - for (const pf of parsed) { - if (!pf || typeof pf.filePath !== 'string' || !wantPaths.has(pf.filePath)) continue; - const flow = sanitizeCallableFlowSites(pf.callableFlowSites); - if (flow === undefined) { - // non-array garbage → distrust the file, re-extract - rejectedFiles++; - continue; - } - const chains = sanitizeReceiverChains(pf.referenceSites); - if (chains === undefined) { - rejectedFiles++; - continue; - } - if (flow.dropped === 0 && chains.dropped === 0) { - out.set(pf.filePath, pf); - } else { - droppedSites += flow.dropped; - droppedChains += chains.dropped; - filesWithDroppedSites++; - out.set(pf.filePath, { - ...pf, - ...(flow.dropped === 0 ? {} : { callableFlowSites: flow.sites }), - ...(chains.dropped === 0 ? {} : { referenceSites: chains.sites }), - }); + if (Array.isArray(parsed)) { + for (const pf of parsed) { + if (!pf || typeof pf.filePath !== 'string' || !wantPaths.has(pf.filePath)) continue; + const flow = sanitizeCallableFlowSites(pf.callableFlowSites); + if (flow === undefined) { + // non-array garbage → distrust the file, re-extract + rejectedFiles++; + continue; + } + const chains = sanitizeReceiverChains(pf.referenceSites); + if (chains === undefined) { + rejectedFiles++; + continue; + } + if (flow.dropped === 0 && chains.dropped === 0) { + out.set(pf.filePath, pf); + } else { + droppedSites += flow.dropped; + droppedChains += chains.dropped; + filesWithDroppedSites++; + out.set(pf.filePath, { + ...pf, + ...(flow.dropped === 0 ? {} : { callableFlowSites: flow.sites }), + ...(chains.dropped === 0 ? {} : { referenceSites: chains.sites }), + }); + } } } - // Every few shards, reclaim the transient pre-intern parse churn before it - // piles up against the heap limit (~5 GB avoidable on the kernel), and - // yield so the GC + any pending I/O can run. - if ((i & 7) === 7) { - forceGc(); - await new Promise((resolve) => setImmediate(resolve)); - } + await maybeYieldAndGc(crossedBudget); } if (droppedSites > 0 || droppedChains > 0) { // Facts for the dropped sites are omitted this run (the file itself is @@ -595,7 +771,10 @@ export const persistDurableParsedFileShardSync = ( mkdirSync(dir, { recursive: true }); createdDurableDirs.add(dir); } - writeFileSync(path.join(dir, `${chunkHash}-w${threadId}-${shardSeq}.json`), payload, 'utf-8'); + const dest = path.join(dir, `${chunkHash}-w${threadId}-${shardSeq}.json`); + dropPathSidecarSync(dest); + writeFileSync(dest, payload, 'utf-8'); + writeShardPathsSidecarSync(dest, parsedFiles); }; /** @@ -624,7 +803,27 @@ export const restoreDurableParsedFileShard = async ( const dst = getParsedFileStoreDir(runStoragePath); await fs.mkdir(dst, { recursive: true }); for (const name of shards) { - await fs.copyFile(path.join(src, name), path.join(dst, name)); + const srcJson = path.join(src, name); + const dstJson = path.join(dst, name); + await dropPathSidecar(dstJson); + await fs.copyFile(srcJson, dstJson); + try { + await fs.copyFile(shardPathsSidecarPath(srcJson), shardPathsSidecarPath(dstJson)); + } catch (copyErr) { + if (!isEnoent(copyErr)) { + warnSidecarIo( + copyErr, + srcJson, + 'parsedfile-store: durable path sidecar copy failed; JSON remains authoritative', + ); + continue; + } + try { + await fs.unlink(shardPathsSidecarPath(dstJson)); + } catch (err) { + ignoreMissingSidecarUnlink(err, dstJson); + } + } } return shards.length; }; diff --git a/gitnexus/test/integration/cli-e2e.test.ts b/gitnexus/test/integration/cli-e2e.test.ts index ffff7dcfd..9998a1425 100644 --- a/gitnexus/test/integration/cli-e2e.test.ts +++ b/gitnexus/test/integration/cli-e2e.test.ts @@ -1126,7 +1126,10 @@ describe('CLI end-to-end', () => { }); return; } - if (stage === 'proof' && /Refresh complete: 1 changed, 8 re-parsed,/.test(output)) { + if ( + stage === 'proof' && + /Refresh complete: 1 changed, 1 re-parsed, 0 affected dependent\(s\)/.test(output) + ) { const meta = JSON.parse( fs.readFileSync(path.join(repo, '.gitnexus', 'gitnexus.json'), 'utf8'), ); @@ -1214,7 +1217,9 @@ describe('CLI end-to-end', () => { return; } expect(stage).toBe('stopping'); - expect(transcript).toContain('Refresh complete: 1 changed, 8 re-parsed,'); + expect(transcript).toContain( + 'Refresh complete: 1 changed, 1 re-parsed, 0 affected dependent(s)', + ); resolve(); }); }); diff --git a/gitnexus/test/integration/parse-impl-clone-skip.test.ts b/gitnexus/test/integration/parse-impl-clone-skip.test.ts index 1fd155cf0..f34c8c3fd 100644 --- a/gitnexus/test/integration/parse-impl-clone-skip.test.ts +++ b/gitnexus/test/integration/parse-impl-clone-skip.test.ts @@ -32,6 +32,7 @@ import { pathToFileURL } from 'node:url'; import { createKnowledgeGraph } from '../../src/core/graph/graph.js'; import { runChunkedParseAndResolve } from '../../src/core/ingestion/pipeline-phases/parse-impl.js'; import { _captureLogger } from '../../src/core/logger.js'; +import { parseCacheBucketId } from '../../src/storage/parse-cache.js'; // file:// URL of the BUILT production result-delivery helper, imported by the // ESM test worker so it exercises the REAL postResultCloneSafe wiring (the @@ -140,10 +141,17 @@ parentPort.on('message', (msg) => { }); `; -const FIXTURE_FILES = { - 'src/good_a.ts': 'export function good_a() { return 1; }\n', - 'src/poison.ts': 'export function poison() { return 2; }\n', - 'src/good_c.ts': 'export function good_c() { return 3; }\n', +const POISON_PATH = 'src/poison.ts'; +/** Pinned same-bucket fixtures (sha256(path) mod 128 of poison.ts). */ +const GOOD_A_PATH = 'src/good_a_16.ts'; +const GOOD_C_PATH = 'src/good_c_51.ts'; +const GOOD_A_NAME = path.basename(GOOD_A_PATH, '.ts'); +const GOOD_C_NAME = path.basename(GOOD_C_PATH, '.ts'); + +const FIXTURE_FILES: Record = { + [GOOD_A_PATH]: 'export function good_a() { return 1; }\n', + [POISON_PATH]: 'export function poison() { return 2; }\n', + [GOOD_C_PATH]: 'export function good_c() { return 3; }\n', }; const nodeNames = (graph: ReturnType): Set => { @@ -165,6 +173,11 @@ const nodeNames = (graph: ReturnType): Set const STRICT = process.env.GITNEXUS_STRICT_CLONE === '1'; describe.skipIf(STRICT)('#2112: worker result clone-safety integration (POOL_SIZE=1)', () => { + it('pins survivors into the same parse-cache bucket as poison.ts', () => { + expect(parseCacheBucketId(GOOD_A_PATH)).toBe(parseCacheBucketId(POISON_PATH)); + expect(parseCacheBucketId(GOOD_C_PATH)).toBe(parseCacheBucketId(POISON_PATH)); + }); + let tempDir: string; let repoDir: string; @@ -228,8 +241,8 @@ describe.skipIf(STRICT)('#2112: worker result clone-safety integration (POOL_SIZ const graph = await runWith(writeWorker(CLONE_SAFE_WORKER)); const names = nodeNames(graph); // Survivors AND the sanitized poison file are all present — the run did not abort. - expect(names.has('good_a')).toBe(true); - expect(names.has('good_c')).toBe(true); + expect(names.has(GOOD_A_NAME)).toBe(true); + expect(names.has(GOOD_C_NAME)).toBe(true); // The poison node is delivered with its legitimate data intact (only the // leaked native `toString` was stripped), so it still lands in the graph. expect(names.has('poison')).toBe(true); @@ -256,8 +269,8 @@ describe.skipIf(STRICT)('#2112: worker result clone-safety integration (POOL_SIZ // rejects; with it, all files (incl. the sanitized poison node) are present. const graph = await runWith(writeWorker(GETTER_WORKER)); const names = nodeNames(graph); - expect(names.has('good_a')).toBe(true); - expect(names.has('good_c')).toBe(true); + expect(names.has(GOOD_A_NAME)).toBe(true); + expect(names.has(GOOD_C_NAME)).toBe(true); expect(names.has('poison')).toBe(true); }); diff --git a/gitnexus/test/integration/parse-impl-env-reads.test.ts b/gitnexus/test/integration/parse-impl-env-reads.test.ts index 1ccdffb7b..2d93e12dd 100644 --- a/gitnexus/test/integration/parse-impl-env-reads.test.ts +++ b/gitnexus/test/integration/parse-impl-env-reads.test.ts @@ -28,6 +28,7 @@ import path from 'node:path'; import { runChunkedParseAndResolve } from '../../src/core/ingestion/pipeline-phases/parse-impl.js'; import { createKnowledgeGraph } from '../../src/core/graph/graph.js'; +import { PARSE_CACHE_VERSION, parseCacheBucketId } from '../../src/storage/parse-cache.js'; const ORIGINAL_BUDGET = process.env.GITNEXUS_CHUNK_BYTE_BUDGET; @@ -123,12 +124,12 @@ describe('parse-impl chunkByteBudget resolution (U14 / F7)', () => { expect(chunks).toBe(3); }); - it('default-fallback: large built-in budget keeps the fixture in a single chunk', async () => { - // Both option and env unset → falls through to DEFAULT_CHUNK_BYTE_BUDGET - // (2 MB). The fixture totals well under that, so exactly one chunk. + it('default-fallback: 2 MB budget packs by bucket, not one sequential mega-chunk', async () => { delete process.env.GITNEXUS_CHUNK_BYTE_BUDGET; - const chunks = await countChunksFromProgress(repoPath, ['a.ts', 'b.ts', 'c.ts']); - expect(chunks).toBe(1); + const files = ['a.ts', 'b.ts', 'c.ts']; + const expectedBuckets = new Set(files.map((f) => `typescript\0${parseCacheBucketId(f)}`)); + const chunks = await countChunksFromProgress(repoPath, files); + expect(chunks).toBe(expectedBuckets.size); }); it('per-call: two back-to-back runs with different option values observe their own values, not the previous call', async () => { @@ -146,6 +147,32 @@ describe('parse-impl chunkByteBudget resolution (U14 / F7)', () => { chunkByteBudget: 10 * 1024 * 1024, }); expect(small).toBe(3); - expect(large).toBe(1); + const expectedBuckets = new Set(files.map((f) => `typescript\0${parseCacheBucketId(f)}`)); + expect(large).toBe(expectedBuckets.size); + }); + + it('workerPoolSize 1 vs 2 produce the same cache keys when budget is unset (#3088)', async () => { + delete process.env.GITNEXUS_CHUNK_BYTE_BUDGET; + const files = ['a.ts', 'b.ts', 'c.ts']; + const keysForPool = async (workerPoolSize: number): Promise => { + const parseCache = { + version: PARSE_CACHE_VERSION, + entries: new Map(), + usedKeys: new Set(), + }; + const graph = createKnowledgeGraph(); + await runChunkedParseAndResolve( + graph, + scanned(repoPath, files), + files, + files.length, + repoPath, + Date.now(), + () => {}, + { workerPoolSize, parseCache }, + ); + return [...parseCache.usedKeys].sort(); + }; + expect(await keysForPool(1)).toEqual(await keysForPool(2)); }); }); diff --git a/gitnexus/test/integration/parse-impl-quarantine-cache-skip.test.ts b/gitnexus/test/integration/parse-impl-quarantine-cache-skip.test.ts index 0bb8dbec6..949f0d75e 100644 --- a/gitnexus/test/integration/parse-impl-quarantine-cache-skip.test.ts +++ b/gitnexus/test/integration/parse-impl-quarantine-cache-skip.test.ts @@ -73,7 +73,11 @@ import { pathToFileURL } from 'node:url'; import { createKnowledgeGraph } from '../../src/core/graph/graph.js'; import { runChunkedParseAndResolve } from '../../src/core/ingestion/pipeline-phases/parse-impl.js'; -import { computeChunkHash, fileContentHash } from '../../src/storage/parse-cache.js'; +import { + computeChunkHash, + fileContentHash, + packParseCacheChunks, +} from '../../src/storage/parse-cache.js'; import type { ParseWorkerResult } from '../../src/core/ingestion/workers/parse-worker.js'; /** @@ -180,6 +184,41 @@ const FIXTURE_FILES = { 'src/good_c.ts': 'export function good_c() { return 3; }\n', }; +const POISON_PATH = 'src/poison.ts'; +const DEFAULT_TEST_CHUNK_BUDGET = 2 * 1024 * 1024; + +const resolveTestChunkByteBudget = (): number => { + const env = Number(process.env.GITNEXUS_CHUNK_BYTE_BUDGET); + if (Number.isFinite(env) && env > 0) return env; + return DEFAULT_TEST_CHUNK_BUDGET; +}; + +const hashPacks = ( + scanned: { path: string; size: number }[], +): { poison: string; others: string[] } => { + const packs = packParseCacheChunks( + scanned.map((file) => ({ + path: file.path, + size: file.size, + language: 'typescript', + })), + resolveTestChunkByteBudget(), + ); + const hashOf = (pack: string[]) => + computeChunkHash( + pack.map((p) => ({ + filePath: p, + contentHash: fileContentHash(FIXTURE_FILES[p as keyof typeof FIXTURE_FILES]), + })), + ); + const poisonPack = packs.find((paths) => paths.includes(POISON_PATH)); + if (!poisonPack) throw new Error('poison.ts was not packed'); + return { + poison: hashOf(poisonPack), + others: packs.filter((paths) => !paths.includes(POISON_PATH)).map(hashOf), + }; +}; + describe('U20: parse-impl quarantine + chunk-cache integration (PR #1693 Codex finding)', () => { let tempDir: string; let repoDir: string; @@ -216,16 +255,7 @@ describe('U20: parse-impl quarantine + chunk-cache integration (PR #1693 Codex f size: statSync(path.join(repoDir, rel)).size, })); - // The chunk hash is computed from EVERY file's content hash. The - // load-bearing U2 assertion below checks `parseCache.entries.has` - // against this exact value, so we compute it the same way - // parse-impl does. - const expectedChunkHash = computeChunkHash( - filePaths.map((p) => ({ - filePath: p, - contentHash: fileContentHash(FIXTURE_FILES[p as keyof typeof FIXTURE_FILES]), - })), - ); + const expectedChunkHash = hashPacks(scanned).poison; const parseCache = { version: 'test', @@ -293,23 +323,20 @@ describe('U20: parse-impl quarantine + chunk-cache integration (PR #1693 Codex f // against a fresh-quarantine pool. expect(parseCache.entries.has(expectedChunkHash)).toBe(false); expect(parseCache.usedKeys.has(expectedChunkHash)).toBe(true); - expect(parseCache.entries.size).toBe(0); + for (const hash of hashPacks(scanned).others) { + expect(parseCache.entries.has(hash)).toBe(true); + } }); - it('cross-run: unchanged fixture re-dispatches on a second pass because the cache was empty', async () => { - // First pass: same setup as the previous test. Cache stays empty - // because poison.ts triggered quarantine. + it('cross-run: unchanged fixture re-dispatches the poison pack because that pack was not cached', async () => { + // First pass: same setup as the previous test. The poison pack is not + // cached; other packs may be. const filePaths = Object.keys(FIXTURE_FILES); const scanned = filePaths.map((rel) => ({ path: rel, size: statSync(path.join(repoDir, rel)).size, })); - const expectedChunkHash = computeChunkHash( - filePaths.map((p) => ({ - filePath: p, - contentHash: fileContentHash(FIXTURE_FILES[p as keyof typeof FIXTURE_FILES]), - })), - ); + const expectedChunkHash = hashPacks(scanned).poison; const parseCache = { version: 'test', @@ -369,6 +396,9 @@ describe('U20: parse-impl quarantine + chunk-cache integration (PR #1693 Codex f // load-bearing cross-run protection. expect(parseCache.entries.has(expectedChunkHash)).toBe(false); expect(parseCache.usedKeys.has(expectedChunkHash)).toBe(true); + for (const hash of hashPacks(scanned).others) { + expect(parseCache.entries.has(hash)).toBe(true); + } // Worker path ran again; surviving files in the graph; poison // still absent per the U20 contract (workers are the sole // resilience layer, no sequential reparse). diff --git a/gitnexus/test/unit/incremental-orchestration.test.ts b/gitnexus/test/unit/incremental-orchestration.test.ts index eb792b701..b0c44917b 100644 --- a/gitnexus/test/unit/incremental-orchestration.test.ts +++ b/gitnexus/test/unit/incremental-orchestration.test.ts @@ -848,7 +848,7 @@ describe('runFullAnalysis — incremental orchestration', () => { deletedFiles: 0, writeMode: 'incremental', }); - expect(incremental.incrementalStats?.reparsedFiles).toBe(7); + expect(incremental.incrementalStats?.reparsedFiles).toBe(1); expect( querySpy.mock.calls.some( ([query]) => diff --git a/gitnexus/test/unit/incremental-parse-cache.test.ts b/gitnexus/test/unit/incremental-parse-cache.test.ts index 8097c8ca5..14c4c8359 100644 --- a/gitnexus/test/unit/incremental-parse-cache.test.ts +++ b/gitnexus/test/unit/incremental-parse-cache.test.ts @@ -4,8 +4,11 @@ import { tmpdir } from 'os'; import path from 'path'; import { PARSE_CACHE_VERSION, + PARSE_CACHE_BUCKET_COUNT, computeChunkHash, fileContentHash, + packParseCacheChunks, + parseCacheBucketId, loadParseCache, loadParseCacheChunk, persistParseCacheChunk, @@ -240,15 +243,11 @@ describe('PARSE_CACHE_VERSION', () => { // collided, because each re-checked once and neither re-checked after the // other moved — which is why the rule is re-applied AT MERGE, not when the // number is picked. - it('pins SCHEMA_BUMP to 79 so concurrent bumps cannot silently collide (#2766, #3015)', () => { - expect(Number(PARSE_CACHE_VERSION.split('+', 1)[0])).toBe(79); - // The PREVIOUS version must fail the reuse gate, not merely differ from the - // current one — a hardcoded number outside the conflict hunk rebases cleanly - // while being wrong, which is exactly how the 37/38 exact clashes landed. - // Every nearby historical or in-flight value is rejected, including 69, - // which carried the route-table payload before this merge. + it('pins SCHEMA_BUMP to 80 so concurrent bumps cannot silently collide (#2766, #3015, #3088)', () => { + expect(Number(PARSE_CACHE_VERSION.split('+', 1)[0])).toBe(80); + expect(PARSE_CACHE_BUCKET_COUNT).toBe(128); for (const taken of [ - 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72, 73, 74, 75, 76, 77, 78, + 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72, 73, 74, 75, 76, 77, 78, 79, ]) { expect(Number(PARSE_CACHE_VERSION.split('+', 1)[0])).not.toBe(taken); } @@ -260,6 +259,55 @@ describe('PARSE_CACHE_VERSION', () => { }); }); +describe('packParseCacheChunks (#3088)', () => { + const files = [ + { path: 'src/a.ts', size: 100, language: 'typescript' }, + { path: 'src/b.ts', size: 100, language: 'typescript' }, + { path: 'pkg/c.py', size: 100, language: 'python' }, + ]; + const budget = 2 * 1024 * 1024; + const packKey = (chunk: string[]): string => + `${files.find((f) => f.path === chunk[0])?.language ?? 'typescript'}\0${parseCacheBucketId(chunk[0])}`; + + it('is independent of scan order', () => { + expect(packParseCacheChunks(files, budget)).toEqual( + packParseCacheChunks([...files].reverse(), budget), + ); + }); + + it('add/delete only rewrites packs in the affected (language, bucket)', () => { + const a = packParseCacheChunks(files, budget); + const added = { path: 'AAA.ts', size: 150_000, language: 'typescript' }; + const withNew = packParseCacheChunks([...files, added], budget); + const addedKey = packKey([added.path]); + const untouched = (packs: string[][]) => + packs.filter((c) => packKey(c) !== addedKey).map((c) => c.join('|')); + expect(untouched(withNew).sort()).toEqual(untouched(a).sort()); + expect(withNew.some((c) => c.includes(added.path))).toBe(true); + + const withoutB = packParseCacheChunks( + files.filter((f) => f.path !== 'src/b.ts'), + budget, + ); + const removedKey = packKey(['src/b.ts']); + const leftover = (packs: string[][]) => + packs.filter((c) => packKey(c) !== removedKey).map((c) => c.join('|')); + expect(leftover(withoutB).sort()).toEqual(leftover(a).sort()); + expect(withoutB.every((c) => !c.includes('src/b.ts'))).toBe(true); + }); + + it('parseCacheBucketId uses the full sha256 digest, not an IEEE-754 prefix', () => { + const path = 'src/foo.ts'; + const hex = fileContentHash(path); + const full = Number(BigInt(`0x${hex}`) % BigInt(PARSE_CACHE_BUCKET_COUNT)); + const truncated = Number.parseInt(hex.slice(0, 8), 16) % PARSE_CACHE_BUCKET_COUNT; + expect(parseCacheBucketId(path)).toBe(full); + expect(parseCacheBucketId(path)).toBeGreaterThanOrEqual(0); + expect(parseCacheBucketId(path)).toBeLessThan(PARSE_CACHE_BUCKET_COUNT); + expect(full).not.toBe(truncated); + }); +}); + describe('pruneCache', () => { it('drops entries whose hashes are not in the used-set', () => { const cache: ParseCache = { diff --git a/gitnexus/test/unit/parsedfile-store.test.ts b/gitnexus/test/unit/parsedfile-store.test.ts index 6ce5cd252..d799f2054 100644 --- a/gitnexus/test/unit/parsedfile-store.test.ts +++ b/gitnexus/test/unit/parsedfile-store.test.ts @@ -1,5 +1,6 @@ -import { describe, it, expect } from 'vitest'; -import { mkdtemp, rm, readdir, readFile } from 'fs/promises'; +import { describe, it, expect, vi } from 'vitest'; +import { promises as nodeFsPromises } from 'node:fs'; +import { mkdtemp, rm, readdir, readFile, writeFile } from 'fs/promises'; import { tmpdir } from 'os'; import path from 'path'; import type { ParsedFile } from 'gitnexus-shared'; @@ -7,8 +8,12 @@ import { clearParsedFileStore, persistParsedFileChunk, persistParsedFileShardSync, + persistDurableParsedFileShardSync, + restoreDurableParsedFileShard, loadParsedFilesForPaths, getParsedFileStoreDir, + getDurableParsedFileDir, + parsedFileLoadGc, } from '../../src/storage/parsedfile-store.js'; /** @@ -250,6 +255,15 @@ describe('parsedfile-store', () => { 'utf-8', ); expect(syncBytes).toBe(asyncBytes); + const asyncPaths = await readFile( + path.join(getParsedFileStoreDir(asyncDir), 'shard.json.paths'), + 'utf-8', + ); + const syncPaths = await readFile( + path.join(getParsedFileStoreDir(syncDir), 'shard.json.paths'), + 'utf-8', + ); + expect(syncPaths).toBe(asyncPaths); } finally { await rm(asyncDir, { recursive: true, force: true }); await rm(syncDir, { recursive: true, force: true }); @@ -614,4 +628,254 @@ describe('parsedfile-store receiverChain sanitation', () => { await rm(dir, { recursive: true, force: true }); } }); + + it('writes a .json.paths sidecar and skips JSON for non-intersecting shards (#3087)', async () => { + const dir = await mkdtemp(path.join(tmpdir(), 'pfstore-sidecar-')); + try { + await persistParsedFileChunk(dir, 'chunk-0', [makeParsedFile('a.c')]); + await persistParsedFileChunk(dir, 'chunk-1', [makeParsedFile('b.c')]); + const storeDir = getParsedFileStoreDir(dir); + const names = await readdir(storeDir); + expect(names.sort()).toEqual([ + 'chunk-0.json', + 'chunk-0.json.paths', + 'chunk-1.json', + 'chunk-1.json.paths', + ]); + const readSpy = vi.spyOn(nodeFsPromises, 'readFile'); + try { + const loaded = await loadParsedFilesForPaths(dir, new Set(['b.c'])); + expect([...loaded.keys()]).toEqual(['b.c']); + const jsonReads = readSpy.mock.calls.filter(([p]) => { + const n = String(p); + return n.endsWith('.json') && !n.endsWith('.json.paths'); + }); + expect(jsonReads).toHaveLength(1); + expect(String(jsonReads[0][0])).toMatch(/chunk-1\.json$/); + } finally { + readSpy.mockRestore(); + } + } finally { + await rm(dir, { recursive: true, force: true }); + } + }); + + it('reads a shard when its sidecar is missing or garbage (#3087)', async () => { + const dir = await mkdtemp(path.join(tmpdir(), 'pfstore-sidecar-fb-')); + try { + await persistParsedFileChunk(dir, 'ok', [makeParsedFile('a.c')]); + await persistParsedFileChunk(dir, 'bad', [makeParsedFile('b.c')]); + const storeDir = getParsedFileStoreDir(dir); + await rm(path.join(storeDir, 'ok.json.paths'), { force: true }); + await writeFile(path.join(storeDir, 'bad.json.paths'), 'not\x00valid', 'utf-8'); + const loaded = await loadParsedFilesForPaths(dir, new Set(['a.c', 'b.c'])); + expect(loaded.has('a.c')).toBe(true); + expect(loaded.has('b.c')).toBe(true); + } finally { + await rm(dir, { recursive: true, force: true }); + } + }); + + it('reads a shard when its sidecar is truncated without a trailing newline (#3087)', async () => { + const dir = await mkdtemp(path.join(tmpdir(), 'pfstore-sidecar-trunc-')); + try { + await persistParsedFileChunk(dir, 'ok', [makeParsedFile('wanted.c')]); + const storeDir = getParsedFileStoreDir(dir); + await writeFile(path.join(storeDir, 'ok.json.paths'), 'unrelated.c', 'utf-8'); + const loaded = await loadParsedFilesForPaths(dir, new Set(['wanted.c'])); + expect(loaded.has('wanted.c')).toBe(true); + } finally { + await rm(dir, { recursive: true, force: true }); + } + }); + + it('reads a shard when its sidecar is a newline-terminated partial listing', async () => { + const dir = await mkdtemp(path.join(tmpdir(), 'pfstore-sidecar-partial-')); + try { + await persistParsedFileChunk(dir, 'ok', [makeParsedFile('wanted.c')]); + const storeDir = getParsedFileStoreDir(dir); + await writeFile(path.join(storeDir, 'ok.json.paths'), 'unrelated.c\n', 'utf-8'); + const loaded = await loadParsedFilesForPaths(dir, new Set(['wanted.c'])); + expect(loaded.has('wanted.c')).toBe(true); + } finally { + await rm(dir, { recursive: true, force: true }); + } + }); + + it('reads a shard when its sidecar contains CR', async () => { + const dir = await mkdtemp(path.join(tmpdir(), 'pfstore-sidecar-cr-')); + try { + await persistParsedFileChunk(dir, 'ok', [makeParsedFile('wanted.c')]); + const storeDir = getParsedFileStoreDir(dir); + await writeFile(path.join(storeDir, 'ok.json.paths'), 'unrelated.c\r\n', 'utf-8'); + const loaded = await loadParsedFilesForPaths(dir, new Set(['wanted.c'])); + expect(loaded.has('wanted.c')).toBe(true); + } finally { + await rm(dir, { recursive: true, force: true }); + } + }); + + it('omits a sidecar when a filePath contains a newline and still loads JSON', async () => { + const dir = await mkdtemp(path.join(tmpdir(), 'pfstore-sidecar-nl-')); + const weird = 'weird\nname.c'; + try { + await persistParsedFileChunk(dir, 'ok', [makeParsedFile(weird)]); + const storeDir = getParsedFileStoreDir(dir); + expect(await readdir(storeDir)).toEqual(['ok.json']); + const loaded = await loadParsedFilesForPaths(dir, new Set([weird])); + expect(loaded.has(weird)).toBe(true); + } finally { + await rm(dir, { recursive: true, force: true }); + } + }); + + it('removes a stale sidecar when a rewritten shard is no longer listing-safe', async () => { + const dir = await mkdtemp(path.join(tmpdir(), 'pfstore-sidecar-stale-')); + const weird = 'weird\nname.c'; + try { + await persistParsedFileChunk(dir, 'ok', [makeParsedFile('safe.c')]); + await persistParsedFileChunk(dir, 'ok', [makeParsedFile(weird)]); + const storeDir = getParsedFileStoreDir(dir); + expect(await readdir(storeDir)).toEqual(['ok.json']); + const loaded = await loadParsedFilesForPaths(dir, new Set([weird])); + expect(loaded.has(weird)).toBe(true); + } finally { + await rm(dir, { recursive: true, force: true }); + } + }); + + it('does not forceGc on a small store (byte budget, not every 8 shards) (#3086)', async () => { + const dir = await mkdtemp(path.join(tmpdir(), 'pfstore-gc-')); + const gc = vi.fn(); + const prev = parsedFileLoadGc.run; + parsedFileLoadGc.run = gc; + try { + for (let i = 0; i < 16; i++) { + await persistParsedFileChunk(dir, `s${i}`, [makeParsedFile(`f${i}.c`)]); + } + await loadParsedFilesForPaths(dir, new Set(Array.from({ length: 16 }, (_, i) => `f${i}.c`))); + expect(gc).not.toHaveBeenCalled(); + } finally { + parsedFileLoadGc.run = prev; + await rm(dir, { recursive: true, force: true }); + } + }); + + it('forceGc when accumulated raw JSON bytes reach parsedFileLoadGc.byteBudget (#3086)', async () => { + const dir = await mkdtemp(path.join(tmpdir(), 'pfstore-gc-pos-')); + const gc = vi.fn(); + const prevRun = parsedFileLoadGc.run; + const prevBudget = parsedFileLoadGc.byteBudget; + parsedFileLoadGc.run = gc; + parsedFileLoadGc.byteBudget = 8; + try { + await persistParsedFileChunk(dir, 's0', [makeParsedFile('f0.c')]); + await loadParsedFilesForPaths(dir, new Set(['f0.c'])); + expect(gc).toHaveBeenCalled(); + } finally { + parsedFileLoadGc.run = prevRun; + parsedFileLoadGc.byteBudget = prevBudget; + await rm(dir, { recursive: true, force: true }); + } + }); + + it('restoreDurableParsedFileShard copies sidecars and returns JSON shard count (#3087)', async () => { + const dir = await mkdtemp(path.join(tmpdir(), 'pfstore-restore-')); + try { + const durable = getDurableParsedFileDir(dir); + persistDurableParsedFileShardSync(durable, 'abc', 1, 0, [makeParsedFile('a.c')]); + const restored = await restoreDurableParsedFileShard(durable, dir, 'abc'); + expect(restored).toBe(1); + const storeDir = getParsedFileStoreDir(dir); + expect(await readdir(storeDir)).toEqual( + expect.arrayContaining(['abc-w1-0.json', 'abc-w1-0.json.paths']), + ); + expect(await readFile(path.join(storeDir, 'abc-w1-0.json.paths'), 'utf-8')).toBe('1\na.c\n'); + const loaded = await loadParsedFilesForPaths(dir, new Set(['a.c'])); + expect(loaded.has('a.c')).toBe(true); + } finally { + await rm(dir, { recursive: true, force: true }); + } + }); + + it('restoreDurableParsedFileShard unlinks a stale dest sidecar when the source has none', async () => { + const dir = await mkdtemp(path.join(tmpdir(), 'pfstore-restore-stale-')); + try { + const durable = getDurableParsedFileDir(dir); + persistDurableParsedFileShardSync(durable, 'abc', 1, 0, [makeParsedFile('a.c')]); + const durableShard = path.join(durable, 'abc', 'abc-w1-0.json'); + await rm(`${durableShard}.paths`, { force: true }); + const storeDir = getParsedFileStoreDir(dir); + await nodeFsPromises.mkdir(storeDir, { recursive: true }); + await writeFile(path.join(storeDir, 'abc-w1-0.json.paths'), 'stale.c\n', 'utf-8'); + const restored = await restoreDurableParsedFileShard(durable, dir, 'abc'); + expect(restored).toBe(1); + await expect(readFile(path.join(storeDir, 'abc-w1-0.json.paths'), 'utf-8')).rejects.toThrow(); + const loaded = await loadParsedFilesForPaths(dir, new Set(['a.c'])); + expect(loaded.has('a.c')).toBe(true); + } finally { + await rm(dir, { recursive: true, force: true }); + } + }); + + it('drops a leftover sidecar before overwriting JSON so load cannot skip new paths', async () => { + const dir = await mkdtemp(path.join(tmpdir(), 'pfstore-sidecar-rewrite-')); + try { + await persistParsedFileChunk(dir, 'ok', [makeParsedFile('stale.c')]); + const origUnlink = nodeFsPromises.unlink.bind(nodeFsPromises); + const origWrite = nodeFsPromises.writeFile.bind(nodeFsPromises); + const order: string[] = []; + const unlinkSpy = vi + .spyOn(nodeFsPromises, 'unlink') + .mockImplementation(async (p, ...rest) => { + order.push(`unlink:${path.basename(String(p))}`); + return origUnlink(p, ...rest); + }); + const writeSpy = vi + .spyOn(nodeFsPromises, 'writeFile') + .mockImplementation(async (p, data, enc) => { + order.push(`write:${path.basename(String(p))}`); + return origWrite(p, data, enc); + }); + try { + await persistParsedFileChunk(dir, 'ok', [makeParsedFile('a.c')]); + } finally { + unlinkSpy.mockRestore(); + writeSpy.mockRestore(); + } + const jsonIdx = order.indexOf('write:ok.json'); + const pathsIdx = order.indexOf('unlink:ok.json.paths'); + expect(pathsIdx).toBeGreaterThanOrEqual(0); + expect(pathsIdx).toBeLessThan(jsonIdx); + const loaded = await loadParsedFilesForPaths(dir, new Set(['a.c'])); + expect(loaded.has('a.c')).toBe(true); + } finally { + await rm(dir, { recursive: true, force: true }); + } + }); + + it('persist still succeeds when the sidecar write fails', async () => { + const dir = await mkdtemp(path.join(tmpdir(), 'pfstore-sidecar-enospc-')); + try { + await persistParsedFileChunk(dir, 'ok', [makeParsedFile('stale.c')]); + const orig = nodeFsPromises.writeFile.bind(nodeFsPromises); + const spy = vi.spyOn(nodeFsPromises, 'writeFile').mockImplementation(async (p, data, enc) => { + if (String(p).endsWith('.paths')) { + throw Object.assign(new Error('ENOSPC'), { code: 'ENOSPC' }); + } + return orig(p, data, enc); + }); + try { + await persistParsedFileChunk(dir, 'ok', [makeParsedFile('a.c')]); + } finally { + spy.mockRestore(); + } + const storeDir = getParsedFileStoreDir(dir); + await expect(readFile(path.join(storeDir, 'ok.json.paths'), 'utf-8')).rejects.toThrow(); + const loaded = await loadParsedFilesForPaths(dir, new Set(['a.c'])); + expect(loaded.has('a.c')).toBe(true); + } finally { + await rm(dir, { recursive: true, force: true }); + } + }); });