mirror of
https://github.com/abhigyanpatwari/GitNexus.git
synced 2026-09-09 22:33:39 +00:00
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 <cursoragent@cursor.com> * 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 <cursoragent@cursor.com> * 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 <cursoragent@cursor.com> * 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 <cursoragent@cursor.com> * 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 <cursoragent@cursor.com> * 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 <cursoragent@cursor.com> * 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 <cursoragent@cursor.com> * 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 <cursoragent@cursor.com> * 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 <cursoragent@cursor.com> * 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 <cursoragent@cursor.com> * 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 <cursoragent@cursor.com> * 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 <cursoragent@cursor.com> * 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 <cursoragent@cursor.com> * 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 <cursoragent@cursor.com> * 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 <cursoragent@cursor.com> * 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 <cursoragent@cursor.com> --------- Co-authored-by: Gergo Magyar <gergomagyar0@gmail.com> Co-authored-by: Cursor <cursoragent@cursor.com> Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>
This commit is contained in:
Gergő Magyar
GitHub
parent
4e51213c82
commit
9718e1247a
11 changed files with 777 additions and 167 deletions
|
|
@ -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. |
|
||||
|
|
|
|||
|
|
@ -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 <N>` 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 <N>` 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;
|
||||
|
||||
|
|
|
|||
|
|
@ -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<string, ParseCachePackFile[]>();
|
||||
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<string>;
|
||||
}
|
||||
|
||||
/** 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);
|
||||
|
||||
|
|
|
|||
|
|
@ -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<void> => {
|
||||
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<void> => {
|
||||
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<void> => {
|
||||
if (forceByteGc) {
|
||||
parsedFileLoadGc.run();
|
||||
bytesSinceGc = 0;
|
||||
shardsSinceYield = 0;
|
||||
await new Promise<void>((resolve) => setImmediate(resolve));
|
||||
return;
|
||||
}
|
||||
shardsSinceYield++;
|
||||
if (shardsSinceYield >= LOAD_YIELD_EVERY_SHARDS) {
|
||||
shardsSinceYield = 0;
|
||||
await new Promise<void>((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 `<count>\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<string, SymbolDefinition>();
|
||||
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<void>((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;
|
||||
};
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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<string, string> = {
|
||||
[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<typeof createKnowledgeGraph>): Set<string> => {
|
||||
|
|
@ -165,6 +173,11 @@ const nodeNames = (graph: ReturnType<typeof createKnowledgeGraph>): Set<string>
|
|||
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);
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -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<string[]> => {
|
||||
const parseCache = {
|
||||
version: PARSE_CACHE_VERSION,
|
||||
entries: new Map(),
|
||||
usedKeys: new Set<string>(),
|
||||
};
|
||||
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));
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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).
|
||||
|
|
|
|||
|
|
@ -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]) =>
|
||||
|
|
|
|||
|
|
@ -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 = {
|
||||
|
|
|
|||
|
|
@ -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 });
|
||||
}
|
||||
});
|
||||
});
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue