diff --git a/docs/runbooks/indexing-parallelism.md b/docs/runbooks/indexing-parallelism.md index f4e9947..fed089c 100644 --- a/docs/runbooks/indexing-parallelism.md +++ b/docs/runbooks/indexing-parallelism.md @@ -116,8 +116,9 @@ loop, third-party libraries) carry `-`. ``` INFO indexer.job [-]: local disk at /tmp: 41.2 GB free of 64.0 GB total; 4 worker(s) x 2.5 GB peak INFO indexer.fetch [acme/widgets]: ... -INFO indexer.job [acme/widgets]: finished acme/widgets in 71.30s -INFO indexer.job [acme/gadgets]: skipped acme/gadgets: already indexed at abc123 (semantics v1) in 0.41s +INFO indexer.job [acme/widgets]: phase timing acme/widgets@main: total=213.32s resolve=0.00s download=12.10s extract=8.40s parse=31.00s embed=88.20s db=64.50s sweep=0.30s other=8.82s +INFO indexer.job [acme/widgets]: finished acme/widgets in 213.74s (resolve=0.42s list=0.00s) +INFO indexer.job [acme/gadgets]: skipped acme/gadgets@main: already indexed at abc123 (semantics v1) in 0.41s ``` **To find the giant:** grep for `finished .* in` and sort by the elapsed number. @@ -130,6 +131,59 @@ accept the duration. line against the sum of the per-repo elapsed times. If the total is already close to the slowest single repo, the pool is not the bottleneck. +### 2.1 Finding the dominant phase, not just the dominant repo + +Every **indexed** branch emits one `phase timing` line accounting for its entire +wall clock. Skipped, failed, and conflicted branches emit none — there is nothing +to attribute. + +``` +grep 'phase timing' run.log # one line per indexed branch +``` + +The nine fields are fixed, always present, always in this order, always `%.2fs`. +A phase that did not run prints `0.00s` rather than disappearing, so the line +never changes shape between a semantic-on and a semantic-off run and every grep +you write keeps working. Read the largest field; that is the branch's bottleneck. + +| Dominant phase | What it means | Which issue addresses it | +|---|---|---| +| `download` / `extract` | archive I/O bound | #106 (single-pass in-memory ingestion) | +| `parse` | GIL-bound tree-sitter extraction | #108 (process-pool extraction) | +| `embed` | serial AI Gateway round trips | #107 (concurrent embedding) | +| `db` | per-file round trips | #105 (batched writes) | +| any of the above, on **unchanged** content | redundant work | #104 (file-level delta indexing) | + +Four fields need interpretation before you act on them: + +- **`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 + `resolve=`. The `list=` on the same line is the branch-listing API call, which + is `0.00s` unless the repo has `branches:` globs configured (it is not called + at all otherwise) and which is paginated — on a monorepo with hundreds of + branches it is a real, and otherwise invisible, cost. The elapsed value in + `finished … in Xs` is measured on its own clock and is deliberately not + reconciled against the parenthesised numbers. +- **`other=` is the unattributed residual**, `total` minus every measured phase, + clamped at zero. It is dominated by the temp-dir teardown — an `rm -rf` of a + freshly extracted multi-GB tree — plus the pre-flight disk check. It exists so + the line has no silently missing time; a large `other` means something real is + happening outside every instrumented phase and is worth chasing. +- **`embed=` covers chunking as well as the network.** It spans `iter_chunks` + (CPU/GIL-bound) *and* the serial AI Gateway round trips. #107 addresses only + the round trips, so before routing work there, confirm the phase is + network-bound rather than chunking-bound (a follow-up may split it into + `chunk=`/`embed=`). +- **`db=` excludes parse and sweep, but the walk still happens inside the + transaction.** Files stream lazily through `index_repo`'s open transaction for + bounded memory, so file production is timed separately and subtracted from + `db`; the sweep is subtracted too. On the **non-semantic** path, though, the + directory walk itself materializes inside that open transaction (`parse.py`'s + `rglob`, on the first item). That is long-standing behavior which this + instrumentation merely makes visible for the first time — it is not a new + regression. + --- ## 3. The three limits, and why raising concurrency is a bad trade diff --git a/indexer/AGENTS.md b/indexer/AGENTS.md index eb5c590..f894434 100644 --- a/indexer/AGENTS.md +++ b/indexer/AGENTS.md @@ -14,13 +14,14 @@ The serverless indexing job (`code-search-index`, wired as a `python_wheel_task` | `chunk_store.py` | `write_chunks`: delete-and-reinsert one file's rows in the `chunks` table (no natural key). 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), safe extraction (`filter="data"`, bomb check) capped at `MAX_EXTRACTED_BYTES` (2 GB), `assert_disk_headroom` (2.5 GB per worker, both caps alive at once). `RateLimitError` is deliberately narrow: 429 always; 403 only with `Retry-After` or `X-RateLimit-Remaining: 0` — other 403s are permission failures. | | `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. 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), 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=… extract=… 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`: 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 (2.5 GB peak per worker), 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`. 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 array-union upsert on `(repo_id, path, content_sha)` with branch-membership union, delete-and-reinsert `symbols` AND `reference_edges` (both keyed only by `file_id`, no natural key), optional `chunk_writer` call, 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]]`; the `reference_edges` delete runs unconditionally, even when a file's `FileExtraction.edges` is empty, so stale rows never survive a re-index. 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). | +| `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 array-union upsert on `(repo_id, path, content_sha)` with branch-membership union, delete-and-reinsert `symbols` AND `reference_edges` (both keyed only by `file_id`, no natural key), optional `chunk_writer` call, 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]]`; the `reference_edges` delete runs unconditionally, even when a file's `FileExtraction.edges` is empty, so stale rows never survive a re-index. 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()`. | +| `timing.py` | Per-phase wall-clock accounting for one branch (#103). `PhaseTimer` accumulates `phase -> seconds`; `install_timer`/`reset_timer`/`current_timer` carry one ambiently in a `ContextVar` (the same idiom as `job.py`'s `_repo_ctx`, for the same cross-module attribution problem); `record(phase, seconds)` is a **no-op when no timer is installed and never raises**, so `index_repo` stays callable directly. `_CLOCK` (default `time.monotonic`) is the single clock source every asserted duration reads — via `now()`, or via `PhaseTimer.clock` captured at construction — and is the seam tests patch; never read `time.monotonic()` directly for a number that appears on those lines. Stdlib only. | ## For AI Agents @@ -38,7 +39,7 @@ The serverless indexing job (`code-search-index`, wired as a `python_wheel_task` - **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. ### Testing Requirements -- `make test` (`pytest -m "unit or observability"`, no external deps): `tests/unit/test_branches.py`, `test_chunk_store.py`, `test_chunking.py`, `test_edges.py`, `test_fetch.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_chunk_writer.py`, `test_semantics_version_tripwire.py`. +- `make test` (`pytest -m "unit or observability"`, no external deps): `tests/unit/test_branches.py`, `test_chunk_store.py`, `test_chunking.py`, `test_edges.py`, `test_fetch.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_chunk_writer.py`, `test_semantics_version_tripwire.py`, `test_timing.py`. - `make test-integration` (needs Postgres): `tests/integration/test_store.py`, `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. diff --git a/indexer/job.py b/indexer/job.py index 83853be..3a0f1d8 100644 --- a/indexer/job.py +++ b/indexer/job.py @@ -79,6 +79,18 @@ loop as symbols. Flag-off: no chunking, no embedder, no import of ``app.embed``'s lazy ``databricks-sdk`` dependency. +Every INDEXED branch also emits one ``phase timing`` line accounting for its whole +wall clock -- resolve / download / extract / parse / embed / db / sweep, plus an +``other`` residual so no time is silently unattributed. The fields are fixed and +unconditional (a phase that did not run prints ``0.00s``) so the line stays +greppable whether or not semantic indexing is on. Skipped, failed, and conflicted +branches emit no such line. ``sweep`` is measured inside ``indexer.store`` and +reaches this module through the ambient timer in :mod:`indexer.timing` rather +than through the ``index_fn`` seam, which keeps that seam's signature (and every +fake of it) unchanged. The repo-level costs that sit outside every branch's +total -- the default branch's HEAD resolve and the branch listing -- are reported +as ``resolve=``/``list=`` on the per-repo ``finished`` line instead. + Logging is INFO only. The GitHub token is read via an injected client and is never logged, and this module never lowers root/SDK/httpx log levels (see the redaction test + source-level tripwire). @@ -93,7 +105,7 @@ import sys import tempfile import time -from collections.abc import Callable +from collections.abc import Callable, Iterable, Iterator from concurrent.futures import ThreadPoolExecutor, as_completed from contextvars import ContextVar from dataclasses import dataclass, field @@ -118,7 +130,7 @@ resolve_branch_head, resolve_ref, ) -from indexer.languages import Chunk, IndexCounts, ParsedFile +from indexer.languages import Chunk, FileExtraction, IndexCounts, ParsedFile from indexer.parse import iter_chunks, iter_source_files from indexer.repo_config import RepoConfig, effective_workers, load_config, normalize_repo from indexer.resolve import MAX_REPOS, RepoEntry, resolve_repos @@ -131,6 +143,7 @@ reconcile_retired_branches, ) from indexer.symbols import extract_file +from indexer.timing import PhaseTimer, install_timer, now, reset_timer logger = logging.getLogger("indexer.job") @@ -887,7 +900,15 @@ def _index_one_inner( """ name = normalize_repo(entry.name) org, repo = name.split("/", 1) + # Timed with two plain locals off the same clock the per-branch PhaseTimer + # reads -- no timer is installed here, because these costs are REPO-scoped + # and belong to no branch's total (they are reported on the `finished` line + # below instead). Without them the default branch's resolve would show as + # `resolve=0.00s` on its phase line, and a glob-configured monorepo's + # paginated branch listing would appear in no field on no line at all. + resolve_started = now() default_branch, default_head_sha = resolve_ref(http_client, org, repo) + resolve_elapsed = now() - resolve_started # An override matched to THIS repo by resolve_repos wins outright over the global # cap -- it is not a floor/ceiling blend, since a repo big enough to need one @@ -904,7 +925,9 @@ def _index_one_inner( # The common case -- no branches: configured -- needs no GitHub branches API # call at all: resolve_branches ignores all_branches entirely when globs is # empty (it always resolves to just [default_branch]). + list_started = now() all_branches = list_branches(http_client, org, repo) if entry.branch_globs else [] + list_elapsed = now() - list_started resolution = resolve_branches( default_branch, all_branches, sorted(entry.branch_globs), repo=name ) @@ -934,10 +957,48 @@ def _index_one_inner( # assertions, and a nonzero-by-construction timing field would break them. # This is the instrument the "throughput measured on the first production # run" promise depends on -- without it that promise is unfalsifiable. - logger.info("finished %s in %.2fs", name, time.monotonic() - started) + # `resolve=`/`list=` are the repo-scoped costs that appear in no branch's + # `phase timing` total. Printed unconditionally (`list=0.00s` whenever the + # repo has no `branches:` globs and the endpoint is never called) so the line + # never drifts in shape. The elapsed value keeps its own time.monotonic() + # reading -- it is not reconciled against the parenthesised numbers. + logger.info( + "finished %s in %.2fs (resolve=%.2fs list=%.2fs)", + name, + time.monotonic() - started, + resolve_elapsed, + list_elapsed, + ) return RepoOutcome(name=name, discovery_complete=resolution.complete, outcomes=outcomes) +def _timed_items( + items: Iterable[tuple[ParsedFile, FileExtraction]], timer: PhaseTimer +) -> Iterator[tuple[ParsedFile, FileExtraction]]: + """Charge each item's PRODUCTION to ``parse``, leaving its consumption to ``db``. + + ``items`` is consumed lazily inside ``index_repo``'s open transaction (the + bounded-memory invariant), so timing ``index_fn`` alone fuses parse and db + into one number, and materializing the generator to separate them would break + that invariant. This wrapper times only the ``next()`` calls -- the walk, the + read, the decode, and the tree-sitter extraction -- so the DML between them + stays attributable to ``db``. Never build a list here. + + The post-``StopIteration`` charge is not bookkeeping pedantry: on the + non-semantic path the final ``next()`` is where the directory walk finishes. + """ + it = iter(items) + while True: + t0 = timer.clock() + try: + item = next(it) + except StopIteration: + timer.add("parse", timer.clock() - t0) + return + timer.add("parse", timer.clock() - t0) + yield item + + def _index_one_branch( name: str, *, @@ -967,11 +1028,30 @@ def _index_one_branch( matched, else ``cfg.semantic_max_chunks_per_repo``. Taken as a parameter rather than read from ``cfg`` directly so every branch of a repo enforces the SAME resolved cap without recomputing (or risking drift on) the override lookup. + + Every phase of the indexed path is wall-clocked into ``timer`` and reported on + one ``phase timing`` line at the end (see the module docstring). ``timer`` is + also installed as the AMBIENT timer for the duration, which is how + ``indexer.store``'s sweep -- behind the ``index_fn`` seam, in another module -- + lands in this branch's numbers without changing that seam's signature. The + ``finally: reset_timer(token)`` is mandatory for the same reason + ``_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. """ + # 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 + # glob-configured repo report a repo-cumulative total and silently inflate + # `other` by every preceding branch's wall clock. + timer = PhaseTimer() + branch_started = timer.clock() + timer_token = install_timer(timer) try: + t0 = timer.clock() head_sha = ( default_head_sha if is_default else resolve_branch_head(http_client, org, repo, branch) ) + timer.add("resolve", timer.clock() - t0) # The skip seam: after the immutable HEAD SHA is known, before anything # is downloaded. Both halves must match -- a stored NULL version never does. @@ -993,15 +1073,35 @@ def _index_one_branch( # being written to) and BEFORE the first byte is downloaded. Raising # here is caught below, costing this branch alone. assert_disk_headroom(tmp_path, repo=f"{name}@{branch}") + + t0 = timer.clock() tar_path = download_tarball(http_client, org, repo, head_sha, tmp_path) + timer.add("download", timer.clock() - t0) + + t0 = timer.clock() root = extract_tarball(tar_path, tmp_path / "extracted") + timer.add("extract", timer.clock() - t0) chunk_writer: ChunkWriter | None = None if cfg.semantic_enabled and embed_fn is not None: # Chunking/embedding needs the full file list up front -- unlike # the lazy items generator below, it cannot stream through # index_repo's open transaction. + # + # This walk is charged to `parse` explicitly: it is the same + # rglob + stat + read + decode work that _timed_items charges on + # the non-semantic path, and leaving it unwrapped would dump the + # entire file-walk cost of the PRODUCTION (semantic-on) path into + # `other`, which is exactly the number this instrumentation + # exists to route work by. + t0 = timer.clock() files = list(iter_source_files(root)) + timer.add("parse", timer.clock() - t0) + + # In a `finally`, unlike every other phase wrap: the degrade path + # below still burned this time (a downed embedder can burn a lot + # of it before it gives up) and must still be reported. + t0 = timer.clock() try: chunk_writer = _precompute_chunk_writer(files, embed_fn, max_chunks_per_repo) except Exception: @@ -1018,11 +1118,23 @@ def _index_one_branch( exc_info=True, ) chunk_writer = None + finally: + timer.add("embed", timer.clock() - t0) items = ((pf, extract_file(pf)) for pf in files) else: # Lazy generator: files stream through the open transaction (bounded memory). items = ((pf, extract_file(pf)) for pf in iter_source_files(root)) + # Parse time is INSIDE the db window (items are produced lazily as + # index_repo consumes them), so `db` subtracts only the parse accrued + # DURING that window -- never the phase total, which on the semantic + # path already holds the eager walk above. `sweep` is windowed the + # same way for the same reason (store.py records it from inside + # index_fn) -- kept symmetric with `parse` so a future sweep call + # site outside this window can't silently mis-window `db`. + parse_before = timer.total("parse") + sweep_before = timer.total("sweep") + t0 = timer.clock() with engine.connect() as conn: counts = index_fn( conn, @@ -1030,9 +1142,62 @@ def _index_one_branch( branch=branch, is_default=is_default, head_sha=head_sha, - items=items, + items=_timed_items(items, timer), chunk_writer=chunk_writer, ) + db_wall = timer.clock() - t0 + + # Emitted HERE -- after the TemporaryDirectory teardown (an rm -rf of a + # possibly multi-GB extracted tree, which `other` must include) and from + # inside the worker, where _index_one's _repo_ctx still resolves the + # [%(repo)s] field. The drain loop on the main thread would render `[-]`. + # + # `index_fn` above already committed this branch's transaction -- `counts` + # is proof of that. This block is measurement ONLY, so it gets its own + # try/except: per timing.py's own principle ("instrumentation must never + # be able to fail the work it measures"), a bug here must degrade to a + # missing log line, never to reclassifying an already-committed branch as + # `failed` (which would also flip the run's exit code and gate off the + # post-fan-out reconciliation checkpoint, which requires zero failures). + try: + db = max( + 0.0, + db_wall + - (timer.total("parse") - parse_before) + - (timer.total("sweep") - sweep_before), + ) + total = timer.clock() - branch_started + other = max( + 0.0, + total + - timer.total("resolve") + - timer.total("download") + - timer.total("extract") + - timer.total("parse") + - timer.total("embed") + - db + - timer.total("sweep"), + ) + # One format string, no branches: a phase that did not run prints 0.00s + # rather than vanishing, so the line stays greppable and field-stable + # whether or not semantic indexing is on. + logger.info( + "phase timing %s@%s: total=%.2fs resolve=%.2fs download=%.2fs extract=%.2fs " + "parse=%.2fs embed=%.2fs db=%.2fs sweep=%.2fs other=%.2fs", + name, + branch, + total, + timer.total("resolve"), + timer.total("download"), + timer.total("extract"), + timer.total("parse"), + timer.total("embed"), + db, + timer.total("sweep"), + other, + ) + except Exception: + logger.warning("phase timing unavailable for %s@%s", name, branch, exc_info=True) return BranchOutcome(branch=branch, status="indexed", counts=counts) except StaleIndexError as exc: # The repo_branches row for THIS branch changed under this worker, so @@ -1065,6 +1230,12 @@ def _index_one_branch( except Exception: logger.exception("failed to index %s@%s", name, branch) return BranchOutcome(branch=branch, status="failed") + finally: + # Mandatory, exactly like _index_one's _repo_ctx reset: ThreadPoolExecutor + # reuses worker threads without resetting their context, so a leaked timer + # would attribute this branch's sweep to the NEXT branch (or repo) that + # lands on this thread -- silently wrong numbers, which is worse than none. + reset_timer(timer_token) def _positive_int(raw: str) -> int: diff --git a/indexer/store.py b/indexer/store.py index f85f0df..84e110a 100644 --- a/indexer/store.py +++ b/indexer/store.py @@ -29,6 +29,7 @@ from app.db.models import INDEX_SEMANTICS_VERSION, File, ReferenceEdge, Repo, RepoBranch, Symbol from indexer.hashing import content_sha from indexer.languages import FileExtraction, IndexCounts, ParsedFile +from indexer.timing import now, record logger = logging.getLogger("indexer.store") @@ -242,6 +243,14 @@ def index_repo( if chunk_writer is not None: chunk_writer(conn, repo_id, file_id, pf) + # Timed into indexer.job's ambient per-branch PhaseTimer, if one is + # installed -- a no-op otherwise, so a direct index_repo call (tests, + # scripts) is unaffected. Deliberately NOT a return value: IndexCounts is + # a frozen dataclass compared by value in existing assertions, and NOT a + # new index_repo parameter: that signature is an injected seam whose + # fakes would all have to grow one. No log record is emitted here; the + # number surfaces on job.py's single `phase timing` line. + sweep_started = now() swept = _sweep_membership( conn, name=name, @@ -250,6 +259,7 @@ def index_repo( seen_paths=seen_paths, seen_shas=seen_shas, ) + record("sweep", now() - sweep_started) _stamp_repo_branch( conn, diff --git a/indexer/timing.py b/indexer/timing.py new file mode 100644 index 0000000..4a7f477 --- /dev/null +++ b/indexer/timing.py @@ -0,0 +1,117 @@ +"""Ambient per-phase wall-clock accounting for one branch's indexing pipeline. + +``indexer.job`` measures most of a branch's phases (resolve / download / extract / +parse / embed / db) inline, but ``sweep`` runs deep inside +:func:`indexer.store.index_repo`, behind the injected ``index_fn`` seam and a +frozen ``IndexCounts`` return type. Threading a timer through that seam would +force every existing ``index_fn`` fake to grow a parameter, turning a log-only +change into a storage-contract change -- so the timer is carried *ambiently* in a +:class:`contextvars.ContextVar`, exactly mirroring ``indexer.job``'s ``_repo_ctx`` ++ ``RepoLogFilter`` idiom for the same cross-module attribution problem. + +Two properties are load-bearing: + +* **Default-unset is a no-op.** :func:`record` with no installed timer does + nothing and never raises, so ``index_repo`` stays callable directly (as + ``tests/integration/test_store.py`` does) with no timer in sight. +* **Install/reset discipline.** ``ThreadPoolExecutor`` reuses worker threads and + does NOT reset their context between tasks, so :func:`install_timer` must + always be paired with :func:`reset_timer` in a ``finally`` -- a leaked timer + would silently attribute one branch's sweep to the next branch's line. + +The ``ContextVar`` does not cross a thread or process boundary: it isolates +concurrent worker threads from each other (which is the property this module +needs today), but a call to :func:`record` from a different thread or a +``ProcessPoolExecutor`` worker than the one that called :func:`install_timer` +reaches a *different* ambient timer (or none), not the installing thread's. +Relevant if a future phase's work is moved off the indexing worker thread -- +e.g. issue #108's process-pool extraction, or issue #107's concurrent +embedding. + +``_CLOCK`` is the module's single clock source. Every interval that appears on +an asserted log line reads it (via :func:`now`, or via +:attr:`PhaseTimer.clock` which captures it at construction), never a bare +``time.monotonic()`` -- mixing a fake clock with the real one makes the residual +``other=`` field meaningless. Tests patch ``indexer.timing._CLOCK`` before +driving ``run()``; that is the only injection point that reaches a +worker-constructed timer. + +Stdlib only (``time`` + ``contextvars``): this module is imported by the +indexing hot path and adds no dependency. +""" + +from __future__ import annotations + +import time +from collections.abc import Callable +from contextvars import ContextVar, Token + +# The single clock source for every duration on the phase-timing and repo +# `finished` log lines. Monotonic (never NTP-adjusted) and patched wholesale by +# tests -- do not read time.monotonic() directly anywhere those numbers are +# asserted against each other. +_CLOCK: Callable[[], float] = time.monotonic + + +def now() -> float: + """Read the current clock, resolving ``_CLOCK`` at call time (test seam).""" + return _CLOCK() + + +class PhaseTimer: + """A per-branch accumulator of ``phase -> seconds``. + + Not thread-safe by design: one timer belongs to exactly one branch, which is + indexed by exactly one worker thread. Cross-thread isolation comes from the + ``ContextVar`` below, not from locking. + """ + + def __init__(self) -> None: + # Captured at construction (not read per call) so one branch's + # arithmetic can never straddle a clock swap mid-flight. There is no + # constructor override for this: `record()` reaches this timer only + # through the ambient ContextVar, always via the module-level + # `_CLOCK`, so a per-instance clock would silently desync from the + # sweep timing recorded through `now()` in indexer/store.py. + self.clock: Callable[[], float] = _CLOCK + self._totals: dict[str, float] = {} + + def add(self, phase: str, seconds: float) -> None: + """Accumulate ``seconds`` into ``phase`` (phases are accrued, not set).""" + self._totals[phase] = self._totals.get(phase, 0.0) + seconds + + def total(self, phase: str) -> float: + """Seconds accrued to ``phase``; ``0.0`` for a phase that never ran.""" + return self._totals.get(phase, 0.0) + + +# Ambient per-thread timer. Default None = "nobody is measuring", which is the +# state every direct index_repo caller (and every non-worker thread) sees. +_timer_ctx: ContextVar[PhaseTimer | None] = ContextVar("phase_timer", default=None) + + +def install_timer(timer: PhaseTimer) -> Token[PhaseTimer | None]: + """Make ``timer`` the ambient one for this context; reset the token in a ``finally``.""" + return _timer_ctx.set(timer) + + +def reset_timer(token: Token[PhaseTimer | None]) -> None: + """Undo :func:`install_timer`. Mandatory -- see the module docstring.""" + _timer_ctx.reset(token) + + +def current_timer() -> PhaseTimer | None: + """The ambient timer, or ``None`` when nothing is measuring.""" + return _timer_ctx.get() + + +def record(phase: str, seconds: float) -> None: + """Accrue ``seconds`` to ``phase`` on the ambient timer, if there is one. + + A no-op when no timer is installed. Instrumentation must never be able to + fail the work it measures. + """ + timer = _timer_ctx.get() + if timer is None: + return + timer.add(phase, seconds) diff --git a/tests/integration/test_store.py b/tests/integration/test_store.py index 194fef5..22cc75e 100644 --- a/tests/integration/test_store.py +++ b/tests/integration/test_store.py @@ -32,6 +32,7 @@ ParsedFile, ) from indexer.store import StaleIndexError, _stamp_repo_branch, index_repo +from indexer.timing import PhaseTimer, install_timer, reset_timer SCHEMA = "test_store" @@ -178,6 +179,38 @@ def test_mark_and_sweep_removes_deleted_file(conn: Connection) -> None: assert _count(conn, "files", "commit = 'sha_second'") == 1 +@pytest.mark.integration +def test_index_repo_records_the_sweep_phase(conn: Connection) -> None: + """The sweep duration reaches ``indexer.job``'s per-branch line ambiently. + + ``index_repo`` returns a frozen ``IndexCounts`` and takes no timer parameter -- + the sweep's cost travels out through the ``indexer.timing`` ContextVar instead, + which is what keeps the ``index_fn`` seam (and every fake of it) unchanged. + ``tests/unit/test_job.py`` pins the arithmetic against a fake ``index_fn``; + this pins the real cross-module wiring against the real sweep and real SQL. + """ + _index_default(conn, name="acme/widgets", head_sha="sha_first", items=_items(MAIN, UTIL)) + conn.rollback() + + timer = PhaseTimer() + token = install_timer(timer) + try: + # Re-run without util.py at a new SHA: the same scenario as + # test_mark_and_sweep_removes_deleted_file, so the counts are unchanged. + counts = _index_default( + conn, name="acme/widgets", head_sha="sha_second", items=_items(MAIN) + ) + finally: + reset_timer(token) + + assert counts == IndexCounts(files=1, symbols=1, swept=1, edges=0) + assert timer.total("sweep") > 0.0 + # index_repo measures the sweep and nothing else -- every other phase is + # job.py's to record. + assert timer.total("db") == 0.0 + assert timer.total("parse") == 0.0 + + @pytest.mark.integration def test_sweep_is_repo_scoped(conn: Connection) -> None: # Two repos indexed at their own SHAs; re-indexing repo A with a new SHA must diff --git a/tests/unit/test_job.py b/tests/unit/test_job.py index 85c1b30..d4ca8dd 100644 --- a/tests/unit/test_job.py +++ b/tests/unit/test_job.py @@ -15,6 +15,7 @@ from __future__ import annotations import base64 +import dataclasses import inspect import io import logging @@ -2266,3 +2267,732 @@ def test_reconcile_unit_level_partial_progress_on_mid_sequence_failure() -> None assert progress.committed_any is True assert progress.purge_blocked is False assert progress.purged_repos == [] + + +# --- per-phase timing instrumentation (#103) -------------------------------- +# Every INDEXED branch emits one `phase timing` record accounting for its whole +# wall clock. The numbers are asserted EXACTLY, never with sleeps: the tests +# below patch `indexer.timing._CLOCK` (the single seam every asserted interval +# reads, captured by PhaseTimer.__init__ inside the worker) with a counter that +# only moves when a fake explicitly moves it. A sleep-based version of these +# assertions would be flaky under CI load and could not pin a residual at all. +# +# `_CLOCK` is a module global, so a fake counter is shared process-wide while a +# run fans out. Any fake-clock test driving more than one repo therefore pins +# `index_concurrency=1`, which gives one worker thread and one deterministic +# advance sequence. + +# Anchored and exhaustive on purpose: this single constant pins the field set, +# the field ORDER, the `%.2f` shape, the mandatory single-space separator, and +# the absence of anything else on the line (no config values, no counts, no +# headers -- the redaction posture). `\d+\.\d\d` also rejects a negative value +# outright, which is what makes the `max(0.0, ...)` clamps in job.py testable. +_TIMING_RE = re.compile( + r"^phase timing (?P[^ @]+)@(?P\S+): " + r"total=(\d+\.\d\d)s resolve=(\d+\.\d\d)s download=(\d+\.\d\d)s " + r"extract=(\d+\.\d\d)s parse=(\d+\.\d\d)s embed=(\d+\.\d\d)s " + r"db=(\d+\.\d\d)s sweep=(\d+\.\d\d)s other=(\d+\.\d\d)s$" +) + + +class _FakeClock: + """A monotonic clock that advances only when a test advances it.""" + + def __init__(self) -> None: + self.t = 0.0 + + def __call__(self) -> float: + return self.t + + def advance(self, seconds: float) -> None: + self.t += seconds + + +def _install_fake_clock(monkeypatch: pytest.MonkeyPatch) -> _FakeClock: + """Patch the one clock seam every asserted duration reads.""" + clock = _FakeClock() + monkeypatch.setattr("indexer.timing._CLOCK", clock) + return clock + + +def _timing_messages(caplog: pytest.LogCaptureFixture) -> list[str]: + return [m for r in caplog.records if (m := r.getMessage()).startswith("phase timing ")] + + +def _phases(message: str) -> dict[str, float]: + """Parse one `phase timing` line into ``{field: seconds}``, pinning its shape first.""" + assert _TIMING_RE.match(message), f"malformed timing line: {message!r}" + body = message.split(": ", 1)[1] + return {k: float(v.removesuffix("s")) for k, v in (f.split("=") for f in body.split(" "))} + + +def _only_phases(caplog: pytest.LogCaptureFixture) -> dict[str, float]: + """The single expected timing line's fields.""" + messages = _timing_messages(caplog) + assert len(messages) == 1, f"expected exactly one timing line, got {messages}" + return _phases(messages[0]) + + +@pytest.mark.unit +def test_indexed_branch_logs_every_phase_field(caplog: pytest.LogCaptureFixture) -> None: + """T1: one record per indexed branch, with all nine fields in a fixed order.""" + idx = _RecordingIndex() + with caplog.at_level(logging.INFO, logger="indexer.job"): + code = _run(_config(repos=["acme/widgets"]), idx) + assert code == 0 + + messages = _timing_messages(caplog) + assert len(messages) == 1 + match = _TIMING_RE.match(messages[0]) + assert match is not None, messages[0] + assert match.group("repo") == "acme/widgets" + assert match.group("branch") == "main" + assert list(_phases(messages[0])) == [ + "total", + "resolve", + "download", + "extract", + "parse", + "embed", + "db", + "sweep", + "other", + ] + + +@pytest.mark.unit +def test_semantic_off_still_logs_embed_zero(caplog: pytest.LogCaptureFixture) -> None: + """T2 (AC2): the format string has no branches -- semantic off zeroes embed, it does + not drop the field. A conditional line would break every grep an operator writes.""" + cfg_off = Settings(semantic_enabled=False) + with caplog.at_level(logging.INFO, logger="indexer.job"): + assert _run(_config(repos=["acme/widgets"]), _RecordingIndex(), cfg=cfg_off) == 0 + off = _only_phases(caplog) + assert off["embed"] == 0.0 + + caplog.clear() + cfg = Settings(semantic_enabled=True, semantic_max_chunks_per_repo=100) + with caplog.at_level(logging.INFO, logger="indexer.job"): + code = _run( + _config(repos=["acme/widgets"]), + _RecordingIndex(), + cfg=cfg, + embed_fn=lambda texts: [[0.0] for _ in texts], + ) + assert code == 0 + on = _only_phases(caplog) + + # Byte-identical field set and order, semantic on or off. + assert list(on) == list(off) + + +@pytest.mark.unit +def test_semantic_on_attributes_embed_time( + caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch +) -> None: + """T3: chunking + embedding is its own phase, and it is NOT double-counted into db.""" + clock = _install_fake_clock(monkeypatch) + + def _embed(texts: list[str]) -> list[list[float]]: + clock.advance(4.0) + return [[0.0] for _ in texts] + + cfg = Settings(semantic_enabled=True, semantic_max_chunks_per_repo=100) + with caplog.at_level(logging.INFO, logger="indexer.job"): + code = _run(_config(repos=["acme/widgets"]), _RecordingIndex(), cfg=cfg, embed_fn=_embed) + assert code == 0 + + phases = _only_phases(caplog) + assert phases["embed"] == 4.0 + assert phases["db"] == 0.0 # the embed happens BEFORE engine.connect(), not inside it + assert phases["total"] == 4.0 + assert phases["other"] == 0.0 + + +@pytest.mark.unit +def test_phase_timing_line_carries_the_repo_context(caplog: pytest.LogCaptureFixture) -> None: + """T4 (AC1): emitted from inside the worker, so `[%(repo)s]` resolves. + + The drain loop's `indexed ...` line runs on the main thread, where the context + var is at its default and the record would render `[-]` -- which is exactly + why the timing line is NOT appended there. + """ + import indexer.job as job + + log_filter = job.RepoLogFilter() + caplog.handler.addFilter(log_filter) + try: + with caplog.at_level(logging.INFO, logger="indexer.job"): + code = _run(_config(repos=["acme/widgets"]), _RecordingIndex()) + finally: + caplog.handler.removeFilter(log_filter) + assert code == 0 + + timing_records = [r for r in caplog.records if r.getMessage().startswith("phase timing ")] + assert len(timing_records) == 1 + assert timing_records[0].repo == "acme/widgets" + + +@pytest.mark.unit +def test_parse_time_is_excluded_from_db_time( + caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch +) -> 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 + + clock = _install_fake_clock(monkeypatch) + real_extract_file = job.extract_file + + def _slow_extract(pf: ParsedFile) -> Any: + clock.advance(2.0) + return real_extract_file(pf) + + monkeypatch.setattr(job, "extract_file", _slow_extract) + + def _index(conn: Any, *, items: Any, **_: Any) -> IndexCounts: + consumed = list(items) + clock.advance(9.0) + return IndexCounts(files=len(consumed), symbols=0, swept=0, edges=0) + + with caplog.at_level(logging.INFO, logger="indexer.job"): + code = _run(_config(repos=["acme/widgets"]), _index) + assert code == 0 + + phases = _only_phases(caplog) + assert phases["parse"] == 4.0 # main.py + README.md, 2.0s each + assert phases["db"] == 9.0 # NOT 13.0 + assert phases["total"] == 13.0 + assert phases["other"] == 0.0 + + +@pytest.mark.unit +def test_skipped_branch_logs_no_timing_line(caplog: pytest.LogCaptureFixture) -> None: + """T6: AC1 says "every indexed (not skipped) branch". A skipped branch has no + phases to report and must not pollute the dominant-phase grep with zeroes.""" + engine = _FakeEngine( + stamps={("acme/widgets", "main"): ("sha_widgets", INDEX_SEMANTICS_VERSION)} + ) + idx = _RecordingIndex() + with caplog.at_level(logging.INFO, logger="indexer.job"): + code = _run(_config(repos=["acme/widgets"]), idx, engine=engine) + assert code == 0 + assert idx.calls == [] + assert "skipped acme/widgets@main" in caplog.text + assert _timing_messages(caplog) == [] + + +@pytest.mark.unit +def test_failed_branch_logs_no_timing_line(caplog: pytest.LogCaptureFixture) -> None: + """T7: partial timings for a branch that never finished would be misleading.""" + + def _boom(conn: Any, *, items: Any, **_: Any) -> IndexCounts: + raise RuntimeError("boom") + + with caplog.at_level(logging.INFO, logger="indexer.job"): + code = _run(_config(repos=["acme/widgets"]), _boom) + assert code == 1 + assert "failed to index acme/widgets@main" in caplog.text + assert _timing_messages(caplog) == [] + + +@pytest.mark.unit +def test_conflicted_branch_logs_no_timing_line(caplog: pytest.LogCaptureFixture) -> None: + """T20: the StaleIndexError path is classified separately from T7's generic + failure, and it too rolled the branch back -- nothing was indexed, so nothing + is reported.""" + + def _stale(conn: Any, *, name: str, branch: str, items: Any, **_: Any) -> IndexCounts: + raise StaleIndexError(f"repo_branches row for {name}@{branch} changed") + + with caplog.at_level(logging.INFO, logger="indexer.job"): + code = _run(_config(repos=["acme/widgets"]), _stale) + assert code == 0 # a conflict self-heals and does not fail the run + assert "index conflict for acme/widgets@main" in caplog.text + assert _timing_messages(caplog) == [] + + +@pytest.mark.unit +def test_repo_line_reports_resolve_and_branch_listing( + caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch +) -> None: + """T8: the two REPO-scoped costs belong to no branch's total. + + Without `resolve=` the default branch's HEAD resolve would show as + `resolve=0.00s` on its phase line and look like a bug; without `list=` a + glob-configured monorepo's paginated branch listing would appear in no field + on no line at all. + + Deliberately asserts NO relation between the `in ...s` elapsed and the + parenthesised values: the former keeps its own real `time.monotonic()` + reading, so under a fake clock it reads ~0.00s while the others do not. + """ + import indexer.job as job + + clock = _install_fake_clock(monkeypatch) + real_resolve_ref = job.resolve_ref + real_list_branches = job.list_branches + + def _resolve_ref(*args: Any, **kwargs: Any) -> tuple[str, str]: + clock.advance(5.0) + return real_resolve_ref(*args, **kwargs) + + def _list_branches(*args: Any, **kwargs: Any) -> list[str]: + clock.advance(3.0) + return real_list_branches(*args, **kwargs) + + monkeypatch.setattr(job, "resolve_ref", _resolve_ref) + monkeypatch.setattr(job, "list_branches", _list_branches) + + # No `branches:` globs -> the branches endpoint is never called at all. + with caplog.at_level(logging.INFO, logger="indexer.job"): + assert _run(_config(repos=["acme/widgets"]), _RecordingIndex()) == 0 + finished = re.search( + r"finished acme/widgets in ([0-9.]+)s \(resolve=([0-9.]+)s list=([0-9.]+)s\)", caplog.text + ) + assert finished is not None, caplog.text + assert float(finished.group(2)) == 5.0 + assert float(finished.group(3)) == 0.0 + # The pre-existing unanchored assertion on this line still matches. + assert re.search(r"finished acme/widgets in ([0-9.]+)s", caplog.text) is not None + + caplog.clear() + github = _GitHub(branches={"acme/widgets": ["main"]}) + with caplog.at_level(logging.INFO, logger="indexer.job"): + code = _run( + _config(repos=["acme/widgets"], branches=["*"], index_concurrency=1), + _RecordingIndex(), + github=github, + ) + assert code == 0 + globbed = re.search( + r"finished acme/widgets in ([0-9.]+)s \(resolve=([0-9.]+)s list=([0-9.]+)s\)", caplog.text + ) + assert globbed is not None, caplog.text + assert float(globbed.group(2)) == 5.0 + assert float(globbed.group(3)) == 3.0 + # ...and none of it leaked into the branch's own total. + assert _only_phases(caplog)["total"] == 0.0 + + +@pytest.mark.unit +def test_timings_do_not_leak_between_branches_of_one_repo( + caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch +) -> None: + """T9: branch 2's numbers are its own, not cumulative. + + The trap this guards is reusing `_index_one_branch`'s `started` parameter for + `total=`: that value is set once per REPO and passed unchanged to every + branch, so branch 2 would report branch 1's wall clock too and silently + inflate `other`. Also guards the ContextVar reset between branches. + """ + import indexer.job as job + + clock = _install_fake_clock(monkeypatch) + real_download = job.download_tarball + + def _download(*args: Any, **kwargs: Any) -> Any: + clock.advance(5.0) + return real_download(*args, **kwargs) + + monkeypatch.setattr(job, "download_tarball", _download) + + github = _GitHub(branches={"acme/widgets": ["main", "feature"]}) + with caplog.at_level(logging.INFO, logger="indexer.job"): + code = _run( + _config(repos=["acme/widgets"], branches=["feature"], index_concurrency=1), + _RecordingIndex(), + github=github, + ) + assert code == 0 + + messages = _timing_messages(caplog) + assert len(messages) == 2 + branches = {_TIMING_RE.match(m).group("branch") for m in messages} # type: ignore[union-attr] + assert branches == {"main", "feature"} + for message in messages: + phases = _phases(message) + assert phases["download"] == 5.0 + assert phases["total"] == 5.0 # not 10.0 on the second branch + assert phases["other"] == 0.0 + + +@pytest.mark.unit +def test_timings_do_not_leak_between_repos_on_a_reused_worker_thread( + caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch +) -> None: + """T10: ThreadPoolExecutor reuses worker threads without resetting their + context. With one worker and two repos, repo 2 runs on the same thread that + just finished repo 1 -- a leaked timer would show up here as doubled numbers.""" + import indexer.job as job + + clock = _install_fake_clock(monkeypatch) + real_download = job.download_tarball + + def _download(*args: Any, **kwargs: Any) -> Any: + clock.advance(3.0) + return real_download(*args, **kwargs) + + monkeypatch.setattr(job, "download_tarball", _download) + + with caplog.at_level(logging.INFO, logger="indexer.job"): + code = _run( + _config(repos=["acme/widgets", "acme/gadgets"], index_concurrency=1), + _RecordingIndex(), + ) + assert code == 0 + + messages = _timing_messages(caplog) + assert len(messages) == 2 + assert {_TIMING_RE.match(m).group("repo") for m in messages} == { # type: ignore[union-attr] + "acme/widgets", + "acme/gadgets", + } + for message in messages: + phases = _phases(message) + assert phases["download"] == 3.0 + assert phases["total"] == 3.0 + assert phases["other"] == 0.0 + + +@pytest.mark.unit +def test_timer_reset_fires_on_every_branch_exit_path(monkeypatch: pytest.MonkeyPatch) -> None: + """Directly guards the `finally: reset_timer(token)` invariant itself. + + T9/T10 assert branch totals aren't cumulative, but `_index_one_branch` + unconditionally does `install_timer(PhaseTimer())` at entry -- which + shadows a prior leaked timer regardless of whether `reset_timer` actually + ran. So T9/T10 pass identically whether or not the `finally` fires, and + cannot detect its removal. This test spies on `indexer.timing.reset_timer` + directly and asserts it is called exactly once per branch attempt, + including the failed and conflicted paths -- the actual invariant the + module docstring calls "mandatory". + """ + import indexer.job as job + import indexer.timing as timing + + calls = 0 + real_reset = timing.reset_timer + + def _spy_reset(token: Any) -> None: + nonlocal calls + calls += 1 + real_reset(token) + + monkeypatch.setattr(job, "reset_timer", _spy_reset) + + def _ok(conn: Any, *, items: Any, **_: Any) -> IndexCounts: + return IndexCounts(files=len(list(items)), symbols=0, swept=0, edges=0) + + def _boom(conn: Any, *, items: Any, **_: Any) -> IndexCounts: + raise RuntimeError("boom") + + def _stale(conn: Any, *, name: str, branch: str, items: Any, **_: Any) -> IndexCounts: + raise StaleIndexError(f"repo_branches row for {name}@{branch} changed") + + _run(_config(repos=["acme/widgets"]), _ok) + assert calls == 1 + + _run(_config(repos=["acme/widgets"]), _boom) + assert calls == 2 + + _run(_config(repos=["acme/widgets"]), _stale) + assert calls == 3 + + +@pytest.mark.unit +def test_phase_timing_failure_does_not_fail_an_already_committed_branch( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + """A bug in the phase-timing arithmetic/log call must degrade to a missing + log line, never to reclassifying an already-committed branch as `failed`. + + By the point this block runs, `index_fn` has already returned `counts` -- + this branch's transaction is committed. Flipping it to `failed` here would + report a real committed index as a failure, flip the run's exit code, and + gate off `_decide_reconciliation` (which requires zero failures), all for a + bug in pure measurement code. Forces the failure at the `logger.info` call + itself (the last statement in the guarded block) so a passing test proves + the whole block is covered, not just the arithmetic above it. + """ + import indexer.job as job + + real_info = job.logger.info + + def _boom_on_phase_timing(msg: object, *args: Any, **kwargs: Any) -> None: + if isinstance(msg, str) and msg.startswith("phase timing "): + raise RuntimeError("boom") + real_info(msg, *args, **kwargs) + + monkeypatch.setattr(job.logger, "info", _boom_on_phase_timing) + + idx = _RecordingIndex() + with caplog.at_level(logging.WARNING, logger="indexer.job"): + code = _run(_config(repos=["acme/widgets"]), idx) + + assert code == 0 + assert idx.calls == ["acme/widgets"] + # No well-formed `phase timing ...: total=...` line was emitted (the raise + # happened inside the logger.info call itself) -- but the module's own + # "phase timing unavailable" fallback below shares the "phase timing " + # prefix, so check the anchored data-line shape rather than the prefix. + assert not any(_TIMING_RE.match(r.getMessage()) for r in caplog.records) + assert "failed to index" not in caplog.text + assert "phase timing unavailable for acme/widgets@main" in caplog.text + + +@pytest.mark.unit +def test_index_counts_fields_are_unchanged() -> None: + """T11 (AC3): a tripwire that timing never leaks into the frozen return type. + + `IndexCounts` is compared by value in existing assertions across the unit and + integration suites; a timing field would be nonzero by construction and break + every one of them. The sweep duration reaches job.py ambiently for exactly + this reason. + """ + assert [f.name for f in dataclasses.fields(IndexCounts)] == [ + "files", + "symbols", + "swept", + "edges", + ] + + +@pytest.mark.unit +def test_timing_line_contains_no_values_beyond_durations( + caplog: pytest.LogCaptureFixture, +) -> None: + """T12 (AC3 / redaction): the anchored regex matches the WHOLE message, so + nothing but the repo, the branch, and the nine durations can ever appear -- + no config values, no counts, no headers -- and no field can go negative.""" + cfg = Settings(semantic_enabled=True, semantic_max_chunks_per_repo=100) + config = _config(repos=["acme/widgets"], semantic_max_chunks_per_repo={"acme/widgets": 7}) + with caplog.at_level(logging.INFO, logger="indexer.job"): + code = _run( + config, + _RecordingIndex(), + cfg=cfg, + embed_fn=lambda texts: [[0.0] for _ in texts], + ) + assert code == 0 + + messages = _timing_messages(caplog) + assert len(messages) == 1 + assert _TIMING_RE.fullmatch(messages[0]) is not None, messages[0] + + +@pytest.mark.unit +def test_sweep_time_is_subtracted_from_db_time( + caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch +) -> None: + """T16: the cross-module sweep hook, pinned without Postgres. + + A fake `index_fn` runs on the worker thread inside the installed timer's + context, so its `record("sweep", ...)` reaches exactly the timer the real + `store.py` hook would. Without this test the subtraction term is zero in + every unit test and the arithmetic is unproven under `make test`. + """ + import indexer.timing as timing + + clock = _install_fake_clock(monkeypatch) + + def _index(conn: Any, *, items: Any, **_: Any) -> IndexCounts: + consumed = list(items) + clock.advance(2.0) # the sweep really burned this time... + timing.record("sweep", 2.0) # ...and store.py reports it ambiently + clock.advance(6.0) # the rest of the transaction + return IndexCounts(files=len(consumed), symbols=0, swept=1, edges=0) + + with caplog.at_level(logging.INFO, logger="indexer.job"): + code = _run(_config(repos=["acme/widgets"]), _index) + assert code == 0 + + phases = _only_phases(caplog) + assert phases["sweep"] == 2.0 + assert phases["db"] == 6.0 # NOT 8.0 + assert phases["total"] == 8.0 + assert phases["other"] == 0.0 + + +@pytest.mark.unit +def test_semantic_eager_walk_is_charged_to_parse( + caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch +) -> None: + """T18: on the semantic (production) path the file list is materialized up front, + OUTSIDE the db window and outside _timed_items. Leaving that walk unwrapped would + dump the whole rglob + stat + read + decode cost into `other` -- the exact number + that routes work between the epic's parse/IO/db issues.""" + import indexer.job as job + + clock = _install_fake_clock(monkeypatch) + real_iter_source_files = job.iter_source_files + + def _slow_walk(root: Any) -> Any: + # iter_source_files is a generator function: paying the cost at call + # time (before the first `next()`) would let this fake pass even if + # the real wrap only timed the call and not the iteration -- the + # exact mis-scoped-wrap bug this test exists to catch. Pay per + # yielded file instead, matching where the real cost (rglob walk, + # stat, read, decode) is actually incurred. + for pf in real_iter_source_files(root): + clock.advance(3.0) # 2 files in _DEFAULT_FILES -> 6.0s total + yield pf + + monkeypatch.setattr(job, "iter_source_files", _slow_walk) + + cfg = Settings(semantic_enabled=True, semantic_max_chunks_per_repo=100) + with caplog.at_level(logging.INFO, logger="indexer.job"): + code = _run( + _config(repos=["acme/widgets"]), + _RecordingIndex(), + cfg=cfg, + embed_fn=lambda texts: [[0.0] for _ in texts], + ) + assert code == 0 + + phases = _only_phases(caplog) + assert phases["parse"] == 6.0 + assert phases["other"] == 0.0 # not swallowed by the residual + assert phases["db"] == 0.0 # and not misattributed to the transaction + assert phases["total"] == 6.0 + + +@pytest.mark.unit +def test_embed_degrade_path_still_logs_a_timing_line( + caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch +) -> None: + """T19: a downed embedder degrades to a core index -- and it burned real time + getting there, so `embed=` must still report it. That is why the embed wrap is + the one phase recorded in a `finally`.""" + clock = _install_fake_clock(monkeypatch) + + def _down(_texts: list[str]) -> list[list[float]]: + clock.advance(7.0) + raise RuntimeError("serving endpoint unavailable") + + idx = _RecordingIndex() + cfg = Settings(semantic_enabled=True, semantic_max_chunks_per_repo=100) + with caplog.at_level(logging.INFO, logger="indexer.job"): + code = _run(_config(repos=["acme/widgets"]), idx, cfg=cfg, embed_fn=_down) + assert code == 0 + assert idx.calls == ["acme/widgets"] # the core index still ran + assert idx.chunk_writer is None # ...without chunks + + phases = _only_phases(caplog) + assert phases["embed"] == 7.0 + assert phases["total"] == 7.0 + assert phases["other"] == 0.0 + + +@pytest.mark.unit +def test_other_is_the_exact_unattributed_residual( + caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch +) -> None: + """T17: pins the RESIDUAL, not the sum identity. + + `other` is *defined* as `total - sum(phases)`, so asserting + `total == sum(phases) + other` is a tautology that holds for any values and can + never detect a mis-measured phase. This test instead drives a distinct known + amount through every one of the seven phases and requires `other == 0.00s`, + then re-runs with one deliberately UNINSTRUMENTED advance and requires `other` + to equal exactly that amount. + """ + import indexer.job as job + + clock = _install_fake_clock(monkeypatch) + real_resolve_branch_head = job.resolve_branch_head + real_download_tarball = job.download_tarball + real_extract_tarball = job.extract_tarball + real_iter_source_files = job.iter_source_files + real_assert_disk_headroom = job.assert_disk_headroom + + def _resolve_branch_head(*args: Any, **kwargs: Any) -> str: + clock.advance(1.0) + return real_resolve_branch_head(*args, **kwargs) + + def _download_tarball(*args: Any, **kwargs: Any) -> Any: + clock.advance(2.0) + return real_download_tarball(*args, **kwargs) + + def _extract_tarball(*args: Any, **kwargs: Any) -> Any: + clock.advance(4.0) + return real_extract_tarball(*args, **kwargs) + + def _iter_source_files(root: Any) -> Any: + # Pay per yielded file, not at call time -- iter_source_files is a + # generator function, so a call-time advance would pass even for a + # wrap that only times the call and not the iteration. + for pf in real_iter_source_files(root): + clock.advance(4.0) # 2 files in _DEFAULT_FILES -> 8.0s total + yield pf + + def _embed(texts: list[str]) -> list[list[float]]: + clock.advance(16.0) + return [[0.0] for _ in texts] + + def _index(conn: Any, *, items: Any, **_: Any) -> IndexCounts: + import indexer.timing as timing + + consumed = list(items) + clock.advance(32.0) # the transaction's own DML + clock.advance(64.0) # the sweep, which store.py reports ambiently + timing.record("sweep", 64.0) + return IndexCounts(files=len(consumed), symbols=0, swept=0, edges=0) + + monkeypatch.setattr(job, "resolve_branch_head", _resolve_branch_head) + monkeypatch.setattr(job, "download_tarball", _download_tarball) + monkeypatch.setattr(job, "extract_tarball", _extract_tarball) + monkeypatch.setattr(job, "iter_source_files", _iter_source_files) + + cfg = Settings(semantic_enabled=True, semantic_max_chunks_per_repo=100) + # "main" is stamped current so it is SKIPPED, leaving exactly one indexed + # branch -- and a non-default one, which is the only way `resolve=` is + # non-zero (the default branch's HEAD came from the repo-level resolve_ref). + engine = _FakeEngine( + stamps={("acme/widgets", "main"): ("sha_widgets", INDEX_SEMANTICS_VERSION)} + ) + github = _GitHub(branches={"acme/widgets": ["main", "feature"]}) + config = _config(repos=["acme/widgets"], branches=["feature"], index_concurrency=1) + + with caplog.at_level(logging.INFO, logger="indexer.job"): + code = _run(config, _index, cfg=cfg, embed_fn=_embed, github=github, engine=engine) + assert code == 0 + + phases = _only_phases(caplog) + assert phases == { + "total": 127.0, + "resolve": 1.0, + "download": 2.0, + "extract": 4.0, + "parse": 8.0, + "embed": 16.0, + "db": 32.0, + "sweep": 64.0, + "other": 0.0, + } + + # Second leg: advance the clock at a point NOTHING instruments (the pre-flight + # disk check). Every phase must be unchanged and `other` must be exactly it. + def _assert_disk_headroom(*args: Any, **kwargs: Any) -> None: + clock.advance(9.0) + real_assert_disk_headroom(*args, **kwargs) + + monkeypatch.setattr(job, "assert_disk_headroom", _assert_disk_headroom) + caplog.clear() + with caplog.at_level(logging.INFO, logger="indexer.job"): + code = _run(config, _index, cfg=cfg, embed_fn=_embed, github=github, engine=engine) + assert code == 0 + + phases = _only_phases(caplog) + assert phases == { + "total": 136.0, + "resolve": 1.0, + "download": 2.0, + "extract": 4.0, + "parse": 8.0, + "embed": 16.0, + "db": 32.0, + "sweep": 64.0, + "other": 9.0, + } diff --git a/tests/unit/test_timing.py b/tests/unit/test_timing.py new file mode 100644 index 0000000..7c02a09 --- /dev/null +++ b/tests/unit/test_timing.py @@ -0,0 +1,86 @@ +"""Unit tests for indexer.timing: the ambient per-phase accumulator. + +Two properties the whole instrumentation rests on are pinned here, away from the +job pipeline that consumes them: ``record`` is a no-op when nobody is measuring +(so ``index_repo`` stays callable directly), and one thread's totals are +invisible to another (so fan-out cannot cross-attribute a phase). +""" + +from __future__ import annotations + +import threading + +import pytest + +from indexer.timing import PhaseTimer, current_timer, install_timer, record, reset_timer + + +@pytest.mark.unit +def test_timer_accumulates_and_is_a_noop_when_unset() -> None: + """Default-unset must be silent, not merely harmless. + + ``tests/integration/test_store.py`` calls ``index_repo`` directly with no + timer in context; the sweep hook in ``store.py`` runs there regardless, so + ``record`` with nothing installed has to do nothing AND not raise. + """ + assert current_timer() is None + record("sweep", 1.0) # must not raise + assert current_timer() is None + + timer = PhaseTimer() + assert timer.total("sweep") == 0.0 # an unrecorded phase reads as zero, not KeyError + + token = install_timer(timer) + try: + assert current_timer() is timer + record("sweep", 1.5) + record("sweep", 0.25) # accrues, never overwrites + record("parse", 2.0) + finally: + reset_timer(token) + + assert timer.total("sweep") == 1.75 + assert timer.total("parse") == 2.0 + assert timer.total("download") == 0.0 + + # The reset really uninstalled it: a later record goes nowhere. + assert current_timer() is None + record("sweep", 100.0) + assert timer.total("sweep") == 1.75 + + +@pytest.mark.unit +def test_timer_is_isolated_per_thread() -> None: + """Each thread starts with a fresh context, so timers cannot cross-attribute. + + This is what lets one ``PhaseTimer`` per branch be correct while four worker + threads index four repos concurrently. + """ + totals: dict[str, float] = {} + start = threading.Barrier(2) + + def worker(label: str, amount: float) -> None: + # A new thread inherits no context: nothing is installed here yet. + assert current_timer() is None + timer = PhaseTimer() + token = install_timer(timer) + try: + start.wait(timeout=5) # force real overlap, not sequential execution + record("db", amount) + start.wait(timeout=5) + totals[label] = timer.total("db") + finally: + reset_timer(token) + + threads = [ + threading.Thread(target=worker, args=("a", 3.0)), + threading.Thread(target=worker, args=("b", 7.0)), + ] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=10) + assert not thread.is_alive() + + assert totals == {"a": 3.0, "b": 7.0} + assert current_timer() is None # and nothing leaked back to the main thread