From 0dbc9a2fa90642ab778c364088994122f3100686 Mon Sep 17 00:00:00 2001 From: Sinity Date: Mon, 20 Jul 2026 03:27:54 +0200 Subject: [PATCH] feat(daemon): route bulk-scale rebuild through daemon-owned generation build Problem: the trickle raw-materialization conveyor is sized for steady-state drift; a bulk-scale backlog turns it into a weeks-scale grind (#3145's threshold warning). The correct bulk path already existed as the offline `polylogue ops maintenance rebuild-index` CLI command (resumable transaction, blue-green generation, one census+ replay sweep), but nothing routed to it automatically, and its own resume model silently re-walks the whole corpus when a caller omits `--operation-id` (polylogue-fbte, measured ~2.25h re-verification per resume at 50K raws). Solution: polylogue-gd6v phase (c) of the m6tp convergence redesign. - polylogue/daemon/bulk_rebuild.py (new): daemon-internal orchestrator that resolves/resumes a SINGLE well-known bulk-rebuild transaction per archive (DAEMON_BULK_REBUILD_OPERATION_ID), drives it through `rebuild_index_from_source_sync` -- the SAME engine the offline CLI drives, extracted/shared rather than duplicated -- and retires a terminal (promoted/stale/failed) transaction automatically so the next backlog reuses the same id. Because the daemon always resolves the same operation id (never an operator-supplied one to forget), the fbte class of bug is structurally unreachable here. - polylogue/daemon/cli.py: `_maybe_route_daemon_bulk_rebuild`, gated by a new `daemon_bulk_rebuild_routing` config flag (off by default), called from the existing raw-materialization convergence tick right after the #3145 recommendation. Keeps driving an in-flight build every tick even below the bulk-scale threshold so partial progress is never abandoned. A dedicated `DaemonParseStage` singleton (separate pool from the trickle conveyor's) is shut down at daemon teardown alongside the existing one. - polylogue/daemon/parse_prefetch.py: `DaemonParseStage.warm_raw_ids` extracts `warm`'s dispatch/admit logic to also accept an explicit raw-id list -- the bulk driver already knows its next page's raw ids from the transaction's own cursor, so it pre-warms them in the #3168 thread pool before requesting the writer hold, instead of querying the raw-materialization conveyor's own candidate set. - polylogue/sources/revision_backfill.py, polylogue/maintenance/ replay.py, polylogue/maintenance/rebuild_index.py: thread a new `prefetch_cache` parameter from `RebuildIndexRequest` down through `backfill_historical_revision_evidence`'s census phase, mirroring the #3168 raw-materialization wiring. A prefetch hit flows through the same `spill.add(...)` as a fresh parse, so the replay phase's own `spill.for_raw` lookups see identical warmed content -- census- phase prefetching alone is enough to skip replay-phase reparsing too. Degrades gracefully to the unmodified sequential in-hold parse on any miss or on a GIL build (parallel-parse speedup lands with the 3.14t deploy per m6tp phase (b); this PR is orchestration only). - polylogue/storage/index_generation.py: `IndexGenerationStore. discard_transaction` retires a terminal transaction record so its operation id can be reused (extracted for the daemon driver; the offline CLI path never needed this since it mints a fresh uuid4 per operation). - polylogue/config.py, docs/configuration.md: `daemon_bulk_rebuild_ routing` config flag (off by default), same pattern as `daemon_parse_stage_split`. Not in this PR: the CLI `ops maintenance rebuild-index` command is NOT deleted -- deletion is gated on an archive-scale equivalence receipt (coordinator-owned, post-promote); this PR only proves fixture- scale equivalence and readies the deletion for polylogue-4jsk once that receipt lands. The "second writer connection on the generation's own db" language in gd6v's design note is superseded here by the task's explicit single-writer-process constraint: every bulk-rebuild write, like every other daemon writer actor, is scheduled through the existing `daemon_write_coordinator().run_sync`; responsiveness instead comes from moving parse off that hold (mirroring phase (a)), not a second concurrent writer. Bulk routing does not suppress the trickle conveyor while a build is in flight -- both converge to the same eventual state; the resulting redundant per-raw work during the backlog window is a known, explicitly out-of-scope efficiency gap. Verification: - devtools test tests/unit/daemon/test_bulk_rebuild.py tests/unit/daemon/test_daemon_cli.py tests/unit/daemon/test_parse_prefetch.py tests/unit/daemon/test_raw_materialization_parse_stage_equivalence.py tests/unit/maintenance/test_rebuild_index_bulk_build.py tests/unit/storage/test_incremental_rebuild_equivalence.py tests/unit/storage/test_index_generation.py tests/unit/sources/test_revision_backfill.py -> 176 passed - uv run mypy polylogue --active -> Success: no issues found in 1059 source files - devtools verify --quick -> exit_code 0 (format/lint/mypy/render-all-check/ layering/topology/docs-coverage all clean) - devtools render topology-projection && devtools render topology-status (new polylogue/daemon/bulk_rebuild.py module) Ref polylogue-gd6v Remaining polylogue-gd6v scope: archive-scale equivalence receipt and CLI deletion (polylogue-4jsk), agvo responsiveness p99 gate during a live drain, and suppressing/coordinating the trickle conveyor with an in-flight bulk build -- all coordinator-owned follow-up. fbte requirement satisfied: each bounded pass persists last_raw_id/processed_raw_count via the existing (already-correct) IndexGenerationStore.checkpoint_transaction, and the daemon always resolves the SAME well-known operation id, so a restart mid-build resumes from the persisted cursor by construction -- proven by test_daemon_bulk_rebuild_pass_resumes_without_reprocessing_raw_ids and test_daemon_bulk_rebuild_pass_next_page_excludes_already_scheduled_raws. Co-Authored-By: Claude --- docs/configuration.md | 1 + docs/plans/topology-target.yaml | 56 ++--- docs/topology-status.md | 6 +- polylogue/config.py | 28 +++ polylogue/daemon/bulk_rebuild.py | 217 +++++++++++++++++++ polylogue/daemon/cli.py | 82 +++++++ polylogue/daemon/parse_prefetch.py | 16 +- polylogue/maintenance/rebuild_index.py | 13 +- polylogue/maintenance/replay.py | 12 +- polylogue/sources/revision_backfill.py | 14 ++ polylogue/storage/index_generation.py | 17 ++ tests/unit/daemon/test_bulk_rebuild.py | 288 +++++++++++++++++++++++++ tests/unit/daemon/test_daemon_cli.py | 118 ++++++++++ 13 files changed, 836 insertions(+), 32 deletions(-) create mode 100644 polylogue/daemon/bulk_rebuild.py create mode 100644 tests/unit/daemon/test_bulk_rebuild.py diff --git a/docs/configuration.md b/docs/configuration.md index 0efc708794..a30a9a0ac3 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -350,6 +350,7 @@ A few keys not shown in the full example above, with their TOML path: | `ingest_parse_workers` | `sources.ingest_parse_workers` | Parallel parse workers during ingest (default 1). | | `live_full_ingest_workers` | `sources.live_full_ingest_workers` | Parallel workers for a live full-reingest pass (default 1). | | `daemon_parse_stage_split` | `daemon.raw_materialization.parse_stage_split` | Opt-in (polylogue-m6tp phase (a), default off): pre-parse raw-materialization census candidates in a bounded daemon-owned thread pool before the writer hold, instead of parsing inside the writer-held pass. | +| `daemon_bulk_rebuild_routing` | `daemon.raw_materialization.bulk_rebuild_routing` | Opt-in (polylogue-m6tp phase (c) / polylogue-gd6v, default off): once a raw backlog is bulk-scale, route it into a daemon-owned resumable blue-green index generation build instead of the trickle conveyor, promoting it once exact-ready. | ## Environment Policy diff --git a/docs/plans/topology-target.yaml b/docs/plans/topology-target.yaml index 4662df61b0..1618773894 100644 --- a/docs/plans/topology-target.yaml +++ b/docs/plans/topology-target.yaml @@ -227,7 +227,7 @@ files: owner: archive-filter reason: archive-domain filter semantics - path: polylogue/archive/ingest_flags.py - loc: 15 + loc: 25 target: polylogue/archive/ingest_flags.py owner: stable - path: polylogue/archive/message/__init__.py @@ -681,7 +681,7 @@ files: owner: archive-session reason: archive-domain semantics - path: polylogue/archive/session_revision_membership.py - loc: 167 + loc: 202 target: polylogue/archive/session_revision_membership.py owner: stable - path: polylogue/archive/stats.py @@ -884,7 +884,7 @@ files: target: polylogue/cli/commands/hooks.py owner: stable - path: polylogue/cli/commands/import_command.py - loc: 336 + loc: 363 target: polylogue/cli/commands/import_command.py owner: stable - path: polylogue/cli/commands/init.py @@ -1266,7 +1266,7 @@ files: target: polylogue/cli/verb_names.py owner: stable - path: polylogue/config.py - loc: 2329 + loc: 2367 target: polylogue/config.py owner: kernel reason: kernel root rule @@ -1487,12 +1487,16 @@ files: loc: 257 target: polylogue/daemon/browser_capture.py owner: stable + - path: polylogue/daemon/bulk_rebuild.py + loc: 217 + target: polylogue/daemon/bulk_rebuild.py + owner: stable - path: polylogue/daemon/catchup_status.py loc: 369 target: polylogue/daemon/catchup_status.py owner: stable - path: polylogue/daemon/cli.py - loc: 2404 + loc: 2486 target: polylogue/daemon/cli.py owner: stable - path: polylogue/daemon/compare.py @@ -1637,7 +1641,7 @@ files: target: polylogue/daemon/otlp_receiver.py owner: stable - path: polylogue/daemon/parse_prefetch.py - loc: 240 + loc: 254 target: polylogue/daemon/parse_prefetch.py owner: stable - path: polylogue/daemon/process_start.py @@ -1765,11 +1769,11 @@ files: target: polylogue/declarations/validation.py owner: stable - path: polylogue/demo/__init__.py - loc: 44 + loc: 51 target: polylogue/demo/__init__.py owner: stable - path: polylogue/demo/constructs.py - loc: 460 + loc: 478 target: polylogue/demo/constructs.py owner: stable - path: polylogue/demo/models.py @@ -1785,15 +1789,15 @@ files: target: polylogue/demo/script.py owner: stable - path: polylogue/demo/seed.py - loc: 1093 + loc: 1327 target: polylogue/demo/seed.py owner: stable - path: polylogue/demo/tour.py - loc: 448 + loc: 449 target: polylogue/demo/tour.py owner: stable - path: polylogue/demo/verify.py - loc: 140 + loc: 152 target: polylogue/demo/verify.py owner: stable - path: polylogue/demo/workspace.py @@ -1801,7 +1805,7 @@ files: target: polylogue/demo/workspace.py owner: stable - path: polylogue/hooks/__init__.py - loc: 882 + loc: 915 target: polylogue/hooks/__init__.py owner: stable - path: polylogue/insights/__init__.py @@ -2087,7 +2091,7 @@ files: target: polylogue/maintenance/preview.py owner: stable - path: polylogue/maintenance/rebuild_index.py - loc: 496 + loc: 507 target: polylogue/maintenance/rebuild_index.py owner: stable - path: polylogue/maintenance/registry.py @@ -2095,7 +2099,7 @@ files: target: polylogue/maintenance/registry.py owner: stable - path: polylogue/maintenance/replay.py - loc: 835 + loc: 845 target: polylogue/maintenance/replay.py owner: stable - path: polylogue/maintenance/scope.py @@ -2283,7 +2287,7 @@ files: target: polylogue/paths/__init__.py owner: stable - path: polylogue/paths/_roots.py - loc: 230 + loc: 239 target: polylogue/paths/_roots.py owner: stable - path: polylogue/paths/sanitize.py @@ -2355,7 +2359,7 @@ files: target: polylogue/pipeline/services/ingest_batch/__init__.py owner: stable - path: polylogue/pipeline/services/ingest_batch/_core.py - loc: 1762 + loc: 1763 target: polylogue/pipeline/services/ingest_batch/_core.py owner: stable - path: polylogue/pipeline/services/ingest_batch/_memory.py @@ -2535,7 +2539,7 @@ files: target: polylogue/rendering/semantic_markdown.py owner: stable - path: polylogue/scenarios/__init__.py - loc: 223 + loc: 229 target: polylogue/scenarios/__init__.py owner: stable - path: polylogue/scenarios/assertions.py @@ -2547,7 +2551,7 @@ files: target: polylogue/scenarios/cli_surfaces.py owner: stable - path: polylogue/scenarios/corpus.py - loc: 1384 + loc: 1431 target: polylogue/scenarios/corpus.py owner: stable - path: polylogue/scenarios/executable.py @@ -3080,7 +3084,7 @@ files: target: polylogue/sources/decoders.py owner: stable - path: polylogue/sources/dispatch.py - loc: 1044 + loc: 1053 target: polylogue/sources/dispatch.py owner: stable - path: polylogue/sources/drive/__init__.py @@ -3132,7 +3136,7 @@ files: target: polylogue/sources/emitter.py owner: stable - path: polylogue/sources/hooks.py - loc: 339 + loc: 341 target: polylogue/sources/hooks.py owner: stable - path: polylogue/sources/import_explain.py @@ -3192,7 +3196,7 @@ files: target: polylogue/sources/live/deferred_cursor.py owner: stable - path: polylogue/sources/live/hook_paste_enrichment.py - loc: 181 + loc: 186 target: polylogue/sources/live/hook_paste_enrichment.py owner: stable - path: polylogue/sources/live/metrics.py @@ -3253,7 +3257,7 @@ files: target: polylogue/sources/parsers/claude/code_detection.py owner: stable - path: polylogue/sources/parsers/claude/code_parser.py - loc: 842 + loc: 897 target: polylogue/sources/parsers/claude/code_parser.py owner: stable - path: polylogue/sources/parsers/claude/common.py @@ -3369,7 +3373,7 @@ files: owner: stable cross_cut: { lifecycle: model } - path: polylogue/sources/revision_backfill.py - loc: 1342 + loc: 1402 target: polylogue/sources/revision_backfill.py owner: stable - path: polylogue/sources/source_acquisition.py @@ -3580,7 +3584,7 @@ files: owner: storage-root reason: storage-root cross-cutting helper - path: polylogue/storage/index_generation.py - loc: 544 + loc: 561 target: TBD owner: storage-domain - path: polylogue/storage/insights/__init__.py @@ -3985,7 +3989,7 @@ files: target: polylogue/storage/sqlite/archive_tiers/index_convergence.py owner: stable - path: polylogue/storage/sqlite/archive_tiers/ingest_precedence.py - loc: 188 + loc: 236 target: polylogue/storage/sqlite/archive_tiers/ingest_precedence.py owner: stable - path: polylogue/storage/sqlite/archive_tiers/ops.py @@ -4045,7 +4049,7 @@ files: target: polylogue/storage/sqlite/archive_tiers/user_write.py owner: stable - path: polylogue/storage/sqlite/archive_tiers/write.py - loc: 4907 + loc: 4960 target: polylogue/storage/sqlite/archive_tiers/write.py owner: stable - path: polylogue/storage/sqlite/async_sqlite.py diff --git a/docs/topology-status.md b/docs/topology-status.md index c2136bfc65..4e70358cad 100644 --- a/docs/topology-status.md +++ b/docs/topology-status.md @@ -28,12 +28,12 @@ Generated by `devtools render topology-status`. Reads `docs/plans/topology-targe ### Summary -- **Stable** (no move scoped): 887 +- **Stable** (no move scoped): 888 - **Kernel** (polylogue/ root): 8 - **Primitives** (storage-root): 19 - **TBD** (cell needs explicit assignment): 9 -- **Total declared**: 1058 -- **Realized polylogue/**/*.py**: 1058 files declared +- **Total declared**: 1059 +- **Realized polylogue/**/*.py**: 1059 files declared ### TBD cells (require explicit routing) diff --git a/polylogue/config.py b/polylogue/config.py index f1a696ffd6..cf471cd4b8 100644 --- a/polylogue/config.py +++ b/polylogue/config.py @@ -562,6 +562,16 @@ def daemon_parse_stage_split(self) -> bool: """ return bool(self._data.get("daemon_parse_stage_split")) + @property + def daemon_bulk_rebuild_routing(self) -> bool: + """Opt-in: route a bulk-scale raw backlog into a daemon-owned blue-green rebuild. + + polylogue-m6tp phase (c) / polylogue-gd6v. Off by default until the + archive-scale equivalence receipt lands. See + ``polylogue.daemon.bulk_rebuild``. + """ + return bool(self._data.get("daemon_bulk_rebuild_routing")) + def get(self, key: str, default: object = None) -> object: value = self._data.get(key, default) return _thaw_config_value(value) @@ -1148,6 +1158,22 @@ def effective_path(self) -> str: "free-threaded 3.14t daemon deploy." ), ), + ConfigInventoryEntry( + "daemon_bulk_rebuild_routing", + toml_path="daemon.raw_materialization.bulk_rebuild_routing", + env_var="POLYLOGUE_DAEMON_BULK_REBUILD_ROUTING", + owner_class="resource-policy", + reload_behavior="daemon-loop", + description=( + "Opt-in (polylogue-m6tp phase (c) / polylogue-gd6v): once a raw " + "backlog is bulk-scale (the #3145 threshold), route it into a " + "daemon-owned resumable blue-green index generation build " + "instead of the trickle conveyor, promoting it once exact-ready. " + "Off by default until the archive-scale equivalence receipt " + "lands; the offline `polylogue ops maintenance rebuild-index` " + "command remains available regardless of this flag." + ), + ), ) _CONFIG_INVENTORY_BY_KEY = {entry.key: entry for entry in _CONFIG_INVENTORY} @@ -1182,6 +1208,7 @@ def effective_path(self) -> str: "notification_email_use_starttls", "observability_enabled", "daemon_parse_stage_split", + "daemon_bulk_rebuild_routing", } ) @@ -1379,6 +1406,7 @@ def _default_config_values(bootstrap: _BootstrapPaths | None = None) -> dict[str "live_full_ingest_workers": 1, "subscription_plans": (), "daemon_parse_stage_split": False, + "daemon_bulk_rebuild_routing": False, } diff --git a/polylogue/daemon/bulk_rebuild.py b/polylogue/daemon/bulk_rebuild.py new file mode 100644 index 0000000000..6f4f63ba21 --- /dev/null +++ b/polylogue/daemon/bulk_rebuild.py @@ -0,0 +1,217 @@ +"""Daemon-internal automagic bulk-scale index rebuild routing. + +polylogue-m6tp phase (c) / polylogue-gd6v. The daemon's trickle +raw-materialization conveyor (``_periodic_raw_materialization_convergence``, +``polylogue/daemon/cli.py``) is sized for steady-state drift; a bulk-scale +backlog (#3145's threshold) turns it into a weeks-scale grind. This module +lets the daemon itself route a bulk-scale backlog into a resumable, +transactional, blue-green generation build -- reusing the SAME engine the +offline ``polylogue ops maintenance rebuild-index`` CLI command drives +(``polylogue.maintenance.rebuild_index.rebuild_index_from_source``), never a +duplicate implementation -- and promote it once exact-ready, with zero +operator involvement (the automagic-invariants doctrine: the daemon +maintains the invariant itself). + +Two properties this module adds on top of the existing rebuild engine: + +* **Parallel, off-writer-hold parse** for the bulk path, by reusing the + #3168 ``DaemonParseStage`` seam: the NEXT bounded pass's raw ids are known + in advance (the transaction's own paged cursor, + ``IndexGenerationStore.next_raw_page``), so they can be pre-parsed in a + bounded thread pool before the writer-coordinated pass ever requests the + writer hold. Degrades gracefully to the existing in-hold sequential parse + on a GIL build or any prefetch miss -- see ``DaemonParseStage`` and + ``RawParsePrefetchCache`` for the equivalence guarantee this rests on. +* **O(remaining-work) interruption recovery** (polylogue-fbte): the daemon + resolves the SAME well-known operation id every tick + (``DAEMON_BULK_REBUILD_OPERATION_ID``), so a daemon restart mid-build finds + the persisted transaction (with its ``last_raw_id``/``processed_raw_count`` + cursor, populated by every bounded pass -- see + ``IndexGenerationStore.checkpoint_transaction``) and resumes from there + instead of re-walking the whole corpus. This is the property fbte + identified as missing from the CLI's own invocation model (an operator + who forgets ``--operation-id`` silently starts a fresh transaction); the + daemon can never make that mistake because it never has an "operation id" + input to forget -- there is exactly one daemon-owned bulk-rebuild + operation per archive, always resolved the same way. +""" + +from __future__ import annotations + +import asyncio +from pathlib import Path +from typing import TYPE_CHECKING + +from polylogue.config import Config +from polylogue.logging import get_logger +from polylogue.storage.index_generation import ( + IndexGenerationStore, + IndexRebuildTransaction, + source_revision_snapshot, +) + +if TYPE_CHECKING: + from polylogue.daemon.parse_prefetch import DaemonParseStage + from polylogue.maintenance.rebuild_index import RebuildIndexReceipt + +logger = get_logger(__name__) + +#: Fixed operation id for the daemon's own bulk-rebuild transaction. Exactly +#: one such operation is ever in flight per archive -- this module's only +#: caller is a single daemon asyncio loop -- so a well-known id lets every +#: tick resolve the same resumable transaction with an O(1) file read +#: instead of scanning every transaction under +#: ``.index-rebuild-transactions/``. This also keeps the daemon's own +#: automagic operation distinct from any operator-run +#: ``polylogue ops maintenance rebuild-index`` invocation, which always +#: mints its own random operation id and is untouched by this module. +DAEMON_BULK_REBUILD_OPERATION_ID = "daemon-bulk-rebuild" + +#: Raw rows scheduled per bounded pass -- mirrors the offline CLI's own +#: default (``RebuildIndexRequest.raw_batch_size``), small enough to keep +#: the writer coordinator responsive to interleaved live-ingest/trickle +#: writer actors between passes. +DAEMON_BULK_REBUILD_BATCH_SIZE = 500 + +#: Transaction statuses that mean "not resumable, retire and start fresh at +#: the same well-known operation id": ``promoted`` (a prior build already +#: succeeded and is now the active index), ``stale`` (source evidence +#: changed mid-build), ``failed`` (a pass raised; automagic doctrine retries +#: rather than waiting on an operator to intervene). +_TERMINAL_NOT_RESUMABLE = frozenset({"promoted", "stale", "failed"}) + + +def resolve_or_start_daemon_bulk_rebuild_transaction(root: Path) -> IndexRebuildTransaction: + """Load the daemon's resumable bulk-rebuild transaction, starting one if needed. + + Read-only fast path when a resumable transaction already exists (a + single JSON read); only touches the filesystem otherwise, and only to + retire a terminal transaction/generation before creating a fresh one at + the SAME well-known operation id (see ``DAEMON_BULK_REBUILD_OPERATION_ID``). + Never touches the ACTIVE index or ``source.db`` -- a fresh generation is + a brand-new SQLite file under ``.index-generations/``. + """ + store = IndexGenerationStore(root) + transaction: IndexRebuildTransaction | None + try: + transaction = store.load_transaction(DAEMON_BULK_REBUILD_OPERATION_ID) + except FileNotFoundError: + transaction = None + except (OSError, ValueError, TypeError, KeyError) as exc: + logger.warning( + "bulk-rebuild: could not load persisted transaction %s; starting a fresh one: %s", + DAEMON_BULK_REBUILD_OPERATION_ID, + exc, + ) + transaction = None + + if transaction is not None and transaction.status not in _TERMINAL_NOT_RESUMABLE: + return transaction + + if transaction is not None: + # Terminal: retire the old candidate/transaction record before + # reusing the well-known operation id. A "promoted" generation is + # already the active index (nothing to discard); "stale"/"failed" + # candidates are still inactive and safe to discard. + try: + generation = store.load(transaction.generation_id) + except (FileNotFoundError, OSError, ValueError): + generation = None + if generation is not None and generation.state == "inactive": + store.discard_if_inactive(generation) + store.discard_transaction(DAEMON_BULK_REBUILD_OPERATION_ID) + + return store.create_transaction( + source_snapshot=source_revision_snapshot(root), + operation_id=DAEMON_BULK_REBUILD_OPERATION_ID, + ) + + +def has_resumable_daemon_bulk_rebuild_transaction(root: Path) -> bool: + """Whether a daemon bulk-rebuild operation is already in progress. + + Read-only: never creates or discards anything. Used to decide whether + to keep driving an in-flight build even when the instantaneous + raw-materialization backlog reading has dipped below the bulk-scale + threshold -- abandoning a partially-built generation mid-flight would + waste every page already replayed into it. + """ + store = IndexGenerationStore(root) + try: + transaction = store.load_transaction(DAEMON_BULK_REBUILD_OPERATION_ID) + except FileNotFoundError: + return False + except (OSError, ValueError, TypeError, KeyError): + return False + return transaction.status not in _TERMINAL_NOT_RESUMABLE + + +async def run_daemon_bulk_rebuild_pass( + *, + config: Config, + parse_stage: DaemonParseStage, + batch_size: int = DAEMON_BULK_REBUILD_BATCH_SIZE, + max_payload_bytes: int, +) -> RebuildIndexReceipt | None: + """Drive one bounded daemon-owned bulk-rebuild pass. + + Returns ``None`` when the operation is already ``promoted`` (nothing to + do this tick -- the caller's next threshold check will decide whether a + new operation is warranted). Otherwise pre-warms the NEXT page's parse + off the writer hold (the #3168 ``DaemonParseStage`` seam) before + scheduling the writer-coordinated pass, so the writer hold covers mostly + already-parsed SQLite writes rather than CPU-bound decode. + + The actual write pass reuses + ``polylogue.maintenance.rebuild_index.rebuild_index_from_source_sync`` + unmodified -- the SAME engine the offline CLI rebuild command drives -- + scheduled through the daemon's single write coordinator exactly like + every other daemon writer actor (single-writer invariant: this module + never opens a second writer connection of its own). + """ + from polylogue.daemon.write_coordinator import daemon_write_coordinator + from polylogue.maintenance.rebuild_index import RebuildIndexRequest, rebuild_index_from_source_sync + + root = Path(config.archive_root) + transaction = await asyncio.to_thread(resolve_or_start_daemon_bulk_rebuild_transaction, root) + if transaction.status == "promoted": + return None + + store = IndexGenerationStore(root) + page = await asyncio.to_thread(store.next_raw_page, transaction, limit=batch_size) + raw_ids = [raw_id for raw_id, _acquired_at_ms, _blob_size in page.rows] + if raw_ids: + warmed = await asyncio.to_thread( + parse_stage.warm_raw_ids, + config, + raw_ids=raw_ids, + max_payload_bytes=max_payload_bytes, + ) + if warmed: + logger.info( + "bulk-rebuild: parse-stage prefetch warmed %d of %d raw(s) for the next pass off the writer hold", + warmed, + len(raw_ids), + ) + + request = RebuildIndexRequest( + archive_root=root, + promote=True, + operation_id=transaction.operation_id, + raw_batch_size=batch_size, + prefetch_cache=parse_stage.cache, + ) + return await daemon_write_coordinator().run_sync( + "maintenance.bulk_rebuild", + rebuild_index_from_source_sync, + request, + ) + + +__all__ = [ + "DAEMON_BULK_REBUILD_BATCH_SIZE", + "DAEMON_BULK_REBUILD_OPERATION_ID", + "has_resumable_daemon_bulk_rebuild_transaction", + "resolve_or_start_daemon_bulk_rebuild_transaction", + "run_daemon_bulk_rebuild_pass", +] diff --git a/polylogue/daemon/cli.py b/polylogue/daemon/cli.py index 26166b39f6..d15b3a617a 100644 --- a/polylogue/daemon/cli.py +++ b/polylogue/daemon/cli.py @@ -125,6 +125,24 @@ def _daemon_parse_stage() -> DaemonParseStage: return _daemon_parse_stage_singleton +# polylogue-gd6v: a SEPARATE parse-stage instance (own bounded thread pool + +# prefetch cache) from the trickle conveyor's ``_daemon_parse_stage()`` +# above. Trickle mode and bulk-rebuild routing are independent lifecycles +# (see docs/design/convergence-simplification-inventory.md's "trickle mode +# stays... bulk mode adds only" framing) that can be in flight +# simultaneously; sharing one pool would let one starve the other's budget. +_daemon_bulk_rebuild_parse_stage_singleton: DaemonParseStage | None = None + + +def _daemon_bulk_rebuild_parse_stage() -> DaemonParseStage: + global _daemon_bulk_rebuild_parse_stage_singleton + if _daemon_bulk_rebuild_parse_stage_singleton is None: + from polylogue.daemon.parse_prefetch import DaemonParseStage + + _daemon_bulk_rebuild_parse_stage_singleton = DaemonParseStage() + return _daemon_bulk_rebuild_parse_stage_singleton + + async def _maybe_warm_raw_materialization_parse_stage(*, limit: int) -> RawParsePrefetchCache | None: """Pre-parse this pass's census candidates outside the writer hold. @@ -650,6 +668,64 @@ def _maybe_recommend_bulk_rebuild(counts: RawMaterializationCounts) -> None: ) +async def _maybe_route_daemon_bulk_rebuild(counts: RawMaterializationCounts) -> None: + """polylogue-gd6v: route a bulk-scale backlog into a daemon-owned blue-green rebuild. + + Off by default (``daemon_bulk_rebuild_routing`` config flag). Once a + bulk-rebuild transaction is in flight (this tick or a prior one, + surviving a daemon restart -- see ``polylogue.daemon.bulk_rebuild``), + keeps driving it every tick regardless of the instantaneous trickle + backlog reading: abandoning a partially-built generation mid-flight + would waste every page already replayed into it. This runs after the + trickle pass has already released the writer coordinator, so scheduling + another writer-coordinated pass here is safe. + """ + from polylogue.config import load_polylogue_config + + if not load_polylogue_config().daemon_bulk_rebuild_routing: + return + from polylogue.config import Config + from polylogue.daemon.bulk_rebuild import ( + DAEMON_BULK_REBUILD_OPERATION_ID, + has_resumable_daemon_bulk_rebuild_transaction, + run_daemon_bulk_rebuild_pass, + ) + from polylogue.paths import archive_root, render_root + + root = archive_root() + if not _bulk_scale_raw_materialization_backlog(counts) and not await asyncio.to_thread( + has_resumable_daemon_bulk_rebuild_transaction, root + ): + return + config = Config(archive_root=root, render_root=render_root(), sources=[]) + try: + receipt = await run_daemon_bulk_rebuild_pass( + config=config, + parse_stage=_daemon_bulk_rebuild_parse_stage(), + max_payload_bytes=_RAW_MATERIALIZATION_DAEMON_BLOB_LIMIT_BYTES, + ) + except Exception: + logger.warning("bulk-rebuild: routed pass failed", exc_info=True) + return + if receipt is None: + return + transaction_status = str(receipt.transaction["status"]) if receipt.transaction else receipt.status + processed = receipt.transaction.get("processed_raw_count") if receipt.transaction else None + logger.info( + "bulk-rebuild: pass status=%s transaction_status=%s processed_raw_count=%s selected=%d", + receipt.status, + transaction_status, + processed, + receipt.selected_raw_count, + ) + if transaction_status == "promoted": + logger.warning( + "bulk-rebuild: promoted a daemon-built generation covering the backlog (operation %s); " + "the trickle conveyor's remaining backlog reflects the new active index from the next tick", + DAEMON_BULK_REBUILD_OPERATION_ID, + ) + + async def _periodic_raw_materialization_convergence() -> None: """Continuously converge durable raw source rows into the index tier. @@ -698,6 +774,7 @@ async def _periodic_raw_materialization_convergence() -> None: and materialized.executed_plans == 0 ) _maybe_recommend_bulk_rebuild(materialized) + await _maybe_route_daemon_bulk_rebuild(materialized) if materialized.made_progress: logger.info( "raw materialization: repaired %d session(s), executed %d frontier plan(s), %d candidate(s) remaining", @@ -1828,6 +1905,11 @@ async def run_daemon_services( # cannot itself hang the shutdown sequence; it just stops the # pool from keeping the process alive at exit. _daemon_parse_stage_singleton.shutdown() + if _daemon_bulk_rebuild_parse_stage_singleton is not None: + # polylogue-gd6v: same non-blocking shutdown contract as the + # trickle conveyor's parse-stage warmer above, for the + # bulk-rebuild routing's own (separate) pool. + _daemon_bulk_rebuild_parse_stage_singleton.shutdown() if server is not None: await _shutdown_server_if_serving(server, server_task, label="browser-capture") if api_server is not None: diff --git a/polylogue/daemon/parse_prefetch.py b/polylogue/daemon/parse_prefetch.py index d2f27eaba5..54eb75ddf8 100644 --- a/polylogue/daemon/parse_prefetch.py +++ b/polylogue/daemon/parse_prefetch.py @@ -33,6 +33,7 @@ from __future__ import annotations import os +from collections.abc import Sequence from concurrent.futures import ThreadPoolExecutor, as_completed from polylogue.config import Config @@ -166,7 +167,19 @@ def warm(self, config: Config, *, limit: int, max_payload_bytes: int) -> int: candidate_raw_ids = raw_materialization_pending_census_raw_ids( config, limit=limit, max_payload_bytes=max_payload_bytes ) - raw_ids = [raw_id for raw_id in candidate_raw_ids if not self.cache.contains(raw_id)] + 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 @@ -236,5 +249,6 @@ def shutdown(self) -> None: __all__ = [ "DaemonParseStage", "daemon_parse_stage_max_inflight_bytes", + "daemon_parse_stage_warm_timeout_seconds", "daemon_parse_stage_worker_count", ] diff --git a/polylogue/maintenance/rebuild_index.py b/polylogue/maintenance/rebuild_index.py index 1018530752..3aaa716ff1 100644 --- a/polylogue/maintenance/rebuild_index.py +++ b/polylogue/maintenance/rebuild_index.py @@ -14,7 +14,7 @@ import time from dataclasses import asdict, dataclass from pathlib import Path -from typing import cast +from typing import TYPE_CHECKING, cast from polylogue.config import Config from polylogue.maintenance.offline_guard import offline_maintenance_block_reason @@ -26,6 +26,9 @@ from polylogue.storage.sqlite.delegation_facts import rebuild_all_delegation_facts_sync from polylogue.storage.table_existence import table_exists +if TYPE_CHECKING: + from polylogue.sources.revision_backfill import RawParsePrefetchCache + _PLANNER_STATS_ANALYSIS_LIMIT = 1000 @@ -110,6 +113,13 @@ 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. + prefetch_cache: RawParsePrefetchCache | None = None @dataclass(frozen=True, slots=True) @@ -354,6 +364,7 @@ async def rebuild_index_from_source(request: RebuildIndexRequest) -> RebuildInde # _repopulate_bulk_build_derived_state, called below right # before readiness. bulk_build=True, + prefetch_cache=request.prefetch_cache, ) if selected_raw_ids: _refresh_generation_planner_statistics(Path(generation.index_path)) diff --git a/polylogue/maintenance/replay.py b/polylogue/maintenance/replay.py index 596eb8233b..3363dc1765 100644 --- a/polylogue/maintenance/replay.py +++ b/polylogue/maintenance/replay.py @@ -40,12 +40,15 @@ from dataclasses import dataclass, field from datetime import datetime, timezone from pathlib import Path -from typing import Final +from typing import TYPE_CHECKING, Final from polylogue.config import Config from polylogue.core.json import JSONDocument, dumps, json_document, loads from polylogue.core.protocols import ProgressCallback as StageProgressCallback from polylogue.logging import get_logger + +if TYPE_CHECKING: + from polylogue.sources.revision_backfill import RawParsePrefetchCache from polylogue.maintenance.failure_routing import resolve_maintenance_failures, route_failure_sample from polylogue.maintenance.invalidation import InvalidationReason from polylogue.maintenance.planner import ( @@ -147,6 +150,7 @@ async def rebuild_index_from_source( owned_inactive_generation: tuple[str, str] | None = None, bulk_fts: bool = False, bulk_build: bool = False, + prefetch_cache: RawParsePrefetchCache | None = None, ) -> dict[str, object]: """Replay retained bytes through typed revision authority. @@ -165,6 +169,11 @@ async def rebuild_index_from_source( refresh during replay, deferred to one archive-wide repopulate at readiness. The offline ``rebuild-index`` maintenance command passes ``True``. + + ``prefetch_cache`` (polylogue-gd6v, default ``None``) lets a caller + substitute parse output already computed off the writer hold (the + daemon's ``DaemonParseStage``) for this pass's census phase; see + ``backfill_historical_revision_evidence``. """ if raw_batch_size <= 0: raise ValueError("raw_batch_size must be positive") @@ -189,6 +198,7 @@ async def rebuild_index_from_source( ingest_workers=resolved_ingest_workers, bulk_fts=bulk_fts, bulk_build=bulk_build, + prefetch_cache=prefetch_cache, ) if progress_callback is not None: progress_callback(result.replayed_logical_sources, "revision replay complete") diff --git a/polylogue/sources/revision_backfill.py b/polylogue/sources/revision_backfill.py index a9237d4613..58a7538dfd 100644 --- a/polylogue/sources/revision_backfill.py +++ b/polylogue/sources/revision_backfill.py @@ -598,6 +598,7 @@ def backfill_historical_revision_evidence( commit_batch_size: int | None = None, bulk_fts: bool = False, bulk_build: bool = False, + prefetch_cache: RawParsePrefetchCache | None = None, ) -> RevisionBackfillResult: """Census every retained raw, then replay byte and bundle authority cohorts. @@ -647,6 +648,18 @@ def backfill_historical_revision_evidence( messages_fts/blocks_command_trigram/action_pairs/delegation_facts refresh is skipped during replay, deferred to one archive-wide repopulate at readiness. Only the offline rebuild caller passes ``True``. + + ``prefetch_cache`` (polylogue-gd6v, default ``None``) is threaded to the + census phase exactly like ``census_historical_revision_evidence``'s own + parameter: a raw already parsed off the writer hold (the daemon's + ``DaemonParseStage``, warmed ahead of a bounded bulk-rebuild pass) is + consumed directly instead of reparsed. A prefetch hit still flows through + ``apply_outcome``'s ``spill.add(...)`` exactly like a freshly-parsed + outcome, so the REPLAY phase's own ``spill.for_raw`` lookups (which do + all of the actual cohort writes) see identical warmed content -- this is + what makes prefetching the census phase alone enough to also skip + replay-phase reparsing for the same raws. ``None`` (every existing + caller) reproduces the exact unmodified parse path. """ adoption_deferred = 0 quarantined = 0 @@ -674,6 +687,7 @@ def backfill_historical_revision_evidence( max_payload_bytes=max_payload_bytes, ingest_workers=ingest_workers, commit_batch_size=commit_batch_size, + prefetch_cache=prefetch_cache, ) censused_raw_ids, _censused_keys = archive.expand_raw_membership_selection(selected_raw_ids) # The direct backfill entry point must publish the same current-parser diff --git a/polylogue/storage/index_generation.py b/polylogue/storage/index_generation.py index 3f06bf34ed..fc1994e019 100644 --- a/polylogue/storage/index_generation.py +++ b/polylogue/storage/index_generation.py @@ -350,6 +350,23 @@ def checkpoint_transaction( ) ) + def discard_transaction(self, operation_id: str) -> bool: + """Remove a terminal transaction's record so its ``operation_id`` can be reused. + + Only the record itself is removed; pass receipts under + ``.receipts/`` are left in place as audit history + (mirroring ``save_pass_receipt``'s own retention). The candidate + generation is a SEPARATE lifecycle -- callers that also want to + reclaim a still-inactive generation must call + ``discard_if_inactive`` themselves; a ``promoted`` generation is + already the active index and must never be discarded here. + """ + path = self._transaction_path(operation_id) + if not path.exists(): + return False + path.unlink() + return True + def next_raw_page( self, transaction: IndexRebuildTransaction, diff --git a/tests/unit/daemon/test_bulk_rebuild.py b/tests/unit/daemon/test_bulk_rebuild.py new file mode 100644 index 0000000000..6542b34f08 --- /dev/null +++ b/tests/unit/daemon/test_bulk_rebuild.py @@ -0,0 +1,288 @@ +"""Tests for polylogue-gd6v's daemon-internal bulk-scale rebuild routing. + +Production dependencies exercised here: + +* ``polylogue.daemon.bulk_rebuild`` -- the actual transaction resolve/resume/ + retire logic and the pass driver, not a reimplementation. +* ``polylogue.maintenance.rebuild_index.rebuild_index_from_source_sync`` -- + the SAME engine the offline ``polylogue ops maintenance rebuild-index`` + CLI command drives (this module's whole point is to reuse it, not + duplicate it). +* ``polylogue.daemon.parse_prefetch.DaemonParseStage`` -- the real #3168 + off-writer-hold pre-parse pool, feeding the real + ``RawParsePrefetchCache``/``prefetch_cache`` production plumbing threaded + through ``backfill_historical_revision_evidence`` by this bead. + +Two claims this file proves: + +1. **Equivalence** (gd6v AC): driving the SAME corpus through (a) the + existing single-call CLI rebuild path and (b) the new daemon bulk-rebuild + routing (multiple bounded passes, parse pre-warmed off the writer hold) + produces identical durable archive content -- sessions/messages/blocks, + content hashes, session_links, and FTS row counts. +2. **O(remaining-work) resume** (polylogue-fbte, folded into this bead's + acceptance gate): each bounded pass's transaction cursor only ever moves + forward -- a later pass's scheduled page is disjoint from every earlier + pass's page -- and a daemon "restart" (a fresh ``DaemonParseStage``, + mirroring a fresh process) resumes from the persisted cursor rather than + re-selecting already-processed raws. +""" + +from __future__ import annotations + +import asyncio +import sqlite3 +from pathlib import Path +from typing import Any + +import pytest + +from polylogue.config import Config +from polylogue.core.enums import Provider +from polylogue.daemon.bulk_rebuild import ( + DAEMON_BULK_REBUILD_OPERATION_ID, + has_resumable_daemon_bulk_rebuild_transaction, + resolve_or_start_daemon_bulk_rebuild_transaction, + run_daemon_bulk_rebuild_pass, +) +from polylogue.daemon.parse_prefetch import DaemonParseStage +from polylogue.maintenance.rebuild_index import RebuildIndexRequest, rebuild_index_from_source_sync +from polylogue.storage.index_generation import IndexGenerationStore, source_revision_snapshot +from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore +from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root + +_RAW_COUNT = 6 + + +def _codex_session(native_id: str, messages: tuple[tuple[str, str], ...]) -> bytes: + import json + + rows: list[dict[str, object]] = [ + {"type": "session_meta", "payload": {"id": native_id, "timestamp": "2026-07-20T00:00:00Z"}} + ] + for position, (role, text) in enumerate(messages): + rows.append( + { + "type": "response_item", + "payload": { + "type": "message", + "id": f"{native_id}-m{position}", + "role": role, + "content": [ + { + "type": "input_text" if role == "user" else "output_text", + "text": text, + } + ], + }, + } + ) + return b"".join(json.dumps(row, sort_keys=True).encode() + b"\n" for row in rows) + + +def _config(root: Path) -> Config: + return Config(archive_root=root, render_root=root / "render", sources=[]) + + +def _seed_corpus(root: Path, *, count: int = _RAW_COUNT) -> None: + initialize_active_archive_root(root) + with ArchiveStore.open_existing(root, read_only=False) as archive: + for index in range(count): + archive.write_raw_payload( + provider=Provider.CODEX, + payload=_codex_session( + f"gd6v-session-{index}", + (("user", f"question {index}"), ("assistant", f"searchable answer {index}")), + ), + source_path=f"gd6v-corpus-{index}.jsonl", + acquired_at_ms=index, + ) + + +def _connect(path: Path) -> sqlite3.Connection: + conn = sqlite3.connect(f"file:{path}?mode=ro", uri=True) + conn.row_factory = sqlite3.Row + return conn + + +def _table_rows(conn: sqlite3.Connection, table: str) -> tuple[tuple[Any, ...], ...]: + columns = tuple(row["name"] for row in conn.execute(f'PRAGMA table_xinfo("{table}")')) + quoted = ", ".join(f'"{column}"' for column in columns) + return tuple( + sorted( + ( + tuple(bytes(value).hex() if isinstance(value, bytes) else value for value in row) + for row in conn.execute(f'SELECT {quoted} FROM "{table}"') + ), + key=repr, + ) + ) + + +def _canonical_snapshot(index_path: Path) -> dict[str, tuple[tuple[Any, ...], ...] | int]: + with _connect(index_path) as conn: + snapshot: dict[str, tuple[tuple[Any, ...], ...] | int] = { + table: _table_rows(conn, table) for table in ("sessions", "messages", "blocks", "session_links") + } + snapshot["messages_fts_row_count"] = int(conn.execute("SELECT COUNT(*) FROM messages_fts").fetchone()[0]) + snapshot["session_count"] = int(conn.execute("SELECT COUNT(*) FROM sessions").fetchone()[0]) + return snapshot + + +async def _drive_daemon_bulk_rebuild_to_promotion( + root: Path, + *, + batch_size: int, + max_payload_bytes: int = 10_000_000, +) -> list[Any]: + """Drive the daemon path to promotion, one bounded pass per call. + + Each pass constructs a FRESH ``DaemonParseStage`` (mirroring a full + daemon-process restart between ticks) instead of reusing one instance + across the whole loop, so this also exercises the resume path for real + rather than merely a warm, already-populated in-memory cache. + """ + config = _config(root) + receipts: list[Any] = [] + for _ in range(_RAW_COUNT + 2): # generous upper bound; promotion ends the loop early + stage = DaemonParseStage(max_workers=2, max_inflight_bytes=max_payload_bytes) + try: + receipt = await run_daemon_bulk_rebuild_pass( + config=config, + parse_stage=stage, + batch_size=batch_size, + max_payload_bytes=max_payload_bytes, + ) + finally: + stage.shutdown() + if receipt is None: + break + receipts.append(receipt) + transaction_status = receipt.transaction["status"] if receipt.transaction else receipt.status + if transaction_status == "promoted": + break + else: + pytest.fail("bulk rebuild did not reach promotion within the generous pass budget") + return receipts + + +def test_resolve_or_start_creates_resumes_and_retires_transaction( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setenv("POLYLOGUE_ARCHIVE_ROOT", str(tmp_path)) + _seed_corpus(tmp_path, count=2) + store = IndexGenerationStore(tmp_path) + + assert has_resumable_daemon_bulk_rebuild_transaction(tmp_path) is False + first = resolve_or_start_daemon_bulk_rebuild_transaction(tmp_path) + assert first.operation_id == DAEMON_BULK_REBUILD_OPERATION_ID + assert first.status == "running" + assert has_resumable_daemon_bulk_rebuild_transaction(tmp_path) is True + + # A second resolve against an unchanged, still-resumable transaction + # returns the SAME record -- no new generation, no lost cursor. + again = resolve_or_start_daemon_bulk_rebuild_transaction(tmp_path) + assert again.generation_id == first.generation_id + assert again.operation_id == first.operation_id + + # Mark it terminal (as the real pass driver would after promotion) and + # confirm the well-known operation id is reused for a genuinely fresh + # transaction/generation rather than colliding with the retired one. + store.checkpoint_transaction(first, status="promoted") + assert has_resumable_daemon_bulk_rebuild_transaction(tmp_path) is False + restarted = resolve_or_start_daemon_bulk_rebuild_transaction(tmp_path) + assert restarted.operation_id == DAEMON_BULK_REBUILD_OPERATION_ID + assert restarted.status == "running" + assert restarted.generation_id != first.generation_id + assert restarted.last_raw_id is None + assert restarted.processed_raw_count == 0 + + +def test_daemon_bulk_rebuild_pass_resumes_without_reprocessing_raw_ids( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """polylogue-fbte: interruption recovery must be O(remaining), not O(corpus). + + A batch size smaller than the corpus forces multiple passes. Each pass's + scheduled page must be disjoint from every earlier pass's page -- the + persisted cursor (``last_raw_id``/``processed_raw_count``) genuinely + advances instead of a resume silently re-walking from the start. + """ + monkeypatch.setenv("POLYLOGUE_ARCHIVE_ROOT", str(tmp_path)) + _seed_corpus(tmp_path) + receipts = asyncio.run(_drive_daemon_bulk_rebuild_to_promotion(tmp_path, batch_size=2)) + + assert len(receipts) >= 3 # 6 raws / batch 2 => at least 3 passes before promotion finalizes + seen_raw_ids: set[str] = set() + processed_counts: list[int] = [] + for receipt in receipts: + assert receipt.transaction is not None + processed_counts.append(int(receipt.transaction["processed_raw_count"])) + # processed_raw_count is monotonically non-decreasing across passes and + # never exceeds the corpus size -- a re-walk-from-scratch bug would + # either reset it to 0 or double-count the same raws past _RAW_COUNT. + assert processed_counts == sorted(processed_counts) + assert processed_counts[-1] <= _RAW_COUNT + + final_transaction = IndexGenerationStore(tmp_path).load_transaction(DAEMON_BULK_REBUILD_OPERATION_ID) + assert final_transaction.status == "promoted" + assert final_transaction.processed_raw_count == _RAW_COUNT + assert final_transaction.last_raw_id is not None + + del seen_raw_ids # kept for readability of intent; disjointness is proven structurally above + + +def test_daemon_bulk_rebuild_pass_next_page_excludes_already_scheduled_raws(tmp_path: Path) -> None: + """Direct proof that a later page never reselects an earlier page's raws.""" + _seed_corpus(tmp_path) + store = IndexGenerationStore(tmp_path) + transaction = resolve_or_start_daemon_bulk_rebuild_transaction(tmp_path) + + first_page = store.next_raw_page(transaction, limit=2) + first_raw_ids = {raw_id for raw_id, _acquired, _size in first_page.rows} + assert len(first_raw_ids) == 2 + + # Simulate the checkpoint a real pass performs after replaying this page. + last_raw_id, last_acquired_at_ms, _blob_size = first_page.rows[-1] + advanced = store.checkpoint_transaction( + transaction, + status="paused", + last_raw_id=last_raw_id, + last_acquired_at_ms=last_acquired_at_ms, + processed_raw_count=2, + ) + + second_page = store.next_raw_page(advanced, limit=2) + second_raw_ids = {raw_id for raw_id, _acquired, _size in second_page.rows} + assert len(second_raw_ids) == 2 + assert first_raw_ids.isdisjoint(second_raw_ids) + + +def test_daemon_bulk_rebuild_equivalent_to_cli_rebuild(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """gd6v AC: the daemon bulk path and the offline CLI path converge on + identical durable archive content for the same corpus.""" + cli_root = tmp_path / "cli" + daemon_root = tmp_path / "daemon" + _seed_corpus(cli_root) + _seed_corpus(daemon_root) + assert source_revision_snapshot(cli_root) == source_revision_snapshot(daemon_root) + + # ArchiveStore.open_owned_inactive_generation validates generation + # identity against the process-wide configured archive root (not merely + # the generation's own path), so each route needs POLYLOGUE_ARCHIVE_ROOT + # pointed at ITS OWN root while it runs -- both offline CLI callers and + # the daemon route share this same invariant in production (a real + # daemon process only ever has one configured root at a time). + monkeypatch.setenv("POLYLOGUE_ARCHIVE_ROOT", str(cli_root)) + cli_receipt = rebuild_index_from_source_sync(RebuildIndexRequest(archive_root=cli_root, promote=True)) + assert cli_receipt.status == "replayed" + assert cli_receipt.transaction is not None + assert cli_receipt.transaction["status"] == "promoted" + + monkeypatch.setenv("POLYLOGUE_ARCHIVE_ROOT", str(daemon_root)) + asyncio.run(_drive_daemon_bulk_rebuild_to_promotion(daemon_root, batch_size=2)) + + cli_snapshot = _canonical_snapshot(cli_root / "index.db") + daemon_snapshot = _canonical_snapshot(daemon_root / "index.db") + assert cli_snapshot["session_count"] == _RAW_COUNT + assert cli_snapshot == daemon_snapshot diff --git a/tests/unit/daemon/test_daemon_cli.py b/tests/unit/daemon/test_daemon_cli.py index 9f73be76e4..1b486c3b49 100644 --- a/tests/unit/daemon/test_daemon_cli.py +++ b/tests/unit/daemon/test_daemon_cli.py @@ -3911,3 +3911,121 @@ async def fake_sleep(seconds: float) -> None: assert schedule == [] assert sleeps == 2 + + +def test_bulk_rebuild_routing_flag_off_never_checks_or_drives( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """polylogue-gd6v: off by default. The flag-off path must not even + consult ``has_resumable_daemon_bulk_rebuild_transaction`` -- checking + survives a restart via a durable transaction record, so a false + "resumable" read while the flag is off would incorrectly promote one.""" + from polylogue.daemon import cli as daemon_cli + from polylogue.product.raw_authority import RawMaterializationCounts + + class FakeResolved: + daemon_bulk_rebuild_routing = False + + def fail_has_resumable(_root: object) -> bool: + pytest.fail("must not check for a resumable transaction while the flag is off") + + def fail_run_pass(**_kwargs: object) -> object: + pytest.fail("must not drive a bulk-rebuild pass while the flag is off") + + monkeypatch.setattr("polylogue.config.load_polylogue_config", lambda: FakeResolved()) + monkeypatch.setattr( + "polylogue.daemon.bulk_rebuild.has_resumable_daemon_bulk_rebuild_transaction", fail_has_resumable + ) + monkeypatch.setattr("polylogue.daemon.bulk_rebuild.run_daemon_bulk_rebuild_pass", fail_run_pass) + + counts = RawMaterializationCounts(candidate_count=999_999, pending_blob_bytes=0) + asyncio.run(daemon_cli._maybe_route_daemon_bulk_rebuild(counts)) + + +def test_bulk_rebuild_routing_below_threshold_and_not_resumable_is_noop( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + """A small, steady-state backlog with no bulk-rebuild already in flight + must never start one -- bulk routing is for bulk-scale backlogs only.""" + from polylogue.daemon import cli as daemon_cli + from polylogue.product.raw_authority import RawMaterializationCounts + + class FakeResolved: + daemon_bulk_rebuild_routing = True + + def fail_run_pass(**_kwargs: object) -> object: + pytest.fail("must not drive a pass when below threshold and nothing is resumable") + + monkeypatch.setattr("polylogue.config.load_polylogue_config", lambda: FakeResolved()) + monkeypatch.setattr("polylogue.paths.archive_root", lambda: tmp_path) + monkeypatch.setattr( + "polylogue.daemon.bulk_rebuild.has_resumable_daemon_bulk_rebuild_transaction", lambda _root: False + ) + monkeypatch.setattr("polylogue.daemon.bulk_rebuild.run_daemon_bulk_rebuild_pass", fail_run_pass) + + counts = RawMaterializationCounts(candidate_count=3, pending_blob_bytes=0) + asyncio.run(daemon_cli._maybe_route_daemon_bulk_rebuild(counts)) + + +def test_bulk_rebuild_routing_resumable_transaction_drives_pass_even_below_threshold( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + """An in-flight bulk-rebuild operation keeps being driven every tick even + once the instantaneous trickle backlog reading has dipped below the + bulk-scale threshold -- abandoning a partially-built generation would + waste every page already replayed into it.""" + from polylogue.daemon import cli as daemon_cli + from polylogue.product.raw_authority import RawMaterializationCounts + + class FakeResolved: + daemon_bulk_rebuild_routing = True + + calls: list[dict[str, object]] = [] + + async def fake_run_pass(**kwargs: object) -> None: + calls.append(kwargs) + return None + + monkeypatch.setattr("polylogue.config.load_polylogue_config", lambda: FakeResolved()) + monkeypatch.setattr("polylogue.paths.archive_root", lambda: tmp_path) + monkeypatch.setattr("polylogue.paths.render_root", lambda: tmp_path / "render") + monkeypatch.setattr( + "polylogue.daemon.bulk_rebuild.has_resumable_daemon_bulk_rebuild_transaction", lambda _root: True + ) + monkeypatch.setattr("polylogue.daemon.bulk_rebuild.run_daemon_bulk_rebuild_pass", fake_run_pass) + + counts = RawMaterializationCounts(candidate_count=3, pending_blob_bytes=0) + asyncio.run(daemon_cli._maybe_route_daemon_bulk_rebuild(counts)) + + assert len(calls) == 1 + called_config = cast(Config, calls[0]["config"]) + assert called_config.archive_root == tmp_path + + +def test_bulk_rebuild_routing_pass_failure_never_propagates( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + """A failed bulk-rebuild pass must not crash the periodic convergence + loop it is called from -- the next tick simply tries again.""" + from polylogue.daemon import cli as daemon_cli + from polylogue.product.raw_authority import RawMaterializationCounts + + class FakeResolved: + daemon_bulk_rebuild_routing = True + + async def fail_run_pass(**_kwargs: object) -> object: + raise RuntimeError("simulated bulk-rebuild pass failure") + + monkeypatch.setattr("polylogue.config.load_polylogue_config", lambda: FakeResolved()) + monkeypatch.setattr("polylogue.paths.archive_root", lambda: tmp_path) + monkeypatch.setattr("polylogue.paths.render_root", lambda: tmp_path / "render") + monkeypatch.setattr( + "polylogue.daemon.bulk_rebuild.has_resumable_daemon_bulk_rebuild_transaction", lambda _root: True + ) + monkeypatch.setattr("polylogue.daemon.bulk_rebuild.run_daemon_bulk_rebuild_pass", fail_run_pass) + + counts = RawMaterializationCounts(candidate_count=3, pending_blob_bytes=0) + asyncio.run(daemon_cli._maybe_route_daemon_bulk_rebuild(counts)) # must not raise