From 4ffc848044f00644a2b3b801973d50c07ee26503 Mon Sep 17 00:00:00 2001 From: Sinity Date: Sat, 1 Aug 2026 00:39:13 +0200 Subject: [PATCH] test(maintenance): pin pipeline-decode auto-engagement on both rebuild routes Problem: polylogue-2cuv reported parse_s + apply_s == total_s exactly (zero overlap) on the real full rebuild, with spill_load (2830s SERIAL pickle.loads/reparse on the writer thread) as 22% of a 12633s pass. The bead asked for a bounded-memory producer/consumer pipeline overlapping parse decode with apply on both the daemon bulk-rebuild route and the offline route. Re-verification against current master found the structural fix already shipped: PR #3478 added `_ReplaySpillPrefetcher` (Lever A), a background thread that decodes upcoming replay cohorts' parsed sessions while the single writer applies the current cohort, wired into `backfill_historical_revision_evidence` via the `pipeline_decode` parameter (auto-engages under `parallel_threads_effective()` once a pass has >=8 cohorts). PR #3485 (`_lineage_aware_replay_order`, landed same day) is consumed by both the prefetcher and the writer loop from the identical `ordered_logical_keys` sequence, so lineage ordering survives pipelining. Both `polylogue/maintenance/rebuild_index.py` (offline CLI, `promote=True`) and `polylogue/daemon/bulk_rebuild.py::run_daemon_bulk_rebuild_pass` (daemon route) call the SAME `rebuild_index_from_source_sync` -> `maintenance/replay.py::rebuild_index_from_source` -> `backfill_historical_revision_evidence` chain, so the pipeline auto-engages identically on both -- there is no separate per-route wiring to add. Direct measurement (ad hoc synthetic corpus, not committed -- `spill_load` before=3.988s -> after=0.063s, `spill_prefetch.decode_concurrent` (hidden behind apply)=2.901s, 72.7% of the corpus's spill_load moved off the writer's critical path) confirms the mechanism is live, not merely present in source. Existing tests already pin outcome parity: `test_pipelined_ decode_matches_serial_archive_state` in `tests/unit/sources/test_revision_backfill.py` asserts byte-identical `RevisionBackfillResult` and full index content manifest between `pipeline_decode=False` and `pipeline_decode=True` runs over both the sqlite-spill and reparse-fallback decode lanes. What changed: added one new test proving the pipeline auto-engages from the PRODUCTION entry point both routes share (`rebuild_index_from_source_sync`), not only from the lower-level `backfill_historical_revision_evidence` calls the existing unit tests already exercised directly. Anti-vacuity: the test shrinks the spill's RAM cache tiers to force every `for_raw` to miss RAM, sizes the corpus at `_PIPELINE_DECODE_MIN_COHORTS + 4` independent raws (each its own logical cohort), and asserts `spill_prefetch.consumed > 0` in the receipt's `stage_timings_s` -- a parameter dropped anywhere in the `rebuild_index_from_source_sync` -> ... -> `backfill_historical_revision_ evidence` threading chain makes this assertion fail with 0, not a wrong number. What was NOT changed: no new pipeline plumbing -- #3478 already built it. The remaining fully-serial stage is the up-front `census` classification pass (still ahead of the replay loop by construction: cohort membership must be known before the replay loop can order/classify cohorts), which `_ReplaySpillPrefetcher` does not and structurally cannot touch -- pipelining THAT would mean overlapping census-page N+1 with replay-page N's apply, a materially larger redesign than the spill_load producer/consumer pipeline this bead asked for. Left as a residual finding, not implemented here. Verification: python -m devtools test tests/unit/maintenance/test_rebuild_parse_apply_split.py -> 5 passed python -m devtools test tests/unit/sources/test_revision_backfill.py -k pipelined_decode -> 3 passed (pre-existing equivalence proofs, confirmed still green) python -m devtools test tests/benchmarks/test_rebuild_cost_model.py -k "not full_population" -> 4 passed python -m devtools verify --quick -> exit 0 Ref polylogue-2cuv Co-Authored-By: Claude --- .../test_rebuild_parse_apply_split.py | 85 +++++++++++++++++++ 1 file changed, 85 insertions(+) diff --git a/tests/unit/maintenance/test_rebuild_parse_apply_split.py b/tests/unit/maintenance/test_rebuild_parse_apply_split.py index 3e74678622..4364682e37 100644 --- a/tests/unit/maintenance/test_rebuild_parse_apply_split.py +++ b/tests/unit/maintenance/test_rebuild_parse_apply_split.py @@ -27,12 +27,14 @@ from __future__ import annotations +import time from pathlib import Path import pytest from polylogue.config import Config from polylogue.maintenance.rebuild_index import RebuildIndexRequest, rebuild_index_from_source_sync +from polylogue.sources import revision_backfill 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 @@ -157,3 +159,86 @@ def _spy_warm_raw_ids(self: CensusParseStage, config: Config, *, raw_ids: list[s # 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) + + +def _give_replay_spill_prefetcher_a_head_start(monkeypatch: pytest.MonkeyPatch) -> None: + """Deterministic race pin, mirroring ``test_revision_backfill.py``'s + identically-named helper: production makes no ordering promise between + the background decode worker and the writer, so a tiny corpus can let + the writer finish before the worker buffers anything, making + ``spill_prefetch.consumed`` a coin flip. Give the worker a bounded head + start before the writer's own replay loop begins.""" + original_start_phase = revision_backfill._ReplaySpillPrefetcher.start_phase + + def start_phase_with_head_start( + self: revision_backfill._ReplaySpillPrefetcher, + ordered_keys: object, + extra_members: object, + ) -> None: + original_start_phase(self, ordered_keys, extra_members) # type: ignore[arg-type] + worker = self._thread + for _ in range(1000): # bounded ~10s; normally exits in milliseconds + with self._lock: + if len(self._buffer) >= 2: + break + if worker is None or not worker.is_alive(): + break + time.sleep(0.01) + + monkeypatch.setattr(revision_backfill._ReplaySpillPrefetcher, "start_phase", start_phase_with_head_start) + + +def test_rebuild_index_from_source_sync_auto_engages_pipelined_decode( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """polylogue-2cuv: BOTH production rebuild routes -- the offline CLI + (``polylogue ops maintenance rebuild-index``) and the daemon bulk-rebuild + loop (``daemon/bulk_rebuild.py::run_daemon_bulk_rebuild_pass``) -- drive + this exact ``rebuild_index_from_source_sync`` entry point (the daemon via + its own write coordinator, unmodified). Both therefore inherit the SAME + ``pipeline_decode`` auto-engagement inside ``backfill_historical_revision_ + evidence`` (Lever A / PR #3478's ``_ReplaySpillPrefetcher``) with zero + per-route wiring -- there is no separate knob either route could forget + to set. + + Anti-vacuity: this drives the real production entry point with a corpus + sized at/above ``_PIPELINE_DECODE_MIN_COHORTS`` (independent raws, so + every raw is its own logical cohort) and shrinks the spill's RAM cache + tiers to 1 byte so every replay ``for_raw`` misses RAM and must go + through the prefetcher-or-inline decode fork. If pipeline_decode were + hardcoded ``False`` somewhere between ``rebuild_index_from_source_sync`` + and ``backfill_historical_revision_evidence`` (e.g. a parameter dropped + while threading a future kwarg), ``spill_prefetch.consumed`` would never + appear in the stage timings and this assertion would fail -- proving the + auto-engagement path is actually reached from the production route, not + only from the lower-level unit tests that call + ``backfill_historical_revision_evidence`` directly. + """ + monkeypatch.setattr(revision_backfill._ParsedSessionSpill, "_DECODED_CACHE_MIN_TREE_BYTES", 1) + monkeypatch.setattr(revision_backfill._ParsedSessionSpill, "_DECODED_CACHE_MAX_TREE_BYTES", 1) + monkeypatch.setattr(revision_backfill._ParsedSessionSpill, "_WHALE_CACHE_MAX_TREE_BYTES", 1) + _give_replay_spill_prefetcher_a_head_start(monkeypatch) + + root = tmp_path / "archive" + cohort_count = revision_backfill._PIPELINE_DECODE_MIN_COHORTS + 4 + raw_ids = build_independent_raw_corpus(root, raw_count=cohort_count, avg_payload_bytes=20_000) + monkeypatch.setenv("POLYLOGUE_ARCHIVE_ROOT", str(root)) + + 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" + stage_timings_s = receipt.replay["stage_timings_s"] + assert isinstance(stage_timings_s, dict) + assert stage_timings_s.get("spill_prefetch.consumed", 0.0) > 0, ( + "rebuild_index_from_source_sync did not auto-engage the background " + "replay-spill prefetcher (Lever A) for a cohort count above " + "_PIPELINE_DECODE_MIN_COHORTS -- census+spill_load decode is no " + "longer proven to overlap the writer's apply work on this route" + ) + assert receipt.selected_raw_count == len(raw_ids)