Indexing performance integration arc (#110)

app/AGENTS.md

@@ -12,7 +12,7 @@ The MCP server Databricks App: a FastMCP streamable-HTTP service exposing the in
 | `main.py` | FastMCP app: tool registration via `create_app()` factory, per-session `lifespan`, process-scoped engine singleton (`get_engine()`, `threading.Lock` + `atexit` dispose), `anyio` off-loop dispatch under a pool-sized `CapacityLimiter(5)`, `/health` (zero-DB) and `/ready` (`SELECT 1 FROM repos LIMIT 1` grant probe) |
 | `service.py` | Shared payload builders (`search_code_payload`, `list_repos_payload`, `get_file_payload`, `clamp_limit`): merges grep + symbol legs into the zoekt-parity envelope, base64url pagination cursors (`encode_cursor`/`decode_cursor`, `CursorError` never swallowed), `commit:` prefix resolution against `repo_branches`, permalink-branch selection |
 | `config.py` | `pydantic-settings` `Settings` with `CODE_SEARCH_` env prefix (timeouts, row limits, semantic tunables); unprefixed `LAKEBASE_ENDPOINT` via `validation_alias`; `SEMANTIC_EMBEDDING_DIM = 1024` single source of truth; `get_settings()` is `lru_cache`d once per process |
-| `embed.py` | `EmbedFn` seam (texts → unit-normalized 1024-dim vectors) and `databricks_embedder`: POSTs to the AI Gateway MLflow embeddings route via the SDK's raw API client; lazy SDK import; per-batch count check (`EmbeddingCountMismatchError`) and dim check (`EmbeddingDimMismatchError`) fail loudly instead of misaligning vectors |
+| `embed.py` | `EmbedFn` seam (texts → unit-normalized 1024-dim vectors) and `databricks_embedder`: POSTs to the AI Gateway MLflow embeddings route via the SDK's raw API client; lazy SDK import; per-batch count check (`EmbeddingCountMismatchError`) and dim check (`EmbeddingDimMismatchError`) fail loudly instead of misaligning vectors; batches dispatch through an order-preserving `ThreadPoolExecutor.map` (never `as_completed`) at `concurrency` (#107, indexer-only — the query path is one batch and stays serial) |
 | `app.yaml` | Databricks App runtime config: `ln -sf . app` symlink so `app.` imports resolve at the uploaded working-dir root; shell-form command so `DATABRICKS_APP_PORT` expands; sets only `LAKEBASE_ENDPOINT` |
 | `requirements.txt` | Deploy-time lockfile exported by `uv export --no-dev --no-hashes --no-emit-project` — regenerate, never hand-edit |
 | `__init__.py` | Empty package marker |

app/config.py

@@ -93,18 +93,38 @@ class Settings(BaseSettings):
     # in-process memory rather than a DB lock; exceeding it fails loudly.
     #
     # Sized against the ACTUAL buffer cost, not a round number: the vectors are held as
-    # Python float lists, ~32 B per element (24 B float object + 8 B list pointer), so at
-    # dim=1024 each chunk costs ~32 KB -- 8000 chunks is ~260 MB of vectors plus ~16 MB of
-    # chunk text. A larger ceiling (e.g. 50k -> ~1.6 GB) would OOM the job container before
-    # this loud check could ever fire, which would defeat the point of having a ceiling.
-    # A repo that legitimately exceeds this needs a temp-table staging path, not a bigger
-    # buffer.
+    # Python float lists, ~32 B per element (24 B float object + 8 B list pointer) structural,
+    # but ~40.1 KB/chunk RESIDENT once measured (issue #109; pymalloc overhead/fragmentation --
+    # use this figure for headroom arithmetic) -- 8000 chunks is ~313 MiB of vectors resident
+    # plus ~16 MB of chunk text. #109 also derived a per-worker chunk-cap ceiling from a full
+    # container-memory model (~73,300 chunks at the pinned N=2 semantic-worker count this cap
+    # is evaluated at, ~36,700 at the shipped N=4 -- see docs/perf/issue-109-measurements.md
+    # §12): a larger ceiling well past that (e.g. 50k, ~1.9 GiB resident) risks OOMing the job
+    # container before this loud check could ever fire, which would defeat the point of having
+    # a ceiling. A repo that legitimately exceeds this needs a temp-table staging path, not a
+    # bigger buffer.
+    #
+    # Scope note (#104): under file-level delta indexing this cap is enforced against
+    # whatever ONE RUN embeds (changed/new + membership-only files), not a branch's whole
+    # corpus -- a branch can legitimately drift above this number between full reindexes
+    # (a semantics bump, or its first index), re-enforced in full at each of those. This is
+    # a deliberate, accepted trade-off (see indexer/job.py's module docstring and
+    # docs/runbooks/indexing-parallelism.md §4.1), not a bug; the constant is unchanged.
     semantic_max_chunks_per_repo: int = 8000
 
     # Chunk size bound (tokens) fed to the embedding model. Distinct from MAX_FILE_BYTES,
     # which bounds file ingestion, not embedding-chunk granularity.
     semantic_chunk_max_tokens: int = 512
 
+    # In-flight embedding requests per worker (#107). The indexer clamps to 4 workers when
+    # semantic is on (indexer/repo_config.py:effective_workers -- issue #109 raised this from
+    # 2), so total in-flight gateway requests are workers x concurrency: 4 x 4 = 16 at this
+    # default, 4 x 8 = 32 at the config.yaml-enforced ceiling of 8 -- the latter now EXCEEDS
+    # the SDK's 20-connection pool (pool_block=True, so exceeding it silently serializes
+    # rather than erroring, so this is a real-concurrency cap, not a correctness one). Setting
+    # this to 1 restores today's fully serial embed() and spawns no thread pool.
+    semantic_embedding_concurrency: int = 4
+
 
 @lru_cache(maxsize=1)
 def get_settings() -> Settings:

app/db/models.py

@@ -45,7 +45,22 @@
 ``indexer/symbols.py``, to ``indexer/parse.py``'s chunking, or to
 ``indexer/languages.py``'s extraction contract. A bump forces every repo to
 re-index once, because a repo's stored ``repos.index_semantics_version`` no
-longer matches. The CI tripwire enforces the bump obligation.
+longer matches. The CI tripwire enforces the bump obligation for those three
+files.
+
+**The obligation extends past the tripwire's reach (#104).** Swapping the
+embedding MODEL (``app/embed.py``) or changing ``SEMANTIC_EMBEDDING_DIM``
+(``app/config.py``) also requires a bump, but the tripwire does not watch
+either file (``app/embed.py`` deliberately -- it would otherwise fire on
+unrelated retry/batching edits) -- this is a reviewed convention, not a
+machine-enforced one. Before file-level delta indexing (issue #104) a missed
+bump here was self-limiting: the next HEAD move re-embedded a branch's whole
+corpus regardless. Under delta indexing only a CHANGED file re-embeds, so a
+missed bump now leaves every unchanged file's vectors silently stale forever
+-- exactly the failure this version column exists to prevent. This is not a
+new kind of case: version ``2`` was minted for precisely this reason (turning
+semantic search on by default, so every already-indexed branch had to
+re-index once for ``chunks`` to backfill).
 
 Migrations must never import this constant -- see
 ``app/alembic/versions/0002_index_semantics_version.py``.

app/embed.py

@@ -18,6 +18,7 @@
 from __future__ import annotations
 
 from collections.abc import Callable, Sequence
+from concurrent.futures import ThreadPoolExecutor
 from typing import Any
 
 from app.config import SEMANTIC_EMBEDDING_DIM, Settings
@@ -50,13 +51,27 @@ def _assert_dims(vectors: Sequence[list[float]], dim: int) -> None:
 
 
 def _query_batch(
-    client: Any, endpoint: str, model: str, batch: list[str], *, max_retries: int
+    client: Any,
+    endpoint: str,
+    model: str,
+    batch: list[str],
+    *,
+    max_retries: int,
+    ordinal: int = 0,
+    offset: int = 0,
 ) -> list[list[float]]:
     """Query one batch, retrying up to ``max_retries`` times (small, bounded).
 
     ``endpoint`` is the gateway route path (e.g. ``/ai-gateway/mlflow/v1/embeddings``),
     POSTed via the SDK's raw API client; the response is OpenAI-shaped
     (``{"data": [{"embedding": [...]}, ...]}``).
+
+    ``ordinal``/``offset`` identify this batch's position in the caller's flat text
+    list (batch index and its starting position, respectively) purely so a count
+    mismatch can name the offending batch -- serially that batch was recoverable
+    from context (the one after the last success), but under concurrent dispatch
+    (:func:`databricks_embedder`) it is not. Both default to 0 so ``_query_batch``
+    stays directly callable in isolation.
     """
     last_exc: Exception | None = None
     for _attempt in range(max_retries + 1):
@@ -69,7 +84,8 @@ def _query_batch(
             # confusing IndexError (or silent corruption) much further downstream.
             if len(vectors) != len(batch):
                 raise EmbeddingCountMismatchError(
-                    f"embedder returned {len(vectors)} vectors for {len(batch)} texts"
+                    f"embedder returned {len(vectors)} vectors for {len(batch)} texts "
+                    f"(batch {ordinal}, texts[{offset}:{offset + len(batch)}])"
                 )
             return vectors
         except EmbeddingCountMismatchError:
@@ -89,6 +105,7 @@ def databricks_embedder(
     batch_size: int = 64,
     timeout: float = 20.0,
     max_retries: int = 2,
+    concurrency: int = 1,
 ) -> EmbedFn:
     """Build an :data:`EmbedFn` backed by the AI Gateway embeddings route ``endpoint``.
 
@@ -105,6 +122,19 @@ def databricks_embedder(
     client's concern. When omitted, the real ``WorkspaceClient`` is built with a
     ``Config`` carrying ``http_timeout_seconds=timeout`` (the raw API client has no
     per-call timeout).
+
+    ``concurrency`` dispatches up to that many batches at once via a
+    ``ThreadPoolExecutor``. It deliberately does NOT mirror this file's usual
+    default-mirroring convention (every other parameter here matches its
+    ``Settings`` twin): the concurrent path must be opt-in at the call site that
+    knows it is the indexer. ``get_embedder`` supplies the real value from
+    ``Settings.semantic_embedding_concurrency``; every direct caller and existing
+    unit test keeps today's serial semantics untouched. Do not "fix" this default
+    to match -- a higher default would make ``tests/unit/test_embed.py``'s
+    single-threaded fakes (e.g. ``_FakeApiClient.batches.append``) nondeterministic
+    under concurrent append. ``concurrency <= 1``, or a text list short enough to
+    produce only one batch, constructs no pool and spawns no thread -- the query
+    path (one text, one batch) is a strict no-op.
     """
     if client is None:
         from databricks.sdk import WorkspaceClient  # lazy: see module docstring
@@ -113,10 +143,33 @@ def databricks_embedder(
         client = WorkspaceClient(config=_SdkConfig(http_timeout_seconds=timeout))
 
     def embed(texts: list[str]) -> list[list[float]]:
-        vectors: list[list[float]] = []
-        for i in range(0, len(texts), batch_size):
-            batch = texts[i : i + batch_size]
-            vectors.extend(_query_batch(client, endpoint, model, batch, max_retries=max_retries))
+        batches = [texts[i : i + batch_size] for i in range(0, len(texts), batch_size)]
+        workers = max(1, min(concurrency, len(batches)))
+
+        def run_batch(indexed: tuple[int, list[str]]) -> list[list[float]]:
+            ordinal, batch = indexed
+            return _query_batch(
+                client,
+                endpoint,
+                model,
+                batch,
+                max_retries=max_retries,
+                ordinal=ordinal,
+                offset=ordinal * batch_size,
+            )
+
+        if workers == 1:
+            per_batch = [run_batch(item) for item in enumerate(batches)]
+        else:
+            # .map() yields in SUBMISSION order regardless of completion order --
+            # never as_completed(), which yields in completion order and would
+            # silently reorder vectors across files. See the module docstring's
+            # EmbeddingCountMismatchError note and tests/unit/test_embed.py's
+            # test_embed_module_never_uses_as_completed.
+            with ThreadPoolExecutor(max_workers=workers, thread_name_prefix="embed") as pool:
+                per_batch = list(pool.map(run_batch, enumerate(batches)))
+
+        vectors = [v for batch_vectors in per_batch for v in batch_vectors]
         _assert_dims(vectors, dim)
         return vectors
 
@@ -140,4 +193,5 @@ def get_embedder(cfg: Settings) -> EmbedFn:
         dim=cfg.semantic_embedding_dim,
         batch_size=cfg.semantic_embedding_batch_size,
         timeout=cfg.semantic_embedding_timeout_s,
+        concurrency=cfg.semantic_embedding_concurrency,
     )

config.yaml

@@ -4,16 +4,32 @@
 version: 1
 
 # How many repos the job indexes at once. Default 4, max 8, min 1.
-# This is a DISK bound, not a CPU one: each worker holds its 500 MB tarball and
-# its 2 GB extraction alive at the same time, so budget 2.5 GB per worker —
-# 10 GB at the default 4, 20 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
-# 20 GB of disk. Raise it knowing that.
-# 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).
+# This is a DISK bound, not a CPU one: each worker holds its 500 MB tarball on
+# 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.
+# 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 4 (issue #109 raised it from
+# 2, after re-deriving the memory model and confirming empirically against the
+# live dev job at N=4: peak self+children RSS landed at ~83% of the 0.7*container
+# memory budget, comfortably under and with more margin than N=2's own ~90%) — a
+# MEMORY bound, since embedding materialises a whole repo's chunks (structural
+# ~32 KB/chunk, resident ~40.1 KB/chunk measured; see effective_workers'
+# docstring and docs/perf/issue-109-measurements.md for the full derivation).
 # 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.
@@ -64,13 +80,23 @@ connections:
 # effective cap as `per-repo override OR global`.
 #
 # Mind the memory math before raising one: buffered vectors are ~32 KB/chunk
-# (dim=1024, Python float-list storage), so 8000 ≈ 260 MB resident for the
-# duration of that repo's write. With semantic on, at most 2 workers run
-# concurrently (indexer/repo_config.py's effective_workers clamp), so a large
-# override multiplies straight into the job container's peak memory — e.g. two
-# repos overridden to 20000 concurrently is ≈1 GB just in vectors, on top of
-# the base per-worker cost. A repo that legitimately needs far more than that
-# needs the temp-table staging path (follow-up), not a bigger override.
+# structural (dim=1024, Python float-list storage) but ~40.1 KB/chunk RESIDENT
+# (measured, issue #109 — includes pymalloc overhead/fragmentation; use this
+# figure for headroom arithmetic), so 8000 ≈ 313 MiB resident for the duration of
+# that repo's write. With semantic on, at most 4 workers run concurrently
+# (indexer/repo_config.py's effective_workers clamp, raised from 2 by #109), so
+# a large override multiplies straight into the job container's peak memory —
+# e.g. two repos overridden to 20000 concurrently is ≈1.6 GB just in vectors, on
+# top of the base per-worker cost. A repo that legitimately needs far more than
+# that needs the temp-table staging path (follow-up), not a bigger override.
+#
+# The derived per-worker chunk-cap ceiling, from a full container-memory model
+# (issue #109; docs/perf/issue-109-measurements.md §12 — pinned at N=2 there
+# only to break a circularity in solving for C from a formula whose dominant
+# term IS C, not a claim about the adopted concurrency): ≈73,300 chunks at
+# N=2, ≈36,700 at the shipped N=4 (both halve/double with N). The current
+# global default of 8000 uses well under a quarter of either budget, so it is
+# NOT the binding constraint and was left unchanged.
 # semantic_max_chunks_per_repo:
 #   "acme/huge-monorepo": 20000
 
@@ -87,16 +113,29 @@ connections:
 # (the default is 8000). It is NOT the per-repo `semantic_max_chunks_per_repo` MAP
 # above — that spot-overrides individual repos and still wins over this global. Use
 # this to raise the floor everyone inherits; use the map for the outliers. The same
-# ~32 KB/chunk memory math and 2-worker clamp above apply here, magnified: raising
-# the global lifts the buffer cost for EVERY concurrently-indexing repo at once.
+# ~32 KB structural / ~40.1 KB resident per-chunk memory math and 4-worker clamp
+# above apply here, magnified: raising the global lifts the buffer cost for EVERY
+# concurrently-indexing repo at once.
 #
 # `enabled: false` makes the job a true semantic no-op (no embedder built, no
-# chunking, the 2-worker memory clamp not applied) even if the env says enabled —
+# chunking, the 4-worker memory clamp not applied) even if the env says enabled —
 # the fastest way to turn semantic off for the job alone.
+#
+# `embedding_concurrency` (#107) is in-flight embedding requests PER WORKER, sent
+# via a ThreadPoolExecutor that preserves submission order — vectors always come
+# back in the order their texts were sent, regardless of which request finishes
+# first. Total in-flight gateway requests for the job is workers x concurrency:
+# 4 x 4 = 16 at this default, 4 x 8 = 32 at the max of 8 — the latter now EXCEEDS
+# the SDK's 20-connection pool (issue #109 raised workers from 2 to 4; this
+# combination was not possible before). Lower embedding_concurrency if raising it
+# alongside a near-ceiling index_concurrency. Set embedding_concurrency to 1 to
+# restore fully serial embedding (no thread pool spawned at all) if you need to
+# roll back.
 # semantic:
 #   enabled: true
 #   max_chunks_per_repo: 8000
 #   embedding_endpoint: /ai-gateway/mlflow/v1/embeddings
 #   embedding_model: system.ai.gte-large-en
 #   embedding_batch_size: 64
 #   embedding_timeout_s: 20.0
+#   embedding_concurrency: 4