diff --git a/docs/maintenance.md b/docs/maintenance.md index 62f1375a44..c3b94e9bb7 100644 --- a/docs/maintenance.md +++ b/docs/maintenance.md @@ -38,6 +38,31 @@ Restart health and runtime-consumer convergence are the final lifecycle proof and are recorded by the durable train lifecycle API, not inferred from this command's migration result alone. +### Rebuild deployment-currency preflight + +Before a managed `rebuild-index`, confirm that the package selected for the +operation owns the live durable schemas. The read-only preflight deliberately +checks `source.db` and `user.db` only: `index.db` may be behind because the +rebuild is the supported way to replace that derived tier. + +```bash +polylogue ops maintenance rebuild-index --preflight --output-format json +``` + +It emits `rebuild-schema-currency` JSON with each durable tier's observed and +package-expected `user_version`, and exits nonzero when either differs. The +execution route repeats this check before it consumes the schema-inference +receipt, acquires archive ownership, or creates a candidate generation. + +For a safe deployment recovery, first choose the exact target package commit. +With the daemon stopped, create a fresh verified full-evidence backup, run +`migrate-tier source` and `migrate-tier user` when the target package requires +them, then deploy that exact package. Run the preflight above and require a +ready result before invoking `polylogue ops maintenance rebuild-index`; use +that blue-green command rather than `ops reset --index` for an active managed +generation. Restart the daemon only after the rebuilt generation is promoted +and the post-deploy status shows no durable-tier mismatch. + For the conceptual model behind derived insights and the FTS / blob substrate, see [architecture.md](architecture.md) and [internals.md](internals.md). For daemon ownership of the inline diff --git a/polylogue/cli/commands/maintenance/_rebuild_index.py b/polylogue/cli/commands/maintenance/_rebuild_index.py index ed2e128aae..8f7eedf4d3 100644 --- a/polylogue/cli/commands/maintenance/_rebuild_index.py +++ b/polylogue/cli/commands/maintenance/_rebuild_index.py @@ -388,6 +388,11 @@ def _rebuild_index_selection_plan( "single-writer path; unsupported with --daemon." ), ) +@click.option( + "--preflight", + is_flag=True, + help="Read-only: report whether durable source/user tiers match this package before rebuilding index.db.", +) def rebuild_index_command( only_missing: bool, raw_ids: tuple[str, ...], @@ -404,6 +409,7 @@ def rebuild_index_command( use_daemon: bool, daemon_url: str, shard_count: int, + preflight: bool, ) -> None: """Inspect or execute an authority-safe source-to-index rebuild. @@ -441,6 +447,22 @@ def rebuild_index_command( raise click.UsageError("resumed rebuild budgets are durable; omit pass budget options with --operation-id") root = archive_root() + if preflight: + from polylogue.maintenance.rebuild_index import rebuild_schema_currency_preflight + + payload = rebuild_schema_currency_preflight(root) + if output_format == "json": + click.echo(json.dumps(payload, indent=2, sort_keys=True)) + else: + click.echo(f"Archive root: {root}") + for tier in cast(list[dict[str, object]], payload["tiers"]): + click.echo( + f"{tier['tier']}.db: {tier['actual_user_version']} (package expects " + f"{tier['expected_user_version']}; {tier['status']})" + ) + if payload["status"] != "ready": + raise click.ClickException("rebuild schema currency preflight failed; migrate or deploy before rebuilding") + return if use_daemon: payload = _run_daemon_rebuild( daemon_url, diff --git a/polylogue/maintenance/rebuild_index.py b/polylogue/maintenance/rebuild_index.py index 2805347939..8bea29f817 100644 --- a/polylogue/maintenance/rebuild_index.py +++ b/polylogue/maintenance/rebuild_index.py @@ -66,6 +66,61 @@ class RebuildDerivedStateProvenanceError(RebuildProvenanceError): """A derived-state stage was blocked by a failed provenance recheck.""" +class RebuildSchemaCurrencyError(RuntimeError): + """The durable tiers do not match the package that would rebuild them.""" + + def __init__(self, diagnostic: dict[str, object]) -> None: + self.diagnostic = diagnostic + blocked = diagnostic["blocking_tiers"] + assert isinstance(blocked, list) + detail = ", ".join( + f"{item['tier']}.db:{item['actual_user_version']}!={item['expected_user_version']}" + for item in blocked + if isinstance(item, dict) + ) + super().__init__(f"rebuild schema currency preflight failed: {detail}") + + +def rebuild_schema_currency_preflight(root: Path) -> dict[str, object]: + """Report whether durable source evidence matches this runtime package. + + ``index.db`` is intentionally absent: rebuilding it is the operation's + purpose, while a source/user mismatch means this package can interpret or + write durable evidence using a schema it does not own. + """ + from polylogue.storage.archive_readiness import probe_archive_tier + from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier + + checks: list[dict[str, object]] = [] + for tier in (ArchiveTier.SOURCE, ArchiveTier.USER): + probe = probe_archive_tier(tier, root / f"{tier.value}.db") + checks.append( + { + "tier": tier.value, + "path": probe.path, + "actual_user_version": probe.user_version, + "expected_user_version": probe.expected_user_version, + "status": probe.version_status, + } + ) + blocking = [check for check in checks if check["status"] != "ok"] + return { + "kind": "rebuild-schema-currency", + "archive_root": str(root), + "status": "ready" if not blocking else "blocked", + "tiers": checks, + "blocking_tiers": blocking, + } + + +def require_rebuild_schema_currency(root: Path) -> dict[str, object]: + """Reject a rebuild before it consumes evidence or creates a generation.""" + diagnostic = rebuild_schema_currency_preflight(root) + if diagnostic["status"] != "ready": + raise RebuildSchemaCurrencyError(diagnostic) + return diagnostic + + @dataclass(frozen=True, slots=True) class RebuildProvenanceContext: """Validated evidence shared by every mutation in one rebuild pass. @@ -1047,6 +1102,7 @@ async def rebuild_index_from_source(request: RebuildIndexRequest) -> RebuildInde log_mapped_bytes_budget_check(logger, check_mapped_bytes_budget_against_cgroup_limit()) validate_rebuild_index_request(request) root = request.archive_root + require_rebuild_schema_currency(root) consumed_evidence = _validate_rebuild_provenance_receipt(root, request.schema_inference_receipt_path) location = ArchiveLocation.resolve(root) # The joined raw-frontier projection is rooted at the co-located active diff --git a/tests/unit/cli/test_archive_maintenance_cli.py b/tests/unit/cli/test_archive_maintenance_cli.py index fd9222171d..4f4bff0761 100644 --- a/tests/unit/cli/test_archive_maintenance_cli.py +++ b/tests/unit/cli/test_archive_maintenance_cli.py @@ -1990,6 +1990,31 @@ def test_rebuild_index_force_write_option_is_retired(cli_runner: CliRunner) -> N assert "--force-write" in result.output +def test_rebuild_index_preflight_reports_durable_schema_currency( + cli_workspace: dict[str, Path], cli_runner: CliRunner +) -> None: + root = cli_workspace["archive_root"] + with sqlite3.connect(root / "source.db") as conn: + conn.execute("DROP INDEX idx_raw_failure_disposition_receipts_disposed_at") + conn.execute("DROP TABLE raw_failure_disposition_receipts") + conn.execute("PRAGMA user_version = 28") + + result = cli_runner.invoke( + cli, + ["--plain", "ops", "maintenance", "rebuild-index", "--preflight", "--output-format", "json"], + catch_exceptions=False, + ) + + assert result.exit_code == 1 + payload = json.loads(result.stdout) + assert payload["kind"] == "rebuild-schema-currency" + assert payload["status"] == "blocked" + assert payload["blocking_tiers"][0]["tier"] == "source" + assert payload["blocking_tiers"][0]["actual_user_version"] == 28 + assert payload["blocking_tiers"][0]["expected_user_version"] == 29 + assert "migrate or deploy before rebuilding" in result.stderr + + def test_rebuild_index_daemon_path_posts_the_real_selection_request( cli_workspace: dict[str, Path], cli_runner: CliRunner, monkeypatch: pytest.MonkeyPatch ) -> None: diff --git a/tests/unit/maintenance/test_rebuild_index_ownership.py b/tests/unit/maintenance/test_rebuild_index_ownership.py index ab5b43fcf8..35f3605fce 100644 --- a/tests/unit/maintenance/test_rebuild_index_ownership.py +++ b/tests/unit/maintenance/test_rebuild_index_ownership.py @@ -16,10 +16,15 @@ import pytest -from polylogue.maintenance.rebuild_index import RebuildIndexRequest, rebuild_index_from_source_sync +from polylogue.maintenance.rebuild_index import ( + RebuildIndexRequest, + RebuildSchemaCurrencyError, + rebuild_index_from_source_sync, +) from polylogue.storage.archive_identity import ArchiveLocation, ArchiveOwnershipError, OwnedArchiveLocation -from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database +from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root, initialize_archive_database from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier +from tests.infra.rebuild_receipt import write_valid_rebuild_receipt def _init_empty_source(root: Path) -> None: @@ -27,6 +32,41 @@ def _init_empty_source(root: Path) -> None: initialize_archive_database(root / "source.db", ArchiveTier.SOURCE) +def test_rebuild_rejects_source_schema_behind_runtime_before_candidate_creation(tmp_path: Path) -> None: + """A real v28 source tier must not reach the v29 rebuild package. + + The test builds ordinary file-backed archive tiers, removes exactly v29's + additive objects, and supplies a valid rebuild receipt. The production + rebuild route used to accept this archive and return ``empty-source``. + """ + root = tmp_path / "archive" + initialize_active_archive_root(root) + with sqlite3.connect(root / "source.db") as conn: + conn.execute("DROP INDEX idx_raw_failure_disposition_receipts_disposed_at") + conn.execute("DROP TABLE raw_failure_disposition_receipts") + conn.execute("PRAGMA user_version = 28") + receipt_path = write_valid_rebuild_receipt(root, tmp_path / "schema-inference-receipt.json") + + with pytest.raises(RebuildSchemaCurrencyError) as exc_info: + rebuild_index_from_source_sync( + RebuildIndexRequest(archive_root=root, schema_inference_receipt_path=receipt_path) + ) + + diagnostic = exc_info.value.diagnostic + assert diagnostic["status"] == "blocked" + assert diagnostic["blocking_tiers"] == [ + { + "tier": "source", + "path": str(root / "source.db"), + "actual_user_version": 28, + "expected_user_version": 29, + "status": "mismatch", + } + ] + assert not (root / ".index-generations").exists() + assert not (root / ".index-rebuild-transactions").exists() + + def test_rebuild_refuses_when_archive_location_already_owned(tmp_path: Path) -> None: """A concurrent holder of the archive-location ownership lock must block an offline rebuild before any generation directory or SQLite tier is