diff --git a/docs/DEPLOYMENT_OPERATOR_GUIDE.md b/docs/DEPLOYMENT_OPERATOR_GUIDE.md index ea12cc7..3741fc3 100644 --- a/docs/DEPLOYMENT_OPERATOR_GUIDE.md +++ b/docs/DEPLOYMENT_OPERATOR_GUIDE.md @@ -431,6 +431,14 @@ SQLite's backup API; non-SQLite databases require `--database-backup-command`, `--database-restore-check-command`, and `--database-restore-command`. +Every non-dry run also owns the database-fenced +`core:module-lifecycle:deployment` recovery operation. The run record includes +its operation id and status. A supervised run reaches durable `succeeded` only +after restart and health verification. `recovery_required` or `outcome_unknown` +blocks another lifecycle mutation until the recorded operation is reconciled; +do not bypass this by deleting `install.lock`. See +[`MODULE_LIFECYCLE_RECOVERY.md`](MODULE_LIFECYCLE_RECOVERY.md). + Database hook commands receive: - `GOVOPLAN_INSTALLER_RUN_DIR` diff --git a/docs/DOCUMENTATION_MAP.md b/docs/DOCUMENTATION_MAP.md index 7fb8532..4cc739e 100644 --- a/docs/DOCUMENTATION_MAP.md +++ b/docs/DOCUMENTATION_MAP.md @@ -19,6 +19,7 @@ operator, and roadmap pages. | Institutional context and governed references | `INSTITUTIONAL_CONTEXT_CONTRACT.md` | Shared temporal, actor/representation, institution, mandate, service, party, decision, evidence, legal-basis, information-governance, presentation, and geo DTO/provider contracts. | | Postbox E2EE target architecture | `POSTBOX_E2EE_ARCHITECTURE.md` | Strategic encrypted postbox/mailbox model, key ownership, role mailbox semantics, and retraction limits. | | Shared state, runtime coordination, and recovery | `STATE_AND_RECOVERY_CONTRACT.md` | State profiles, object storage, node registration/drain, fenced leases, migration ordering, and recovery evidence. | +| Module lifecycle recovery | `MODULE_LIFECYCLE_RECOVERY.md` | Installer/live-graph recovery modes, deployment fence, evidence, retry blocking, and operator reconciliation. | ## Release And Operations diff --git a/docs/MODULE_ARCHITECTURE.md b/docs/MODULE_ARCHITECTURE.md index 7872d54..bb8a640 100644 --- a/docs/MODULE_ARCHITECTURE.md +++ b/docs/MODULE_ARCHITECTURE.md @@ -1346,6 +1346,11 @@ The package install-plan API records operator intent only: default; successful uninstalls are removed from saved startup state by default. Use `--no-activate-installed-modules` or `--keep-uninstalled-modules-in-desired` only for staged rollout workflows. +- Every non-dry installer and live active-graph mutation acquires the + deployment-wide `core:module-lifecycle:deployment` lease and records a Core + recovery operation. Unresolved effects block later lifecycle changes. The + operation modes and operator reconciliation contract are defined in + `MODULE_LIFECYCLE_RECOVERY.md`. - `govoplan-module-installer --supervise --migrate --health-url http://127.0.0.1:8000/health --restart-command ''` is the preferred disruptive-change path. It applies the plan, optionally runs migrations in a fresh Python process after a fresh-process manifest diff --git a/docs/MODULE_LIFECYCLE_RECOVERY.md b/docs/MODULE_LIFECYCLE_RECOVERY.md new file mode 100644 index 0000000..d7dc5f5 --- /dev/null +++ b/docs/MODULE_LIFECYCLE_RECOVERY.md @@ -0,0 +1,60 @@ +# Module Lifecycle Recovery + +Package changes and live module-graph changes use Core's durable recovery +ledger. The local `install.lock` still prevents duplicate work in one runtime +directory; the database lease `core:module-lifecycle:deployment` is the +deployment-wide authority across API, installer, worker, and scheduler nodes. + +## Declared Boundaries + +| Operation | Recovery mode | Completion condition | +| --- | --- | --- | +| `module-lifecycle.pre-migration` | compensation | package, WebUI, manifest, and desired-graph evidence match | +| `module-lifecycle.post-migration` | forward recovery | migration tasks, manifests, desired graph, restart, and health are verified | +| `module-retirement.destroy-data` | snapshot restore | a hashed, restore-checked backup exists and retirement state is verified | +| `module-runtime.apply-graph` | compensation | hooks, capability contexts, active graph, and workflow contributions match | + +The installer prepares the recovery operation before it captures the database +snapshot. A full database restore therefore retains the prepared operation and +its fence instead of erasing the fact that a mutation was attempted. Backup +artifacts are hashed and sized before any package, migration, or retirement +effect starts. + +Every command boundary records the command source and canonical hashes of the +redacted command/result records. Credentials, database URLs, command output, +and package-registry secrets are never copied into recovery evidence. + +## Failure And Retry Rules + +- A conclusive failure before effects is terminal `failed`. +- A command or compensatable effect that started but did not complete is + `recovery_required`. +- A lost or unexpected outcome after a migration/external boundary is + `outcome_unknown`. +- A verified package/database rollback becomes `recovered`. +- A supervised install becomes `succeeded` only after restart and all configured + health probes succeed. + +An unresolved lifecycle operation blocks every later lifecycle mutation on the +same deployment fence, even after its execution lease is released. Operators +must inspect the checkpoint chain and run record, restore or complete the +declared recovery path, and explicitly reconcile the operation. A new install +must not be used as an implicit retry. + +Live graph changes use the same fence. A non-migrating hook or registry failure +restores the prior in-process graph and records verified compensation. A failure +after migrations begin remains unresolved because restoring the process-local +registry does not reverse database schema effects. + +## Operator Evidence + +The installer run record contains the recovery operation id, mode, plan hash, +and current lifecycle status. The Ops recovery view is authoritative for the +durable state and evidence-chain result. Keep both the run directory and the +state-service backup evidence until the operation is terminal and the normal +retention policy permits removal. + +Run the module installer rollback drill and recovery-runtime test matrix before +enabling lifecycle mutation in a new deployment. Shared-state deployments must +still use immutable release images; the ledger does not make in-place package +mutation across replicas safe. diff --git a/src/govoplan_core/commands/module_installer.py b/src/govoplan_core/commands/module_installer.py index 0ecbf89..a5711be 100644 --- a/src/govoplan_core/commands/module_installer.py +++ b/src/govoplan_core/commands/module_installer.py @@ -1,6 +1,7 @@ from __future__ import annotations import argparse +from importlib.metadata import PackageNotFoundError, version import json from pathlib import Path import sys @@ -33,6 +34,10 @@ from govoplan_core.core.module_installer_notifications import ( installer_notification_priority, installer_notification_subject, ) +from govoplan_core.core.runtime_coordination import ( + bind_process_runtime_identity, + runtime_identity, +) from govoplan_core.core.module_license import issue_module_license, module_license_diagnostics from govoplan_core.core.module_package_catalog import sign_module_package_catalog, validate_module_package_catalog from govoplan_core.core.module_management import ( @@ -107,11 +112,27 @@ def _build_parser() -> argparse.ArgumentParser: def main() -> int: args = _build_parser().parse_args() runtime_dir = args.runtime_dir or default_installer_runtime_dir(args.database_url) + bind_process_runtime_identity( + runtime_identity( + settings, + software_version=_core_version(), + role="installer", + ) + ) try: return _dispatch_command(args=args, runtime_dir=runtime_dir) except ModuleInstallerError as exc: print(f"error: {exc}", file=sys.stderr) return 1 + finally: + bind_process_runtime_identity(None) + + +def _core_version() -> str: + try: + return version("govoplan-core") + except PackageNotFoundError: + return "development" def _dispatch_command(*, args: argparse.Namespace, runtime_dir: Path) -> int: diff --git a/src/govoplan_core/core/lifecycle.py b/src/govoplan_core/core/lifecycle.py index a8c9bbd..a846195 100644 --- a/src/govoplan_core/core/lifecycle.py +++ b/src/govoplan_core/core/lifecycle.py @@ -7,6 +7,11 @@ from threading import RLock from fastapi import APIRouter, Depends, FastAPI, HTTPException, Request, status from govoplan_core.core.module_management import ModuleManagementError, REQUIRED_PLATFORM_MODULES, plan_desired_enabled_modules +from govoplan_core.core.module_lifecycle_recovery import ( + ModuleLifecycleRecovery, + begin_runtime_graph_recovery, + canonical_sha256, +) from govoplan_core.core.modules import ModuleContext, ModuleManifest from govoplan_core.core.registry import PlatformRegistry from govoplan_core.core.runtime import configure_runtime @@ -99,30 +104,110 @@ class ModuleLifecycleManager: next_set = set(plan.enabled_modules) activated = tuple(module_id for module_id in plan.enabled_modules if module_id not in previous_set) deactivated = tuple(module_id for module_id in previous if module_id not in next_set) + graph_changes = bool(activated or deactivated) + recovery: ModuleLifecycleRecovery | None = None + if graph_changes or migrate: + from govoplan_core.db.session import get_database - if migrate: - self._migrate(plan.enabled_modules) + with get_database().session() as recovery_session: + recovery = begin_runtime_graph_recovery( + recovery_session, + previous_modules=previous, + requested_modules=plan.enabled_modules, + migrate=migrate, + ) - mounted = tuple(module_id for module_id in plan.enabled_modules if self._mount_module_router(module_id)) + old_manifests = { + manifest.id: manifest for manifest in self.registry.manifests() + } + try: + if recovery is not None: + recovery.checkpoint( + kind="runtime-graph-effect-started", + summary="Runtime module graph entered its mutation boundary", + evidence={ + "activated_sha256": canonical_sha256(activated), + "deactivated_sha256": canonical_sha256(deactivated), + "migrate": migrate, + }, + effect_started=True, + ) - old_manifests = {manifest.id: manifest for manifest in self.registry.manifests()} - for module_id in deactivated: - hook = old_manifests[module_id].on_deactivate - if hook is not None: - hook(self.context) + if migrate: + self._migrate(plan.enabled_modules) - self.registry.replace(self.available_modules[module_id] for module_id in plan.enabled_modules) - self.configure_runtime() + mounted = tuple(module_id for module_id in plan.enabled_modules if self._mount_module_router(module_id)) - for module_id in activated: - hook = self.available_modules[module_id].on_activate - if hook is not None: - hook(self.context) + for module_id in deactivated: + hook = old_manifests[module_id].on_deactivate + if hook is not None: + hook(self.context) - self.reconcile_workflow_definitions() + self.registry.replace(self.available_modules[module_id] for module_id in plan.enabled_modules) + self.configure_runtime() - if self._app is not None: - self._app.openapi_schema = None + for module_id in activated: + hook = self.available_modules[module_id].on_activate + if hook is not None: + hook(self.context) + + reconciliation = self.reconcile_workflow_definitions() + + if self._app is not None: + self._app.openapi_schema = None + + if recovery is not None: + from govoplan_core.db.session import get_database + + with get_database().session() as recovery_session: + recovery.succeed( + recovery_session, + evidence={ + "active_graph_sha256": canonical_sha256( + self.active_module_ids() + ), + "mounted_graph_sha256": canonical_sha256( + self.mounted_module_ids() + ), + "workflow_reconciliation_sha256": canonical_sha256( + reconciliation + ), + }, + commit_projection=False, + ) + except Exception as exc: + self.registry.replace(old_manifests.values()) + self.configure_runtime() + if self._app is not None: + self._app.openapi_schema = None + if recovery is not None: + from govoplan_core.db.session import get_database + + recovery.unresolved( + summary="Runtime graph mutation did not reach verified completion", + evidence={ + "error_type": type(exc).__name__, + "previous_graph_sha256": canonical_sha256(previous), + "registry_restored": True, + "migrate": migrate, + }, + outcome_unknown=migrate, + ) + if not migrate: + with get_database().session() as recovery_session: + recovery.recovered( + recovery_session, + evidence={ + "active_graph_sha256": canonical_sha256( + self.active_module_ids() + ), + "previous_graph_restored": ( + self.active_module_ids() == previous + ), + }, + summary="Previous runtime module graph was restored", + ) + raise return ModuleLifecycleResult( enabled_modules=plan.enabled_modules, diff --git a/src/govoplan_core/core/module_installer.py b/src/govoplan_core/core/module_installer.py index da2c2d5..379df8f 100644 --- a/src/govoplan_core/core/module_installer.py +++ b/src/govoplan_core/core/module_installer.py @@ -27,6 +27,12 @@ from sqlalchemy.orm import Session from govoplan_core.core.maintenance import saved_maintenance_mode from govoplan_core.core.events import current_event_trace +from govoplan_core.core.module_lifecycle_recovery import ( + ModuleLifecycleRecovery, + ModuleLifecycleRecoveryError, + begin_module_installer_recovery, + canonical_sha256, +) from govoplan_core.core.module_management import ( PROTECTED_MODULES, ModuleInstallPlan, @@ -268,6 +274,11 @@ class ModuleInstallerRunResult: return_code: int = 0 error: str | None = None rollback: dict[str, object] | None = None + recovery: ModuleLifecycleRecovery | None = field( + default=None, + repr=False, + compare=False, + ) def as_dict(self) -> dict[str, object]: payload: dict[str, object] = { @@ -293,6 +304,7 @@ class _ModuleInstallRunState: result_commands: tuple[str, ...] record_redactions: tuple[str, ...] record: dict[str, Any] + recovery: ModuleLifecycleRecovery | None = None def default_installer_runtime_dir(database_url: str | None = None, *, cwd: Path | None = None) -> Path: @@ -503,6 +515,7 @@ def run_module_install_plan( remove_uninstalled_modules_from_desired: bool = True, dry_run: bool = False, request_context: Mapping[str, object] | None = None, + finalize_recovery: bool = True, ) -> ModuleInstallerRunResult: maintenance_mode = saved_maintenance_mode(session) effective_runtime_dir = runtime_dir or default_installer_runtime_dir(database_url) @@ -520,6 +533,7 @@ def run_module_install_plan( raise ModuleInstallerError("Install preflight is blocked: " + "; ".join(issue.message for issue in preflight.issues if issue.severity == "blocker")) state = _prepare_module_install_run( + session=session, plan=plan, preflight=preflight, database_url=database_url, @@ -550,6 +564,7 @@ def run_module_install_plan( if failed_error is not None: return _failed_module_install_run_result( + session=session, state=state, plan=plan, executed=executed, @@ -569,11 +584,13 @@ def run_module_install_plan( remove_uninstalled_modules_from_desired=remove_uninstalled_modules_from_desired, executed=executed, state=state, + finalize_recovery=finalize_recovery, ) def _prepare_module_install_run( *, + session: Session, plan: ModuleInstallPlan, preflight: ModuleInstallerPreflight, database_url: str, @@ -605,13 +622,39 @@ def _prepare_module_install_run( verify_modules=True, ) record_redactions = _installer_secret_redactions(database_url) - record = _initial_module_install_record( - run_id=run_id, - plan=plan, - preflight=preflight, - commands=commands, - record_redactions=record_redactions, - snapshot=_snapshot_environment( + recovery: ModuleLifecycleRecovery | None = None + if not dry_run: + try: + recovery = begin_module_installer_recovery( + session, + run_id=run_id, + plan=tuple(item.as_dict() for item in plan.items), + command_count=len(commands), + migrate_database=migrate_database, + destructive_retirement=_destructive_retirement_requested(plan), + snapshot_sha256=None, + backup_reference=( + f"module-installer:{run_id}:database-backup" + if _destructive_retirement_requested(plan) + else None + ), + request_context_sha256=canonical_sha256(dict(request_context or {})), + ) + recovery.checkpoint( + kind="snapshot-started", + summary="Installer environment snapshot started before package effects", + evidence={ + "run_id": run_id, + "database_backup_expected": bool( + migrate_database or _destructive_retirement_requested(plan) + ), + }, + ) + except ModuleLifecycleRecoveryError as exc: + raise ModuleInstallerError(str(exc)) from exc + + try: + snapshot = _snapshot_environment( run_dir, webui_root=webui_root, database_url=database_url, @@ -619,7 +662,32 @@ def _prepare_module_install_run( database_backup_command=database_backup_command, database_restore_command=database_restore_command, database_restore_check_command=database_restore_check_command, - ), + ) + except Exception as exc: + if recovery is not None: + recovery.unresolved( + summary="Installer snapshot preparation failed before package effects", + evidence={"snapshot_error_type": type(exc).__name__}, + outcome_unknown=False, + ) + raise + + if recovery is not None: + recovery.checkpoint( + kind="snapshot-verified", + summary="Installer environment snapshot and backup evidence were verified", + evidence={ + "snapshot_sha256": canonical_sha256(snapshot), + **_database_backup_recovery_evidence(snapshot), + }, + ) + record = _initial_module_install_record( + run_id=run_id, + plan=plan, + preflight=preflight, + commands=commands, + record_redactions=record_redactions, + snapshot=snapshot, build_webui=build_webui, migrate_database=migrate_database, activate_installed_modules=activate_installed_modules, @@ -627,6 +695,8 @@ def _prepare_module_install_run( dry_run=dry_run, request_context=request_context, ) + if recovery is not None: + record["recovery"] = _module_lifecycle_recovery_record(recovery) record_path = run_dir / "record.json" _write_json(record_path, record) return _ModuleInstallRunState( @@ -637,6 +707,7 @@ def _prepare_module_install_run( result_commands=_command_displays(commands, redactions=record_redactions), record_redactions=record_redactions, record=record, + recovery=recovery, ) @@ -673,6 +744,40 @@ def _initial_module_install_record( return record +def _module_lifecycle_recovery_record( + recovery: ModuleLifecycleRecovery, + *, + status: str = "running", +) -> dict[str, object]: + return { + "operation_id": recovery.operation_id, + "operation_type": recovery.operation_type, + "mode": recovery.mode.value, + "plan_sha256": recovery.plan_sha256, + "replayed": recovery.replayed, + "status": status, + } + + +def _database_backup_recovery_evidence( + snapshot: Mapping[str, object], +) -> dict[str, object]: + backup = snapshot.get("database_backup") + if not isinstance(backup, Mapping): + return {"database_backup_present": False} + sha256 = str(backup.get("artifact_sha256") or "").strip() + return { + "database_backup_present": True, + "database_backup_type": str(backup.get("type") or "unknown"), + "database_backup_sha256": sha256 or "unavailable", + "database_backup_size_bytes": int(backup.get("size_bytes") or 0), + "database_backup_reference": ( + f"sha256:{sha256}" if sha256 else "unavailable" + ), + "restore_check_sha256": canonical_sha256(backup.get("restore_check")), + } + + def _execute_module_install_run( *, session: Session, @@ -685,14 +790,91 @@ def _execute_module_install_run( failed_error: str | None = None with _installer_lock(effective_runtime_dir): try: + if state.recovery is not None: + state.recovery.checkpoint( + kind="effects-starting", + summary="Installer acquired local and distributed execution fences", + evidence={ + "command_count": len(state.commands), + "destructive_retirement": _destructive_retirement_requested(plan), + }, + ) + if _destructive_retirement_requested(plan): + state.recovery.checkpoint( + kind="retirement-effect-started", + summary="Destructive module retirement entered its effect boundary", + evidence={ + "retirement_plan_sha256": canonical_sha256( + [ + item.as_dict() + for item in plan.items + if item.destroy_data + ] + ), + }, + effect_started=True, + ) _execute_module_install_retirements(session=session, plan=plan, available=available, state=state) - for command in state.commands: - executed.append(_run_module_install_command(command, state=state)) + for index, command in enumerate(state.commands): + if state.recovery is not None: + command_record = _command_record( + command, + redactions=state.record_redactions, + ) + state.recovery.checkpoint( + kind="command-effect-started", + summary="Installer command entered its effect boundary", + evidence={ + "command_index": index, + "command_source": str(command.get("source") or "unknown"), + "command_sha256": canonical_sha256(command_record), + }, + effect_started=True, + ) + command_result = _run_module_install_command(command, state=state) + executed.append(command_result) + if state.recovery is not None: + state.recovery.checkpoint( + kind="command-result-verified", + summary="Installer command returned a conclusive successful result", + evidence={ + "command_index": index, + "return_code": int(command_result["return_code"]), + "result_sha256": canonical_sha256(command_result), + }, + ) state.record["commands"] = executed _write_json(state.record_path, state.record) except Exception as exc: failed_error = _redact_installer_text(str(exc), redactions=state.record_redactions) _rollback_session_after_module_install_error(session, exc) + if state.recovery is not None: + outcome_unknown = not isinstance(exc, ModuleInstallerError) + try: + state.recovery.unresolved( + summary="Module installer effects did not reach verified completion", + evidence={ + "error_type": type(exc).__name__, + "completed_command_count": len(executed), + }, + outcome_unknown=outcome_unknown, + ) + state.record["recovery"] = _module_lifecycle_recovery_record( + state.recovery, + status=( + "outcome_unknown" + if outcome_unknown + else "recovery_required" + if state.recovery.effect_started + else "failed" + ), + ) + except Exception as recovery_exc: + state.record["recovery_error"] = type(recovery_exc).__name__ + failed_error = ( + f"{failed_error}; recovery ledger transition failed: " + f"{type(recovery_exc).__name__}" + ) return executed, failed_error @@ -740,6 +922,7 @@ def _rollback_session_after_module_install_error(session: Session, exc: Exceptio def _failed_module_install_run_result( *, + session: Session, state: _ModuleInstallRunState, plan: ModuleInstallPlan, executed: list[dict[str, object]], @@ -765,6 +948,7 @@ def _failed_module_install_run_result( commands=state.result_commands, return_code=1, error=failed_error, + recovery=state.recovery, ) rollback = rollback_module_install_run( run_id=state.run_id, @@ -775,6 +959,30 @@ def _failed_module_install_run_result( database_url=database_url, ) _update_run_record(state.record_path, {"destructive_retirement_rollback": rollback.as_dict()}) + if rollback.return_code == 0 and state.recovery is not None: + try: + state.recovery.recovered( + session, + evidence={ + "rollback_return_code": rollback.return_code, + "rollback_sha256": canonical_sha256(rollback.as_dict()), + }, + summary="Verified rollback restored the pre-install module state", + ) + _update_run_record( + state.record_path, + { + "recovery": _module_lifecycle_recovery_record( + state.recovery, + status="recovered", + ) + }, + ) + except Exception as recovery_exc: + _update_run_record( + state.record_path, + {"recovery_error": type(recovery_exc).__name__}, + ) return ModuleInstallerRunResult( run_id=state.run_id, status="rolled-back" if rollback.return_code == 0 else "failed", @@ -783,6 +991,7 @@ def _failed_module_install_run_result( return_code=1, error=failed_error, rollback=rollback.as_dict(), + recovery=state.recovery, ) @@ -795,6 +1004,7 @@ def _applied_module_install_run_result( remove_uninstalled_modules_from_desired: bool, executed: list[dict[str, object]], state: _ModuleInstallRunState, + finalize_recovery: bool, ) -> ModuleInstallerRunResult: save_module_install_plan(session, tuple(_mark_applied(item) for item in plan.items)) if activate_installed_modules or remove_uninstalled_modules_from_desired: @@ -806,14 +1016,50 @@ def _applied_module_install_run_result( ) save_desired_enabled_modules(session, next_desired) state.record["desired_enabled_after"] = list(next_desired) - session.commit() + recovery_evidence = { + "command_count": len(executed), + "command_results_sha256": canonical_sha256(executed), + "desired_graph_sha256": canonical_sha256( + state.record.get("desired_enabled_after", list(desired_enabled)) + ), + "plan_projection_sha256": canonical_sha256( + [item.as_dict() for item in plan.items] + ), + } + if state.recovery is not None and finalize_recovery: + state.recovery.succeed( + session, + evidence=recovery_evidence, + commit_projection=True, + ) + recovery_status = "succeeded" + else: + session.commit() + recovery_status = "awaiting_supervisor" if state.recovery is not None else None + if state.recovery is not None: + state.recovery.checkpoint( + kind="local-projection-committed", + summary="Package and desired-graph projections await runtime health verification", + evidence=recovery_evidence, + ) state.record.update({ "status": "applied", "finished_at": datetime.now(tz=UTC).isoformat(), "commands": executed, }) + if state.recovery is not None and recovery_status is not None: + state.record["recovery"] = _module_lifecycle_recovery_record( + state.recovery, + status=recovery_status, + ) _write_json(state.record_path, state.record) - return ModuleInstallerRunResult(run_id=state.run_id, status="applied", record_path=state.record_path, commands=state.result_commands) + return ModuleInstallerRunResult( + run_id=state.run_id, + status="applied", + record_path=state.record_path, + commands=state.result_commands, + recovery=state.recovery, + ) def supervise_module_install_plan( @@ -864,6 +1110,7 @@ def supervise_module_install_plan( remove_uninstalled_modules_from_desired=remove_uninstalled_modules_from_desired, dry_run=False, request_context=request_context, + finalize_recovery=False, ) supervisor: dict[str, object] = { "started_at": datetime.now(tz=UTC).isoformat(), @@ -938,6 +1185,27 @@ def supervise_module_install_plan( "status": "ok", "finished_at": datetime.now(tz=UTC).isoformat(), }) + if result.recovery is not None: + result.recovery.succeed( + session, + evidence={ + "restart_results_sha256": canonical_sha256(restart_results), + "health_results_sha256": canonical_sha256(supervisor.get("health")), + "runtime_health_verified": True, + }, + commit_projection=False, + ) + supervisor["recovery_operation_id"] = result.recovery.operation_id + supervisor["recovery_status"] = "succeeded" + _update_run_record( + result.record_path, + { + "recovery": _module_lifecycle_recovery_record( + result.recovery, + status="succeeded", + ) + }, + ) _update_run_record(result.record_path, {"supervisor": supervisor}) return result @@ -3264,6 +3532,31 @@ def _rollback_after_supervisor_failure( session.commit() supervisor["rollback"] = rollback.as_dict() + if rollback.return_code == 0 and result.recovery is not None: + try: + result.recovery.recovered( + session, + evidence={ + "rollback_sha256": canonical_sha256(rollback.as_dict()), + "desired_graph_restored": True, + }, + summary="Supervisor rollback restored package and desired module state", + ) + supervisor["recovery_operation_id"] = result.recovery.operation_id + supervisor["recovery_status"] = "recovered" + _update_run_record( + result.record_path, + { + "recovery": _module_lifecycle_recovery_record( + result.recovery, + status="recovered", + ) + }, + ) + except Exception as recovery_exc: + supervisor["recovery_status"] = "reconciliation-failed" + supervisor["recovery_error"] = type(recovery_exc).__name__ + rollback_restart = _run_restart_commands(restart_commands) if rollback_restart: supervisor["rollback_restart_commands"] = rollback_restart @@ -3284,6 +3577,7 @@ def _rollback_after_supervisor_failure( return_code=1, error=reason, rollback=rollback.as_dict(), + recovery=result.recovery, ) @@ -3698,10 +3992,13 @@ def _snapshot_sqlite_database(run_dir: Path, database_url: str | None) -> dict[s raise ModuleInstallerError( f"SQLite backup failed its restore-readiness integrity check: {integrity}" ) + artifact_sha256 = _sha256_file(backup_path) return { "type": "sqlite", "source": str(db_path), "path": backup_path.name, + "artifact_sha256": artifact_sha256, + "size_bytes": backup_path.stat().st_size, "restore_check": { "type": "sqlite_integrity_check", "result": integrity, @@ -3744,6 +4041,12 @@ def _snapshot_external_database( payload["database_url_secret"] = database_url_secret if result.returncode != 0: raise ModuleInstallerError(f"Database backup command failed ({result.returncode}): {_redact_installer_text(backup_command, redactions=redactions)}") + if not backup_path.is_file() or backup_path.stat().st_size <= 0: + raise ModuleInstallerError( + "Database backup command did not create a non-empty backup artifact." + ) + payload["artifact_sha256"] = _sha256_file(backup_path) + payload["size_bytes"] = backup_path.stat().st_size if restore_check_command: restore_check = _run_database_hook( restore_check_command, diff --git a/src/govoplan_core/core/module_lifecycle_recovery.py b/src/govoplan_core/core/module_lifecycle_recovery.py new file mode 100644 index 0000000..0ce189c --- /dev/null +++ b/src/govoplan_core/core/module_lifecycle_recovery.py @@ -0,0 +1,457 @@ +from __future__ import annotations + +from dataclasses import dataclass +import hashlib +import json +from typing import Mapping, Sequence +from uuid import uuid4 + +from sqlalchemy.exc import SQLAlchemyError +from sqlalchemy.orm import Session, sessionmaker + +from govoplan_core.core.recovery import ( + RecoveryGuaranteeError, + RecoveryMode, + RecoveryOperation, + RecoveryPlan, + RecoveryStatus, +) +from govoplan_core.core.recovery_runtime import ( + DurableRecoveryOperation, + RecoveryOperationBusy, + RecoveryOperationStateConflict, + begin_durable_recovery_operation, + claim_durable_recovery_operation, +) +from govoplan_core.core.runtime_coordination import process_runtime_identity + + +class ModuleLifecycleRecoveryError(RuntimeError): + pass + + +@dataclass(frozen=True, slots=True) +class ModuleLifecycleRecoveryDeclaration: + operation_type: str + mode: RecoveryMode + resources: tuple[str, ...] + verification: tuple[str, ...] + + +MODULE_LIFECYCLE_RECOVERY_OPERATIONS = ( + ModuleLifecycleRecoveryDeclaration( + operation_type="module-lifecycle.pre-migration", + mode=RecoveryMode.COMPENSATION, + resources=("postgresql", "package-environment", "webui-bundle", "filesystem"), + verification=( + "verify the canonical install plan and immutable package references", + "verify the package and WebUI snapshots before mutation", + "verify the installed manifests and desired module graph", + ), + ), + ModuleLifecycleRecoveryDeclaration( + operation_type="module-lifecycle.post-migration", + mode=RecoveryMode.FORWARD_RECOVERY, + resources=( + "postgresql", + "package-environment", + "webui-bundle", + "runtime-nodes", + ), + verification=( + "verify the backup reference and migration execution evidence", + "verify migration heads and installed module manifests", + "verify the desired graph and runtime health before completion", + ), + ), + ModuleLifecycleRecoveryDeclaration( + operation_type="module-retirement.destroy-data", + mode=RecoveryMode.SNAPSHOT_RESTORE, + resources=("postgresql", "object-storage", "package-environment"), + verification=( + "verify the pinned backup artifact and restore-readiness evidence", + "verify the retirement provider result and remaining migration state", + "verify the installed manifests and desired module graph", + ), + ), + ModuleLifecycleRecoveryDeclaration( + operation_type="module-runtime.apply-graph", + mode=RecoveryMode.COMPENSATION, + resources=("postgresql", "runtime-nodes", "module-registry"), + verification=( + "verify the requested graph against available module contracts", + "verify activation and deactivation hooks completed", + "verify the active graph and workflow contribution reconciliation", + ), + ), +) + + +_DECLARATIONS = { + item.operation_type: item for item in MODULE_LIFECYCLE_RECOVERY_OPERATIONS +} + + +def canonical_sha256(value: object) -> str: + encoded = json.dumps( + value, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=True, + default=str, + ).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +def lifecycle_session_factory(session: Session) -> sessionmaker[Session]: + bind = session.get_bind() + if bind is None: + raise ModuleLifecycleRecoveryError( + "Module lifecycle recovery requires a database bind" + ) + return sessionmaker(bind=bind, expire_on_commit=False) + + +@dataclass(slots=True) +class ModuleLifecycleRecovery: + operation: DurableRecoveryOperation | None + operation_id: str + operation_type: str + mode: RecoveryMode + plan_sha256: str + replayed: bool + effect_started: bool = False + + def checkpoint( + self, + *, + kind: str, + summary: str, + evidence: Mapping[str, object], + effect_started: bool = False, + ) -> None: + if self.operation is None: + return + self.effect_started = self.effect_started or effect_started + self.operation.checkpoint( + kind=kind, + summary=summary, + evidence={ + **dict(evidence), + "effect_started": self.effect_started, + "plan_sha256": self.plan_sha256, + }, + ) + + def succeed( + self, + session: Session, + *, + evidence: Mapping[str, object], + commit_projection: bool, + ) -> None: + if self.operation is None: + return + terminal = { + "verified": True, + "checks": { + **dict(evidence), + "plan_sha256": self.plan_sha256, + "effect_started": self.effect_started, + }, + } + if commit_projection: + self.operation.commit_verified_success(session, evidence=terminal) + else: + self.operation.succeed(evidence=terminal) + + def unresolved( + self, + *, + summary: str, + evidence: Mapping[str, object], + outcome_unknown: bool, + ) -> None: + if self.operation is None: + return + if not self.effect_started: + self.operation.fail( + summary=summary, + evidence={ + "verified": True, + "checks": { + **dict(evidence), + "effect_started": False, + }, + }, + ) + return + self.operation.unresolved( + status=( + RecoveryStatus.OUTCOME_UNKNOWN + if outcome_unknown + else RecoveryStatus.RECOVERY_REQUIRED + ), + summary=summary, + evidence={ + **dict(evidence), + "effect_started": True, + "plan_sha256": self.plan_sha256, + }, + failure_summary=( + "Inspect the installer run record and affected state services " + "before retrying or restoring" + ), + ) + + def recovered( + self, + session: Session, + *, + evidence: Mapping[str, object], + summary: str, + ) -> None: + state = session.get(RecoveryOperation, self.operation_id) + if state is None: + raise ModuleLifecycleRecoveryError( + "Module lifecycle recovery operation is unavailable" + ) + if state.status == RecoveryStatus.RECOVERED.value: + return + try: + handle = claim_durable_recovery_operation( + lifecycle_session_factory(session), + identity=process_runtime_identity(), + operation_id=self.operation_id, + lease_ttl_seconds=900, + ) + except RecoveryOperationStateConflict as exc: + if exc.status == RecoveryStatus.RECOVERED.value: + return + raise ModuleLifecycleRecoveryError( + f"Module lifecycle recovery is already {exc.status}" + ) from exc + except (RecoveryOperationBusy, RecoveryGuaranteeError, RuntimeError) as exc: + raise ModuleLifecycleRecoveryError( + "Module lifecycle recovery authority is unavailable" + ) from exc + session.expire_all() + state = session.get(RecoveryOperation, self.operation_id) + if state is None: + raise ModuleLifecycleRecoveryError( + "Module lifecycle recovery operation is unavailable" + ) + recovery_evidence = { + "verified": True, + "checks": { + **dict(evidence), + "plan_sha256": self.plan_sha256, + }, + } + if state.status == RecoveryStatus.OUTCOME_UNKNOWN.value: + handle.resolve_unknown( + effect_occurred=False, + evidence=recovery_evidence, + summary=summary, + ) + else: + handle.compensate( + failure_summary=summary, + failure_evidence={ + "effect_started": self.effect_started, + "plan_sha256": self.plan_sha256, + }, + recovery_evidence=recovery_evidence, + ) + + +def begin_module_installer_recovery( + session: Session, + *, + run_id: str, + plan: Sequence[Mapping[str, object]], + command_count: int, + migrate_database: bool, + destructive_retirement: bool, + snapshot_sha256: str | None, + backup_reference: str | None, + request_context_sha256: str, +) -> ModuleLifecycleRecovery: + operation_type = ( + "module-retirement.destroy-data" + if destructive_retirement + else "module-lifecycle.post-migration" + if migrate_database + else "module-lifecycle.pre-migration" + ) + declaration = _DECLARATIONS[operation_type] + plan_sha256 = canonical_sha256([dict(item) for item in plan]) + recovery_plan = RecoveryPlan( + mode=declaration.mode, + preconditions=( + "maintenance mode and installer preflight are current", + "package references and the requested module graph are pinned", + "the deployment-wide module lifecycle fence is owned", + ), + compensation_steps=( + "restore the Python and WebUI package snapshots", + "restore the prior desired module graph", + "verify installed manifests and runtime health", + ) + if declaration.mode == RecoveryMode.COMPENSATION + else (), + forward_recovery_steps=( + "inspect migration task and command evidence", + "complete or repair migrations under the same deployment fence", + "verify migration heads, manifests, desired graph, and runtime health", + ) + if declaration.mode == RecoveryMode.FORWARD_RECOVERY + else (), + verification_steps=declaration.verification, + backup_reference=( + backup_reference + if declaration.mode == RecoveryMode.SNAPSHOT_RESTORE + else None + ), + ) + if declaration.mode == RecoveryMode.SNAPSHOT_RESTORE and not backup_reference: + raise ModuleLifecycleRecoveryError( + "Destructive module retirement requires verified backup evidence" + ) + session.commit() + try: + started = begin_durable_recovery_operation( + lifecycle_session_factory(session), + identity=process_runtime_identity(), + module_id="core", + operation_type=operation_type, + idempotency_key=f"module-installer:{run_id}", + request={ + "run_id": run_id, + "plan_sha256": plan_sha256, + "command_count": command_count, + "migrate_database": migrate_database, + "destructive_retirement": destructive_retirement, + "snapshot_expected": True, + "request_context_sha256": request_context_sha256, + }, + recovery_plan=recovery_plan, + precondition_evidence={ + "plan_sha256": plan_sha256, + "snapshot_sha256": snapshot_sha256 or "pending", + "request_context_sha256": request_context_sha256, + "command_count": command_count, + "backup_reference_present": bool(backup_reference), + }, + lease_resource_key="core:module-lifecycle:deployment", + lease_ttl_seconds=900, + resource_type="module_installer_run", + resource_id=run_id, + metadata={ + "resources": list(declaration.resources), + "migrate_database": migrate_database, + "destructive_retirement": destructive_retirement, + }, + block_unresolved_resource=True, + ) + except RecoveryOperationBusy as exc: + raise ModuleLifecycleRecoveryError( + "Another runtime owns the deployment module lifecycle fence" + ) from exc + except RecoveryOperationStateConflict as exc: + raise ModuleLifecycleRecoveryError( + f"Module installer recovery is already {exc.status}" + ) from exc + except (RecoveryGuaranteeError, RuntimeError, SQLAlchemyError, ValueError) as exc: + raise ModuleLifecycleRecoveryError( + "The recovery ledger is unavailable; module mutation did not start" + ) from exc + return ModuleLifecycleRecovery( + operation=started.operation, + operation_id=started.operation_id, + operation_type=operation_type, + mode=declaration.mode, + plan_sha256=plan_sha256, + replayed=started.replayed, + ) + + +def begin_runtime_graph_recovery( + session: Session, + *, + previous_modules: Sequence[str], + requested_modules: Sequence[str], + migrate: bool, +) -> ModuleLifecycleRecovery: + declaration = _DECLARATIONS["module-runtime.apply-graph"] + plan = { + "previous_modules": sorted(set(previous_modules)), + "requested_modules": sorted(set(requested_modules)), + "migrate": migrate, + } + plan_sha256 = canonical_sha256(plan) + session.commit() + try: + started = begin_durable_recovery_operation( + lifecycle_session_factory(session), + identity=process_runtime_identity(), + module_id="core", + operation_type=declaration.operation_type, + idempotency_key=f"module-runtime:{uuid4()}", + request={**plan, "plan_sha256": plan_sha256}, + recovery_plan=RecoveryPlan( + mode=declaration.mode, + preconditions=( + "the requested graph passed module contract validation", + "the deployment-wide module lifecycle fence is owned", + ), + compensation_steps=( + "restore the previous in-process active registry", + "reconfigure capability contexts from the previous graph", + ), + verification_steps=declaration.verification, + ), + precondition_evidence={ + "plan_sha256": plan_sha256, + "previous_graph_sha256": canonical_sha256( + sorted(set(previous_modules)) + ), + "requested_graph_sha256": canonical_sha256( + sorted(set(requested_modules)) + ), + }, + lease_resource_key="core:module-lifecycle:deployment", + lease_ttl_seconds=300, + resource_type="module_runtime_graph", + resource_id=plan_sha256, + metadata={"resources": list(declaration.resources)}, + block_unresolved_resource=True, + ) + except (RecoveryOperationBusy, RecoveryOperationStateConflict) as exc: + raise ModuleLifecycleRecoveryError( + "Another lifecycle mutation is active or unresolved" + ) from exc + except (RecoveryGuaranteeError, RuntimeError, SQLAlchemyError, ValueError) as exc: + raise ModuleLifecycleRecoveryError( + "The recovery ledger is unavailable; the active graph was unchanged" + ) from exc + return ModuleLifecycleRecovery( + operation=started.operation, + operation_id=started.operation_id, + operation_type=declaration.operation_type, + mode=declaration.mode, + plan_sha256=plan_sha256, + replayed=started.replayed, + ) + + +__all__ = [ + "MODULE_LIFECYCLE_RECOVERY_OPERATIONS", + "ModuleLifecycleRecovery", + "ModuleLifecycleRecoveryDeclaration", + "ModuleLifecycleRecoveryError", + "begin_module_installer_recovery", + "begin_runtime_graph_recovery", + "canonical_sha256", + "lifecycle_session_factory", +] diff --git a/src/govoplan_core/core/recovery_runtime.py b/src/govoplan_core/core/recovery_runtime.py index d5eaf2f..009e16b 100644 --- a/src/govoplan_core/core/recovery_runtime.py +++ b/src/govoplan_core/core/recovery_runtime.py @@ -512,6 +512,7 @@ def begin_durable_recovery_operation( resource_type: str | None = None, resource_id: str | None = None, metadata: dict[str, Any] | None = None, + block_unresolved_resource: bool = False, ) -> DurableRecoveryStart: if lease_ttl_seconds < 1: raise ValueError("Recovery lease TTL must be at least one second") @@ -539,6 +540,31 @@ def begin_durable_recovery_operation( RecoveryOperation.idempotency_key == idempotency_key, ) ).scalar_one_or_none() + if block_unresolved_resource: + blocking = session.execute( + select(RecoveryOperation).where( + RecoveryOperation.installation_id == identity.installation_id, + RecoveryOperation.lease_resource_key == lease_resource_key, + RecoveryOperation.status.not_in( + ( + RecoveryStatus.SUCCEEDED.value, + RecoveryStatus.REJECTED.value, + RecoveryStatus.FAILED.value, + RecoveryStatus.RECOVERED.value, + RecoveryStatus.MANUAL_INTERVENTION.value, + ) + ), + ) + ).scalars().first() + if blocking is not None and ( + existing is None or blocking.id != existing.id + ): + release_lease(session, claim) + session.commit() + raise RecoveryOperationStateConflict( + blocking.id, + blocking.status, + ) operation = plan_recovery_operation( session, installation_id=identity.installation_id, diff --git a/tests/test_module_system.py b/tests/test_module_system.py index da2652f..f0d039e 100644 --- a/tests/test_module_system.py +++ b/tests/test_module_system.py @@ -20,7 +20,7 @@ from pathlib import Path from types import SimpleNamespace from unittest.mock import patch -from sqlalchemy import Column, Integer, MetaData, Table, create_engine, insert, inspect +from sqlalchemy import Column, Integer, MetaData, Table, create_engine, insert, inspect, select from sqlalchemy.orm import Session # Keep the default app import side effect from bootstrapping a development DB. @@ -93,6 +93,13 @@ from govoplan_core.core.configuration_packages import ( validate_configuration_package_catalog, ) from govoplan_core.core.module_license import issue_module_license, module_license_decision, module_license_diagnostics, validate_module_license +from govoplan_core.core.recovery import ( + RecoveryCheckpoint, + RecoveryOperation, + RecoveryStatus, + verify_recovery_evidence_chain, +) +from govoplan_core.core.runtime_coordination import DistributedLease from govoplan_core.core.module_package_catalog import ( module_package_catalog, record_module_package_catalog_acceptance, @@ -2607,7 +2614,15 @@ finally: settings = _settings(root) configure_database(settings.database_url) database = get_database() - Base.metadata.create_all(bind=database.engine, tables=[SystemSettings.__table__]) + Base.metadata.create_all( + bind=database.engine, + tables=[ + SystemSettings.__table__, + DistributedLease.__table__, + RecoveryOperation.__table__, + RecoveryCheckpoint.__table__, + ], + ) metadata = MetaData() table = Table("retirement_example", metadata, Column("id", Integer, primary_key=True)) metadata.create_all(bind=database.engine) @@ -2658,6 +2673,9 @@ finally: database_url=settings.database_url, runtime_dir=root / "installer", ) + recovery = session.execute(select(RecoveryOperation)).scalar_one() + self.assertEqual(RecoveryStatus.SUCCEEDED.value, recovery.status) + self.assertTrue(verify_recovery_evidence_chain(session, recovery.id)) self.assertEqual("applied", result.status) self.assertFalse(inspect(database.engine).has_table("retirement_example")) @@ -2813,7 +2831,15 @@ finally: settings = _settings(root) configure_database(settings.database_url) database = get_database() - Base.metadata.create_all(bind=database.engine, tables=[SystemSettings.__table__]) + Base.metadata.create_all( + bind=database.engine, + tables=[ + SystemSettings.__table__, + DistributedLease.__table__, + RecoveryOperation.__table__, + RecoveryCheckpoint.__table__, + ], + ) def fake_run(*_args, **kwargs): argv = tuple(_args[0]) if _args else () @@ -2846,11 +2872,78 @@ finally: restored_desired = saved_desired_enabled_modules(session, ("tenancy", "access")) restored_plan = saved_module_install_plan(session) + recovery = session.execute(select(RecoveryOperation)).scalar_one() + self.assertEqual(RecoveryStatus.RECOVERED.value, recovery.status) + self.assertTrue(verify_recovery_evidence_chain(session, recovery.id)) self.assertEqual("rolled-back", result.status) self.assertEqual(("tenancy", "access"), restored_desired) self.assertEqual(("planned",), tuple(item.status for item in restored_plan.items)) + def test_module_installer_blocks_after_unresolved_package_effect(self) -> None: + root = Path(tempfile.mkdtemp(prefix="govoplan-installer-unresolved-", dir=_TEST_ROOT)) + settings = _settings(root) + configure_database(settings.database_url) + database = get_database() + Base.metadata.create_all( + bind=database.engine, + tables=[ + SystemSettings.__table__, + DistributedLease.__table__, + RecoveryOperation.__table__, + RecoveryCheckpoint.__table__, + ], + ) + + def fail_package_install(*args, **_kwargs): + argv = tuple(args[0]) if args else () + if any("govoplan-example==0.1.4" in str(item) for item in argv): + return SimpleNamespace(returncode=1, stdout="", stderr="install failed") + return SimpleNamespace(returncode=0, stdout="", stderr="") + + with database.session() as session: + save_maintenance_mode(session, MaintenanceMode(enabled=True)) + plan = save_module_install_plan(session, [{ + "module_id": "example", + "action": "install", + "python_package": "govoplan-example", + "python_ref": "govoplan-example==0.1.4", + }]) + session.commit() + + with patch( + "govoplan_core.core.module_installer.subprocess.run", + side_effect=fail_package_install, + ): + result = run_module_install_plan( + session=session, + plan=plan, + available=available_module_manifests(), + current_enabled=("tenancy", "access"), + desired_enabled=("tenancy", "access"), + database_url=settings.database_url, + runtime_dir=root / "installer", + ) + + self.assertEqual("failed", result.status) + recovery = session.execute(select(RecoveryOperation)).scalar_one() + self.assertEqual(RecoveryStatus.RECOVERY_REQUIRED.value, recovery.status) + self.assertTrue(verify_recovery_evidence_chain(session, recovery.id)) + + with self.assertRaisesRegex( + module_installer_module.ModuleInstallerError, + "already recovery_required", + ): + run_module_install_plan( + session=session, + plan=plan, + available=available_module_manifests(), + current_enabled=("tenancy", "access"), + desired_enabled=("tenancy", "access"), + database_url=settings.database_url, + runtime_dir=root / "installer", + ) + def test_module_installer_external_database_backup_command_is_recorded(self) -> None: root = Path(tempfile.mkdtemp(prefix="govoplan-installer-external-backup-", dir=_TEST_ROOT)) settings = _settings(root) @@ -4216,6 +4309,15 @@ finally: app, _settings_obj = self._app_for_modules(()) lifecycle = getattr(app.state, "govoplan_lifecycle", None) self.assertIsNotNone(lifecycle) + database = get_database() + Base.metadata.create_all( + bind=database.engine, + tables=[ + DistributedLease.__table__, + RecoveryOperation.__table__, + RecoveryCheckpoint.__table__, + ], + ) with TestClient(app) as client: response = client.get("/api/v1/platform/modules") diff --git a/tests/test_recovery_runtime.py b/tests/test_recovery_runtime.py index b5de02a..808dc66 100644 --- a/tests/test_recovery_runtime.py +++ b/tests/test_recovery_runtime.py @@ -50,7 +50,13 @@ def _identity(node: str, incarnation: str) -> RuntimeIdentity: ) -def _start(factory, identity, *, key: str = "build-1"): +def _start( + factory, + identity, + *, + key: str = "build-1", + block_unresolved_resource: bool = False, +): return begin_durable_recovery_operation( factory, identity=identity, @@ -68,6 +74,7 @@ def _start(factory, identity, *, key: str = "build-1"): lease_resource_key="campaign:build:version-1", resource_type="campaign_version", resource_id="version-1", + block_unresolved_resource=block_unresolved_resource, ) @@ -273,6 +280,36 @@ def test_other_runtime_cannot_use_an_active_fence() -> None: engine.dispose() +def test_unresolved_predecessor_can_block_new_effects_on_same_resource() -> None: + engine, factory = _fixture() + try: + started = _start( + factory, + _identity("worker-1", "incarnation-1"), + block_unresolved_resource=True, + ) + assert started.operation is not None + started.operation.unresolved( + status=RecoveryStatus.OUTCOME_UNKNOWN, + summary="Provider outcome is unknown", + evidence={"request_sent": True}, + failure_summary="Reconcile before retry", + ) + + with pytest.raises( + RecoveryOperationStateConflict, + match="outcome_unknown", + ): + _start( + factory, + _identity("worker-2", "incarnation-2"), + key="build-2", + block_unresolved_resource=True, + ) + finally: + engine.dispose() + + def test_expired_crash_fence_is_taken_over_as_recovery_required() -> None: engine, factory = _fixture() try: diff --git a/tests/test_workflow_contribution_lifecycle.py b/tests/test_workflow_contribution_lifecycle.py index c8c309f..fbab1e3 100644 --- a/tests/test_workflow_contribution_lifecycle.py +++ b/tests/test_workflow_contribution_lifecycle.py @@ -10,11 +10,18 @@ from govoplan_core.core.modules import ( ModuleManifest, ) from govoplan_core.core.registry import PlatformRegistry +from govoplan_core.core.recovery import RecoveryCheckpoint, RecoveryOperation +from govoplan_core.core.runtime_coordination import ( + DistributedLease, + RuntimeIdentity, + bind_process_runtime_identity, +) from govoplan_core.core.workflows import ( CAPABILITY_WORKFLOW_DEFINITION_CONTRIBUTIONS, WorkflowDefinitionContribution, ) -from govoplan_core.db.session import configure_database, reset_database +from govoplan_core.db.base import Base +from govoplan_core.db.session import configure_database, get_database, reset_database class _ContributionProvider: @@ -30,8 +37,28 @@ class _ContributionProvider: class WorkflowContributionLifecycleTests(unittest.TestCase): def setUp(self) -> None: configure_database("sqlite:///:memory:") + database = get_database() + Base.metadata.create_all( + bind=database.engine, + tables=[ + DistributedLease.__table__, + RecoveryOperation.__table__, + RecoveryCheckpoint.__table__, + ], + ) + bind_process_runtime_identity( + RuntimeIdentity( + installation_id="test-installation", + node_id="test-node", + incarnation="test-incarnation", + role="test", + software_version="test", + composition_hash="0" * 64, + ) + ) def tearDown(self) -> None: + bind_process_runtime_identity(None) reset_database(dispose=True) def test_active_graph_change_reconciles_module_workflow_baselines(self) -> None: