diff --git a/config.yaml b/config.yaml index 98e99a7..28eff07 100644 --- a/config.yaml +++ b/config.yaml @@ -8,13 +8,23 @@ version: 1 # disk. That is the only on-disk artifact — the archive is streamed in memory and # never extracted — so budget 0.5 GB per worker: 2 GB at the default 4, 4 GB at # the ceiling of 8. -# Returns at the ceiling are sublinear: symbol extraction does not parallelise -# (measured 0.95x on 4 threads), so you buy far less than 8x for a hard linear -# 4 GB of disk. Raise it knowing that. +# Symbol extraction does not parallelise across THREADS (measured 0.95x on 4 +# threads — the tree walk is GIL-serialized), which is why extraction now runs +# in its own shared process pool instead — see extract_processes below. Raising +# this knob buys disk-bound repo fan-out, not extraction throughput. # With semantic indexing enabled this is clamped to 2 — a MEMORY bound, since # embedding materialises a whole repo's chunks (~0.5-0.8 GB per worker). # index_concurrency: 4 +# How many worker PROCESSES the job uses to extract symbols/edges (issue #108). +# Independent of index_concurrency above: this is a CPU knob (a shared, spawn- +# based process pool decoupled from the per-repo worker threads), not a disk or +# memory one. Default (unset) derives from the runtime's affinity/cgroup-aware +# CPU count, clamped to 8. Setting this to 1 restores fully serial, in-process +# extraction and spawns no process pool at all — the rollback switch if the +# pool ever misbehaves in this runtime. +# extract_processes: 4 + connections: - type: github # orgs / users / repos are UNIONED, then deduplicated by canonical org/repo. diff --git a/docs/perf/issue-108-measurements.md b/docs/perf/issue-108-measurements.md new file mode 100644 index 0000000..463ba65 --- /dev/null +++ b/docs/perf/issue-108-measurements.md @@ -0,0 +1,139 @@ +# Issue #108 — process-pool symbol/edge extraction: measurements + +Two measurement rounds, per the plan's §8.4 requirement. The **planning-time** +probe (§1) established the initial GO decision, against an on-disk-walk baseline +that no longer exists in production. The **shipped** measurement (§2) re-runs the +comparison through the real `indexer.ingest.iter_tar_source_files` tarball source +via `scripts/measure_extraction_pool.py`, and is the number this PR reports as +AC1's evidence — per the plan, it supersedes §1 rather than sitting beside it as +an equal alternative. + +**§2's result requires a stop-and-report, not a silent GO**: see §2.3. + +## 1. Planning-time probe (superseded — on-disk-walk baseline) + +Throwaway `/tmp` probes, not committed to the repo. **Environment:** Linux, 12 +cores, Python 3.12 (`.venv`), `spawn`, 2 MB batches. **Corpus:** this repo +including `.venv/` at planning time — 9605 indexable files / 97.4 MB, 5064 files +/ 65.8 MB qualifying. + +| Path | Wall clock | Speedup | +|---|---|---| +| serial in-process (on-disk walk) | 9.84 s | 1.00x | +| pool, 2 processes | 5.45 s | 1.81x | +| pool, 4 processes | 2.91 s | **3.38x** | +| pool, 8 processes | 2.06 s | 4.78x | + +Output parity `identical=True` at every process count. This number used a +**batch-submit harness** (`executor.map` over a pre-built list), not the shipped +bounded-look-ahead generator, and a source that read from an already-extracted +tree — both superseded by #106's single-pass tarball source. Recorded here for +provenance only; **do not cite as AC1's evidence.** + +## 2. Shipped measurement (real tarball, real `stream()`) + +`scripts/measure_extraction_pool.py --tarball repo.tar.gz --processes 2,4,8 +--repeat 3`, driving the actual `ExtractionPool.stream()` a production branch +calls, over `indexer.ingest.iter_tar_source_files`. + +**Environment:** Linux, 12 cores (`os.sched_getaffinity`; no cgroup-v2 quota on +this box), Python 3.12.13 (`.venv`), `spawn`, 2 MB batches (the default +`_BATCH_BYTES`). + +**Corpus:** a real GitHub-codeload-shaped tarball of this repository's own +working tree at `5f290c6` (the branch point), including `.venv/` as a +site-packages-heavy large-repo proxy (`.venv/bin`'s one absolute-target symlink +excluded — `iter_tar_source_files` correctly rejects it, matching production +behavior for such a member). 4547 indexable files / 54.5 MB, of which **2749 +files / 45.7 MB qualify for parsing** (60% of files, 84% of bytes — a Python- +and-JS-heavy tree). Smaller than the planning-time corpus (the repo has grown +since, and `.venv/bin` is excluded), but the same shape of proxy. + +### 2.1 Result + +``` +$ uv run python scripts/measure_extraction_pool.py --tarball repo.tar.gz --processes 2,4,8 --repeat 3 +loading repo.tar.gz ... +4547 indexable files / 54.5 MB, 2749 qualify for parsing / 45.7 MB (ingest: 0.79s -- serial in the parent, both arms below) + + path extract_s total_s speedup note + serial 6.563 7.355 1.00x + pool x2 3.423 4.215 1.74x identical + pool x4 1.913 2.705 2.72x identical + pool x8 1.422 2.214 3.32x identical + +ingest (serial, shared by both arms): 0.79s of 7.35s serial total (11%) -- the floor AC1's 'combined' speedup above cannot cross, however many processes extract_processes uses. +``` + +Reproduced across three separate invocations (best-of-3 per data point each +time); the pattern is stable, not noise: + +| Run | pool x2 | pool x4 | pool x8 | +|---|---|---|---| +| repeat=1 | 1.81x | 2.75x | 3.22x | +| repeat=3 (a) | 1.73x | 2.75x | 3.22x | +| repeat=3 (b, tabulated above) | 1.74x | **2.72x** | 3.32x | + +Output parity: **`identical=True` at every process count, every run** — the +pooled result list compared element-wise against the serial `extract_file` list. + +### 2.2 Where the two numbers in the table come from + +- **`extract_s`** — the pool's own wall clock (`ExtractionPool.stream()` alone), + comparable to the planning-time probe's number. +- **`total_s`** — `extract_s` plus the ONE shared `ingest` cost (identical in + both arms, since both read the same pre-loaded `ParsedFile` list): this is the + number that maps onto a real branch's `phase timing … parse=…` field, because + `_timed_items` charges `ExtractionPool.stream()`'s pull time — which includes + blocking on a future — entirely to `parse` (proven by + `tests/unit/test_job.py::test_pool_engaged_attributes_stream_production_to_parse`, + T10). **`total_s`'s speedup is AC1's evidence**, not `extract_s`'s. + +`extract_s` alone clears 3x at 4 processes (6.563 / 1.913 = **3.43x**, ~86% +parallel efficiency for the CPU-bound work). `total_s` does not, because ingest +is a **fixed, unavoidably serial** 0.79s added to both the numerator and +denominator (Amdahl's law: at an 11% serial fraction, even the CPU-bound part +scaling perfectly to infinity caps combined speedup at 1/0.11 ≈ 9.1x, and at 4x +*ideal* extraction speedup the combined ceiling is 1/(0.11 + 0.89/4) ≈ 3.03x — +so the measured 2.72x reflects real, good parallel efficiency running into a +structural ceiling, not a bug in this implementation). + +### 2.3 The go/no-go decision: STOP AND REPORT, per §8.4 + +**AC1 ("≥3x on 4+ cores") is NOT met at 4 processes on the shipped measurement: +2.72–2.75x combined, reproduced across three runs.** At 8 processes it clears +the bar (3.22–3.32x), but AC1 asks for the bar to be cleared starting at 4, not +only eventually at 8. + +Per the plan's binding instruction (§8.4, R12), this is the exact condition +under which the executor must **stop and report to the operator rather than +quietly shipping a change whose stated acceptance criterion is a measurement it +missed**, and must not retune the corpus until the number cooperates. It has not +been retuned: this is the corpus described in §2's "Corpus" paragraph, measured +as found. + +This is a real, structural finding, not a defect in the pool's own parallel +efficiency (§2.2's Amdahl arithmetic shows the CPU-bound part scales well). It +is a consequence of #106: the file source is now a single serial +gzip-decompress + decode + filter pass that this pool cannot touch (§4.7.1 of +the plan), and on this corpus that pass is 11% of the serial total — enough to +keep the *combined* number under 3x at 4 processes even though the *extraction* +number alone clears it comfortably (3.43x). The plan names the likely +implication directly: *"it may mean the win now lives in parallelizing +*ingest*, which is a different issue."* + +**This blocks a clean GO claim for AC1 as literally worded.** The design itself +(D1–D8: fork-safety, `BrokenProcessPool` blast-radius containment, the +terminable preflight probe, bounded look-ahead, order preservation) is sound and +independently valuable — a shared process pool doubles-plus extraction +throughput at 4 processes and more than triples it at 8, which is a real win for +any dominant-repo run — but AC1's specific "≥3x on 4+ cores" bar is not met at +the "4" endpoint on this corpus. Reported here rather than adjusted to look +better. + +## 3. Serial-fraction split (for #109) + +The number that tells #109 whether raising `extract_processes` past 4 is worth +anything at all: **ingest is 0.79s of the 7.355s serial total (≈11%) on this +corpus.** #109 should treat ~9x as this pool's asymptotic combined-speedup +ceiling on a similarly-shaped corpus, not the naive "N cores → Nx" expectation. diff --git a/docs/runbooks/indexing-parallelism.md b/docs/runbooks/indexing-parallelism.md index f9ce380..ee82ebd 100644 --- a/docs/runbooks/indexing-parallelism.md +++ b/docs/runbooks/indexing-parallelism.md @@ -125,7 +125,11 @@ INFO indexer.job [acme/gadgets]: skipped acme/gadgets@main: already indexed at a Wall-clock for the whole run is bounded below by the single slowest repo, so if one repo dominates, raising `index_concurrency` will not help — that is Amdahl's law asserting itself at the repo level, and the fix is to exclude the repo or -accept the duration. +accept the duration. **Still true for `index_concurrency` specifically** — but +since #108 a single dominant repo is no longer bounded by one thread's `parse` +time: its files parse on every `extract_processes` core via the shared +extraction pool (§3), so raising *that* knob can move the giant's own duration, +even though raising `index_concurrency` cannot. **To decide whether tuning is worth it:** compare the total on the completion line against the sum of the per-repo elapsed times. If the total is already @@ -163,15 +167,29 @@ promise that the field set never changes across releases.) `#104` narrows the **db** and **embed** costs to a branch's actual delta, not its size — but it does NOT touch `parse`: extraction still runs on every file every run (tree-sitter must produce a `FileExtraction` before -`index_repo` can classify it), so an all-unchanged branch on a large repo -still pays its full `download`+`parse` cost. See `indexer.store`'s -`delta write set …` line (below) to tell "this branch is genuinely mostly-new" -from "this branch is mostly-unchanged but still parsing everything" — the -latter is exactly the case `#108` (process-pool extraction) or a future -extraction-skip step would address next. - -Four fields need interpretation before you act on them: - +`index_repo` can classify it, and `index_repo`'s classification happens +*downstream* of extraction — the `FileExtraction` for an unchanged file is +computed and then discarded), so an all-unchanged branch on a large repo still +pays its full `download`+`parse` cost. `#108` (process-pool extraction) makes +that cost cheaper by spreading it across cores — it does NOT skip it: every +file is still parsed every run, including the unchanged ones. See +`indexer.store`'s `delta write set …` line (below) to tell "this branch is +genuinely mostly-new" from "this branch is mostly-unchanged but still parsing +everything on every core" — deferring extraction past classification for the +unchanged fraction is a real, larger follow-up win, deliberately left out of +`#108`'s scope (it would change `index_repo`'s `items` contract). + +Five fields need interpretation before you act on them: + +- **`parse=` can RISE on a single branch after #108, even though the RUN's + total wall clock falls.** The shared extraction pool (§3) is shared across + every `index_concurrency` repo worker, so time a branch spends blocked in + `ExtractionPool.stream()` now includes queueing behind OTHER branches' + batches, not just this branch's own extraction. Reading one branch's `parse=` + in isolation and comparing it to a pre-#108 run will therefore sometimes look + like a regression when it is not — compare **run totals and the dominant + repo** instead (see `docs/perf/issue-108-measurements.md` for the shape of + that comparison). - **`resolve=0.00s` on a default branch is expected, not a bug.** That branch's HEAD SHA came from the repo-level resolve, which happens once per repo outside every branch's total and is reported on the repo's `finished` line as @@ -300,6 +318,7 @@ source edit and redeploy, never a 2am incident-response lever. | `index_concurrency` | 1..8, default **4** | Repos in flight | | `MAX_TARBALL_BYTES` | 500 MB | The compressed download, per worker | | `MAX_EXTRACTED_BYTES` | 2 GB | The streamed uncompressed content, per branch | +| `extract_processes` (#108) | 1..8, default: derived (affinity/cgroup, capped 8) | Symbol/edge extraction worker processes, shared across ALL `index_concurrency` repo workers | Only the first of the two byte caps is a **disk** cap. Since #106 the tarball is streamed once, in memory, and is never extracted, so `MAX_EXTRACTED_BYTES` is a @@ -316,13 +335,17 @@ the compressed tarball is the only artifact on disk, so peak local disk is | **4 (default)** | **2 GB** | | 8 (ceiling) | 4 GB | -**Returns at the ceiling are sublinear; the disk cost is not.** Symbol -extraction was measured at **0.95x on 4 threads** — the tree walk is -GIL-serialized and is ~56% of extraction time, so Amdahl's law caps the speedup -well below 8x. Meanwhile the 4 GB is a hard, linear, unavoidable cost. Raise -`index_concurrency` to 8 only knowing you are buying a fraction of a speedup -with a doubling of disk. (#106 lowered these numbers by 5x but deliberately did -**not** move the default of 4; re-deriving it is #109's job.) +**`index_concurrency` no longer bounds extraction throughput; the disk cost is +still linear.** Symbol extraction was measured at **0.95x on 4 threads** — the +tree walk is GIL-serialized, so raising `index_concurrency` never bought +extraction speedup, only more repos in flight at once. That measurement is +exactly *why* extraction now runs in its own shared **process** pool instead +(`extract_processes`, below) — decoupled from `index_concurrency` entirely. +Raise `index_concurrency` only for repo-level (disk-bound) fan-out; raise +`extract_processes` for CPU-bound extraction throughput. The 4 GB disk figure +above is still a hard, linear, unavoidable cost of `index_concurrency` alone. +(#106 lowered these numbers by 5x but deliberately did **not** move the default +of 4; re-deriving it is #109's job.) **Semantic indexing clamps the pool to 2**, regardless of `index_concurrency`. That clamp is a *memory* bound, not a CPU one: embedding materialises a whole @@ -347,6 +370,84 @@ under the SDK's 20-connection pool. thread pool spawned. See `docs/runbooks/semantic-enablement.md` §4 for the full in-flight/memory arithmetic and the 429 posture. +### Extraction process pool (#108) + +Symbol/edge extraction runs in a **shared, `spawn`-based process pool** — one +pool per run, built once and shared across every `index_concurrency` repo +worker, so a single dominant repo's files parse on every available core instead +of being bound to one thread. This is a **CPU** knob, entirely independent of +`index_concurrency`'s disk bound and the semantic clamp above: it adds no new +corpus writer and does not change per-branch/per-repo sequencing (see §1.1) — +"process pool" reads like a concurrency change to anyone who has internalized +this runbook's §1.1, and it is not one. + +`extract_processes` (unset, the default) derives from the runtime: +affinity/cgroup-aware CPU count, capped at 8 — the same ceiling as every other +parallelism knob here. **Set it to `1` to disable the pool entirely** — fully +serial, in-process extraction, no worker processes spawned at all — mirroring +`embedding_concurrency: 1`'s rollback shape above. This is the 2am escape +hatch if the pool ever misbehaves in this runtime; it is config-only, needs no +redeploy, no migration, and no re-index. + +**How to tell whether the pool engaged**, once per run beside the disk line: + +``` +INFO indexer.extract_pool [-]: symbol extraction: 4 process(es) (spawn); pool preflight ok +``` + +Absent that line (or a `WARNING` in its place, also from the `indexer.extract_pool` +logger), extraction ran in-process — either by config (`extract_processes: 1`) +or because the pool degraded. Three WARNING shapes to recognize: + +- **`extraction pool preflight failed: ...`** — the pool never engaged for this + run at all (a bad sandbox, `/dev/shm` too small, an unguarded `__main__` + under `spawn`). The run proceeds at today's (pre-#108) speed. (A killed probe + worker may also print a stderr line like `resource_tracker: There appear to + be 1 leaked semaphore objects to clean up at shutdown` — harmless noise from + the kill path, not a separate problem.) +- **`... rebuilt the pool (generation N, rebuild M/3)`** — a worker died + (`BrokenProcessPool`, e.g. a native crash in a grammar, an OOM kill). The + branch(es) in flight at that moment failed (up to `index_concurrency` + branches, semantic-clamped to 2 — **not just one**: the pool is shared, so a + break can surface on every repo worker holding a future at that instant). + Each failed branch re-indexes on its next run (it never got a stamp — the + same self-healing property §1 describes). The pool is rebuilt and later + branches run normally. +- **`... rebuild budget (3) exhausted; latching to in-process extraction`** — a + deterministically poisoned file re-broke the pool three times running. The + rest of THIS run finishes in-process (slower, but correct and complete); the + next run gets a fresh pool. + +**Small repos engage fewer workers than `extract_processes` requests, and that +is not a bug.** Batching is by 2 MB of aggregate qualifying-file content, not +file count — a repo with less than `extract_processes x 2 MB` of parseable +content never fills every worker. Such repos are fast regardless; watching one +show fewer active workers than configured does not mean the pool "isn't +working". + +**Ingestion stays serial — the pool cannot parallelize it.** Since #106 the tar +stream is decompressed, decoded, and filtered in ONE pass on the repo-worker +thread (`indexer.ingest.iter_tar_source_files`); the pool only parallelizes +`extract_file` itself. On a real measured corpus this serial pass was ~11% of +the pre-#108 serial total — small, but by Amdahl's law it is a hard ceiling on +the *combined* (ingest + extract) speedup this pool can ever deliver, however +many processes `extract_processes` uses. See +`docs/perf/issue-108-measurements.md` for the full measurement and why it means +AC1's "≥3x on 4+ cores" bar is met at 8 processes but not at 4 on that corpus — +a real, structural finding, not a defect in the pool's own parallel efficiency +(extraction alone scales at ~86% efficiency at 4 processes; the ceiling is the +serial ingest pass, not the pool). + +**Memory**, added to the peak-usage arithmetic #109 inherits: each extraction +worker process peaks at roughly 121 MB RSS (measured, all 7 grammars touched), +so the pool adds `extract_processes x ~121 MB` on top of everything above — +~484 MB at the default-derived 4, ~1.0 GB at the ceiling of 8. The pool's own +bounded look-ahead window (scaled by `2 x extract_processes`: up to 2 MB of +aggregate content per repo worker, plus a file-count backstop for near-zero- +byte files) adds a few tens of MB per in-flight branch on top of that, and +sits alongside (not instead of) the batched-write consumer's own up-to-8-MiB +retention (#105, §2.3) for the same in-flight branch. + ### The connection pool follows the workers Each worker holds exactly one connection, so the engine is built with diff --git a/indexer/AGENTS.md b/indexer/AGENTS.md index 4477922..ab62cb5 100644 --- a/indexer/AGENTS.md +++ b/indexer/AGENTS.md @@ -10,16 +10,17 @@ The serverless indexing job (`code-search-index`, wired as a `python_wheel_task` | File | Description | |------|-------------| | `__init__.py` | Empty package marker. | +| `extract_pool.py` | `ExtractionPool` (#108): a shared, `spawn`-based `ProcessPoolExecutor` for symbol/edge extraction — one pool per run, built in `job.run()` and threaded down to every branch, so a dominant repo's files parse on every `extract_processes` core instead of one thread's worth. `stream(files)` batches qualifying files (per `_needs_parse`, mirroring `extract_file`'s own short-circuit) by aggregate content bytes (`_BATCH_BYTES`, 2 MB), bounds look-ahead by two independent budgets scaled by `2 x extract_processes` (pending content bytes, and pending file count — a backstop for near-zero-byte files, which the byte budget alone would not throttle), and yields `(ParsedFile, FileExtraction)` pairs in source order. `extract_processes: 1` (or a failed preflight probe) degrades `stream()` to plain in-process `extract_file` calls — no pool built at all. A `BrokenProcessPool` raises `ExtractionPoolError` (chained), caught by `_index_one_branch`'s existing broad `except`; the supervisor rebuilds once per generation (CAS under a lock, bounded by `MAX_POOL_REBUILDS = 3`) then latches to in-process. The preflight probe (`build_extraction_pool`) is a bare, terminable `multiprocessing.Process` run BEFORE any executor exists — never `executor.submit()` + timeout, which cannot be cancelled and hangs the interpreter at exit. Not watched by the semantics tripwire: it changes *where* extraction runs, never *what* is extracted. | | `branches.py` | Resolves a connection's branch globs against a repo's real branch list via `fnmatchcase`, returning a typed `BranchResolution` (`branches`, `complete`, `dropped`, `cap`). Default branch always included and always first; empty globs means default-branch-only with no GitHub branches API call, always `complete=True`. `SOFT_BRANCH_CAP = 20` truncates (loud warning naming the cap and dropped branches, not a failure; sets `complete=False`, blocking reconciliation for that repo — #61); no override flag — fix the config. | | `bulk.py` | `insert_rows` (#105): param-budgeted multi-row `INSERT` helper shared by `store.py` and `chunk_store.py` — explicit `.values([...])` rather than executemany, so the round-trip count is exactly one statement per slice on every driver. `PARAM_BUDGET` (30,000) and `CHUNK_PARAM_BUDGET` (6,000, payload- not param-bounded — each chunk row carries a 1024-float vector). No-op on an empty row list. | | `chunk_store.py` | `write_chunks_batch` (#105): delete-and-reinsert MANY files' rows in the `chunks` table in one `DELETE ... WHERE file_id = ANY(:ids)` (over every id in the call, including zero-chunk files) plus one param-budgeted bulk insert via `bulk.insert_rows`. `write_chunks` is now a one-element wrapper over it (signature/return unchanged). Takes a live `Connection`; never opens an engine, never calls the embedder — vectors arrive precomputed. No `repo_id` parameter: `chunks` is scoped by `file_id` only. | | `fetch.py` | All GitHub HTTP: paginated org/user repo enumeration (`RepoMeta`), branch listing, `resolve_ref`/`resolve_branch_head` (branch name -> immutable SHA), streamed tarball download capped at `MAX_TARBALL_BYTES` (500 MB), `assert_disk_headroom` (0.5 GB per worker — the compressed tarball is the only artifact written to disk). No extraction: `MAX_EXTRACTED_BYTES` moved to `ingest.py` with #106 and is now a work cap, not a disk cap. `RateLimitError` is deliberately narrow: 429 always; 403 only with `Retry-After` or `X-RateLimit-Remaining: 0` — other 403s are permission failures. | | `ingest.py` | `iter_tar_source_files` (#106): the job's only file source. ONE forward pass over the downloaded tarball (`tarfile.open(mode="r:*")` driven by `tf.next()`; no `getmembers()`, no `extractall()`, nothing written to disk), yielding the same `ParsedFile` stream `parse.iter_source_files` would have yielded from an extracted tree — same filter chain, reusing `parse._looks_binary` by import so the binary-sniff rule cannot fork. `size` is `member.size` (the archive-declared length, which exceeds `len(content)` after NUL stripping); `lang` is the RELATIVE path's lowered suffix. Yields in ARCHIVE order (deterministic per commit; nothing downstream depends on the sequence — see the module docstring's D5 note). `tf.members.clear()` is the FIRST statement of the loop body, since the `TarFile` now outlives `engine.connect()`. Replaces `filter="data"` with explicit, probe-verified rules: raw-`..`-component and absolute member names raise, absolute/escaping link targets raise for hardlinks and symlinks alike, special files and benign hardlinks skip with a WARNING, duplicate names are first-wins, and a zero-member archive raises rather than silently stamping the branch as indexed. `MAX_EXTRACTED_BYTES` (2 GB) is enforced incrementally over pre-filter member sizes. | | `hashing.py` | `content_sha`: canonical SHA-256 hex of content (`None` -> empty string). Single source of truth for `files.content_sha`; must stay byte-identical to the `0003` migration's SQL backfill forever or cross-branch dedup silently breaks (`tests/integration/test_content_sha_parity.py` is the gate). | -| `job.py` | Entry point and orchestration: `main()`/`run()`, `ThreadPoolExecutor` sized by `effective_workers` (unit of work = one repo, all branches sequential), batched `repo_branches` stamp read, per-branch skip-if-unchanged, per-branch `BranchOutcome` classification (`indexed`/`skipped`/`conflict`/`failed`), semantic precompute (`_precompute_chunk_writer`), `ContextVar`-based `[repo]` log attribution, and per-phase timing instrumentation (one `phase timing repo@branch: total=… resolve=… download=… parse=… embed=… db=… sweep=… other=…` INFO line per INDEXED branch, emitted from inside the worker so `[repo]` resolves; fixed unconditional field set; `_timed_items` charges lazy item production to `parse` so it is not fused into `db`; `resolve=`/`list=` on the per-repo `finished` line carry the repo-scoped costs that belong to no branch's total). Returns 1 if any branch failed; conflicts self-heal and do not fail the run. **Reconciliation checkpoint**: after every worker joins, `_decide_reconciliation` gates on zero failures/conflicts, every repo accounted for, and every repo's `discovery_complete`; on a pass, `_reconcile` reuses the pre-fan-out stamp snapshot to compute each repo's retired branches and the run's full desired repo set, then calls the injected `reconcile_retired_fn`/`reconcile_removed_fn` seams (default: the real `indexer.store` primitives) on one post-fan-out connection — retired branches first, repo purge second. `MAX_PURGE_SHRINK_FRACTION = 0.5` (hardcoded, no config knob) withholds ONLY the purge, as a logged incident signal, if it would remove a strict majority of currently stored repos; retired-branch cleanup on survivors still applies. `ReconcileProgress.committed_any` drives an honest "partially reconciled" vs "left stale" failure message — never the raw exception, only its phase/repo/type name. Never called from worker-thread code (`_index_one*`) — a source-level tripwire test enforces it. | +| `job.py` | Entry point and orchestration: `main()`/`run()`, `ThreadPoolExecutor` sized by `effective_workers` (unit of work = one repo, all branches sequential), a shared `ExtractionPool` (#108) built once per run and threaded through `_index_one` → `_index_one_inner` → `_index_one_branch` (`run(..., extraction_pool=None)` injectable, `owns_pool` mirrors `owns_engine`/`owns_http`, shut down in `finally` before `engine.dispose()`), batched `repo_branches` stamp read, per-branch skip-if-unchanged, per-branch `BranchOutcome` classification (`indexed`/`skipped`/`conflict`/`failed`), semantic precompute (`_precompute_chunk_writer`), `ContextVar`-based `[repo]` log attribution, and per-phase timing instrumentation (one `phase timing repo@branch: total=… resolve=… download=… parse=… embed=… db=… sweep=… other=…` INFO line per INDEXED branch, emitted from inside the worker so `[repo]` resolves; fixed unconditional field set; `_timed_items` charges lazy item production to `parse` so it is not fused into `db`; `resolve=`/`list=` on the per-repo `finished` line carry the repo-scoped costs that belong to no branch's total). Returns 1 if any branch failed; conflicts self-heal and do not fail the run. **Reconciliation checkpoint**: after every worker joins, `_decide_reconciliation` gates on zero failures/conflicts, every repo accounted for, and every repo's `discovery_complete`; on a pass, `_reconcile` reuses the pre-fan-out stamp snapshot to compute each repo's retired branches and the run's full desired repo set, then calls the injected `reconcile_retired_fn`/`reconcile_removed_fn` seams (default: the real `indexer.store` primitives) on one post-fan-out connection — retired branches first, repo purge second. `MAX_PURGE_SHRINK_FRACTION = 0.5` (hardcoded, no config knob) withholds ONLY the purge, as a logged incident signal, if it would remove a strict majority of currently stored repos; retired-branch cleanup on survivors still applies. `ReconcileProgress.committed_any` drives an honest "partially reconciled" vs "left stale" failure message — never the raw exception, only its phase/repo/type name. Never called from worker-thread code (`_index_one*`) — a source-level tripwire test enforces it. | | `languages.py` | Single source of truth shared by `parse.py` and `symbols.py`: `EXT_TO_LANG` (8 extensions -> tree-sitter language names), `SYMBOL_KINDS` (node type -> symbol kind per language), `EDGE_NODE_KINDS` (node type -> reference-edge kind per language, Python-only until #85), `MAX_FILE_BYTES` (1 MB), `SEMANTIC_CHUNK_MAX_CHARS` (2000, ~4 chars/token), and the frozen dataclasses `ParsedFile`, `Chunk`, `ExtractedSymbol`, `ExtractedEdge`, `FileExtraction`, `IndexCounts` (now carries `edges`). | | `parse.py` | `iter_source_files`: **no production caller since #106** — retained deliberately as the extraction ORACLE the streaming path (`ingest.py`) is pinned against by `tests/unit/test_ingest_parity.py`, and undeletable anyway without editing a `SEMANTICS_PATHS` module and forcing an `INDEX_SEMANTICS_VERSION` bump. Walk an extracted tree, yield every text file (unknown extensions kept with `lang=None`, since grep runs over all files); skips `.git/`, symlinks, files > `MAX_FILE_BYTES` (stat before read), NUL-sniffed binaries, and UTF-8 decode failures; strips surviving NULs (Postgres `text` rejects them). `iter_chunks`: deterministic line-aligned chunking, no overlap, no mid-line splits, 1-based inclusive line ranges. | -| `repo_config.py` | Pydantic schema for `config.yaml` (`RepoConfig` / `GitHubConnection` / `ExcludeRules`), `normalize_repo` (URL/SSH/bare -> canonical `org/repo`, GitHub hosts only), `parse_config` (pure) vs `read_workspace_config` (SDK I/O, wraps every failure in `ConfigError` with HTTP status so 404 "never synced" stays distinguishable from 403 "no permission"), and `effective_workers` (the semantic clamp). A connection with no `orgs`/`users`/`repos` fails validation. `index_concurrency`: 1-8, default 4 — a disk bound (0.5 GB peak per worker since #106), not a CPU one. `semantic_max_chunks_per_repo`: optional top-level `{"org/repo": N}` map overriding `app.config.Settings.semantic_max_chunks_per_repo` per repo; keys canonicalised via `normalize_repo` and rejected on collision post-casefold, values `>= 1`. `SemanticOverrides.embedding_concurrency` (#107): optional `1..8`, overlays `Settings.semantic_embedding_concurrency` (in-flight embedding requests per worker; job-only, inert on the MCP/webui query path). Deliberately import-light: pydantic + PyYAML + stdlib only. | +| `repo_config.py` | Pydantic schema for `config.yaml` (`RepoConfig` / `GitHubConnection` / `ExcludeRules`), `normalize_repo` (URL/SSH/bare -> canonical `org/repo`, GitHub hosts only), `parse_config` (pure) vs `read_workspace_config` (SDK I/O, wraps every failure in `ConfigError` with HTTP status so 404 "never synced" stays distinguishable from 403 "no permission"), and `effective_workers` (the semantic clamp). A connection with no `orgs`/`users`/`repos` fails validation. `index_concurrency`: 1-8, default 4 — a disk bound (0.5 GB peak per worker since #106), not a CPU one. `semantic_max_chunks_per_repo`: optional top-level `{"org/repo": N}` map overriding `app.config.Settings.semantic_max_chunks_per_repo` per repo; keys canonicalised via `normalize_repo` and rejected on collision post-casefold, values `>= 1`. `SemanticOverrides.embedding_concurrency` (#107): optional `1..8`, overlays `Settings.semantic_embedding_concurrency` (in-flight embedding requests per worker; job-only, inert on the MCP/webui query path). `extract_processes` (#108): optional `1..8`, sizes the shared extraction-pool process count (`indexer.extract_pool.derive_process_count`); `None` (default) derives from affinity/cgroup CPU count capped at 8; `1` is the kill switch (no pool built at all). A CPU knob, independent of `index_concurrency`'s disk bound and the semantic clamp. Deliberately import-light: pydantic + PyYAML + stdlib only. | | `resolve.py` | `resolve_repos`: enumerate org/user selectors, apply `ExcludeRules` to enumerated repos only (explicit `repos` entries always win, unfiltered), dedup case-insensitively keeping first-seen spelling, union each connection's `branches:` globs into per-repo `RepoEntry.branch_globs`, then fail fast: `EmptyConfigError` on zero repos (indexing nothing must not exit 0), `RepoCeilingError` above `MAX_REPOS` (500, overridable via `--max_repos`). Also matches `RepoConfig.semantic_max_chunks_per_repo` case-insensitively onto each resolved repo's `RepoEntry.semantic_max_chunks` (`None` if unmatched); an override key matching no resolved repo logs a WARNING (typo guard). Enumerator params are a test seam only, not provider dispatch. | | `store.py` | `index_repo`: the single atomic unit of work for one (repo, branch) inside `with conn.begin():` — repos upsert, `repo_branches` CAS-baseline read, per-file classification (unchanged / membership-only / changed-new), membership sweep (strip this branch from unseen rows, delete rows with empty `branches`; skipped with a WARNING on an empty seen-set), then CAS stamp (raises `StaleIndexError` on mismatch, rolling everything back). `items` is `Iterable[tuple[ParsedFile, FileExtraction]]`. **Changed/new files are batched (#105)**, not written one at a time: `_flush_file_batch` accumulates up to `_BATCH_MAX_FILES` files or `_BATCH_MAX_CONTENT_BYTES` bytes of `pf.size` (whichever trips first, post-append — a single oversized file is never split), then issues ONE multi-row `files` upsert (`SET` uses `excluded.*` for every per-row column, NEVER a Python literal from one file — the trap that would collapse a batch's conflicting rows onto the last file's values), maps ids back from `RETURNING` by `(path, content_sha)` — never by row order, and RAISES on a missing id rather than warn-and-skip (a wrong id here would attach one file's symbols to another file's row) — then ONE `DELETE ... WHERE file_id = ANY(:ids)` + bulk insert each for `symbols`/`reference_edges` (`reference_edges` unconditional, even when a batch's `FileExtraction.edges` is all-empty, so stale rows never survive a re-index), then ONE `chunk_writer` call for the batch. An intra-batch duplicate `(path, content_sha)` is deduped first (last occurrence wins, one WARNING) — Postgres raises `ON CONFLICT DO UPDATE command cannot affect row a second time` and poisons the transaction otherwise. The sweep call site is wall-clocked into `indexer.timing`'s ambient `PhaseTimer` (`record("sweep", …)`) — never into `IndexCounts` (frozen, compared by value) and never through a new `index_repo` parameter (that signature is an injected seam); `store.py` emits **no** log record for it. Pure DML, no TEMP tables (job role has no TEMP privilege on Lakebase). `reconcile_retired_branches` and `reconcile_removed_repos`: the desired-state storage primitives `job.py`'s reconciliation checkpoint calls — each its own `conn.begin()`, repo-scoped membership subtraction (never delete-by-path) and an exact-match repo purge respectively; neither decides WHAT is retired/desired, only applies a caller-supplied set (see `job.py`'s row). | | `symbols.py` | `extract_file` (#84): one tree-sitter parse, one full-tree walk, emitting both `symbols` and `reference_edges` candidates (`FileExtraction`) in the same pass — nested definitions captured, named nodes only for symbols, kinds from `SYMBOL_KINDS`/`EDGE_NODE_KINDS`, 1-based lines. Stack entries carry the innermost named enclosing definition so edges attribute to it in O(1) with no second walk. `extract_symbols` is a thin wrapper (`extract_file(pf).symbols`) kept for the existing call sites. Python-only edge helpers (`_python_call_target`, `_python_import_edges`) are the seam #85 generalizes to other languages. Parser cache is per-thread (`threading.local`) as insurance against a future GIL-releasing `parse()`. | @@ -40,9 +41,10 @@ The serverless indexing job (`code-search-index`, wired as a `python_wheel_task` - **Test seams are injected callables**, not registries: `run()` accepts `workspace_client`, `http_client`, `engine`, `index_fn`, `embed_fn`, `config_loader`, `reconcile_retired_fn`, `reconcile_removed_fn`; `resolve_repos` accepts enumerators; `store.index_repo`/`reconcile_retired_branches`/`reconcile_removed_repos` and `chunk_store.write_chunks` take a live `Connection`. `resolve_repos` itself is deliberately NOT injectable in `run()` — tests drive it through `httpx.MockTransport`. - **The schema is provider-extensible; the code is GitHub-only.** No `type`-keyed dispatch anywhere — adding GitLab means a new union member plus a new fetch implementation, not a provider abstraction. - **Import discipline**: `repo_config.py` stays import-light (no httpx/SQLAlchemy/SDK at module level); `normalize_repo` is imported from `repo_config`, never from `job` (import cycle); `app.embed`'s databricks-sdk dependency is only imported when semantic is on. +- **Extraction runs in a shared, `spawn`-based process pool (#108)**: workers are pure `ParsedFile -> FileExtraction` — no logging, no DB, no network, and no `indexer.timing.record()` (the ambient `ContextVar` does not cross a process boundary, as `timing.py`'s own docstring states, naming this issue). `stream()`'s source generator (`indexer.ingest.iter_tar_source_files`, a single-pass `TarFile`) is pulled STRICTLY from the calling thread — never add a prefetch/warm-up thread to advance it; a second thread interleaving `tf.next()` calls corrupts a mid-stream `TarFile` silently, not as an exception. A broken pool fails the branches in flight (up to `effective_workers` of them, not just one — the pool is shared), self-heals by rebuilding (generation-CAS'd, bounded), then latches to in-process extraction. Never `multiprocessing.set_start_method(...)` — that mutates global interpreter state a Databricks `python_wheel_task` wrapper may depend on; pass `spawn` as an explicit context object instead. ### Testing Requirements -- `make test` (`pytest -m "unit or observability"`, no external deps): `tests/unit/test_branches.py`, `test_bulk.py`, `test_chunk_store.py`, `test_chunking.py`, `test_edges.py`, `test_fetch.py`, `test_ingest.py`, `test_ingest_parity.py`, `test_job.py`, `test_job_redaction.py`, `test_languages.py`, `test_parse.py`, `test_repo_config.py`, `test_resolve.py`, `test_symbols.py`, `test_store_batching.py`, `test_store_chunk_writer.py`, `test_store_delta.py`, `test_semantics_version_tripwire.py`, `test_timing.py`. +- `make test` (`pytest -m "unit or observability"`, no external deps): `tests/unit/test_branches.py`, `test_bulk.py`, `test_chunk_store.py`, `test_chunking.py`, `test_edges.py`, `test_extract_pool.py`, `test_fetch.py`, `test_ingest.py`, `test_ingest_parity.py`, `test_job.py`, `test_job_redaction.py`, `test_languages.py`, `test_parse.py`, `test_repo_config.py`, `test_resolve.py`, `test_symbols.py`, `test_store_batching.py`, `test_store_chunk_writer.py`, `test_store_delta.py`, `test_semantics_version_tripwire.py`, `test_timing.py`. `test_job.py`'s shared `_config()` helper pins `extract_processes=1` (the pool kill switch) by default, so its ~150 `run()`-driving tests spawn no worker processes; only `test_extract_pool.py` and a handful of dedicated pool-lifecycle tests in `test_job.py` opt into a real pool. - `make test-integration` (needs Postgres): `tests/integration/test_store.py`, `test_store_batching.py` (#105, batch-size parity/excluded-trap/rollback/statement-count — creates no `chunks` table, runs locally), `test_chunk_batching.py` (#105, batched `write_chunks_batch` on bare `vector`, runs locally), `test_store_chunk_writer.py`, `test_content_sha_parity.py` (the hard gate on `hashing.py` vs the migration backfill), `test_reconcile.py` (the `reconcile_retired_branches`/`reconcile_removed_repos` storage primitives directly), and `test_job_reconcile.py` (#59: the real `run()` end to end — real primitives, real engine, only GitHub HTTP faked). - Schema tests must stay fast: changes to `repo_config.py` must not add heavy imports. @@ -51,7 +53,7 @@ The serverless indexing job (`code-search-index`, wired as a `python_wheel_task` - `pg_insert(...).on_conflict_do_update(...).returning(...)` with a no-op `SET` on conflict — `DO NOTHING ... RETURNING` returns no row on conflict and would break the id bootstrap. - Delete-and-reinsert for child rows with no natural key (symbols, chunks). - `fnmatchcase`, never `fnmatch`: plain globs must behave identically on every platform. -- Guardrail constants with config-level fixes, not override flags: `SOFT_BRANCH_CAP` (20), `MAX_REPOS` (500, the one exception — `--max_repos`), `MAX_FILE_BYTES`, `MAX_TARBALL_BYTES` (`fetch.py`, a disk cap), `MAX_EXTRACTED_BYTES` (`ingest.py` since #106, a WORK cap — nothing is written to disk, so it bounds streamed content, not storage), `MAX_PURGE_SHRINK_FRACTION` (0.5, `job.py` — withholds a corpus-wide repo purge above this shrink fraction; fix is staged config removal, never a knob), `_BATCH_MAX_FILES` (500) / `_BATCH_MAX_CONTENT_BYTES` (8 MiB, `store.py` since #105 — the changed/new write-batch bounds; `_BATCH_MAX_FILES * _FILE_UPSERT_COLUMNS < 65535` is asserted at import time against libpq's Bind param ceiling), `PARAM_BUDGET` (30,000) / `CHUNK_PARAM_BUDGET` (6,000, `bulk.py` since #105 — the generic multi-row-insert slicing bounds). +- Guardrail constants with config-level fixes, not override flags: `SOFT_BRANCH_CAP` (20), `MAX_REPOS` (500, the one exception — `--max_repos`), `MAX_FILE_BYTES`, `MAX_TARBALL_BYTES` (`fetch.py`, a disk cap), `MAX_EXTRACTED_BYTES` (`ingest.py` since #106, a WORK cap — nothing is written to disk, so it bounds streamed content, not storage), `MAX_PURGE_SHRINK_FRACTION` (0.5, `job.py` — withholds a corpus-wide repo purge above this shrink fraction; fix is staged config removal, never a knob), `_BATCH_MAX_FILES` (500) / `_BATCH_MAX_CONTENT_BYTES` (8 MiB, `store.py` since #105 — the changed/new write-batch bounds; `_BATCH_MAX_FILES * _FILE_UPSERT_COLUMNS < 65535` is asserted at import time against libpq's Bind param ceiling), `PARAM_BUDGET` (30,000) / `CHUNK_PARAM_BUDGET` (6,000, `bulk.py` since #105 — the generic multi-row-insert slicing bounds), `_BATCH_BYTES` (2 MB, `extract_pool.py` since #108 — the pool's IPC-amortization batch bound; deliberately `len(pf.content)`, NOT `pf.size` like `store.py`'s batch bound above — one bounds a pickled IPC payload, the other a `files` write payload, and they are allowed to diverge since #106 made `size` exceed `len(content)`) / `MAX_POOL_REBUILDS` (3, `extract_pool.py` — the shared pool's rebuild-then-latch budget). - Errors name the fix: exceptions carry the config key or job parameter to change (`exclude.size_mb`, `index_concurrency`, `--max_repos`). - Broad `except Exception` at classification boundaries only (`run()`'s config/resolve handlers, `_index_one_branch`), each with a comment explaining why broad is load-bearing. - Lazy generators through the open transaction for bounded memory (semantic mode is the exception — it materializes the file list because embedding needs all chunk texts up front). @@ -71,6 +73,6 @@ The serverless indexing job (`code-search-index`, wired as a `python_wheel_task` - `pydantic` + `pyyaml` (config schema and parsing) - `tree-sitter` / `tree-sitter-language-pack` (symbol extraction) - `databricks-sdk` (workspace config read + secrets, injected/lazy — never imported at module level here) -- stdlib: `tarfile` (single-pass in-memory streaming in `ingest.py`; no `extractall`, so no `filter="data"` — the rejection rules are explicit there instead), `posixpath` (member-name and link-target safety), `concurrent.futures`, `tempfile`, `shutil`, `fnmatch`, `hashlib`, `argparse`, `contextvars` +- stdlib: `tarfile` (single-pass in-memory streaming in `ingest.py`; no `extractall`, so no `filter="data"` — the rejection rules are explicit there instead), `posixpath` (member-name and link-target safety), `concurrent.futures` (repo-worker thread pool; `ProcessPoolExecutor` for the shared extraction pool since #108), `multiprocessing` (`extract_pool.py`'s `spawn` context and preflight probe, since #108), `tempfile`, `shutil`, `fnmatch`, `hashlib`, `argparse`, `contextvars` diff --git a/indexer/extract_pool.py b/indexer/extract_pool.py new file mode 100644 index 0000000..ee178f6 --- /dev/null +++ b/indexer/extract_pool.py @@ -0,0 +1,635 @@ +"""Shared spawn-based process pool for symbol/edge extraction (issue #108). + +``extract_file`` (``indexer.symbols``) is CPU-bound tree-sitter work that holds +the GIL through ``parse()``, so a repo's own worker THREAD (see ``indexer.job``) +cannot parallelize it. This module decouples extraction from that thread: one +:class:`ExtractionPool` is built once per :func:`indexer.job.run` and shared +across every repo worker, so a single giant repo's files parse on every +available core instead of one thread's worth. + +**Not watched by the semantics tripwire, deliberately.** This module changes +*where* extraction runs, never *what* is extracted -- it imports and calls the +unmodified :func:`indexer.symbols.extract_file` for every file, so output is +identical by construction. It is therefore NOT added to +``tests/unit/test_semantics_version_tripwire.py``'s ``SEMANTICS_PATHS``: that +watch set is for modules that decide extraction *output*, not for a change in +execution topology. See ``indexer/AGENTS.md`` for the ``SEMANTICS_PATHS`` list. + +Four load-bearing design decisions, each with a measured or reproduced failure +mode behind it (see the execution plan for the full writeup): + +* **``spawn``, never ``fork``/``forkserver``.** ``ProcessPoolExecutor`` creates + worker processes lazily on first ``submit()``, so building the pool early does + NOT make ``fork`` safe -- the fork would still happen mid-run, from a process + that by then holds a repo-worker ``ThreadPoolExecutor``, a live SQLAlchemy + engine, an ``httpx.Client``, and (issue #107) a per-embed + ``ThreadPoolExecutor``. Forking a threaded process inherits locks in + indeterminate states -- the classic symptom is a silent stall, not a crash. + ``spawn`` has no inherited-lock or inherited-fd hazard and is passed as an + explicit ``multiprocessing.get_context("spawn")`` object; this module never + calls ``multiprocessing.set_start_method`` (that mutates global interpreter + state a Databricks ``python_wheel_task`` wrapper may depend on). +* **The preflight probe is a bare, terminable ``multiprocessing.Process``, run + BEFORE any ``ProcessPoolExecutor`` exists -- never ``executor.submit()`` + + ``future.result(timeout=...)``.** Reproduced: a timed-out task cannot be + cancelled (``future.cancel()`` returns ``False``, already running), so + ``executor.shutdown(wait=True)`` never returns, and even abandoning the + executor still hangs the interpreter at exit (``concurrent.futures.process`` + registers an atexit hook that joins the executor manager thread). Since + ``resources/job.yml`` sets ``max_concurrent_runs: 1`` with no + ``timeout_seconds``, an executor-based probe that hangs does not degrade to + "a WARNING and a run at today's speed" -- it blocks every queued run + indefinitely until a human intervenes. A bare ``Process`` can be ``kill()``ed + on a timeout, which an executor cannot offer. +* **A broken pool is a per-branch failure, not a run failure, and it + self-heals.** ``ProcessPoolExecutor`` is permanently broken after any worker + dies abnormally: every pending future *and every subsequent submit()* raises + ``BrokenProcessPool``. Because the pool is shared, the true blast radius when + a worker dies is "every repo worker holding a future at that moment" (up to + ``effective_workers`` branches), not one -- state that honestly, not as "one + branch". :class:`ExtractionPool` is a generation-tagged supervisor: it + rebuilds the executor at most once per generation (compare-and-swap on the + generation under a lock, so N concurrent branches hitting the same break + trigger exactly one rebuild), bounded by :data:`MAX_POOL_REBUILDS`. Past that + budget it latches to in-process extraction for the rest of the run and logs a + WARNING -- slower, but correct and complete. +* **No prefetch thread, ever.** Since issue #106 the production file source is + ``indexer.ingest.iter_tar_source_files``, a single forward pass over one open + ``TarFile`` (``tf.members.clear()`` is the first statement of its loop body). + :meth:`ExtractionPool.stream` pulls from that source strictly from the + CALLING thread. A second thread advancing the same generator to "warm" the + next batch would interleave ``tf.next()`` calls on a mid-stream ``TarFile`` + and corrupt output SILENTLY -- not as an exception. Bounded look-ahead + (content-byte and file-count budgets scaled by ``n_processes``, covering + BOTH submitted batches and locally-answered non-qualifying files) is + achieved by pulling more items inside the consumer's own ``next()`` call, + never concurrently. + +No timing, no logging, and no DB/network access happens inside a worker: no +logger is configured in a spawned child (a stray ``logger.info`` there would +silently vanish), and ``indexer.timing``'s ambient ``ContextVar`` does not cross +a process boundary (its own module docstring names this issue by hand). Workers +are pure ``ParsedFile -> FileExtraction``. +""" + +from __future__ import annotations + +import logging +import multiprocessing +import os +import threading +from collections import deque +from collections.abc import Callable, Iterable, Iterator +from concurrent.futures import Future, ProcessPoolExecutor +from concurrent.futures.process import BrokenProcessPool +from pathlib import Path +from typing import TYPE_CHECKING + +from indexer.languages import SYMBOL_KINDS, FileExtraction, ParsedFile +from indexer.symbols import extract_file + +if TYPE_CHECKING: + from indexer.repo_config import RepoConfig + +logger = logging.getLogger("indexer.extract_pool") + +# Aggregate len(pf.content) per submitted batch -- deliberately NOT pf.size +# (indexer/store.py's write-batch bound): since #106 ParsedFile.size is the +# archive-declared length, which exceeds len(content) after NUL stripping. +# store.py bounds a WRITE payload, where the declared size is the honest bound; +# this bounds an IPC payload, which is the string actually pickled. Different +# quantities on purpose -- do not unify them. +_BATCH_BYTES = 2_000_000 + +# Backstop file-count bound, applied at TWO levels: (1) per submitted batch, +# alongside `_BATCH_BYTES` (whichever trips first), and (2) via +# `max_in_flight * _MAX_BATCH_FILES` on the overall look-ahead window -- both +# independent of `_BATCH_BYTES`. Zero- or near-zero-byte files (an empty +# `__init__.py`, a `.gitkeep`) barely move the byte bound, so a corpus +# dominated by them would otherwise defeat it and pickle one enormous batch -- +# mirrors `indexer/store.py`'s own dual byte-and-count bound +# (`_BATCH_MAX_CONTENT_BYTES` / `_BATCH_MAX_FILES`) for the same reason. +_MAX_BATCH_FILES = 2000 + +# Bare-Process preflight probe timeout. Generous on purpose: a healthy runtime +# answers in well under a second (measured ~0.06s); this bound exists only to +# cap the pathological "sandbox silently never reports back" case, and it is +# fully recoverable (kill + latch) either way. +_PROBE_TIMEOUT_S = 60.0 + +# Rebuild-once-per-generation budget for a shared pool that keeps breaking +# (a deterministically poisoned file, a flaky sandbox). Past this, further +# rebuild attempts buy nothing -- latch to in-process and finish the run. +MAX_POOL_REBUILDS = 3 + +_PROBE_SENTINEL = "ok" + + +def _needs_parse(pf: ParsedFile) -> bool: + """Would ``extract_file(pf)`` do any real work, or short-circuit to empty? + + Mirrors ``extract_file``'s own short-circuit (``pf.lang is None`` or the + language has no ``SYMBOL_KINDS`` entry) so the pooled path can answer a + guaranteed-empty file locally instead of paying IPC for it. The two must + stay in exact agreement -- pinned directly by a test that checks every + language in ``EXT_TO_LANG.values()`` plus ``None`` plus a bogus value. + """ + return pf.lang is not None and pf.lang in SYMBOL_KINDS + + +def _extract_batch(batch: list[ParsedFile]) -> list[FileExtraction]: + """Extract every file in ``batch``, in order. Module-level so it is picklable + by qualified name under ``spawn`` -- calls the SAME ``extract_file`` the + in-process path calls, so parity is structural, not re-implemented.""" + return [extract_file(pf) for pf in batch] + + +def _available_cpus() -> int: + """Usable CPU count: affinity-aware, cgroup-v2-quota-aware, floor of 1. + + Prefers ``os.sched_getaffinity(0)`` over ``os.cpu_count()`` (affinity + respects cpuset pinning; ``cpu_count`` reports host cores), then clamps by + the cgroup-v2 CPU quota when ``/sys/fs/cgroup/cpu.max`` is readable and + parses to a smaller number. Every branch here is pure and unit-tested + against synthetic inputs -- this function's REAL return value is never + asserted on in a test (it depends on the test host). + """ + getter = getattr(os, "sched_getaffinity", None) + n = len(getter(0)) if getter is not None else (os.cpu_count() or 1) + quota = _cgroup_cpu_quota() + if quota is not None: + n = min(n, quota) + return max(1, n) + + +def _cgroup_cpu_quota() -> int | None: + """Parse ``/sys/fs/cgroup/cpu.max`` ("$QUOTA $PERIOD" or "max $PERIOD"). + + Returns ``None`` (no clamp) for an unreadable file, an unlimited quota + (``"max"``), or a malformed line -- safe in the correct direction, since + the caller's own ``min(..., 8)`` ceiling still bounds the result. + """ + try: + raw = Path("/sys/fs/cgroup/cpu.max").read_text() + except OSError: + return None + parts = raw.split() + if len(parts) != 2: + return None + quota_str, period_str = parts + if quota_str == "max": + return None + try: + quota = int(quota_str) + period = int(period_str) + except ValueError: + return None + if period <= 0: + return None + return max(1, quota // period) + + +def derive_process_count(config: RepoConfig) -> int: + """Worker-process count for the shared extraction pool. + + ``config.extract_processes`` set explicitly wins outright (``1`` is the + kill switch -- see :func:`build_extraction_pool`). Unset (``None``, the + default) derives from the runtime: affinity/cgroup-aware CPU count, capped + at 8 to match every other parallelism knob in this repo + (``index_concurrency``, ``SemanticOverrides.embedding_concurrency``). + """ + if config.extract_processes is not None: + return config.extract_processes + return min(_available_cpus(), 8) + + +class ExtractionPoolError(RuntimeError): + """A shared extraction pool broke while processing a batch. + + Always chained ``from`` the underlying ``BrokenProcessPool`` and always + fails the branch that raised it -- caught by + ``indexer.job._index_one_branch``'s existing broad ``except Exception``, + exactly like any other extraction failure. The pool itself has already + started (or exhausted its budget for) a rebuild by the time this is + raised; the NEXT branch to call :meth:`ExtractionPool.stream` sees the + rebuilt (or latched-to-in-process) pool, not this one. + """ + + +def _probe_child(queue: "multiprocessing.Queue[str]") -> None: + """Preflight probe worker: exercise the real per-worker import + extraction + path (grammar load included), then report success. Never returns data other + than the bare sentinel -- see :func:`_run_preflight_probe` for why.""" + pf = ParsedFile(path="__preflight__.py", lang="python", size=0, content="def f():\n pass\n") + extract_file(pf) + queue.put(_PROBE_SENTINEL) + + +def _drain(queue: "multiprocessing.Queue[str]") -> str | None: + """Best-effort, non-blocking read of the probe's sentinel. ``None`` on any + failure (empty queue, a queue whose feeder thread never flushed) -- the + caller already knows the child exited 0 by this point; this is corroboration, + not the primary signal.""" + try: + return queue.get_nowait() # type: ignore[no-any-return] + except Exception: + return None + + +def _run_preflight_probe( + *, + target: "Callable[[multiprocessing.Queue[str]], None]" = _probe_child, + timeout: float = _PROBE_TIMEOUT_S, +) -> bool: + """Verify ``spawn``-based multiprocessing actually works here, without ever + constructing a ``ProcessPoolExecutor``. + + A bare, terminable ``multiprocessing.Process`` -- see the module docstring + for why an executor-based probe is actively worse than skipping the probe + entirely. ``join(timeout=...)`` rather than a blocking ``queue.get(timeout= + ...)`` is deliberate: a crashed child is detected the instant it exits, + rather than burning the full timeout waiting on a queue that will never + receive anything. + + ``target``/``timeout`` are test seams only (mirroring this repo's injected- + callable convention -- see ``indexer/AGENTS.md``); production always calls + this with both defaults. Both must be picklable by qualified name under + ``spawn``, so a test override must be a MODULE-LEVEL function, never a + closure or a lambda. + + Every failure mode is caught under a bare ``except Exception``, not a type + list: an unguarded ``__main__`` under ``spawn`` (the serverless hazard this + probe exists to catch -- see the module docstring and ``docs/runbooks``) + makes ``ctx.Process(...).start()`` raise ``RuntimeError`` from + ``multiprocessing.spawn._check_not_importing_main`` in THIS process, not + ``OSError`` -- narrower exception handling here would let that one escape + to the caller instead of latching to in-process extraction. + """ + ctx = multiprocessing.get_context("spawn") + try: + queue: "multiprocessing.Queue[str]" = ctx.Queue() + except Exception: + logger.warning( + "extraction pool preflight failed: could not create an IPC queue " + "(sem_open); falling back to in-process extraction for this run", + exc_info=True, + ) + return False + + process = ctx.Process(target=target, args=(queue,), daemon=True) + try: + process.start() + except Exception: + logger.warning( + "extraction pool preflight failed: could not start a worker process; " + "falling back to in-process extraction for this run", + exc_info=True, + ) + return False + + process.join(timeout=timeout) + if process.is_alive(): + # The disposal an executor cannot offer: kill, then join to reap it. + process.kill() + process.join() + logger.warning( + "extraction pool preflight failed: probe worker did not respond " + "within %.0fs (killed); falling back to in-process extraction for " + "this run", + timeout, + ) + return False + + if process.exitcode != 0: + logger.warning( + "extraction pool preflight failed: probe worker exited with code %s; " + "falling back to in-process extraction for this run", + process.exitcode, + ) + return False + + if _drain(queue) != _PROBE_SENTINEL: + logger.warning( + "extraction pool preflight failed: probe worker exited cleanly but " + "never reported success (missing sentinel); falling back to " + "in-process extraction for this run", + ) + return False + + return True + + +class _FutureHolder: + """One submitted batch's shared future, referenced by every slot in it.""" + + __slots__ = ("future",) + + def __init__(self, future: "Future[list[FileExtraction]]") -> None: + self.future = future + + +class _Slot: + """One file's position in the output stream: either already answered + locally (``local`` set), or awaiting its batch's shared future (``holder`` + + its index within that batch's result list).""" + + __slots__ = ("pf", "local", "holder", "index") + + def __init__(self, pf: ParsedFile, local: FileExtraction | None = None) -> None: + self.pf = pf + self.local = local + self.holder: _FutureHolder | None = None + self.index = 0 + + +class ExtractionPool: + """Supervises a shared, ``spawn``-based extraction pool for one run. + + Owns ``(executor, generation)`` under a lock (D7's supervisor). Built once + per run in ``indexer.job.run`` and shared across every repo worker thread; + ``submit()`` from N concurrent threads into one ``ProcessPoolExecutor`` is + supported (the executor is internally locked). ``n_processes < 2`` means no + executor is ever built -- :meth:`stream` degrades to plain in-process + extraction, exactly today's expression. + """ + + def __init__(self, n_processes: int, *, batch_bytes: int = _BATCH_BYTES) -> None: + self._n_processes = n_processes + self._batch_bytes = batch_bytes + self._lock = threading.Lock() + self._generation = 0 + self._rebuilds = 0 + self._latched = False + self._executor: ProcessPoolExecutor | None = ( + self._new_executor() if n_processes >= 2 else None + ) + + def _new_executor(self) -> ProcessPoolExecutor: + ctx = multiprocessing.get_context("spawn") + return ProcessPoolExecutor(max_workers=self._n_processes, mp_context=ctx) + + def _current(self) -> tuple[ProcessPoolExecutor | None, int]: + with self._lock: + return self._executor, self._generation + + def _handle_broken(self, generation: int) -> None: + """React to a ``BrokenProcessPool`` observed against ``generation``. + + Compare-and-swap on the generation, under the lock, so N concurrent + callers hitting the same break trigger exactly one rebuild (or one + latch): the first caller in advances the generation (or latches); every + other caller sees ``generation != self._generation`` and no-ops. The + caller always raises :class:`ExtractionPoolError` regardless of what + happens here -- this method only updates state for the NEXT call to + :meth:`stream`. + """ + old_executor: ProcessPoolExecutor | None = None + with self._lock: + if self._latched or generation != self._generation: + pass + else: + old_executor = self._executor + if self._rebuilds >= MAX_POOL_REBUILDS: + self._latched = True + self._executor = None + logger.warning( + "extraction pool: rebuild budget (%d) exhausted after " + "repeated BrokenProcessPool; latching to in-process " + "extraction for the rest of this run (rollback: set " + "extract_processes: 1 in config.yaml)", + MAX_POOL_REBUILDS, + ) + else: + self._rebuilds += 1 + self._generation += 1 + self._executor = self._new_executor() + logger.warning( + "extraction pool: a worker died (BrokenProcessPool); " + "rebuilt the pool (generation %d, rebuild %d/%d)", + self._generation, + self._rebuilds, + MAX_POOL_REBUILDS, + ) + # Disposed OUTSIDE the lock, and only with plain shutdown(wait=False): + # a BROKEN executor's manager thread has already terminated, so this + # cannot hang (unlike the healthy-executor teardown in shutdown()). + if old_executor is not None: + old_executor.shutdown(wait=False) + + def stream(self, files: Iterable[ParsedFile]) -> Iterator[tuple[ParsedFile, FileExtraction]]: + """Yield ``(pf, FileExtraction)`` pairs for every file in ``files``, in + source order. + + Pull-driven, strictly from the calling thread: ``files`` (in production, + ``indexer.ingest.iter_tar_source_files``'s single-pass generator) is + never advanced from any other thread. Qualifying files (see + :func:`_needs_parse`) are accumulated into batches bounded by + ``batch_bytes`` of aggregate ``len(pf.content)`` and submitted to the + pool; non-qualifying files are answered locally with + ``FileExtraction([], [])`` and interleaved back at their original + position. Look-ahead is bounded by TWO independent budgets scaled by + ``2 * n_processes`` -- aggregate pending content bytes, and pending + file count (a backstop for a corpus of near-zero-byte files, which the + byte budget alone would not throttle) -- covering both submitted + batches and locally-answered files; pulling more is throttled by + consuming (blocking on) the oldest batch first. + + Raises :class:`ExtractionPoolError` (chained from the underlying + ``BrokenProcessPool``) the first time a submitted batch's worker turns + out to have died -- the caller (one repo branch) fails; every future + call to :meth:`stream` sees whatever :meth:`_handle_broken` decided. + """ + executor, generation = self._current() + if executor is None: + for pf in files: + yield pf, extract_file(pf) + return + yield from self._stream_pooled(files, executor, generation) + + def _stream_pooled( + self, + files: Iterable[ParsedFile], + executor: ProcessPoolExecutor, + generation: int, + ) -> Iterator[tuple[ParsedFile, FileExtraction]]: + # Bounds the total content bytes pulled-but-not-yet-yielded, across + # BOTH local (non-qualifying) and batched slots -- not "in-flight + # batch count". A corpus dominated by non-qualifying files (markdown, + # data, unknown extensions) never submits a batch at all, so a count + # of in-flight batches alone would never throttle it and `pending` + # would grow to the full corpus. Same total budget as a pure + # batch-count scheme (`max_in_flight` batches of up to `_batch_bytes` + # each), just measured directly in bytes so it also covers the + # local-only case. + # + # `max_pending_slots` is a second, independent bound on FILE COUNT: + # zero- or near-zero-byte files (an empty `__init__.py`, a `.gitkeep`) + # contribute ~nothing to `pending_bytes`, so the byte bound alone never + # throttles a corpus dominated by them. Mirrors `indexer/store.py`'s + # own dual byte-and-count bound (`_BATCH_MAX_CONTENT_BYTES` / + # `_BATCH_MAX_FILES`) for the same reason: whichever limit is tighter + # for a given corpus shape is the one that actually binds. + max_in_flight = max(1, 2 * self._n_processes) + max_pending_bytes = max_in_flight * self._batch_bytes + max_pending_slots = max_in_flight * _MAX_BATCH_FILES + source = iter(files) + pending: deque[_Slot] = deque() + in_flight: deque[_FutureHolder] = deque() + pending_bytes = 0 + batch_files: list[ParsedFile] = [] + batch_slots: list[_Slot] = [] + batch_bytes = 0 + exhausted = False + + def flush_batch() -> None: + nonlocal batch_files, batch_slots, batch_bytes + if not batch_files: + return + try: + future = executor.submit(_extract_batch, batch_files) + except BrokenProcessPool as exc: + self._handle_broken(generation) + raise ExtractionPoolError( + f"shared extraction pool (generation {generation}) broke while " + "submitting a batch -- a worker likely crashed or was OOM-killed. " + "This branch failed and will re-index on the next run. Rollback: " + "set extract_processes: 1 in config.yaml to disable pooled " + "extraction." + ) from exc + holder = _FutureHolder(future) + for i, slot in enumerate(batch_slots): + slot.holder = holder + slot.index = i + in_flight.append(holder) + batch_files = [] + batch_slots = [] + batch_bytes = 0 + + def pull_one() -> bool: + """Pull exactly one item from ``source``. ``False`` at EOF (which + also flushes any still-accumulating batch, so nothing is stranded + unsubmitted when the source runs dry).""" + nonlocal batch_bytes, pending_bytes, exhausted + try: + pf = next(source) + except StopIteration: + exhausted = True + flush_batch() + return False + pending_bytes += len(pf.content) + if not _needs_parse(pf): + pending.append(_Slot(pf, local=FileExtraction(symbols=[], edges=[]))) + return True + slot = _Slot(pf) + pending.append(slot) + batch_files.append(pf) + batch_slots.append(slot) + batch_bytes += len(pf.content) + if batch_bytes >= self._batch_bytes or len(batch_files) >= _MAX_BATCH_FILES: + flush_batch() + return True + + try: + while True: + while ( + not exhausted + and pending_bytes < max_pending_bytes + and len(pending) < max_pending_slots + ): + pull_one() + # Force out a still-accumulating partial batch ONLY when it is + # about to be popped next (its first slot is the front of + # `pending`): a long run of non-qualifying (local) files pulled + # AFTER this batch's own files can exhaust the budget while the + # batch itself is still under its own _batch_bytes threshold, + # and since the batch's slots were pulled BEFORE those local + # files, they sit AHEAD of them in `pending` -- reaching the + # front, unflushed, before the local files do. Guarding on + # "is it actually the front" (rather than flushing + # unconditionally every outer-loop iteration) keeps a batch + # accumulating normally toward its full _batch_bytes whenever + # something else is ahead of it in the queue, so this does not + # fragment steady-state batching into one-file submissions. + if pending and pending[0].local is None and pending[0].holder is None: + flush_batch() + if not pending: + return + + slot = pending.popleft() + pending_bytes -= len(slot.pf.content) + if slot.local is not None: + yield slot.pf, slot.local + continue + + holder = slot.holder + # Invariant: every non-local slot has a holder by this point, + # because the unconditional flush_batch() call directly above + # empties `batch_files`/`batch_slots` before we ever pop from + # `pending` -- there is never an unflushed batch slot inside + # `pending` at the moment we reach this line. + assert holder is not None, "unflushed batch slot reached the front of pending" + try: + results = holder.future.result() + except BrokenProcessPool as exc: + self._handle_broken(generation) + raise ExtractionPoolError( + f"shared extraction pool (generation {generation}) broke while " + "processing a batch -- a worker likely crashed or was OOM-killed. " + "This branch failed and will re-index on the next run. Rollback: " + "set extract_processes: 1 in config.yaml to disable pooled " + "extraction." + ) from exc + finally: + if in_flight and in_flight[0] is holder: + in_flight.popleft() + yield slot.pf, results[slot.index] + finally: + # Cleanup on abandonment: `files` is consumed inside index_repo's + # open transaction, so any exception in that loop (a DB error, + # StaleIndexError, a chunk_writer failure) abandons this generator + # with however many batches the look-ahead budgets above still + # left submitted and unconsumed. Cancel every + # one we haven't consumed -- a future already running cannot be + # cancelled (returns False) but finishes in milliseconds and its + # discarded result costs nothing. + for holder in in_flight: + holder.future.cancel() + + def shutdown(self) -> None: + """Tear down the executor, if one exists. ``wait=True``, and safe to + call unconditionally: by the time ``indexer.job.run`` reaches its + ``finally``, the repo-worker ``ThreadPoolExecutor``'s own ``with`` block + has already exited (every branch joined), so no branch holds a future + against this pool -- the one condition that makes ``wait=True`` safe + (see the module's execution-plan writeup). Never + ``cancel_futures=True``, matching this repo's existing executor-teardown + discipline (``indexer/job.py``). + """ + with self._lock: + executor = self._executor + self._executor = None + self._latched = True + if executor is not None: + executor.shutdown(wait=True) + + +def build_extraction_pool(config: RepoConfig) -> ExtractionPool: + """Derive this run's process count, preflight it, and build the pool. + + ``extract_processes: 1`` (explicit or, on a single-core host, derived) is + the kill switch: no preflight probe runs and no executor is ever built -- + :meth:`ExtractionPool.stream` degrades straight to in-process extraction. + For ``>= 2``, the preflight probe (see :func:`_run_preflight_probe`) MUST + pass before any ``ProcessPoolExecutor`` is constructed; a failing probe logs + a WARNING and returns an already-latched, in-process-only pool instead of + raising -- a degraded run at today's speed, never a failed one. + """ + n = derive_process_count(config) + if n < 2: + logger.info("symbol extraction: in-process (extract_processes=1; no pool built)") + return ExtractionPool(n_processes=0) + if not _run_preflight_probe(): + # _run_preflight_probe already logged the WARNING naming the cause. + return ExtractionPool(n_processes=0) + logger.info("symbol extraction: %d process(es) (spawn); pool preflight ok", n) + return ExtractionPool(n_processes=n) diff --git a/indexer/job.py b/indexer/job.py index c5c5c6f..16e8d23 100644 --- a/indexer/job.py +++ b/indexer/job.py @@ -141,6 +141,7 @@ from app.embed import EmbeddingCountMismatchError, EmbedFn, get_embedder from indexer.branches import resolve_branches from indexer.chunk_store import write_chunks_batch +from indexer.extract_pool import ExtractionPool, build_extraction_pool from indexer.fetch import ( REQUIRED_FREE_BYTES, assert_disk_headroom, @@ -165,7 +166,6 @@ reconcile_removed_repos, reconcile_retired_branches, ) -from indexer.symbols import extract_file from indexer.timing import PhaseTimer, install_timer, now, reset_timer logger = logging.getLogger("indexer.job") @@ -309,19 +309,23 @@ def run( reconcile_retired_fn: Callable[..., ReconcileCounts] = reconcile_retired_branches, reconcile_removed_fn: Callable[..., list[str]] = reconcile_removed_repos, shas_fn: Callable[..., ContentShaSets] = read_indexed_shas, + extraction_pool: ExtractionPool | None = None, ) -> int: """Index every configured repo and return a process exit code (0 = all ok). Boundaries are injectable for tests: ``workspace_client`` (secret + config read), ``http_client`` (GitHub HTTP), ``engine`` (DB), ``index_fn`` (the store), ``embed_fn`` (semantic chunking), ``config_loader`` (so orchestration - tests need no SDK fake), and ``reconcile_retired_fn`` / ``reconcile_removed_fn`` + tests need no SDK fake), ``reconcile_retired_fn`` / ``reconcile_removed_fn`` (the storage primitives the post-fan-out checkpoint below invokes -- mirroring - ``index_fn``'s injection so reconciliation tests need no real Postgres either). - ``cfg`` defaults to the process-cached :func:`app.config.get_settings`, then any - ``semantic:`` fields in config.yaml are overlaid onto it before the worker clamp - and embedder build (config.yaml > env > default -- config.yaml is the job's - semantic-config surface; see :class:`indexer.repo_config.SemanticOverrides`). + ``index_fn``'s injection so reconciliation tests need no real Postgres either), + and ``extraction_pool`` (issue #108's shared symbol/edge-extraction pool -- + mirroring ``owns_engine``/``owns_http``, a caller-supplied pool is never + shut down here). ``cfg`` defaults to the process-cached + :func:`app.config.get_settings`, then any ``semantic:`` fields in config.yaml + are overlaid onto it before the worker clamp and embedder build (config.yaml > + env > default -- config.yaml is the job's semantic-config surface; see + :class:`indexer.repo_config.SemanticOverrides`). ``resolve_repos`` is deliberately **not** injectable: the existing ``httpx.MockTransport`` seam already lets tests drive enumeration outcomes @@ -452,6 +456,10 @@ def run( reconciliation_failed = False reconcile_skip_reason = "" progress = ReconcileProgress() + # Computed BEFORE the try below (from the parameter, not the built value) so + # it stays defined even if disk_usage or pool construction never runs -- the + # finally block reads it unconditionally. + owns_pool = extraction_pool is None try: # Inside the try for the same reason as the stamp read below: shutil # .disk_usage raises if tmp is missing or unmounted, and above the try @@ -467,6 +475,27 @@ def run( REQUIRED_FREE_BYTES / 1e9, ) + # Built once per run, shared across every repo worker below (issue + # #108). Wrapped in a bare `except Exception`, not a type list: + # NOTHING about this pool may ever fail the run -- build_extraction_pool + # already degrades its own internal failures (a bad preflight probe, an + # unavailable sandbox) to a WARNING plus an in-process pool, so this + # guard exists only for a truly unexpected failure in that derivation + # itself. `run()` has no handler around the disk-headroom block above + # either, so an uncaught escape here would propagate out of `run()` and + # fail the wheel task after the `finally` below -- this is what prevents + # that. + if extraction_pool is None: + try: + extraction_pool = build_extraction_pool(config) + except Exception: + logger.warning( + "failed to build the shared extraction pool; indexing will " + "run in-process for this run", + exc_info=True, + ) + extraction_pool = ExtractionPool(n_processes=0) + # One batched read for the whole run, BEFORE fan-out. Inside the try so a # failure here still closes the http client and disposes the engine. stamps = _read_stamps(engine, entries) @@ -486,6 +515,7 @@ def run( embed_fn=embed_fn, stamps=stamps, shas_fn=shas_fn, + extraction_pool=extraction_pool, ): entry for entry in entries } @@ -564,6 +594,13 @@ def run( entries=entries, ) finally: + # Shut down BEFORE the http client/engine below, and only ours: an + # injected pool is the caller's to manage (mirrors owns_engine/owns_http). + # By this point the ThreadPoolExecutor `with` block above has already + # exited (every repo worker joined), so no branch holds a future against + # this pool -- the condition that makes shutdown(wait=True) safe. + if owns_pool and extraction_pool is not None: + extraction_pool.shutdown() if owns_http: http_client.close() if owns_engine: @@ -942,6 +979,7 @@ def _index_one( embed_fn: EmbedFn | None, stamps: dict[tuple[str, str], tuple[str | None, int | None]], shas_fn: Callable[..., ContentShaSets], + extraction_pool: ExtractionPool, ) -> RepoOutcome: """Run the full fetch -> parse -> symbols -> store pipeline for every branch of one repo. @@ -968,6 +1006,7 @@ def _index_one( embed_fn=embed_fn, stamps=stamps, shas_fn=shas_fn, + extraction_pool=extraction_pool, ) finally: _repo_ctx.reset(token) @@ -984,6 +1023,7 @@ def _index_one_inner( embed_fn: EmbedFn | None, stamps: dict[tuple[str, str], tuple[str | None, int | None]], shas_fn: Callable[..., ContentShaSets], + extraction_pool: ExtractionPool, ) -> RepoOutcome: """The body of :func:`_index_one`, run with the repo log context already set. @@ -1045,6 +1085,7 @@ def _index_one_inner( started=started, max_chunks_per_repo=max_chunks_per_repo, shas_fn=shas_fn, + extraction_pool=extraction_pool, ) for branch in resolution.branches ] @@ -1113,6 +1154,7 @@ def _index_one_branch( started: float, max_chunks_per_repo: int, shas_fn: Callable[..., ContentShaSets], + extraction_pool: ExtractionPool, ) -> BranchOutcome: """Fetch, parse, and store ONE branch. Never raises -- every failure is classified. @@ -1152,6 +1194,15 @@ def _index_one_branch( ``_index_one``'s ``_repo_ctx`` reset is: ``ThreadPoolExecutor`` reuses worker threads without resetting their context, so a leaked timer would attribute this branch's sweep to the next task on this thread. + + ``extraction_pool`` (issue #108) is the run-shared, ``spawn``-based process + pool ``indexer.extract_pool`` extraction runs through; both ``items`` + expressions below call ``extraction_pool.stream(...)`` rather than + ``extract_file`` directly. A worker death raises + ``indexer.extract_pool.ExtractionPoolError`` out of that stream -- caught by + THIS function's existing broad ``except Exception`` below like any other + extraction failure, with no new ``except`` branch: legibility comes from + ``ExtractionPoolError``'s own message, not from special-casing it here. """ # NOT the `started` parameter: that one is set once per REPO and passed # unchanged to every branch, so reusing it would make branch 2+ of a @@ -1273,10 +1324,12 @@ def _index_one_branch( precompute_failed = True finally: timer.add("embed", timer.clock() - t0) - items = ((pf, extract_file(pf)) for pf in files) + items = extraction_pool.stream(files) else: # Lazy generator: files stream through the open transaction (bounded memory). - items = ((pf, extract_file(pf)) for pf in iter_tar_source_files(tar_path)) + # extraction_pool.stream() is itself pull-driven and single-consumer, so this + # stays lazy end to end -- no full materialization on the semantic-off path. + items = extraction_pool.stream(iter_tar_source_files(tar_path)) # Parse time is INSIDE the db window (items are produced lazily as # index_repo consumes them), so `db` subtracts only the parse accrued diff --git a/indexer/repo_config.py b/indexer/repo_config.py index d863771..38cacf5 100644 --- a/indexer/repo_config.py +++ b/indexer/repo_config.py @@ -264,9 +264,11 @@ class RepoConfig(BaseModel): numbers 5x but deliberately left the default alone; #109 re-derives it.) **Returns at the ceiling are sublinear.** Symbol extraction does not - parallelise (measured at 0.95x on 4 threads), so Amdahl's law caps the - speedup well below 8x while the disk cost stays a hard linear 4 GB. Raise - it only knowing that trade. + parallelise across THREADS (measured at 0.95x on 4 threads -- the tree walk + is GIL-serialized) -- which is exactly why extraction now runs in a shared + process pool instead (``extract_processes``, below); that pool is decoupled + from ``index_concurrency`` entirely. Raise ``index_concurrency`` only + knowing you are buying disk-bound repo fan-out, not extraction throughput. When semantic indexing is on, the effective worker count is clamped to 2 by :func:`effective_workers`. That clamp is a **memory** bound, not a CPU one: @@ -286,11 +288,25 @@ class RepoConfig(BaseModel): the job resolves ``entry.semantic_max_chunks or cfg.semantic_max_chunks_per_repo`` and the ``semantic:`` block only supplies the second operand -- see :class:`SemanticOverrides`. + + ``extract_processes`` (#108) sizes the shared, ``spawn``-based process pool + ``indexer.extract_pool`` runs symbol/edge extraction through -- one pool per + run, shared across every ``index_concurrency`` repo worker, so a single + giant repo's files parse on every available core rather than one thread's + worth. It is a CPU knob, entirely independent of ``index_concurrency``'s + disk bound above. ``None`` (the default) derives from the runtime: + affinity/cgroup-aware CPU count, clamped to 8 -- matching every other + parallelism knob in this repo (``index_concurrency``, + ``SemanticOverrides.embedding_concurrency``). Setting this to ``1`` restores + fully serial, in-process extraction and spawns no process pool at all -- + the rollback switch, exercised the same way #107's + ``embedding_concurrency: 1`` is. """ version: Literal[1] connections: list[Connection] = Field(min_length=1) index_concurrency: int = Field(default=4, ge=1, le=8) + extract_processes: int | None = Field(default=None, ge=1, le=8) # Per-repo override of Settings.semantic_max_chunks_per_repo (app/config.py, default # 8000), keyed by repo. Absent (the default) -> no repo gets an override, and diff --git a/scripts/measure_extraction_pool.py b/scripts/measure_extraction_pool.py new file mode 100644 index 0000000..1534778 --- /dev/null +++ b/scripts/measure_extraction_pool.py @@ -0,0 +1,128 @@ +"""Measure symbol/edge extraction pool scaling against a REAL tarball (#108, AC1). + +Offline companion to :mod:`indexer.extract_pool`. Drives the exact production call +site :mod:`indexer.job` uses: ``indexer.ingest.iter_tar_source_files(tarball)`` -> +``ExtractionPool.stream(...)`` -- never a batch-submit harness (``executor.map`` +over a pre-built batch list) and never an on-disk walk. Since #106 the production +file source is a single serial gzip-decompress + decode + filter pass over one +open ``TarFile``, run once here and shared between the serial and pooled arms, so +this script also reports the ingest/extract split the pool cannot touch (see +``docs/runbooks/indexing-parallelism.md`` §3's "ingestion stays serial" note). + +**The tarball must be real** -- a GitHub codeload archive (``curl -L +https://codeload.github.com/OWNER/REPO/tar.gz/HEAD -o repo.tar.gz`` or +``indexer.fetch.download_tarball`` by hand), not a synthetic fixture. A +site-packages-heavy Python tree (e.g. this repo's own ``.venv/``, tarred up) is a +reasonable large-repo proxy: ``tar czf repo.tar.gz --transform 's,^,repo-abc1234/,' .`` + +Usage: ``uv run python scripts/measure_extraction_pool.py --tarball repo.tar.gz +[--processes 2,4,8] [--batch-bytes 2000000] [--repeat 1]`` +""" + +from __future__ import annotations + +import argparse +import time +from collections.abc import Sequence +from pathlib import Path + +from indexer.extract_pool import _BATCH_BYTES, ExtractionPool +from indexer.ingest import iter_tar_source_files +from indexer.languages import FileExtraction, ParsedFile +from indexer.symbols import extract_file + + +def _load_files(tarball: Path) -> list[ParsedFile]: + """One real, serial pass over the tarball -- the Amdahl floor (§4.7.1) this + pool cannot parallelize. Timed separately by the caller.""" + return list(iter_tar_source_files(tarball)) + + +def _measure_serial( + files: list[ParsedFile], +) -> tuple[float, list[tuple[ParsedFile, FileExtraction]]]: + start = time.perf_counter() + results = [(pf, extract_file(pf)) for pf in files] + return time.perf_counter() - start, results + + +def _measure_pooled( + files: list[ParsedFile], *, n_processes: int, batch_bytes: int +) -> tuple[float, list[tuple[ParsedFile, FileExtraction]]]: + pool = ExtractionPool(n_processes=n_processes, batch_bytes=batch_bytes) + try: + start = time.perf_counter() + results = list(pool.stream(files)) + elapsed = time.perf_counter() - start + finally: + pool.shutdown() + return elapsed, results + + +def main(argv: Sequence[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--tarball", required=True, type=Path, help="a real downloaded repo tarball (.tar.gz)" + ) + parser.add_argument( + "--processes", default="2,4,8", help="comma-separated worker-process counts to measure" + ) + parser.add_argument("--batch-bytes", type=int, default=_BATCH_BYTES) + parser.add_argument("--repeat", type=int, default=1, help="repetitions per data point") + args = parser.parse_args(argv) + process_counts = [int(p) for p in args.processes.split(",")] + + print(f"loading {args.tarball} ...") + ingest_start = time.perf_counter() + files = _load_files(args.tarball) + ingest_elapsed = time.perf_counter() - ingest_start + qualifying = [pf for pf in files if pf.lang is not None] + total_bytes = sum(len(pf.content) for pf in files) + qualifying_bytes = sum(len(pf.content) for pf in qualifying) + print( + f"{len(files)} indexable files / {total_bytes / 1e6:.1f} MB, " + f"{len(qualifying)} qualify for parsing / {qualifying_bytes / 1e6:.1f} MB " + f"(ingest: {ingest_elapsed:.2f}s -- serial in the parent, both arms below)" + ) + + print(f"\n{'path':>14s} {'extract_s':>10s} {'total_s':>10s} {'speedup':>8s} {'note':>9s}") + serial_extract = None + serial_results: list[tuple[ParsedFile, FileExtraction]] = [] + for _ in range(args.repeat): + elapsed, serial_results = _measure_serial(files) + if serial_extract is None or elapsed < serial_extract: + serial_extract = elapsed + assert serial_extract is not None + serial_total = ingest_elapsed + serial_extract + print(f"{'serial':>14s} {serial_extract:>10.3f} {serial_total:>10.3f} {1.0:>7.2f}x") + + for n in process_counts: + best_elapsed = None + identical = True + for _ in range(args.repeat): + elapsed, pooled_results = _measure_pooled( + files, n_processes=n, batch_bytes=args.batch_bytes + ) + identical = identical and pooled_results == serial_results + if best_elapsed is None or elapsed < best_elapsed: + best_elapsed = elapsed + assert best_elapsed is not None + total = ingest_elapsed + best_elapsed + speedup = serial_total / total if total else float("inf") + note = "identical" if identical else "MISMATCH" + print( + f"{f'pool x{n}':>14s} {best_elapsed:>10.3f} {total:>10.3f} " + f"{speedup:>7.2f}x {note:>9s}" + ) + + print( + f"\ningest (serial, shared by both arms): {ingest_elapsed:.2f}s of " + f"{serial_total:.2f}s serial total ({100 * ingest_elapsed / serial_total:.0f}%) -- " + "the floor AC1's 'combined' speedup above cannot cross, however many " + "processes extract_processes uses." + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/unit/test_extract_pool.py b/tests/unit/test_extract_pool.py new file mode 100644 index 0000000..baa3cfb --- /dev/null +++ b/tests/unit/test_extract_pool.py @@ -0,0 +1,626 @@ +"""Unit tests for indexer.extract_pool: the shared process-pool extraction (#108). + +T1-T8 below mirror the execution plan's numbering. T9/T10 (the existing +``test_job.py`` suite's kill-switch pin and phase-timing attribution) live in +``tests/unit/test_job.py`` -- this file covers ``indexer.extract_pool`` in +isolation. +""" + +from __future__ import annotations + +import logging +import threading +import time +from concurrent.futures import Future +from typing import Any + +import pytest + +from indexer import extract_pool +from indexer.extract_pool import ( + _MAX_BATCH_FILES, + MAX_POOL_REBUILDS, + ExtractionPool, + ExtractionPoolError, + _needs_parse, + build_extraction_pool, + derive_process_count, +) +from indexer.languages import EXT_TO_LANG, SYMBOL_KINDS, FileExtraction, ParsedFile +from indexer.repo_config import RepoConfig +from indexer.symbols import extract_file + +# --- fixtures ----------------------------------------------------------------- + +# One real per-language snippet each (reusing test_symbols.py's shapes), so T1's +# parity check exercises every SYMBOL_KINDS language through a real spawn pool. +_LANG_SOURCES: dict[str, str] = { + "python": "class C:\n def m(self):\n pass\ndef top():\n pass\n", + "javascript": "class C {\n m() {}\n}\nfunction top() {}\n", + "typescript": "interface I {}\nclass C {\n m() {}\n}\nfunction top() {}\n", + "tsx": "function top() {}\n", + "go": "package main\nfunc top() {}\nfunc (r R) m() {}\ntype T int\n", + "java": "class C {\n void m() {}\n}\ninterface I {}\n", + "rust": "fn top() {}\nstruct S;\nenum E {}\ntrait T {}\n", +} + + +def _fixture_corpus() -> list[ParsedFile]: + files = [ + ParsedFile(path=f"src/f.{lang}", lang=lang, size=len(content), content=content) + for lang, content in _LANG_SOURCES.items() + ] + files.append(ParsedFile(path="unknown.dat", lang=None, size=3, content="hi\n")) + files.append(ParsedFile(path="doc.md", lang="markdown", size=5, content="# hi\n")) + files.append(ParsedFile(path="empty.py", lang="python", size=0, content="")) + files.append( + ParsedFile( + path="nested.py", + lang="python", + size=0, + content=( + "class Outer:\n" + " def method(self):\n" + " return helper()\n" + "def helper():\n" + " return 1\n" + ), + ) + ) + return files + + +def _python_files(n: int, *, size: int = 200) -> list[ParsedFile]: + padding = "x" * max(size - 20, 1) + return [ + ParsedFile( + path=f"pkg/mod{i}.py", + lang="python", + size=size, + content=f"def f{i}():\n return '{padding}'\n", + ) + for i in range(n) + ] + + +def _config(*, extract_processes: int | None = None) -> RepoConfig: + doc: dict[str, Any] = {"version": 1, "connections": [{"type": "github", "users": ["u"]}]} + if extract_processes is not None: + doc["extract_processes"] = extract_processes + return RepoConfig.model_validate(doc) + + +# Module-level (picklable under spawn) preflight-probe targets for T6a/T6b -- +# see _run_preflight_probe's docstring: a test override must be a module-level +# function, never a closure or lambda, or spawn's re-import cannot find it. +def _crashing_probe_target(queue: Any) -> None: + raise RuntimeError("simulated preflight crash") + + +def _hanging_probe_target(queue: Any) -> None: + time.sleep(3600) + + +# --- T1: parity, real spawn pool ---------------------------------------------- + + +@pytest.mark.unit +def test_stream_parity_against_extract_file_real_spawn_pool() -> None: + """T1 (AC2): real spawn processes, real pickling -- the risk the issue names. + + ``max_workers=2`` is pinned deliberately: each spawned child re-imports + ``tree_sitter_language_pack``, so worker count drives this test's cost more + than corpus size does. A tiny ``batch_bytes`` forces the fixture corpus + (well under the real 2 MB default) across >=3 batches, so submit/collect + round-trips more than once. + """ + files = _fixture_corpus() + pool = ExtractionPool(n_processes=2, batch_bytes=20) + try: + got = list(pool.stream(files)) + finally: + pool.shutdown() + + expected = [(pf, extract_file(pf)) for pf in files] + assert got == expected + + +@pytest.mark.unit +def test_build_extraction_pool_real_spawn_end_to_end() -> None: + """Sanity: the full build_extraction_pool -> preflight -> stream wiring, + through a real spawn pool (not the lower-level ExtractionPool constructor + T1 uses directly).""" + pool = build_extraction_pool(_config(extract_processes=2)) + try: + assert pool._executor is not None + files = _python_files(5) + got = list(pool.stream(files)) + finally: + pool.shutdown() + + expected = [(pf, extract_file(pf)) for pf in files] + assert got == expected + + +# --- T2: routing-predicate agreement ------------------------------------------ + + +@pytest.mark.unit +@pytest.mark.parametrize("lang", [*sorted(EXT_TO_LANG.values()), None, "bogus"]) +def test_needs_parse_agrees_with_extract_file_short_circuit(lang: str | None) -> None: + pf = ParsedFile(path="x", lang=lang, size=0, content="whatever\n") + expect_parse = lang is not None and lang in SYMBOL_KINDS + assert _needs_parse(pf) is expect_parse + if not expect_parse: + assert extract_file(pf) == FileExtraction(symbols=[], edges=[]) + + +# --- T3: bounded look-ahead ---------------------------------------------------- + + +class _CountingSource: + """Wraps a file list, counting how many items have actually been pulled.""" + + def __init__(self, files: list[ParsedFile]) -> None: + self._it = iter(files) + self.pulled = 0 + + def __iter__(self) -> "_CountingSource": + return self + + def __next__(self) -> ParsedFile: + pf = next(self._it) + self.pulled += 1 + return pf + + +class _ImmediateFuture: + def __init__(self, result: list[FileExtraction]) -> None: + self._result = result + + def result(self) -> list[FileExtraction]: + return self._result + + def cancel(self) -> bool: + return False + + +class _ImmediateExecutor: + """Resolves every submitted batch synchronously -- still lets ExtractionPool's + own in_flight bookkeeping (not completion timing) drive the look-ahead bound.""" + + def submit(self, fn: Any, *args: Any) -> _ImmediateFuture: + return _ImmediateFuture(fn(*args)) + + +@pytest.mark.unit +def test_bounded_look_ahead() -> None: + """T3: the issue's explicit no-full-materialization rule. Each file alone + exceeds the tiny batch_bytes, so one file == one batch, and max_in_flight + (2 * n_processes) directly bounds how many files are pulled ahead.""" + total = 20 + files = _python_files(total, size=200) + source = _CountingSource(files) + + pool = ExtractionPool(n_processes=0, batch_bytes=50) + pool._n_processes = 2 + pool._executor = _ImmediateExecutor() # type: ignore[assignment] + max_in_flight = 2 * pool._n_processes + + gen = pool.stream(source) # type: ignore[arg-type] + next(gen) + assert source.pulled <= max_in_flight + assert source.pulled < total + + for _ in gen: + assert source.pulled <= total + assert source.pulled == total + + +@pytest.mark.unit +def test_bounded_look_ahead_for_non_qualifying_files() -> None: + """Regression: non-qualifying (locally-answered) files never enter a batch, + so a look-ahead bound keyed on in-flight BATCH count alone never throttles a + corpus dominated by them -- the whole source would be pulled in one shot. + The bound must cover pending local content bytes too.""" + total = 500 + files = [ + ParsedFile(path=f"doc{i}.md", lang=None, size=2000, content="x" * 2000) + for i in range(total) + ] + source = _CountingSource(files) + + # batch_bytes deliberately tiny: max_pending_bytes (4 * batch_bytes) must be + # well under the corpus's total bytes (500 * 2000 = 1,000,000) for this test + # to actually exercise the throttle rather than trivially fitting everything + # under budget. + pool = ExtractionPool(n_processes=0, batch_bytes=1000) + pool._n_processes = 2 + pool._executor = _ImmediateExecutor() # type: ignore[assignment] + + gen = pool.stream(source) # type: ignore[arg-type] + next(gen) + assert source.pulled < total # bounded -- NOT the whole corpus in one shot + + for _ in gen: + pass + assert source.pulled == total + + +class _RecordingExecutor: + """Resolves synchronously like _ImmediateExecutor, but also records each + submitted batch's file count -- for pinning the per-batch file-count cap.""" + + def __init__(self) -> None: + self.batch_sizes: list[int] = [] + + def submit(self, fn: Any, *args: Any) -> _ImmediateFuture: + self.batch_sizes.append(len(args[0])) + return _ImmediateFuture(fn(*args)) + + +@pytest.mark.unit +def test_batch_is_capped_by_file_count_even_with_huge_batch_bytes() -> None: + """Regression: a corpus of near-zero-byte qualifying files barely moves + batch_bytes, so without a per-batch FILE-COUNT cap a single batch would grow + to the entire look-ahead window (thousands of files, one oversized pickle on + one idle worker) before ever tripping the byte bound.""" + total = _MAX_BATCH_FILES * 2 + 5 + files = _python_files(total, size=1) # near-zero content per file + executor = _RecordingExecutor() + + pool = ExtractionPool(n_processes=0, batch_bytes=10_000_000) # never trips + pool._n_processes = 2 + pool._executor = executor # type: ignore[assignment] + + got = list(pool.stream(files)) + assert got == [(pf, extract_file(pf)) for pf in files] + assert executor.batch_sizes # at least one batch was actually submitted + assert max(executor.batch_sizes) <= _MAX_BATCH_FILES + + +# --- T4: order preservation under out-of-order completion --------------------- + + +class _ReverseOrderExecutor: + """Batch N's future resolves BEFORE batch N-1's -- proves stream() yields in + submission (source) order, not completion order.""" + + def __init__(self, total_batches: int) -> None: + self._total = total_batches + self._n = 0 + + def submit(self, fn: Any, *args: Any) -> "Future[list[FileExtraction]]": + self._n += 1 + idx = self._n + fut: "Future[list[FileExtraction]]" = Future() + + def _worker() -> None: + time.sleep(0.01 * (self._total - idx + 1)) + fut.set_result(fn(*args)) + + threading.Thread(target=_worker, daemon=True).start() + return fut + + +@pytest.mark.unit +def test_order_preservation_under_out_of_order_completion() -> None: + """T4 (D6): later-submitted batches resolving first must not reorder output.""" + files = _python_files(6, size=200) # tiny batch_bytes -> one batch per file + pool = ExtractionPool(n_processes=0, batch_bytes=50) + pool._n_processes = 2 + pool._executor = _ReverseOrderExecutor(len(files)) # type: ignore[assignment] + + got = list(pool.stream(files)) + expected = [(pf, extract_file(pf)) for pf in files] + assert got == expected + + +# --- T5: BrokenProcessPool isolation, two concurrent streams ------------------ + + +class _BrokenFuture: + def __init__(self, barrier: threading.Barrier | None = None) -> None: + self._barrier = barrier + + def result(self) -> list[FileExtraction]: + if self._barrier is not None: + # Rendezvous so two concurrent callers genuinely observe the SAME + # (pre-rebuild) executor before either raises -- without this, one + # thread can race ahead, rebuild, and hand the second thread an + # already-healthy pool, which would defeat the point of this test. + self._barrier.wait(timeout=10.0) + raise extract_pool.BrokenProcessPool("worker died") + + def cancel(self) -> bool: + return False + + +class _AlwaysBreakingExecutor: + def __init__(self, barrier: threading.Barrier | None = None) -> None: + self._barrier = barrier + + def submit(self, fn: Any, *args: Any) -> _BrokenFuture: + return _BrokenFuture(self._barrier) + + def shutdown(self, wait: bool = True) -> None: + pass + + +class _WorkingExecutor: + def submit(self, fn: Any, *args: Any) -> "Future[list[FileExtraction]]": + fut: "Future[list[FileExtraction]]" = Future() + fut.set_result(fn(*args)) + return fut + + def shutdown(self, wait: bool = True) -> None: + pass + + +@pytest.mark.unit +def test_broken_process_pool_isolation_two_concurrent_streams() -> None: + """T5 (AC3): the two-thread shape is the point -- a single-stream test would + pass against a racy rebuild.""" + files = _python_files(3, size=50) + pool = ExtractionPool(n_processes=0, batch_bytes=10) + pool._n_processes = 2 + barrier = threading.Barrier(2) + pool._executor = _AlwaysBreakingExecutor(barrier=barrier) # type: ignore[assignment] + + rebuild_calls = {"n": 0} + + def _new_executor_stub() -> Any: + rebuild_calls["n"] += 1 + # The FIRST rebuild recovers onto a healthy executor (proves "a + # subsequent stream succeeds on the new generation"); every rebuild + # after that keeps breaking, driving toward the rebuild budget. + return _WorkingExecutor() if rebuild_calls["n"] == 1 else _AlwaysBreakingExecutor() + + pool._new_executor = _new_executor_stub # type: ignore[method-assign] + + errors: list[ExtractionPoolError] = [] + lock = threading.Lock() + + def _drive() -> None: + try: + list(pool.stream(files)) + except ExtractionPoolError as exc: + with lock: + errors.append(exc) + + t1 = threading.Thread(target=_drive) + t2 = threading.Thread(target=_drive) + t1.start() + t2.start() + t1.join() + t2.join() + + assert len(errors) == 2 # both concurrent callers saw the failure + assert isinstance(errors[0].__cause__, extract_pool.BrokenProcessPool) + assert "extract_processes: 1" in str(errors[0]) # names the rollback + assert pool._rebuilds == 1 # exactly one rebuild for N concurrent breaks + assert pool._generation == 1 + assert not pool._latched + + # A subsequent stream succeeds on the new (healthy) generation. + got = list(pool.stream(files)) + assert got == [(pf, extract_file(pf)) for pf in files] + + # Simulate a later worker death under the SAME (still healthy) generation, + # driving repeated rebuilds until the budget is exhausted and it latches. + pool._executor = _AlwaysBreakingExecutor() # type: ignore[assignment] + for _ in range(MAX_POOL_REBUILDS - 1): + with pytest.raises(ExtractionPoolError): + list(pool.stream(files)) + assert pool._rebuilds == MAX_POOL_REBUILDS + assert not pool._latched + + with pytest.raises(ExtractionPoolError): + list(pool.stream(files)) + assert pool._latched + assert pool._executor is None + + # Latched: still yields correct results, now in-process. + got_latched = list(pool.stream(files)) + assert got_latched == [(pf, extract_file(pf)) for pf in files] + + +class _BreaksOnSubmitExecutor: + """Unlike _AlwaysBreakingExecutor (which fails at .result()), this fails + inside submit() itself -- the shape a real ProcessPoolExecutor takes once a + worker has already died before the NEXT batch is even dispatched.""" + + def submit(self, fn: Any, *args: Any) -> Any: + raise extract_pool.BrokenProcessPool("worker already dead at submit time") + + def shutdown(self, wait: bool = True) -> None: + pass + + +@pytest.mark.unit +def test_broken_process_pool_raised_from_submit_is_handled() -> None: + """Regression: a real ProcessPoolExecutor can raise BrokenProcessPool from + submit() itself, not only from a future's result() -- verified against a + real ProcessPoolExecutor in review. The supervisor must still record the + break (rebuild-or-latch) rather than let the raw exception escape + unclassified and leave (_rebuilds, _generation, _latched) stale.""" + files = _python_files(3, size=50) + pool = ExtractionPool(n_processes=0, batch_bytes=10) + pool._n_processes = 2 + pool._executor = _BreaksOnSubmitExecutor() # type: ignore[assignment] + + with pytest.raises(ExtractionPoolError) as excinfo: + list(pool.stream(files)) + + assert isinstance(excinfo.value.__cause__, extract_pool.BrokenProcessPool) + assert pool._rebuilds == 1 + assert pool._generation == 1 + assert not pool._latched + + +# --- T6a/T6b: preflight probe ------------------------------------------------- + + +@pytest.mark.unit +def test_preflight_fails_fast_when_ipc_queue_cannot_be_created( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + """T6a: the sem_open/OSError half of "probe child exits nonzero (or + ctx.Queue() raises OSError)". Fails before any child is even started.""" + + class _BrokenContext: + def Queue(self) -> Any: + raise OSError("sem_open: No space left on device") + + monkeypatch.setattr(extract_pool.multiprocessing, "get_context", lambda name: _BrokenContext()) + + with caplog.at_level(logging.WARNING, logger="indexer.extract_pool"): + ok = extract_pool._run_preflight_probe() + + assert ok is False + assert "could not create an IPC queue" in caplog.text + + +@pytest.mark.unit +def test_preflight_fails_fast_when_probe_child_crashes(caplog: pytest.LogCaptureFixture) -> None: + """T6a: a real spawned child that exits nonzero is detected -- and the WARNING + names the cause.""" + with caplog.at_level(logging.WARNING, logger="indexer.extract_pool"): + ok = extract_pool._run_preflight_probe(target=_crashing_probe_target, timeout=10.0) + + assert ok is False + assert "exited with code" in caplog.text + + +@pytest.mark.unit +def test_preflight_hang_is_killed_and_latches(caplog: pytest.LogCaptureFixture) -> None: + """T6b: THE defect this plan was blocked on. A hung probe child must be + killed within the injected timeout, never awaited to completion, and the + test process itself must not hang.""" + start = time.monotonic() + with caplog.at_level(logging.WARNING, logger="indexer.extract_pool"): + ok = extract_pool._run_preflight_probe(target=_hanging_probe_target, timeout=0.5) + elapsed = time.monotonic() - start + + assert ok is False + assert elapsed < 30.0 # nowhere near the child's 3600s sleep + assert "did not respond within" in caplog.text + assert "killed" in caplog.text + + +@pytest.mark.unit +def test_preflight_failure_latches_build_extraction_pool_to_in_process( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """D8: build_extraction_pool degrades to a correct in-process pool, and no + ProcessPoolExecutor is ever constructed, when the preflight probe fails.""" + monkeypatch.setattr(extract_pool, "_run_preflight_probe", lambda: False) + + pool = build_extraction_pool(_config(extract_processes=4)) + + assert pool._executor is None + files = _python_files(3) + assert list(pool.stream(files)) == [(pf, extract_file(pf)) for pf in files] + + +# --- T7: sizing derivation ------------------------------------------------------ + + +@pytest.mark.unit +@pytest.mark.parametrize( + ("content", "expected"), + [ + ("200000 100000\n", 2), + ("100000 100000\n", 1), + ("max 100000\n", None), + ("bogus 100000\n", None), + ("200000 0\n", None), + ("200000\n", None), + ("\n", None), + ], +) +def test_cgroup_cpu_quota_parsing( + monkeypatch: pytest.MonkeyPatch, content: str, expected: int | None +) -> None: + monkeypatch.setattr(extract_pool.Path, "read_text", lambda self: content) + assert extract_pool._cgroup_cpu_quota() == expected + + +@pytest.mark.unit +def test_cgroup_cpu_quota_unreadable_file_returns_none(monkeypatch: pytest.MonkeyPatch) -> None: + def _raise(self: Any) -> str: + raise OSError("no such file") + + monkeypatch.setattr(extract_pool.Path, "read_text", _raise) + assert extract_pool._cgroup_cpu_quota() is None + + +@pytest.mark.unit +def test_available_cpus_prefers_affinity_over_cpu_count(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(extract_pool.os, "sched_getaffinity", lambda pid: {0, 1, 2}, raising=False) + monkeypatch.setattr(extract_pool.os, "cpu_count", lambda: 64) + monkeypatch.setattr(extract_pool, "_cgroup_cpu_quota", lambda: None) + assert extract_pool._available_cpus() == 3 + + +@pytest.mark.unit +def test_available_cpus_falls_back_to_cpu_count_without_affinity( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delattr(extract_pool.os, "sched_getaffinity", raising=False) + monkeypatch.setattr(extract_pool.os, "cpu_count", lambda: 5) + monkeypatch.setattr(extract_pool, "_cgroup_cpu_quota", lambda: None) + assert extract_pool._available_cpus() == 5 + + +@pytest.mark.unit +def test_available_cpus_clamped_by_cgroup_quota(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr( + extract_pool.os, "sched_getaffinity", lambda pid: set(range(16)), raising=False + ) + monkeypatch.setattr(extract_pool, "_cgroup_cpu_quota", lambda: 2) + assert extract_pool._available_cpus() == 2 + + +@pytest.mark.unit +def test_available_cpus_floors_at_one(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(extract_pool.os, "sched_getaffinity", lambda pid: set(), raising=False) + monkeypatch.setattr(extract_pool, "_cgroup_cpu_quota", lambda: None) + assert extract_pool._available_cpus() == 1 + + +@pytest.mark.unit +def test_derive_process_count_explicit_value_wins(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(extract_pool, "_available_cpus", lambda: 1) # would derive to 1 if used + assert derive_process_count(_config(extract_processes=6)) == 6 + + +@pytest.mark.unit +def test_derive_process_count_kill_switch_is_explicit_one() -> None: + assert derive_process_count(_config(extract_processes=1)) == 1 + + +@pytest.mark.unit +def test_derive_process_count_derives_and_clamps_to_eight(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(extract_pool, "_available_cpus", lambda: 64) + assert derive_process_count(_config()) == 8 + + +@pytest.mark.unit +def test_derive_process_count_derives_below_ceiling(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(extract_pool, "_available_cpus", lambda: 3) + assert derive_process_count(_config()) == 3 + + +# --- T8: extract_processes: 1 creates no pool ---------------------------------- + + +@pytest.mark.unit +def test_extract_processes_one_creates_no_pool() -> None: + """T8: the kill switch. No subprocess is ever created (no executor object at + all), and results are identical to calling extract_file directly.""" + pool = build_extraction_pool(_config(extract_processes=1)) + assert pool._executor is None + + files = _python_files(3) + assert list(pool.stream(files)) == [(pf, extract_file(pf)) for pf in files] diff --git a/tests/unit/test_job.py b/tests/unit/test_job.py index 8d1c4a9..8fe93fe 100644 --- a/tests/unit/test_job.py +++ b/tests/unit/test_job.py @@ -25,6 +25,7 @@ import tarfile import threading import time +from concurrent.futures import Future from pathlib import Path from typing import Any, NamedTuple @@ -33,6 +34,7 @@ from app.config import Settings from app.db.models import INDEX_SEMANTICS_VERSION +from indexer.extract_pool import ExtractionPool, ExtractionPoolError from indexer.hashing import content_sha from indexer.job import ( BranchOutcome, @@ -43,7 +45,7 @@ read_github_token, run, ) -from indexer.languages import ExtractedSymbol, IndexCounts, ParsedFile +from indexer.languages import ExtractedSymbol, FileExtraction, IndexCounts, ParsedFile from indexer.repo_config import ConfigError, RepoConfig, load_config from indexer.resolve import RepoEntry from indexer.store import ( @@ -191,6 +193,7 @@ def _repo_meta(full_name: str, **overrides: Any) -> dict[str, Any]: def _config( *, index_concurrency: int | None = None, + extract_processes: int | None = 1, semantic_max_chunks_per_repo: dict[str, int] | None = None, semantic: dict[str, Any] | None = None, **connection: Any, @@ -201,10 +204,21 @@ def _config( job-side semantic-config surface), mirroring the ``semantic_max_chunks_per_repo`` per-repo-map kwarg above -- the two are distinct config surfaces (global INT vs per-repo MAP) and either can be set independently. + + ``extract_processes`` (#108) defaults to ``1`` -- the pool KILL SWITCH -- + deliberately, not to the schema's own ``None`` default. Left at ``None``, every + one of the ~150 ``run()`` tests below would run the preflight probe and spawn + real worker processes on every call, each re-importing + ``tree_sitter_language_pack``: minutes slower and flaky on a constrained CI + runner. Only the pool-specific tests in ``tests/unit/test_extract_pool.py`` + (and the dedicated pool-lifecycle tests here) opt into a real pool by passing + ``extract_processes=`` explicitly. """ doc: dict[str, Any] = {"version": 1, "connections": [{"type": "github", **connection}]} if index_concurrency is not None: doc["index_concurrency"] = index_concurrency + if extract_processes is not None: + doc["extract_processes"] = extract_processes if semantic_max_chunks_per_repo is not None: doc["semantic_max_chunks_per_repo"] = semantic_max_chunks_per_repo if semantic is not None: @@ -395,6 +409,7 @@ def _run( reconcile_retired_fn: Any = _noop_retired_fn, reconcile_removed_fn: Any = _noop_removed_fn, shas_fn: Any = _noop_shas_fn, + extraction_pool: Any = None, ) -> int: """Drive run() with a faked config read but a REAL resolve_repos. @@ -409,6 +424,11 @@ def _run( version-matching, sha-mismatching stamp with semantic indexing on would otherwise reach the REAL ``read_indexed_shas`` against ``_FakeConn``, which cannot answer it (see that fake's docstring). + + ``extraction_pool`` (#108) defaults to ``None``, in which case ``run()`` + builds its own from ``config.extract_processes`` (pinned to ``1``, the kill + switch, by ``_config``'s own default -- see T9). Pass an explicit fake/real + :class:`~indexer.extract_pool.ExtractionPool` to test the pool seam itself. """ wc = _FakeWorkspaceClient("tok") engine = engine if engine is not None else _FakeEngine() @@ -430,6 +450,7 @@ def _run( reconcile_retired_fn=reconcile_retired_fn, reconcile_removed_fn=reconcile_removed_fn, shas_fn=shas_fn, + extraction_pool=extraction_pool, ) @@ -1070,6 +1091,7 @@ def _down(_texts: list[str]) -> list[list[float]]: started=time.monotonic(), max_chunks_per_repo=100, shas_fn=_noop_shas_fn, + extraction_pool=ExtractionPool(n_processes=0), ) assert outcome.status == "indexed" assert outcome.semantic_degraded is True @@ -1576,6 +1598,7 @@ def test_index_one_inner_reports_discovery_complete_for_a_normal_run() -> None: embed_fn=None, stamps={}, shas_fn=_noop_shas_fn, + extraction_pool=ExtractionPool(n_processes=0), ) assert outcome.name == "acme/widgets" assert outcome.discovery_complete is True @@ -1603,6 +1626,7 @@ def test_index_one_inner_reports_discovery_incomplete_when_capped() -> None: embed_fn=None, stamps={}, shas_fn=_noop_shas_fn, + extraction_pool=ExtractionPool(n_processes=0), ) assert outcome.discovery_complete is False assert len(outcome.outcomes) == SOFT_BRANCH_CAP @@ -1631,6 +1655,7 @@ def test_index_one_inner_default_flip_mirror_is_complete() -> None: embed_fn=None, stamps={}, shas_fn=_noop_shas_fn, + extraction_pool=ExtractionPool(n_processes=0), ) assert [o.branch for o in outcome.outcomes] == ["main"] assert outcome.discovery_complete is True @@ -1708,7 +1733,9 @@ def test_two_repos_are_indexed_concurrently_at_concurrency_two() -> None: NOTE this proves CONCURRENCY, not throughput. It would pass identically in a world where fan-out delivers zero speedup — symbol extraction itself measured - 0.95x on 4 threads. Throughput is measured on the first production run, via + 0.95x on 4 THREADS, which is exactly why extraction now runs in a shared + process pool instead (#108) rather than leaning on this thread-level + fan-out for it. Throughput is measured on the first production run, via the duration log lines asserted below. """ idx = _BarrierIndex(parties=2) @@ -1939,6 +1966,7 @@ def test_repo_context_is_reset_even_when_the_repo_fails() -> None: embed_fn=None, stamps={}, shas_fn=_noop_shas_fn, + extraction_pool=ExtractionPool(n_processes=0), ) assert _repo_ctx.get() == "-" @@ -2933,17 +2961,24 @@ def test_parse_time_is_excluded_from_db_time( ) -> None: """T5: items are produced lazily INSIDE index_repo's transaction, so a naive "time index_fn" would report parse and db fused. _timed_items charges only the - production of each item to parse, leaving the DML between them to db.""" - import indexer.job as job + production of each item to parse, leaving the DML between them to db. + + Patches ``indexer.extract_pool.extract_file`` (#108 moved the in-process + extraction call site there from ``indexer.job``) -- with the kill switch + pinned by ``_config``'s default, ``ExtractionPool.stream()`` degrades to + calling exactly that name per file, so this still measures the real + ``_timed_items`` wrap around the pool's in-process path. + """ + import indexer.extract_pool as extract_pool clock = _install_fake_clock(monkeypatch) - real_extract_file = job.extract_file + real_extract_file = extract_pool.extract_file def _slow_extract(pf: ParsedFile) -> Any: clock.advance(2.0) return real_extract_file(pf) - monkeypatch.setattr(job, "extract_file", _slow_extract) + monkeypatch.setattr(extract_pool, "extract_file", _slow_extract) def _index(conn: Any, *, items: Any, **_: Any) -> IndexCounts: consumed = list(items) @@ -3482,3 +3517,200 @@ def _assert_disk_headroom(*args: Any, **kwargs: Any) -> None: "sweep": 64.0, "other": 9.0, } + + +# --- process-pool extraction (#108) ------------------------------------------- + + +@pytest.mark.unit +def test_default_helper_config_creates_no_extraction_subprocess( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """T9: the shared _config()/_run() helper pair pins extract_processes=1, so + the ~150 run() tests in this file spawn no worker processes and run no + preflight probe. Without this pin every one of them would spawn up to 8 + workers, each re-importing tree_sitter_language_pack -- minutes slower and + flaky under CI load.""" + probe_calls = {"n": 0} + + def _tracking_probe(*args: Any, **kwargs: Any) -> bool: + probe_calls["n"] += 1 + return True + + monkeypatch.setattr("indexer.extract_pool._run_preflight_probe", _tracking_probe) + + code = _run(_config(repos=["acme/widgets"]), _RecordingIndex()) + assert code == 0 + assert probe_calls["n"] == 0 + + +@pytest.mark.unit +def test_pool_engaged_attributes_stream_production_to_parse( + caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch +) -> None: + """T10: phase-timing attribution survives the pool. The pool is built in + run(), OUTSIDE branch_started, so it contributes nothing to any branch's + total by itself; only the per-item production _timed_items wraps is + charged, and it still lands on `parse`, never `other`.""" + clock = _install_fake_clock(monkeypatch) + + class _ClockAdvancingExecutor: + """Advances the fake clock once per submitted batch, proportional to + its file count -- mirrors this file's other `_slow_*` timing fakes, + just at the pool's own submit() seam instead of a direct function + patch.""" + + def submit(self, fn: Any, *args: Any) -> "Future[list[FileExtraction]]": + batch = args[0] + clock.advance(1.5 * len(batch)) + fut: "Future[list[FileExtraction]]" = Future() + fut.set_result(fn(*args)) + return fut + + pool = ExtractionPool(n_processes=0, batch_bytes=10_000_000) # one big batch + pool._n_processes = 2 + pool._executor = _ClockAdvancingExecutor() # type: ignore[assignment] + + idx = _RecordingIndex() + cfg = Settings(semantic_enabled=False) + # Both files must be a recognized language (unlike _DEFAULT_FILES' README.md, + # `lang=None`) so BOTH qualify for the pool and land in the SAME one-file- + # exceeds-nothing batch -- otherwise D5's local short-circuit answers one of + # them without ever reaching the executor, undercounting this assertion. + github = _GitHub( + files={"a.py": b"def f():\n return 1\n", "b.py": b"def g():\n return 2\n"} + ) + with caplog.at_level(logging.INFO, logger="indexer.job"): + code = _run( + _config(repos=["acme/widgets"]), idx, cfg=cfg, github=github, extraction_pool=pool + ) + assert code == 0 + + phases = _only_phases(caplog) + assert list(phases) == [ + "total", + "resolve", + "download", + "parse", + "embed", + "db", + "sweep", + "other", + ] + assert phases["parse"] == 3.0 # a.py + b.py (the github= override above) x 1.5s, one batch + assert phases["other"] == 0.0 + assert phases["db"] == 0.0 + assert phases["total"] == 3.0 + + +@pytest.mark.unit +def test_extraction_pool_is_shut_down_even_when_fan_out_raises( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Pool lifecycle: built once per run, shut down in the `finally` even when + something inside the fan-out block raises before any branch runs.""" + import indexer.job as job + + shutdown_calls = {"n": 0} + + class _FakePool: + def shutdown(self) -> None: + shutdown_calls["n"] += 1 + + monkeypatch.setattr(job, "build_extraction_pool", lambda config: _FakePool()) + + def _boom_stamps(engine: Any, entries: Any) -> Any: + raise RuntimeError("boom") + + monkeypatch.setattr(job, "_read_stamps", _boom_stamps) + + with pytest.raises(RuntimeError): + _run(_config(repos=["acme/widgets"]), _RecordingIndex()) + + assert shutdown_calls["n"] == 1 + + +@pytest.mark.unit +def test_pool_is_built_before_the_first_db_read(monkeypatch: pytest.MonkeyPatch) -> None: + """R2 (§10.3 of the plan): the preflight probe is the first process spawn in + run(), and it MUST precede any database read or write. A re-entrant + python_wheel_task child under an unguarded __main__ inherits the parent's + sys.argv and would otherwise become a second corpus writer; the probe's + unguarded-__main__ crash (reproduced: RuntimeError from + multiprocessing.spawn._check_not_importing_main) only bounds that risk if + it runs before the child could touch the database. This pins the ordering + directly so a future refactor (e.g. "only build the pool if a branch needs + it") cannot silently move the pool build after _read_stamps without a test + failing -- do not "fix" a failure here by reordering the assertion.""" + import indexer.job as job + + calls: list[str] = [] + + def _tracking_build(config: Any) -> Any: + calls.append("pool") + return ExtractionPool(n_processes=0) + + real_read_stamps = job._read_stamps + + def _tracking_read_stamps(*args: Any, **kwargs: Any) -> Any: + calls.append("stamps") + return real_read_stamps(*args, **kwargs) + + monkeypatch.setattr(job, "build_extraction_pool", _tracking_build) + monkeypatch.setattr(job, "_read_stamps", _tracking_read_stamps) + + code = _run(_config(repos=["acme/widgets"]), _RecordingIndex()) + assert code == 0 + assert calls == ["pool", "stamps"] + + +@pytest.mark.unit +def test_injected_extraction_pool_is_never_shut_down_by_run() -> None: + """Mirrors owns_engine/owns_http: a CALLER-supplied pool is the caller's to + manage, never torn down inside run().""" + shutdown_calls = {"n": 0} + + class _FakePool: + def stream(self, files: Any) -> Any: + return ((pf, FileExtraction(symbols=[], edges=[])) for pf in files) + + def shutdown(self) -> None: + shutdown_calls["n"] += 1 + + code = _run(_config(repos=["acme/widgets"]), _RecordingIndex(), extraction_pool=_FakePool()) + assert code == 0 + assert shutdown_calls["n"] == 0 + + +@pytest.mark.unit +def test_broken_pool_fails_only_the_in_flight_branch_next_branch_recovers( + caplog: pytest.LogCaptureFixture, +) -> None: + """AC3: a shared pool's break fails the branch it broke under; the run + continues and the NEXT branch of the same repo indexes normally -- the + per-branch classification this design maps ExtractionPoolError onto.""" + call_count = {"n": 0} + + class _FlakyPool: + def stream(self, files: Any) -> Any: + call_count["n"] += 1 + if call_count["n"] == 1: + raise ExtractionPoolError("simulated pool break (generation 0)") + return ((pf, FileExtraction(symbols=[], edges=[])) for pf in files) + + def shutdown(self) -> None: + pass + + idx = _RecordingIndex() + github = _GitHub(branches={"acme/widgets": ["main", "feature"]}) + # Default branch ("main") is always first and always included -- see + # indexer.branches.resolve_branches -- so "main" is the branch that breaks + # and "feature" is the one that recovers. + config = _config(repos=["acme/widgets"], branches=["feature"], index_concurrency=1) + + with caplog.at_level(logging.INFO, logger="indexer.job"): + code = _run(config, idx, github=github, extraction_pool=_FlakyPool()) + + assert code == 1 # the run continues but is not clean + assert "failed to index acme/widgets@main" in caplog.text + assert idx.calls == ["acme/widgets"] # only "feature" reached index_repo diff --git a/tests/unit/test_repo_config.py b/tests/unit/test_repo_config.py index e0d0eb4..5861752 100644 --- a/tests/unit/test_repo_config.py +++ b/tests/unit/test_repo_config.py @@ -213,6 +213,41 @@ def test_index_concurrency_out_of_range_raises_config_error(value: int) -> None: assert "index_concurrency" in str(excinfo.value) +# --- extract_processes (#108) ------------------------------------------------- + + +@pytest.mark.unit +def test_extract_processes_defaults_to_none() -> None: + """None means "derive from the runtime" (indexer.extract_pool.derive_process_count) -- + omitting the field is the supported shape, since it predates #108.""" + assert parse_config(_MINIMAL, source="cfg").extract_processes is None + + +@pytest.mark.unit +@pytest.mark.parametrize("value", [1, 4, 8]) +def test_extract_processes_accepts_in_range(value: int) -> None: + raw = b"version: 1\nconnections:\n - type: github\n users: [u]\nextract_processes: %d\n" % ( + value + ) + + assert parse_config(raw, source="cfg").extract_processes == value + + +@pytest.mark.unit +@pytest.mark.parametrize("value", [0, 9]) +def test_extract_processes_out_of_range_raises_config_error(value: int) -> None: + """Matches index_concurrency's `1..8` ceiling -- every parallelism knob in this + repo shares it, and the pool's own sizing derivation already clamps to 8.""" + raw = b"version: 1\nconnections:\n - type: github\n users: [u]\nextract_processes: %d\n" % ( + value + ) + + with pytest.raises(ConfigError) as excinfo: + parse_config(raw, source="cfg") + + assert "extract_processes" in str(excinfo.value) + + @pytest.mark.unit @pytest.mark.parametrize( ("configured", "semantic_enabled", "expected"),