diff --git a/docs/plans/layering-surface-baseline.json b/docs/plans/layering-surface-baseline.json index 5baf6d6aec..417657f8ab 100644 --- a/docs/plans/layering-surface-baseline.json +++ b/docs/plans/layering-surface-baseline.json @@ -1387,27 +1387,7 @@ { "target": "polylogue/daemon", "file": "polylogue/daemon/parse_prefetch.py", - "import": "polylogue.pipeline.parsed_tree_size" - }, - { - "target": "polylogue/daemon", - "file": "polylogue/daemon/parse_prefetch.py", - "import": "polylogue.sources" - }, - { - "target": "polylogue/daemon", - "file": "polylogue/daemon/parse_prefetch.py", - "import": "polylogue.sources.dispatch" - }, - { - "target": "polylogue/daemon", - "file": "polylogue/daemon/parse_prefetch.py", - "import": "polylogue.sources.revision_backfill" - }, - { - "target": "polylogue/daemon", - "file": "polylogue/daemon/parse_prefetch.py", - "import": "polylogue.storage.repair" + "import": "polylogue.sources.census_parse_stage" }, { "target": "polylogue/daemon", diff --git a/docs/plans/topology-target.yaml b/docs/plans/topology-target.yaml index 24113cf1f1..91b955b3a5 100644 --- a/docs/plans/topology-target.yaml +++ b/docs/plans/topology-target.yaml @@ -1700,7 +1700,7 @@ files: target: polylogue/daemon/otlp_receiver.py owner: stable - path: polylogue/daemon/parse_prefetch.py - loc: 417 + loc: 39 target: polylogue/daemon/parse_prefetch.py owner: stable - path: polylogue/daemon/process_start.py @@ -2190,7 +2190,7 @@ files: target: polylogue/maintenance/raw_authority_reset.py owner: stable - path: polylogue/maintenance/rebuild_index.py - loc: 790 + loc: 864 target: polylogue/maintenance/rebuild_index.py owner: stable - path: polylogue/maintenance/registry.py @@ -3194,6 +3194,10 @@ files: loc: 118 target: polylogue/sources/assembly_gemini.py owner: stable + - path: polylogue/sources/census_parse_stage.py + loc: 446 + target: polylogue/sources/census_parse_stage.py + owner: stable - path: polylogue/sources/cursor.py loc: 196 target: polylogue/sources/cursor.py diff --git a/polylogue/daemon/parse_prefetch.py b/polylogue/daemon/parse_prefetch.py index 312158c5a5..9238276e84 100644 --- a/polylogue/daemon/parse_prefetch.py +++ b/polylogue/daemon/parse_prefetch.py @@ -1,412 +1,34 @@ -"""Daemon-owned parse-stage extraction: parse census candidates off the writer hold. - -polylogue-m6tp phase (a). The raw-materialization conveyor's writer hold -(``DaemonWriteCoordinator.run_sync``) used to cover BOTH the CPU-bound -blob->``ParsedSession`` decode (census parse) and the SQLite writes that -record it, so a large or slow parse extended the writer hold by exactly as -long as the parse took -- starving every other write actor (live ingest, -status snapshots, insight convergence) queued behind the same coordinator. - -This module lets the daemon pre-parse the NEXT pass's candidate raws in a -bounded ``ThreadPoolExecutor`` BEFORE the writer hold is ever requested. The -writer-held pass then finds those results already warmed in a -``RawParsePrefetchCache`` (``polylogue.sources.revision_backfill``) and skips -reparsing them -- see that class's docstring for why a miss (empty cache, -budget-rejected entry, or the flag simply being off) always degrades to the -exact unmodified parse path rather than incorrect behavior. - -Why threads are safe here even on a standard (GIL) build: the polylogue-7mtf -control-run measurement (``parallel_threads_effective`` in -``polylogue.pipeline.services.process_pool``) found threaded parse gives no -GIL-build speedup AND inflates a *concurrently write-holding* thread's commit -latency ~5000x. That hazard is specifically about a parse thread running -WHILE a writer thread is active. This module never does that: ``warm()`` is -called by the conveyor BEFORE it ever asks the write coordinator for the -writer hold, so there is no writer thread to contend with. On a GIL build -this still gives little or no wall-clock parse speedup (CPython serializes -the CPU-bound decode across threads) -- that is expected and is the point of -phase (a): prove the parse/apply seam is correct and equivalence-safe ahead -of the free-threaded 3.14t deploy (phase (b), polylogue-m6tp), which is what -turns the same code path into a real speedup. +"""Daemon-facing re-export of the shared off-writer-hold parse-stage engine. + +polylogue-m6tp phase (a); relocated to substrate at +``polylogue.sources.census_parse_stage`` (polylogue-czq2) so the offline +rebuild engine (``maintenance/rebuild_index.py``) can consume the exact same +``CensusParseStage``/``RawParsePrefetchCache`` machinery this module used to +own exclusively, instead of only the daemon's own bulk-rebuild loop +(``daemon/bulk_rebuild.py``) ever getting a warmed prefetch cache while the +offline CLI and the daemon's own ``/api/maintenance/rebuild-index`` HTTP +route silently threaded ``prefetch_cache=None``. + +Every name below is the SAME object as its ``polylogue.sources. +census_parse_stage`` counterpart -- this module adds no behavior, only +preserves the daemon's existing import path +(``from polylogue.daemon.parse_prefetch import DaemonParseStage``) so every +pre-existing daemon call site and test keeps working unchanged. """ from __future__ import annotations -import os -import threading -from collections.abc import Sequence -from concurrent.futures import ThreadPoolExecutor, as_completed - -from polylogue.config import Config -from polylogue.logging import get_logger -from polylogue.pipeline.parsed_tree_size import ( - effective_physical_memory_bytes, - estimate_parsed_tree_bytes, +from polylogue.sources.census_parse_stage import ( + CensusParseStage as DaemonParseStage, ) -from polylogue.sources import revision_backfill -from polylogue.sources.dispatch import is_stream_record_provider -from polylogue.sources.revision_backfill import RawParsePrefetchCache -from polylogue.storage.repair import ( - raw_materialization_pending_census_raw_ids, - raw_materialization_readonly_descriptors, +from polylogue.sources.census_parse_stage import ( + daemon_parse_stage_max_cached_tree_bytes, + daemon_parse_stage_max_inflight_bytes, + daemon_parse_stage_warm_timeout_seconds, + daemon_parse_stage_worker_count, + estimate_parsed_tree_bytes, ) -logger = get_logger(__name__) - -# Floor/ceiling for the adaptive whale-memory budget below. The original -# fixed 64 MiB default starved bulk-scale warm on whale corpora: measured -# live 2026-07-20 on the 50K-raw archive, a 2000-raw page warmed 139 raws in -# 376s (0.37 raws/s, pool stalled on cache admission) under 64 MiB versus -# 500 raws in 8.8s (56.7 raws/s) with the budget raised — the workers were -# blocked on `try_admit`, not on parsing. The budget's purpose is bounding -# transient memory beside a live daemon, so it scales with the machine -# instead of a one-size constant: 1/16 of physical RAM, clamped to -# [64 MiB, 2 GiB]. -_MIN_MAX_INFLIGHT_BYTES = 64 * 1024 * 1024 # 64 MiB -_MAX_MAX_INFLIGHT_BYTES = 2 * 1024 * 1024 * 1024 # 2 GiB - -# CodeRabbit (PR #3168): as_completed()/future.result() had no timeout, so one -# hung worker (e.g. an unresponsive filesystem read) would block warm() -# forever -- and warm() is awaited directly ahead of run_sync in the periodic -# raw-materialization loop, so a stuck warm pass would stall every subsequent -# drain pass indefinitely, not just this one. 300s (5 min) is generous for -# the happy path (a bounded batch of already-published local blob reads) and -# only ever matters on a genuine hang. On timeout, still-pending raws are -# simply left uncached -- the writer-held pass reparses them normally, the -# same graceful-degradation guarantee as any other prefetch miss. A -# ThreadPoolExecutor cannot forcibly kill a running worker thread, so a truly -# wedged worker keeps occupying one pool slot until it (eventually) returns; -# that is an inherent limitation of thread-based cancellation, not something -# this bound can fix -- the bound's job is only to stop the CONVEYOR LOOP -# from waiting on it forever, which it does. -_DEFAULT_WARM_TIMEOUT_SECONDS = 300.0 - - -def daemon_parse_stage_worker_count() -> int: - """Bounded worker cap for the daemon-owned pre-parse thread pool. - - ``cpu_count - 1`` leaves one core free for the daemon's own event loop, - mirroring ``resolve_parse_worker_count``'s cpu-1 convention (see - ``polylogue.pipeline.services.process_pool``). Override with - ``POLYLOGUE_DAEMON_PARSE_STAGE_WORKERS``. - """ - from polylogue.config import load_polylogue_config - - configured = load_polylogue_config().daemon_parse_stage_workers - if configured is not None and configured > 0: - return configured - return max(1, (os.cpu_count() or 2) - 1) - - -def _physical_memory_bytes() -> int | None: - return effective_physical_memory_bytes() - - -def daemon_parse_stage_max_inflight_bytes() -> int: - """Whale-memory budget for parsed sessions held in the prefetch cache. - - Adaptive: 1/16 of physical RAM clamped to [64 MiB, 2 GiB] (see the - constants above for the measured starvation the old fixed 64 MiB default - caused). Override with ``POLYLOGUE_DAEMON_PARSE_STAGE_MAX_INFLIGHT_BYTES``. - """ - from polylogue.config import load_polylogue_config - - configured = load_polylogue_config().daemon_parse_stage_max_inflight_bytes - if configured is not None and configured > 0: - return configured - physical = _physical_memory_bytes() - if physical is None: - return _MIN_MAX_INFLIGHT_BYTES - return max(_MIN_MAX_INFLIGHT_BYTES, min(_MAX_MAX_INFLIGHT_BYTES, physical // 16)) - - -# polylogue-xb4i: the inflight-bytes budget above (and RawParsePrefetchCache's -# own admission gate) both account raw PAYLOAD bytes, because payload size is -# the only thing known BEFORE a raw is parsed. But a parsed ``ParsedSession`` -# tree resident in the cache is not the same size as the payload it was -# parsed from -- Pydantic model instances, per-block dicts, and Python object -# overhead inflate a compact JSON/JSONL payload substantially. Two earlyoom -# kills (19.3G and 20.2G RSS peaks, 2026-07-20) happened on a whale-dense -# page precisely because the cache retained a whole 2000-raw page of PARSED -# TREES while only the raw payload bytes were budgeted -- clamping the -# inflight (pre-parse) budget did nothing, since the memory pressure came -# from trees already sitting in the cache post-parse, not from parses in -# flight. -# -# Floor/ceiling mirror the inflight-bytes budget's adaptive-RAM shape: 1/8 of -# physical RAM (trees are the bigger of the two budgets since they are what -# actually sits resident) clamped to [256 MiB, 4 GiB]. -_MIN_MAX_CACHED_TREE_BYTES = 256 * 1024 * 1024 # 256 MiB -_MAX_MAX_CACHED_TREE_BYTES = 4 * 1024 * 1024 * 1024 # 4 GiB - - -def daemon_parse_stage_max_cached_tree_bytes() -> int: - """Whole-cache budget for ESTIMATED parsed-tree bytes (not payload bytes). - - Distinct from :func:`daemon_parse_stage_max_inflight_bytes`, which caps - raw PAYLOAD bytes admitted while parses are in flight (the only thing - knowable pre-parse). This is the budget that actually bounds what a - quiet daemon holds resident in ``DaemonParseStage.cache`` between warm() - passes -- see the calibration comment on ``_ESTIMATOR_BYTES_PER_CHAR`` - above for why the two budgets can diverge by 10x+ on the same page. - Adaptive: 1/8 of physical RAM clamped to [256 MiB, 4 GiB]. Override with - ``POLYLOGUE_DAEMON_PARSE_STAGE_MAX_CACHED_TREE_BYTES``. - """ - from polylogue.config import load_polylogue_config - - configured = load_polylogue_config().daemon_parse_stage_max_cached_tree_bytes - if configured is not None and configured > 0: - return configured - physical = _physical_memory_bytes() - if physical is None: - return _MIN_MAX_CACHED_TREE_BYTES - return max(_MIN_MAX_CACHED_TREE_BYTES, min(_MAX_MAX_CACHED_TREE_BYTES, physical // 8)) - - -def daemon_parse_stage_warm_timeout_seconds() -> float: - """Bound on how long ``warm()`` waits for its dispatched workers. - - Override with ``POLYLOGUE_DAEMON_PARSE_STAGE_WARM_TIMEOUT_SECONDS``. See - ``_DEFAULT_WARM_TIMEOUT_SECONDS`` for why this exists and what it does - (and does not) guarantee. - """ - from polylogue.config import load_polylogue_config - - configured = load_polylogue_config().daemon_parse_stage_warm_timeout_seconds - if configured is not None and configured > 0: - return configured - return _DEFAULT_WARM_TIMEOUT_SECONDS - - -class DaemonParseStage: - """Owns the daemon's bounded pre-parse ``ThreadPoolExecutor`` and cache. - - One instance lives for the daemon process's lifetime (created lazily on - first use by the raw-materialization conveyor loop). ``warm`` is - synchronous/blocking -- callers run it off the - event loop (``asyncio.to_thread``), exactly like every other conveyor - pass, and NEVER under ``daemon_write_coordinator().run_sync``: doing so - would defeat the entire point, since the pre-parse must run without the - writer hold held. - """ - - def __init__( - self, - *, - max_workers: int | None = None, - max_inflight_bytes: int | None = None, - warm_timeout_seconds: float | None = None, - max_cached_tree_bytes: int | None = None, - ) -> None: - self._executor = ThreadPoolExecutor( - max_workers=max_workers if max_workers is not None else daemon_parse_stage_worker_count(), - thread_name_prefix="polylogue-parse-stage", - ) - self.cache = RawParsePrefetchCache( - max_inflight_bytes=( - max_inflight_bytes if max_inflight_bytes is not None else daemon_parse_stage_max_inflight_bytes() - ) - ) - self._warm_timeout_seconds = ( - warm_timeout_seconds if warm_timeout_seconds is not None else daemon_parse_stage_warm_timeout_seconds() - ) - # polylogue-xb4i: a SECOND budget tracked alongside ``self.cache``, - # keyed on the same raw_ids but accounting ESTIMATED PARSED-TREE - # bytes instead of the raw cache's payload bytes. ``self.cache`` - # itself is not touched/subclassed (it is a shared type consumed - # directly by other callers -- ``bulk_rebuild.py`` hands - # ``stage.cache`` straight to ``RebuildIndexRequest.prefetch_cache`` - # -- so this stays a side ledger that reconciles against the raw - # cache's own admission/eviction rather than replacing it. - self._max_cached_tree_bytes = ( - max_cached_tree_bytes if max_cached_tree_bytes is not None else daemon_parse_stage_max_cached_tree_bytes() - ) - self._tree_bytes_lock = threading.Lock() - self._tree_bytes_by_raw_id: dict[str, int] = {} - self._cached_tree_bytes_total = 0 - - @property - def cached_tree_bytes_total(self) -> int: - """Sum of estimated parsed-tree bytes currently tracked as cached.""" - with self._tree_bytes_lock: - return self._cached_tree_bytes_total - - def _reconcile_stale_tree_tracking_locked(self) -> None: - """Drop tracking for any raw_id no longer present in ``self.cache``. - - Consumers outside this class (the writer-held pass, via - ``RawParsePrefetchCache.pop``) remove entries from ``self.cache`` - directly -- this class has no hook into that removal, so the tree- - byte ledger can only be reconciled lazily, by checking membership - before making an eviction decision. Cheap: proportional to the - number of currently-tracked entries, each check a dict lookup under - the raw cache's own lock. - """ - stale = [raw_id for raw_id in self._tree_bytes_by_raw_id if not self.cache.contains(raw_id)] - for raw_id in stale: - self._drop_tree_bytes_locked(raw_id) - - def _drop_tree_bytes_locked(self, raw_id: str) -> None: - tree_bytes = self._tree_bytes_by_raw_id.pop(raw_id, None) - if tree_bytes is not None: - self._cached_tree_bytes_total -= tree_bytes - - def _select_eviction_candidate_locked(self) -> str | None: - """Largest entry wins; ties break to the oldest (dict preserves - insertion order, and ``>`` -- not ``>=`` -- means the first-seen - (oldest) entry at the max size is kept as the running candidate).""" - best_id: str | None = None - best_bytes = -1 - for raw_id, tree_bytes in self._tree_bytes_by_raw_id.items(): - if tree_bytes > best_bytes: - best_bytes = tree_bytes - best_id = raw_id - return best_id - - def _register_cached_tree_bytes(self, raw_id: str, tree_bytes: int) -> None: - """Record a newly-admitted raw's estimated tree size and evict - largest-or-oldest entries (via ``self.cache.pop``, which releases - both the raw cache's own payload-byte budget and this ledger's tree- - byte budget) until back under ``self._max_cached_tree_bytes``.""" - evicted: list[str] = [] - with self._tree_bytes_lock: - self._reconcile_stale_tree_tracking_locked() - self._tree_bytes_by_raw_id[raw_id] = tree_bytes - self._cached_tree_bytes_total += tree_bytes - while self._cached_tree_bytes_total > self._max_cached_tree_bytes and self._tree_bytes_by_raw_id: - candidate = self._select_eviction_candidate_locked() - if candidate is None: - break - self._drop_tree_bytes_locked(candidate) - evicted.append(candidate) - for evicted_id in evicted: - self.cache.pop(evicted_id) - if evicted: - logger.info( - "parse-stage prefetch: evicted %d cached parsed tree(s) to stay within the %d-byte " - "estimated-tree-bytes budget after admitting raw_id=%s (%d bytes)", - len(evicted), - self._max_cached_tree_bytes, - raw_id, - tree_bytes, - ) - - def warm(self, config: Config, *, limit: int, max_payload_bytes: int) -> int: - """Pre-parse up to ``limit`` pending census candidates outside any writer hold. - - Returns the number of raws newly admitted to the cache. Read-only - end to end: candidate discovery and descriptor lookup both open - ``mode=ro`` SQLite connections (``polylogue.storage.repair``); - parsing reads only already-published blob bytes via a stateless - ``ArchiveBlobPublisher``, mirroring the production census parse - worker exactly (``census_parse_worker``, the same function the - writer-held path dispatches to a process/thread pool). Nothing here - writes to source.db, index.db, or takes the daemon's writer lease. - """ - candidate_raw_ids = raw_materialization_pending_census_raw_ids( - config, limit=limit, max_payload_bytes=max_payload_bytes - ) - return self.warm_raw_ids(config, raw_ids=candidate_raw_ids, max_payload_bytes=max_payload_bytes) - - def warm_raw_ids(self, config: Config, *, raw_ids: Sequence[str], max_payload_bytes: int) -> int: - """Pre-parse an explicit ``raw_ids`` list outside any writer hold. - - Same read-only, graceful-degradation contract as :meth:`warm`, but - for a caller (polylogue-gd6v's daemon bulk-rebuild routing) that - already knows exactly which raws its next bounded pass will select - -- a resumable rebuild transaction's own paged cursor -- instead of - querying the raw-materialization conveyor's own pending-census - candidate set. :meth:`warm` is now a thin wrapper around this method. - """ - raw_ids = [raw_id for raw_id in raw_ids if not self.cache.contains(raw_id)] - if not raw_ids: - return 0 - archive_root = config.archive_root - descriptors = raw_materialization_readonly_descriptors(archive_root, raw_ids) - blob_root_str = str(archive_root / "blob") - source_db_path_str = str(archive_root / "source.db") - - futures = {} - for raw_id in raw_ids: - descriptor = descriptors.get(raw_id) - if descriptor is None: - continue - provider, blob_hash, source_path, _kind, _size = descriptor - future = self._executor.submit( - revision_backfill.census_parse_worker, - raw_id, - provider.value, - blob_hash, - source_path, - is_stream_record_provider(source_path, str(provider)), - blob_root_str, - source_db_path_str, - ) - futures[future] = raw_id - - warmed = 0 - completed = 0 - try: - for future in as_completed(futures, timeout=self._warm_timeout_seconds): - completed += 1 - raw_id = futures[future] - try: - _raw_id, sessions, error = future.result() - except Exception: - logger.warning("parse-stage prefetch: worker failed for raw_id=%s", raw_id, exc_info=True) - continue - if error is not None or sessions is None: - # Parse failures are intentionally NOT cached: the writer-held - # pass reparses (and correctly quarantines/records) this raw - # exactly as it would with the flag off. Prefetch only ever - # shortcuts the happy path. - continue - _provider, _blob_hash, _source_path, kind, payload_size = descriptors[raw_id] - # polylogue-xb4i: estimate the PARSED TREE size (not payload - # size) before ever admitting to the cache. A tree bigger - # than the whole tree-bytes budget is never retained at all - # -- it does not even occupy a payload-bytes admission slot - # -- so one whale raw can never pin gigabytes of resident - # memory regardless of how much inflight-bytes headroom it - # happened to fit under pre-parse. This is the fix for the - # 2026-07-20 earlyoom kills: the inflight clamp bounded - # PAYLOAD bytes in flight, but the cache retained whatever - # trees resulted from that payload with no size check at all. - tree_bytes = estimate_parsed_tree_bytes(sessions) - if tree_bytes > self._max_cached_tree_bytes: - logger.warning( - "parse-stage prefetch: raw_id=%s estimated parsed-tree bytes %d exceed the " - "whole cache budget %d bytes; never retained -- the writer-held pass reparses " - "it normally, identical to any other prefetch miss", - raw_id, - tree_bytes, - self._max_cached_tree_bytes, - ) - continue - if self.cache.try_admit(raw_id, sessions, payload_bytes=payload_size, revision_kind=kind): - self._register_cached_tree_bytes(raw_id, tree_bytes) - warmed += 1 - except TimeoutError: - # Bounds the CONVEYOR LOOP's wait, not the worker itself -- a - # ThreadPoolExecutor cannot forcibly kill a running thread, so a - # genuinely wedged worker keeps occupying one pool slot until it - # eventually returns (see _DEFAULT_WARM_TIMEOUT_SECONDS). Every raw - # not yet completed is simply left uncached: the writer-held pass - # reparses it normally, identical to any other prefetch miss. - pending = len(futures) - completed - logger.warning( - "parse-stage prefetch: warm() timed out after %.0fs waiting on %d of %d worker(s); " - "leaving unfinished raw(s) uncached for the writer-held pass to reparse normally", - self._warm_timeout_seconds, - pending, - len(futures), - ) - return warmed - - def shutdown(self) -> None: - self._executor.shutdown(wait=False, cancel_futures=True) - - __all__ = [ "DaemonParseStage", "daemon_parse_stage_max_cached_tree_bytes", diff --git a/polylogue/maintenance/rebuild_index.py b/polylogue/maintenance/rebuild_index.py index 00f012229e..88fad987a3 100644 --- a/polylogue/maintenance/rebuild_index.py +++ b/polylogue/maintenance/rebuild_index.py @@ -53,6 +53,58 @@ logger = get_logger(__name__) +#: Passed through to ``CensusParseStage.warm_raw_ids``'s ``max_payload_bytes`` +#: parameter for symmetry with the daemon's own call site +#: (``daemon/bulk_rebuild.py``); the parameter is currently accepted but not +#: consulted inside ``warm_raw_ids`` itself, so this value has no observable +#: effect today, but every caller supplies one so a future budget check does +#: not silently start unbounded for whichever caller forgot to pass it. +_OFFLINE_PREFETCH_WARM_MAX_PAYLOAD_BYTES = 512 * 1024 * 1024 + + +def _warm_offline_prefetch_cache(config: Config, raw_ids: list[str]) -> RawParsePrefetchCache | None: + """Pre-parse this pass's raw ids in a bounded thread pool before replay. + + polylogue-czq2: closes the gap where every rebuild caller EXCEPT the + daemon's own bulk-rebuild loop (``daemon/bulk_rebuild.py``, #3168) left + ``RebuildIndexRequest.prefetch_cache`` at its default ``None`` -- the + offline ``polylogue ops maintenance rebuild-index`` CLI and the daemon's + own ``/api/maintenance/rebuild-index`` HTTP route both construct a + ``RebuildIndexRequest`` without ever threading one, so ``census``'s parse + step (and the ``spill_load`` reload it feeds) always paid the full + unwarmed cost on those routes even though the exact machinery to avoid it + (``CensusParseStage``/``RawParsePrefetchCache``, + ``polylogue.sources.census_parse_stage``) already existed and was fully + wired through ``backfill_historical_revision_evidence``. + + Called from ``_rebuild_index_from_source_owned`` only when + ``request.prefetch_cache is None`` (a caller that already warmed its own + cache off a writer hold it does not yet hold -- the daemon's bulk-rebuild + loop -- is never overridden here). Returns ``None`` for an empty + ``raw_ids`` (nothing to warm); a construction/warm failure is never + raised -- this is a pure optimization over the unmodified parse path, so + any failure here must degrade to that path exactly like an ordinary + prefetch miss, never abort the rebuild pass itself. + """ + if not raw_ids: + return None + from polylogue.sources.census_parse_stage import CensusParseStage + + stage = CensusParseStage() + try: + warmed = stage.warm_raw_ids(config, raw_ids=raw_ids, max_payload_bytes=_OFFLINE_PREFETCH_WARM_MAX_PAYLOAD_BYTES) + logger.info( + "rebuild_index_offline_prefetch_warm", + requested=len(raw_ids), + warmed=warmed, + ) + except Exception: + logger.warning("rebuild_index_offline_prefetch_warm_failed", exc_info=True) + stage.shutdown() + return None + stage.shutdown() + return stage.cache + def _should_refresh_generation_planner_statistics( *, @@ -194,12 +246,16 @@ class RebuildIndexRequest: raw_batch_size: int = 500 pass_byte_budget_mb: float | None = None pass_deadline_seconds: float | None = None - # polylogue-gd6v: daemon-internal callers only (never CLI/HTTP -- there is - # no JSON wire shape for a live cache object). Lets the daemon's bulk - # rebuild routing substitute parse output already computed off the - # writer hold (``DaemonParseStage``) for this pass's census phase. Every - # existing caller leaves this ``None`` and gets the exact unmodified - # parse path. + # polylogue-gd6v: in-process callers only (never CLI/HTTP wire params -- + # there is no JSON shape for a live cache object). Lets a caller that + # already computed parse output off a writer hold (the daemon's own + # ``CensusParseStage``, e.g. ``daemon/bulk_rebuild.py``) substitute it for + # this pass's census phase. Leaving this ``None`` (every CLI/HTTP-facing + # caller) does NOT skip prefetching any more: ``_rebuild_index_from_source_owned`` + # warms one itself for exactly this pass's selected raw ids before + # replaying (see ``_warm_offline_prefetch_cache``) -- a caller only needs + # to set this explicitly to reuse a cache warmed AHEAD of a writer hold it + # does not yet hold, which is what the daemon's bulk-rebuild loop does. prefetch_cache: RawParsePrefetchCache | None = None @@ -574,6 +630,24 @@ async def _rebuild_index_from_source_owned( sources=[], db_path=Path(generation.index_path), ) + # polylogue-czq2: warm THIS pass's own raw ids before replay when + # the caller did not already hand us a prefetch cache (the + # daemon's bulk-rebuild loop is the only caller that does -- + # `resolve_or_start_daemon_bulk_rebuild_transaction` warms off a + # writer hold it does not yet hold, so its cache is left alone + # here). Every other caller (offline CLI, the daemon's own HTTP + # rebuild-index route) gets the same census-parse seam the daemon + # loop always had, closing the gap that let ``spill_load`` pay a + # full serial re-parse/reload cost on those routes. Reads from the + # REAL archive root (`root`), not `generation_root`: source.db and + # the blob store live beside the outer archive, never inside a + # not-yet-promoted generation directory. + effective_prefetch_cache = request.prefetch_cache + if effective_prefetch_cache is None and selected_raw_ids: + warm_config = Config(archive_root=root, render_root=render_root(), sources=[]) + effective_prefetch_cache = await asyncio.to_thread( + _warm_offline_prefetch_cache, warm_config, selected_raw_ids + ) pass_started_at_s = time.perf_counter() replay = await replay_source( config, @@ -599,7 +673,7 @@ async def _rebuild_index_from_source_owned( # _repopulate_bulk_build_derived_state, called below right # before readiness. bulk_build=True, - prefetch_cache=request.prefetch_cache, + prefetch_cache=effective_prefetch_cache, ) pass_elapsed_s = time.perf_counter() - pass_started_at_s processed_before = transaction.processed_raw_count if transaction is not None else None diff --git a/polylogue/sources/census_parse_stage.py b/polylogue/sources/census_parse_stage.py new file mode 100644 index 0000000000..0afefae17b --- /dev/null +++ b/polylogue/sources/census_parse_stage.py @@ -0,0 +1,446 @@ +"""Shared off-writer-hold parse-stage engine: parse census candidates before any writer hold. + +polylogue-m6tp phase (a), relocated to substrate (polylogue-czq2). Originally +lived in ``polylogue.daemon.parse_prefetch`` and was consulted by exactly one +caller (``daemon/bulk_rebuild.py``'s automagic bulk-rebuild routing) even +though the mechanism it provides -- pre-parsing a bounded set of raw ids in a +``ThreadPoolExecutor`` and handing the result to ``RawParsePrefetchCache`` -- +has nothing daemon-specific about it. Every OTHER caller of the shared +rebuild engine (the offline ``polylogue ops maintenance rebuild-index`` CLI, +and the daemon's own ``/api/maintenance/rebuild-index`` HTTP route) threaded +``prefetch_cache=None`` and paid the full serial re-parse/spill-reload cost +this module exists to avoid -- see ``maintenance/rebuild_index.py``'s +``_warm_offline_prefetch_cache`` for the fix that consumes this module +directly instead of only through the daemon's bulk-rebuild loop. + +``polylogue.daemon.parse_prefetch`` re-exports ``DaemonParseStage`` (an alias +of :class:`CensusParseStage` below) and every config-resolution helper from +here unchanged, so every existing daemon caller/test keeps its import path +and behavior byte-identical; this module is the substrate the daemon +consumes, not a daemon-owned implementation detail any more. + +The writer-hold contention this was originally built to avoid does not apply +to the offline CLI or HTTP maintenance route the same way (there is no +sibling write actor sharing a coordinator in a one-shot offline rebuild +process; the HTTP route already sits inside the write bridge's held slot) -- +but the SAME mechanism still avoids the double parse-then-spill round trip +those callers were paying: a raw already popped from the prefetch cache +skips ``_parse_retained_raws``'s own dispatch entirely (see that function's +docstring), so census output flows into ``_ParsedSessionSpill`` exactly once +either way, just without the caller having forgotten to ask for it. + +Why threads are safe here even on a standard (GIL) build: the polylogue-7mtf +control-run measurement (``parallel_threads_effective`` in +``polylogue.pipeline.services.process_pool``) found threaded parse gives no +GIL-build speedup AND inflates a *concurrently write-holding* thread's commit +latency ~5000x. That hazard is specifically about a parse thread running +WHILE a writer thread is active. This module never does that: ``warm()``/ +``warm_raw_ids()`` are called BEFORE any caller ever asks for (or already +holds) a writer hold, so there is no writer thread to contend with. On a GIL +build this still gives little or no wall-clock parse speedup (CPython +serializes the CPU-bound decode across threads) -- that is expected; it still +avoids the process-pool pickle-back round trip ``_parse_unique_retained_raws`` +pays on a GIL build (threads share parsed object graphs by reference), and it +is what turns into a real multi-core speedup on the free-threaded 3.14t +deploy (phase (b), polylogue-m6tp). +""" + +from __future__ import annotations + +import os +import threading +from collections.abc import Sequence +from concurrent.futures import ThreadPoolExecutor, as_completed + +from polylogue.config import Config +from polylogue.logging import get_logger +from polylogue.pipeline.parsed_tree_size import ( + effective_physical_memory_bytes, + estimate_parsed_tree_bytes, +) +from polylogue.sources import revision_backfill +from polylogue.sources.dispatch import is_stream_record_provider +from polylogue.sources.revision_backfill import RawParsePrefetchCache +from polylogue.storage.repair import ( + raw_materialization_pending_census_raw_ids, + raw_materialization_readonly_descriptors, +) + +logger = get_logger(__name__) + +# Floor/ceiling for the adaptive whale-memory budget below. The original +# fixed 64 MiB default starved bulk-scale warm on whale corpora: measured +# live 2026-07-20 on the 50K-raw archive, a 2000-raw page warmed 139 raws in +# 376s (0.37 raws/s, pool stalled on cache admission) under 64 MiB versus +# 500 raws in 8.8s (56.7 raws/s) with the budget raised — the workers were +# blocked on `try_admit`, not on parsing. The budget's purpose is bounding +# transient memory beside a live daemon, so it scales with the machine +# instead of a one-size constant: 1/16 of physical RAM, clamped to +# [64 MiB, 2 GiB]. +_MIN_MAX_INFLIGHT_BYTES = 64 * 1024 * 1024 # 64 MiB +_MAX_MAX_INFLIGHT_BYTES = 2 * 1024 * 1024 * 1024 # 2 GiB + +# CodeRabbit (PR #3168): as_completed()/future.result() had no timeout, so one +# hung worker (e.g. an unresponsive filesystem read) would block warm() +# forever -- and warm() is awaited directly ahead of run_sync in the periodic +# raw-materialization loop, so a stuck warm pass would stall every subsequent +# drain pass indefinitely, not just this one. 300s (5 min) is generous for +# the happy path (a bounded batch of already-published local blob reads) and +# only ever matters on a genuine hang. On timeout, still-pending raws are +# simply left uncached -- the writer-held pass reparses them normally, the +# same graceful-degradation guarantee as any other prefetch miss. A +# ThreadPoolExecutor cannot forcibly kill a running worker thread, so a truly +# wedged worker keeps occupying one pool slot until it (eventually) returns; +# that is an inherent limitation of thread-based cancellation, not something +# this bound can fix -- the bound's job is only to stop the CONVEYOR LOOP +# from waiting on it forever, which it does. +_DEFAULT_WARM_TIMEOUT_SECONDS = 300.0 + + +def daemon_parse_stage_worker_count() -> int: + """Bounded worker cap for the daemon-owned pre-parse thread pool. + + ``cpu_count - 1`` leaves one core free for the daemon's own event loop, + mirroring ``resolve_parse_worker_count``'s cpu-1 convention (see + ``polylogue.pipeline.services.process_pool``). Override with + ``POLYLOGUE_DAEMON_PARSE_STAGE_WORKERS``. + """ + from polylogue.config import load_polylogue_config + + configured = load_polylogue_config().daemon_parse_stage_workers + if configured is not None and configured > 0: + return configured + return max(1, (os.cpu_count() or 2) - 1) + + +def _physical_memory_bytes() -> int | None: + return effective_physical_memory_bytes() + + +def daemon_parse_stage_max_inflight_bytes() -> int: + """Whale-memory budget for parsed sessions held in the prefetch cache. + + Adaptive: 1/16 of physical RAM clamped to [64 MiB, 2 GiB] (see the + constants above for the measured starvation the old fixed 64 MiB default + caused). Override with ``POLYLOGUE_DAEMON_PARSE_STAGE_MAX_INFLIGHT_BYTES``. + """ + from polylogue.config import load_polylogue_config + + configured = load_polylogue_config().daemon_parse_stage_max_inflight_bytes + if configured is not None and configured > 0: + return configured + physical = _physical_memory_bytes() + if physical is None: + return _MIN_MAX_INFLIGHT_BYTES + return max(_MIN_MAX_INFLIGHT_BYTES, min(_MAX_MAX_INFLIGHT_BYTES, physical // 16)) + + +# polylogue-xb4i: the inflight-bytes budget above (and RawParsePrefetchCache's +# own admission gate) both account raw PAYLOAD bytes, because payload size is +# the only thing known BEFORE a raw is parsed. But a parsed ``ParsedSession`` +# tree resident in the cache is not the same size as the payload it was +# parsed from -- Pydantic model instances, per-block dicts, and Python object +# overhead inflate a compact JSON/JSONL payload substantially. Two earlyoom +# kills (19.3G and 20.2G RSS peaks, 2026-07-20) happened on a whale-dense +# page precisely because the cache retained a whole 2000-raw page of PARSED +# TREES while only the raw payload bytes were budgeted -- clamping the +# inflight (pre-parse) budget did nothing, since the memory pressure came +# from trees already sitting in the cache post-parse, not from parses in +# flight. +# +# Floor/ceiling mirror the inflight-bytes budget's adaptive-RAM shape: 1/8 of +# physical RAM (trees are the bigger of the two budgets since they are what +# actually sits resident) clamped to [256 MiB, 4 GiB]. +_MIN_MAX_CACHED_TREE_BYTES = 256 * 1024 * 1024 # 256 MiB +_MAX_MAX_CACHED_TREE_BYTES = 4 * 1024 * 1024 * 1024 # 4 GiB + + +def daemon_parse_stage_max_cached_tree_bytes() -> int: + """Whole-cache budget for ESTIMATED parsed-tree bytes (not payload bytes). + + Distinct from :func:`daemon_parse_stage_max_inflight_bytes`, which caps + raw PAYLOAD bytes admitted while parses are in flight (the only thing + knowable pre-parse). This is the budget that actually bounds what a + quiet daemon holds resident in ``DaemonParseStage.cache`` between warm() + passes -- see the calibration comment on ``_ESTIMATOR_BYTES_PER_CHAR`` + above for why the two budgets can diverge by 10x+ on the same page. + Adaptive: 1/8 of physical RAM clamped to [256 MiB, 4 GiB]. Override with + ``POLYLOGUE_DAEMON_PARSE_STAGE_MAX_CACHED_TREE_BYTES``. + """ + from polylogue.config import load_polylogue_config + + configured = load_polylogue_config().daemon_parse_stage_max_cached_tree_bytes + if configured is not None and configured > 0: + return configured + physical = _physical_memory_bytes() + if physical is None: + return _MIN_MAX_CACHED_TREE_BYTES + return max(_MIN_MAX_CACHED_TREE_BYTES, min(_MAX_MAX_CACHED_TREE_BYTES, physical // 8)) + + +def daemon_parse_stage_warm_timeout_seconds() -> float: + """Bound on how long ``warm()`` waits for its dispatched workers. + + Override with ``POLYLOGUE_DAEMON_PARSE_STAGE_WARM_TIMEOUT_SECONDS``. See + ``_DEFAULT_WARM_TIMEOUT_SECONDS`` for why this exists and what it does + (and does not) guarantee. + """ + from polylogue.config import load_polylogue_config + + configured = load_polylogue_config().daemon_parse_stage_warm_timeout_seconds + if configured is not None and configured > 0: + return configured + return _DEFAULT_WARM_TIMEOUT_SECONDS + + +class CensusParseStage: + """Owns a bounded pre-parse ``ThreadPoolExecutor`` and its prefetch cache. + + In the daemon, one instance lives for the process's lifetime (created + lazily on first use by the raw-materialization conveyor loop, or by + ``daemon/bulk_rebuild.py``'s bulk-rebuild routing). ``warm``/ + ``warm_raw_ids`` are synchronous/blocking -- a daemon caller runs them off + the event loop (``asyncio.to_thread``), exactly like every other conveyor + pass, and NEVER under ``daemon_write_coordinator().run_sync``: doing so + would defeat the entire point, since the pre-parse must run without the + writer hold held. An offline caller (``maintenance/rebuild_index.py``) + instead constructs a short-lived instance scoped to one bounded pass's + raw ids and discards it once ``warm_raw_ids`` returns -- see + ``_warm_offline_prefetch_cache``. + """ + + def __init__( + self, + *, + max_workers: int | None = None, + max_inflight_bytes: int | None = None, + warm_timeout_seconds: float | None = None, + max_cached_tree_bytes: int | None = None, + ) -> None: + self._executor = ThreadPoolExecutor( + max_workers=max_workers if max_workers is not None else daemon_parse_stage_worker_count(), + thread_name_prefix="polylogue-parse-stage", + ) + self.cache = RawParsePrefetchCache( + max_inflight_bytes=( + max_inflight_bytes if max_inflight_bytes is not None else daemon_parse_stage_max_inflight_bytes() + ) + ) + self._warm_timeout_seconds = ( + warm_timeout_seconds if warm_timeout_seconds is not None else daemon_parse_stage_warm_timeout_seconds() + ) + # polylogue-xb4i: a SECOND budget tracked alongside ``self.cache``, + # keyed on the same raw_ids but accounting ESTIMATED PARSED-TREE + # bytes instead of the raw cache's payload bytes. ``self.cache`` + # itself is not touched/subclassed (it is a shared type consumed + # directly by other callers -- ``bulk_rebuild.py`` hands + # ``stage.cache`` straight to ``RebuildIndexRequest.prefetch_cache`` + # -- so this stays a side ledger that reconciles against the raw + # cache's own admission/eviction rather than replacing it. + self._max_cached_tree_bytes = ( + max_cached_tree_bytes if max_cached_tree_bytes is not None else daemon_parse_stage_max_cached_tree_bytes() + ) + self._tree_bytes_lock = threading.Lock() + self._tree_bytes_by_raw_id: dict[str, int] = {} + self._cached_tree_bytes_total = 0 + + @property + def cached_tree_bytes_total(self) -> int: + """Sum of estimated parsed-tree bytes currently tracked as cached.""" + with self._tree_bytes_lock: + return self._cached_tree_bytes_total + + def _reconcile_stale_tree_tracking_locked(self) -> None: + """Drop tracking for any raw_id no longer present in ``self.cache``. + + Consumers outside this class (the writer-held pass, via + ``RawParsePrefetchCache.pop``) remove entries from ``self.cache`` + directly -- this class has no hook into that removal, so the tree- + byte ledger can only be reconciled lazily, by checking membership + before making an eviction decision. Cheap: proportional to the + number of currently-tracked entries, each check a dict lookup under + the raw cache's own lock. + """ + stale = [raw_id for raw_id in self._tree_bytes_by_raw_id if not self.cache.contains(raw_id)] + for raw_id in stale: + self._drop_tree_bytes_locked(raw_id) + + def _drop_tree_bytes_locked(self, raw_id: str) -> None: + tree_bytes = self._tree_bytes_by_raw_id.pop(raw_id, None) + if tree_bytes is not None: + self._cached_tree_bytes_total -= tree_bytes + + def _select_eviction_candidate_locked(self) -> str | None: + """Largest entry wins; ties break to the oldest (dict preserves + insertion order, and ``>`` -- not ``>=`` -- means the first-seen + (oldest) entry at the max size is kept as the running candidate).""" + best_id: str | None = None + best_bytes = -1 + for raw_id, tree_bytes in self._tree_bytes_by_raw_id.items(): + if tree_bytes > best_bytes: + best_bytes = tree_bytes + best_id = raw_id + return best_id + + def _register_cached_tree_bytes(self, raw_id: str, tree_bytes: int) -> None: + """Record a newly-admitted raw's estimated tree size and evict + largest-or-oldest entries (via ``self.cache.pop``, which releases + both the raw cache's own payload-byte budget and this ledger's tree- + byte budget) until back under ``self._max_cached_tree_bytes``.""" + evicted: list[str] = [] + with self._tree_bytes_lock: + self._reconcile_stale_tree_tracking_locked() + self._tree_bytes_by_raw_id[raw_id] = tree_bytes + self._cached_tree_bytes_total += tree_bytes + while self._cached_tree_bytes_total > self._max_cached_tree_bytes and self._tree_bytes_by_raw_id: + candidate = self._select_eviction_candidate_locked() + if candidate is None: + break + self._drop_tree_bytes_locked(candidate) + evicted.append(candidate) + for evicted_id in evicted: + self.cache.pop(evicted_id) + if evicted: + logger.info( + "parse-stage prefetch: evicted %d cached parsed tree(s) to stay within the %d-byte " + "estimated-tree-bytes budget after admitting raw_id=%s (%d bytes)", + len(evicted), + self._max_cached_tree_bytes, + raw_id, + tree_bytes, + ) + + def warm(self, config: Config, *, limit: int, max_payload_bytes: int) -> int: + """Pre-parse up to ``limit`` pending census candidates outside any writer hold. + + Returns the number of raws newly admitted to the cache. Read-only + end to end: candidate discovery and descriptor lookup both open + ``mode=ro`` SQLite connections (``polylogue.storage.repair``); + parsing reads only already-published blob bytes via a stateless + ``ArchiveBlobPublisher``, mirroring the production census parse + worker exactly (``census_parse_worker``, the same function the + writer-held path dispatches to a process/thread pool). Nothing here + writes to source.db, index.db, or takes the daemon's writer lease. + """ + candidate_raw_ids = raw_materialization_pending_census_raw_ids( + config, limit=limit, max_payload_bytes=max_payload_bytes + ) + return self.warm_raw_ids(config, raw_ids=candidate_raw_ids, max_payload_bytes=max_payload_bytes) + + def warm_raw_ids(self, config: Config, *, raw_ids: Sequence[str], max_payload_bytes: int) -> int: + """Pre-parse an explicit ``raw_ids`` list outside any writer hold. + + Same read-only, graceful-degradation contract as :meth:`warm`, but + for a caller (polylogue-gd6v's daemon bulk-rebuild routing) that + already knows exactly which raws its next bounded pass will select + -- a resumable rebuild transaction's own paged cursor -- instead of + querying the raw-materialization conveyor's own pending-census + candidate set. :meth:`warm` is now a thin wrapper around this method. + """ + raw_ids = [raw_id for raw_id in raw_ids if not self.cache.contains(raw_id)] + if not raw_ids: + return 0 + archive_root = config.archive_root + descriptors = raw_materialization_readonly_descriptors(archive_root, raw_ids) + blob_root_str = str(archive_root / "blob") + source_db_path_str = str(archive_root / "source.db") + + futures = {} + for raw_id in raw_ids: + descriptor = descriptors.get(raw_id) + if descriptor is None: + continue + provider, blob_hash, source_path, _kind, _size = descriptor + future = self._executor.submit( + revision_backfill.census_parse_worker, + raw_id, + provider.value, + blob_hash, + source_path, + is_stream_record_provider(source_path, str(provider)), + blob_root_str, + source_db_path_str, + ) + futures[future] = raw_id + + warmed = 0 + completed = 0 + try: + for future in as_completed(futures, timeout=self._warm_timeout_seconds): + completed += 1 + raw_id = futures[future] + try: + _raw_id, sessions, error = future.result() + except Exception: + logger.warning("parse-stage prefetch: worker failed for raw_id=%s", raw_id, exc_info=True) + continue + if error is not None or sessions is None: + # Parse failures are intentionally NOT cached: the writer-held + # pass reparses (and correctly quarantines/records) this raw + # exactly as it would with the flag off. Prefetch only ever + # shortcuts the happy path. + continue + _provider, _blob_hash, _source_path, kind, payload_size = descriptors[raw_id] + # polylogue-xb4i: estimate the PARSED TREE size (not payload + # size) before ever admitting to the cache. A tree bigger + # than the whole tree-bytes budget is never retained at all + # -- it does not even occupy a payload-bytes admission slot + # -- so one whale raw can never pin gigabytes of resident + # memory regardless of how much inflight-bytes headroom it + # happened to fit under pre-parse. This is the fix for the + # 2026-07-20 earlyoom kills: the inflight clamp bounded + # PAYLOAD bytes in flight, but the cache retained whatever + # trees resulted from that payload with no size check at all. + tree_bytes = estimate_parsed_tree_bytes(sessions) + if tree_bytes > self._max_cached_tree_bytes: + logger.warning( + "parse-stage prefetch: raw_id=%s estimated parsed-tree bytes %d exceed the " + "whole cache budget %d bytes; never retained -- the writer-held pass reparses " + "it normally, identical to any other prefetch miss", + raw_id, + tree_bytes, + self._max_cached_tree_bytes, + ) + continue + if self.cache.try_admit(raw_id, sessions, payload_bytes=payload_size, revision_kind=kind): + self._register_cached_tree_bytes(raw_id, tree_bytes) + warmed += 1 + except TimeoutError: + # Bounds the CONVEYOR LOOP's wait, not the worker itself -- a + # ThreadPoolExecutor cannot forcibly kill a running thread, so a + # genuinely wedged worker keeps occupying one pool slot until it + # eventually returns (see _DEFAULT_WARM_TIMEOUT_SECONDS). Every raw + # not yet completed is simply left uncached: the writer-held pass + # reparses it normally, identical to any other prefetch miss. + pending = len(futures) - completed + logger.warning( + "parse-stage prefetch: warm() timed out after %.0fs waiting on %d of %d worker(s); " + "leaving unfinished raw(s) uncached for the writer-held pass to reparse normally", + self._warm_timeout_seconds, + pending, + len(futures), + ) + return warmed + + def shutdown(self) -> None: + self._executor.shutdown(wait=False, cancel_futures=True) + + +#: Back-compat alias: every existing daemon call site and test imports this +#: class as ``DaemonParseStage`` (from ``polylogue.daemon.parse_prefetch``, +#: which re-exports it from here). Keeping the same name here too means +#: ``polylogue.sources.census_parse_stage.DaemonParseStage`` and +#: ``polylogue.daemon.parse_prefetch.DaemonParseStage`` are the exact same +#: object, not two types that happen to look alike. +DaemonParseStage = CensusParseStage + + +__all__ = [ + "CensusParseStage", + "DaemonParseStage", + "daemon_parse_stage_max_cached_tree_bytes", + "daemon_parse_stage_max_inflight_bytes", + "daemon_parse_stage_warm_timeout_seconds", + "daemon_parse_stage_worker_count", + "estimate_parsed_tree_bytes", +] diff --git a/tests/unit/daemon/test_parse_prefetch.py b/tests/unit/daemon/test_parse_prefetch.py index 72f4226a7f..d8634dead0 100644 --- a/tests/unit/daemon/test_parse_prefetch.py +++ b/tests/unit/daemon/test_parse_prefetch.py @@ -239,7 +239,13 @@ def test_max_inflight_bytes_default_is_adaptive_and_clamped(monkeypatch: pytest. with an adequate budget) — the budget must grow on capable machines while keeping the 64 MiB floor semantics on small ones. """ - from polylogue.daemon import parse_prefetch as pp + # polylogue-czq2: the adaptive-budget resolution functions (and the + # ``_physical_memory_bytes`` helper they call) now live in + # ``polylogue.sources.census_parse_stage`` -- ``polylogue.daemon.parse_prefetch`` + # only re-exports the same function objects, so monkeypatching an + # attribute there would rebind the re-export's local name, not the + # global the functions actually read from when they execute. + from polylogue.sources import census_parse_stage as pp monkeypatch.delenv("POLYLOGUE_DAEMON_PARSE_STAGE_MAX_INFLIGHT_BYTES", raising=False) @@ -474,7 +480,13 @@ def test_max_cached_tree_bytes_default_is_adaptive_and_clamped(monkeypatch: pyte range [256 MiB, 4 GiB] and distinct RAM fraction (1/8, not 1/16). A mutation that swapped in the inflight-bytes clamp/fraction, or dropped the env override, would flip these assertions.""" - from polylogue.daemon import parse_prefetch as pp + # polylogue-czq2: the adaptive-budget resolution functions (and the + # ``_physical_memory_bytes`` helper they call) now live in + # ``polylogue.sources.census_parse_stage`` -- ``polylogue.daemon.parse_prefetch`` + # only re-exports the same function objects, so monkeypatching an + # attribute there would rebind the re-export's local name, not the + # global the functions actually read from when they execute. + from polylogue.sources import census_parse_stage as pp monkeypatch.delenv("POLYLOGUE_DAEMON_PARSE_STAGE_MAX_CACHED_TREE_BYTES", raising=False) diff --git a/tests/unit/maintenance/test_rebuild_parse_apply_split.py b/tests/unit/maintenance/test_rebuild_parse_apply_split.py index d930aa646d..3e74678622 100644 --- a/tests/unit/maintenance/test_rebuild_parse_apply_split.py +++ b/tests/unit/maintenance/test_rebuild_parse_apply_split.py @@ -31,7 +31,9 @@ import pytest +from polylogue.config import Config from polylogue.maintenance.rebuild_index import RebuildIndexRequest, rebuild_index_from_source_sync +from polylogue.sources.census_parse_stage import CensusParseStage from polylogue.sources.revision_backfill import split_parse_and_apply_seconds from tests.infra.revision_backfill_benchmark import build_independent_raw_corpus @@ -107,3 +109,51 @@ def test_rebuild_records_parse_apply_split_summing_to_stage_total( # This corpus is genuinely replayed (not empty), so real writer work # happened: apply_s must be strictly positive, not merely non-negative. assert apply_s > 0.0 + + +def test_rebuild_index_from_source_sync_warms_prefetch_cache_when_caller_omits_one( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """polylogue-czq2: the offline CLI route must get the SAME off-writer + census prefetch seam ``daemon/bulk_rebuild.py`` has always had, not just + a caller that remembers to construct its own ``CensusParseStage``. + + Anti-vacuity: this drives the real production entry point + (``rebuild_index_from_source_sync``, the exact function the CLI's + ``rebuild-index`` command and ``polylogued``'s own HTTP maintenance route + call) with a request that leaves ``RebuildIndexRequest.prefetch_cache`` + at its default ``None`` -- exactly what those two callers do today. Before + ``_rebuild_index_from_source_owned`` grew its internal + ``_warm_offline_prefetch_cache`` call, ``CensusParseStage.warm_raw_ids`` + was reached by exactly one caller in the whole codebase + (``daemon/bulk_rebuild.py``): deleting the internal warm call this test + exercises makes ``warm_raw_ids`` unreached again from this route and this + assertion fails, proving the spy is wired to production code, not a + self-authorized double. + """ + root = tmp_path / "archive" + raw_ids = build_independent_raw_corpus(root, raw_count=6, avg_payload_bytes=20_000) + monkeypatch.setenv("POLYLOGUE_ARCHIVE_ROOT", str(root)) + + warmed_raw_id_batches: list[tuple[str, ...]] = [] + real_warm_raw_ids = CensusParseStage.warm_raw_ids + + def _spy_warm_raw_ids(self: CensusParseStage, config: Config, *, raw_ids: list[str], max_payload_bytes: int) -> int: + warmed_raw_id_batches.append(tuple(raw_ids)) + return real_warm_raw_ids(self, config, raw_ids=raw_ids, max_payload_bytes=max_payload_bytes) + + monkeypatch.setattr(CensusParseStage, "warm_raw_ids", _spy_warm_raw_ids) + + receipt = rebuild_index_from_source_sync( + RebuildIndexRequest( + archive_root=root, + promote=True, + raw_batch_size=500, # single page: whole corpus fits in one pass + ) + ) + + assert receipt.status == "replayed" + assert warmed_raw_id_batches, "offline rebuild_index_from_source_sync never warmed a prefetch cache" + # Every raw the census phase went on to process was offered to the warmer + # first -- the exact set this pass selected, not a subset/superset. + assert set(warmed_raw_id_batches[0]) == set(raw_ids)