Fix external deployment recovery preflight

CHANGELOG.md

@@ -4,16 +4,19 @@
 
 - Успешный CI на `push` в `main` запускает release workflow: при настроенных
   registry secrets он собирает образы rootless Kaniko, публикует digest
-  references и сохраняет schema-v3 descriptor с SHA-256 в Gitea artifact;
+  references и сохраняет schema-v4 descriptor с SHA-256 в Gitea artifact;
   sidecar проверяется после извлечения из корня. Следующий workflow проверяет
   source-run identity и, при доступном `gitea-api`, метаданные артефакта только
   в dry-run: ZIP не скачивается, SSH и host apply не выполняются. Подготовлены
   owner-run cutover и rollback с
   проверкой backup, image IDs, RU/KY canaries и recovery journal; CI проверяет
-  synthetic artifact и завершается ошибкой при отсутствии файлов. Полный backup
-  на VPS запрещён, прежние локальные процедуры устарели; внешний
-  Windows-controlled backup/restore ещё не реализован и остаётся обязательным
-  gate до production cutover.
+  synthetic artifact и завершается ошибкой при отсутствии файлов. Внешний
+  Windows-controlled backup/restore реализован в PR, но ещё не установлен и не
+  проверен на Windows или VPS; production cutover остаётся заблокирован до
+  отдельного restore-теста и owner bootstrap. Принимаются только завершённые
+  успешные Gitea runs; pending recovery journal записывается до остановки
+  сервисов, повторно применённый release отклоняется до простоя, а проверка
+  `main` использует явно заданный production checkout.
 - Frontend `0.3.3` / Backend `0.13.2`: управление пользователями отображается
   согласно правам аккаунта, подписи ролей локализованы, повторяющиеся действия
   кабинета убраны, а знак лаборатории приведён к основному бренду.

docs/operations/BACKUP.md

@@ -18,7 +18,9 @@ verified production backup, so production deployment remains blocked.
   not create a local archive or temporary full-size copy on the VPS.
 - The Windows receiver requires the exact D: repository, a real NTFS path,
   restrictive ACLs, and the exact `/.backups/` rule in that checkout's local
-  `.git/info/exclude`. It writes into a private staging directory, checks the
+  `.git/info/exclude`. Before SSH connects, it also requires at least 100 GiB
+  plus 1 MiB of free D: space, matching the protocol's maximum archive size.
+  It writes into a private staging directory, checks the
   stream size and SHA-256, verifies the archive and manifest, then publishes
   the complete backup directory without overwriting an existing ID.
 - Only after Windows confirms the stored archive identity does the VPS proceed

docs/operations/CURRENT_STATE.md

@@ -23,6 +23,12 @@ remains blocked by the backup gate.
   identity. The rollback handshake requires the exact previously pinned
   archive and creates a fresh external backup before image references change.
   Local VPS archive paths fail closed.
+- Review fixes in this PR require Gitea `status=completed` with
+  `conclusion=success`, resolve `main` from `/opt/akyldash/current`, write a
+  durable pending journal before stopping services, and reject an already
+  applied release before downtime. The Windows client checks D: capacity,
+  NTFS/path/ACL, and the local ignore rule before SSH; it requires 100 GiB plus
+  1 MiB free. These code changes remain unverified on Windows and the VPS.
 - Synthetic tests cover archive/protocol integrity, helper download rules,
   artifact/run provenance, receiver publication, release schema, and rollback
   refusal. Windows ACL, NTFS, PowerShell, SSH alias, Gitea helper against the

docs/operations/DEPLOYMENT.md

@@ -62,6 +62,13 @@ only, not data. Old local VPS backup paths and legacy release states without a
 pinned external backup identity are rejected. See [backup and restore](BACKUP.md)
 and [rollback](ROLLBACK.md).
 
+The receiver records a durable pending recovery journal before stopping
+services. A retry of the exact release already recorded in pinned state is
+rejected before backup or downtime. Gitea source verification requires
+`status=completed` and `conclusion=success`, and resolves current `main` from
+the explicit production checkout. The Windows client checks the destination,
+ACL, ignore rule, and full protocol archive capacity before opening SSH.
+
 ## Owner bootstrap still required
 
 After review and merge, the owner must decide whether to install the helper

infra/backup/receive_external.py

@@ -16,7 +16,7 @@ from pathlib import Path
 
 from archive import verify_external_archive
 from external_session import (ArchiveFrameWriter, read_challenge, read_response, read_result,
-                              receive_archive, write_ack, write_request)
+                              MAX_ARCHIVE_BYTES, receive_archive, write_ack, write_request)
 
 DEFAULT_REPOSITORY = Path(r"D:\Dev\akyldash")
 MAX_ERROR_BYTES = 8192
@@ -63,6 +63,11 @@ if (-not $ownerAllowed) { exit 4 }
         raise ValueError(f"backup path ACL is not private (PowerShell status {result.returncode})")
 
 
+def check_capacity(free_bytes: int) -> None:
+    if free_bytes < MAX_ARCHIVE_BYTES + 1024 * 1024:
+        raise ValueError("D: must have room for the maximum 100 GiB archive plus staging metadata")
+
+
 def check_repository_target(repository: Path, backup_id: str, *, allow_existing: bool = False) -> tuple[Path, Path]:
     if not repository.is_absolute() or not repository.is_dir() or _reparse(repository):
         raise ValueError("repository must be an existing real directory on the selected Windows volume")
@@ -70,6 +75,8 @@ def check_repository_target(repository: Path, backup_id: str, *, allow_existing:
         raise ValueError("external backup destination is restricted to D:\\Dev\\akyldash")
     if os.name == "nt" and repository.resolve().as_posix().lower() != "d:/dev/akyldash":
         raise ValueError("external backup destination is restricted to D:\\Dev\\akyldash")
+    if os.name == "nt":
+        check_capacity(shutil.disk_usage(repository).free)
     for parent in repository.parents:
         if _reparse(parent):
             raise ValueError("backup destination has a reparse-point parent")

infra/backup/test_receive_external.py

@@ -13,6 +13,12 @@ from external_session import ArchiveFrameWriter, read_ack, write_header
 
 
 class ExternalReceiverTests(unittest.TestCase):
+    def test_windows_preflight_requires_full_protocol_archive_capacity(self):
+        required = receive_external.MAX_ARCHIVE_BYTES + 1024 * 1024
+        receive_external.check_capacity(required)
+        with self.assertRaisesRegex(ValueError, "maximum 100 GiB archive"):
+            receive_external.check_capacity(required - 1)
+
     def setUp(self):
         self.request = {
             "version": 1, "action": "release", "run_id": "301", "commit": "a" * 40,

infra/deploy/deploy_release.py

@@ -146,7 +146,7 @@ def deploy_external(
     if not re.fullmatch(r"[0-9a-f]{64}", descriptor_sha256):
         raise ValueError("release descriptor digest is invalid")
     from release_handoff import latest_main_sha
-    if latest_main_sha() != manifest["commit"]:
+    if latest_main_sha(ROOT) != manifest["commit"]:
         raise ValueError("main advanced after the external backup; refusing a stale release cutover")
     if not isinstance(current_before_stop, dict) or image_ids(allow_stopped=True) != current_before_stop:
         raise ValueError("stopped services do not match the pre-backup image snapshot")

infra/deploy/external_deploy.py

@@ -50,7 +50,10 @@ def execute(request: dict, *, reader, writer, policy: dict | None = None, api=No
     canary_ru, canary_ky = canaries or owner_canaries()
     if JOURNAL_FILE.exists() or JOURNAL_FILE.is_symlink():
         raise ValueError(f"recovery journal exists at {JOURNAL_FILE}; resolve it before another release")
-    manifest, _manifest_bytes = fetch_release(request, policy=policy, current_main=current_main, **({"api": api} if api is not None else {}))
+    manifest, _manifest_bytes = fetch_release(
+        request, policy=policy, current_main=current_main, checkout=ROOT,
+        **({"api": api} if api is not None else {}),
+    )
 
     with deployment_lock():
         if JOURNAL_FILE.exists() or JOURNAL_FILE.is_symlink():
@@ -65,6 +68,14 @@ def execute(request: dict, *, reader, writer, policy: dict | None = None, api=No
             previous_state = validate_state(read_json(STATE_FILE))
             if previous_state["current"] != current_record:
                 raise ValueError("current deployment does not match the recorded release state")
+            if (previous_state["commit"] == manifest["commit"]
+                    and previous_state["manifest_sha256"] == request["descriptor_sha256"]):
+                raise ValueError("release is already applied")
+        journal = {
+            "schema_version": 2, "status": "external-backup-pending", "commit": manifest["commit"],
+            "manifest_sha256": request["descriptor_sha256"], "previous": current_record,
+        }
+        write_json_atomic(JOURNAL_FILE, journal)
         try:
             compose("stop")
             sources, stopped_revision, stopped_images, deployment_files = compose_sources(COMPOSE_FILE, ENV_FILE)
@@ -78,13 +89,12 @@ def execute(request: dict, *, reader, writer, policy: dict | None = None, api=No
         except Exception as error:
             try:
                 restart_current(current)
+                JOURNAL_FILE.unlink(missing_ok=True)
             except Exception as recovery_error:
-                write_json_atomic(JOURNAL_FILE, {
-                    "schema_version": 2, "status": "external-backup-failed",
-                    "error_type": type(error).__name__,
-                    "recovery_error_type": type(recovery_error).__name__,
-                    "previous": current_record,
-                })
+                journal.update({"status": "external-backup-failed",
+                                "error_type": type(error).__name__,
+                                "recovery_error_type": type(recovery_error).__name__})
+                write_json_atomic(JOURNAL_FILE, journal)
                 raise ValueError(f"external backup failed and current services need recovery; journal at {JOURNAL_FILE}") from error
             raise
 

infra/deploy/external_rollback.py

@@ -51,6 +51,8 @@ def execute(*, reader, writer) -> str:
         current = image_ids()
         if current != state["current"]:
             raise ValueError("current deployment does not match root-pinned release state")
+        journal = {"schema_version": 2, "status": "external-backup-pending", "previous": current}
+        write_json_atomic(JOURNAL_FILE, journal)
         try:
             compose("stop")
             sources, revision, stopped_images, deployment_files = compose_sources(COMPOSE_FILE, ENV_FILE)
@@ -68,13 +70,12 @@ def execute(*, reader, writer) -> str:
                 compose("up", "--detach", "--no-build", "--wait", "--wait-timeout", "300")
                 if image_ids() != current:
                     raise ValueError("restarted image IDs differ from the pre-rollback deployment")
+                JOURNAL_FILE.unlink(missing_ok=True)
             except Exception as recovery_error:
-                write_json_atomic(JOURNAL_FILE, {
-                    "schema_version": 2, "status": "external-backup-failed",
-                    "error_type": type(error).__name__,
-                    "recovery_error_type": type(recovery_error).__name__,
-                    "previous": current,
-                })
+                journal.update({"status": "external-backup-failed",
+                                "error_type": type(error).__name__,
+                                "recovery_error_type": type(recovery_error).__name__})
+                write_json_atomic(JOURNAL_FILE, journal)
                 raise ValueError(f"rollback backup failed and services need recovery; journal at {JOURNAL_FILE}") from error
             raise
 

infra/deploy/release_handoff.py

@@ -15,9 +15,9 @@ RUN_ID_RE = re.compile(r"[0-9]+\Z")
 DESCRIPTOR_RE = re.compile(r"release-descriptor-([0-9]+)-([0-9a-f]{64})\Z")
 
 
-def latest_main_sha() -> str:
+def latest_main_sha(checkout: Path | None = None) -> str:
     result = subprocess.run(
-        ["git", "ls-remote", "--exit-code", "origin", "refs/heads/main"],
+        ["git", "-C", str(checkout or Path.cwd()), "ls-remote", "--exit-code", "origin", "refs/heads/main"],
         check=True,
         text=True,
         capture_output=True,

infra/deploy/release_source.py

@@ -36,13 +36,14 @@ def ensure_private_scratch(path: Path) -> None:
         raise ValueError("artifact scratch directory must be a private root-owned directory")
 
 
-def verify_run(run_id: str, *, stage: str, commit: str, api=api_json, current_main: str | None = None) -> dict:
+def verify_run(run_id: str, *, stage: str, commit: str, api=api_json, current_main: str | None = None,
+               checkout: Path | None = None) -> dict:
     if not run_id.isdigit() or not re.fullmatch(r"[0-9a-f]{40}", commit):
         raise ValueError("Gitea run identity is invalid")
     run = api(f"/repos/{REPOSITORY}/actions/runs/{run_id}")
     if not isinstance(run, dict) or str(run.get("id")) != run_id:
         raise ValueError("Gitea returned a different workflow run")
-    if run.get("status") != "success" or run.get("conclusion") not in {None, "success"}:
+    if run.get("status") not in {"completed", "success"} or run.get("conclusion") != "success":
         raise ValueError("workflow run is not successful")
     workflow_id = run.get("workflow_id")
     if not isinstance(workflow_id, (int, str)) or not str(workflow_id):
@@ -66,7 +67,7 @@ def verify_run(run_id: str, *, stage: str, commit: str, api=api_json, current_ma
         workflow=expected_workflow, workflow_path=f"{Path(path).name}@refs/heads/main", event=event,
         conclusion="success", branch=branch, run_id=run_id, head_sha=commit,
         repository=REPOSITORY, stage="release" if stage == "ci" else "handoff",
-        current_main=current_main or latest_main_sha(),
+        current_main=current_main or latest_main_sha(checkout),
     )
     return run
 
@@ -128,13 +129,14 @@ def validate_artifact_zip(path: Path, *, descriptor_sha256: str, release_run_id:
 
 def fetch_release(request: dict, *, policy: object, helper: str = "gitea-api",
                   temp_dir: Path = Path("/run/akyldash"), env: dict | None = None,
-                  api=api_json, current_main: str | None = None) -> tuple[dict, bytes]:
+                  api=api_json, current_main: str | None = None,
+                  checkout: Path | None = None) -> tuple[dict, bytes]:
     run_id, commit, descriptor_sha256 = (
         request["run_id"], request["commit"], request["descriptor_sha256"]
     )
-    current_main = current_main or latest_main_sha()
+    current_main = current_main or latest_main_sha(checkout)
     release_run = verify_run(run_id, stage="handoff", commit=commit, api=api,
-                             current_main=current_main)
+                             current_main=current_main, checkout=checkout)
     metadata = api(f"/repos/{REPOSITORY}/actions/runs/{run_id}/artifacts")
     tuple_data = descriptor_request(metadata, run_id=run_id, head_sha=commit)
     if tuple_data["descriptor_sha256"] != descriptor_sha256:
@@ -157,5 +159,6 @@ def fetch_release(request: dict, *, policy: object, helper: str = "gitea-api",
             ci_run_id=None, commit=commit, policy=policy,
         )
         source_run_id = manifest["source_run_id"]
-        verify_run(source_run_id, stage="ci", commit=commit, api=api, current_main=current_main)
+        verify_run(source_run_id, stage="ci", commit=commit, api=api, current_main=current_main,
+                   checkout=checkout)
         return manifest, manifest_bytes

infra/deploy/test_external_deploy.py

@@ -0,0 +1,74 @@
+import json
+import tempfile
+import unittest
+from contextlib import nullcontext
+from pathlib import Path
+from unittest.mock import Mock, patch
+
+import external_deploy
+
+
+COMMIT = "a" * 40
+DESCRIPTOR = "d" * 64
+CURRENT = {
+    "backend": {"configured": "registry/backend:current", "image_id": "sha256:" + "1" * 64},
+    "opensearch": {"configured": "registry/search:current", "image_id": "sha256:" + "2" * 64},
+}
+
+
+def release_state(commit: str, descriptor: str) -> dict:
+    return {
+        "schema_version": 1, "commit": commit, "manifest_sha256": descriptor,
+        "current": CURRENT, "previous": CURRENT,
+    }
+
+
+class ExternalDeployTests(unittest.TestCase):
+    def run_execute(self, root: Path, state: dict, *, compose_mock=None):
+        compose_mock = compose_mock or Mock()
+        state_file = root / "state.json"
+        state_file.write_text("{}", encoding="utf-8")
+        journal = root / "journal.json"
+        request = {"version": 1, "action": "release", "run_id": "401",
+                   "commit": COMMIT, "descriptor_sha256": DESCRIPTOR}
+        with (
+            patch.object(external_deploy.os, "geteuid", return_value=0),
+            patch.object(external_deploy, "validate_request"),
+            patch.object(external_deploy, "POLICY_FILE", root / "policy.json"),
+            patch.object(external_deploy, "CANARIES_FILE", root / "canaries.json"),
+            patch.object(external_deploy, "JOURNAL_FILE", journal),
+            patch.object(external_deploy, "STATE_FILE", state_file),
+            patch.object(external_deploy, "fetch_release", return_value=({"commit": COMMIT}, b"")),
+            patch.object(external_deploy, "deployment_lock", return_value=nullcontext()),
+            patch.object(external_deploy, "image_ids", return_value=CURRENT),
+            patch.object(external_deploy, "deploy_release_revision", return_value=COMMIT),
+            patch.object(external_deploy, "read_json", return_value=state),
+            patch.object(external_deploy, "compose", compose_mock),
+        ):
+            result = external_deploy.execute(
+                request, reader=None, writer=None, policy={}, current_main=COMMIT, canaries=("ru", "ky")
+            )
+            return result, compose_mock, journal
+
+    def test_same_verified_release_is_rejected_before_stopping_services(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            compose = Mock()
+            with self.assertRaisesRegex(ValueError, "already applied"):
+                self.run_execute(root, release_state(COMMIT, DESCRIPTOR), compose_mock=compose)
+            self.assertFalse((root / "journal.json").exists())
+            compose.assert_not_called()
+
+    def test_pending_journal_is_durable_before_service_stop(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            compose = Mock(side_effect=KeyboardInterrupt)
+            with self.assertRaises(KeyboardInterrupt):
+                self.run_execute(root, release_state("b" * 40, "e" * 64), compose_mock=compose)
+            journal = json.loads((root / "journal.json").read_text(encoding="utf-8"))
+            self.assertEqual(journal["status"], "external-backup-pending")
+            self.assertEqual(journal["commit"], COMMIT)
+
+
+if __name__ == "__main__":
+    unittest.main()

infra/deploy/test_external_rollback.py

@@ -16,6 +16,44 @@ from release_state import read_json, validate_state
 
 
 class ExternalRollbackTests(unittest.TestCase):
+    def test_pending_journal_is_durable_before_rollback_service_stop(self):
+        manifest = {"git_revision": "a" * 40}
+        manifest_sha = hashlib.sha256(
+            json.dumps(manifest, sort_keys=True, separators=(",", ":")).encode()
+        ).hexdigest()
+        identity = {
+            "backup_id": "akyldash-20261031T120000Z-12345678", "size": 64,
+            "archive_sha256": "b" * 64, "manifest_sha256": manifest_sha, "manifest": manifest,
+        }
+        current = {
+            "backend": {"configured": "registry/backend:current", "image_id": "sha256:" + "1" * 64},
+            "opensearch": {"configured": "registry/search:current", "image_id": "sha256:" + "2" * 64},
+        }
+        state = {
+            "schema_version": 2, "commit": "c" * 40, "manifest_sha256": "d" * 64,
+            "current": current, "previous": current, "previous_backup": identity,
+        }
+        request = io.BytesIO()
+        write_request(request, {"version": 1, "action": "rollback", "data_compatible": True})
+        request.seek(0)
+        with tempfile.TemporaryDirectory() as directory:
+            journal = Path(directory) / "journal.json"
+            with (
+                patch.object(external_rollback.os, "geteuid", return_value=0),
+                patch.object(external_rollback, "STATE_FILE", Path(directory) / "state.json"),
+                patch.object(external_rollback, "JOURNAL_FILE", journal),
+                patch.object(external_rollback, "read_json", return_value=state),
+                patch.object(external_rollback, "deployment_lock", return_value=nullcontext()),
+                patch.object(external_rollback, "nonce", return_value="e" * 64),
+                patch.object(external_rollback, "receive_archive", return_value=(64, "b" * 64)),
+                patch.object(external_rollback, "read_ack"),
+                patch.object(external_rollback, "image_ids", return_value=current),
+                patch.object(external_rollback, "compose", side_effect=KeyboardInterrupt),
+            ):
+                with self.assertRaises(KeyboardInterrupt):
+                    external_rollback.execute(reader=request, writer=io.BytesIO())
+            self.assertEqual(read_json(journal)["status"], "external-backup-pending")
+
     def test_release_state_rejects_boolean_schema_version(self):
         with self.assertRaisesRegex(ValueError, "unsupported schema"):
             validate_state({

infra/deploy/test_release_source.py

@@ -50,7 +50,7 @@ class ReleaseSourceTests(unittest.TestCase):
     def test_verify_run_binds_run_id_workflow_path_event_branch_commit_and_main(self):
         def api(path):
             if path.endswith("/actions/runs/301"):
-                return {"id": 301, "workflow_id": "ci.yaml", "name": "CI", "status": "success",
+                return {"id": 301, "workflow_id": "ci.yaml", "name": "CI", "status": "completed",
                         "conclusion": "success", "head_sha": COMMIT, "head_branch": "main", "event": "push"}
             return {"id": "ci.yaml", "name": "CI", "path": ".gitea/workflows/ci.yaml"}
         run = release_source.verify_run(CI_RUN, stage="ci", commit=COMMIT, api=api, current_main=COMMIT)
@@ -59,6 +59,18 @@ class ReleaseSourceTests(unittest.TestCase):
             release_source.verify_run(CI_RUN, stage="ci", commit=COMMIT,
                                       api=lambda path: {"id": 301, "status": "failure"}, current_main=COMMIT)
 
+    def test_completed_run_requires_success_conclusion(self):
+        responses = {
+            f"/repos/admin/akyldash/actions/runs/{CI_RUN}": {
+                "id": int(CI_RUN), "workflow_id": "ci.yaml", "status": "completed",
+                "conclusion": "cancelled", "head_sha": COMMIT,
+                "head_branch": "main", "event": "push",
+            },
+        }
+        with self.assertRaisesRegex(ValueError, "workflow run is not successful"):
+            release_source.verify_run(CI_RUN, stage="ci", commit=COMMIT,
+                                      api=lambda path: responses[path], current_main=COMMIT)
+
     def test_zip_descriptor_checks_exact_members_digest_source_ids_and_allowlist(self):
         with tempfile.TemporaryDirectory() as directory:
             path = Path(directory) / "artifact.zip"
@@ -120,7 +132,7 @@ class ReleaseSourceTests(unittest.TestCase):
         run_responses = {
             f"/repos/admin/akyldash/actions/runs/{RELEASE_RUN}": {
                 "id": int(RELEASE_RUN), "workflow_id": "release-images.yaml", "name": "Release production images",
-                "status": "success", "conclusion": "success", "head_sha": COMMIT,
+                "status": "completed", "conclusion": "success", "head_sha": COMMIT,
                 "head_branch": "main", "event": "workflow_run",
             },
             "/repos/admin/akyldash/actions/workflows/release-images.yaml": {
@@ -128,7 +140,7 @@ class ReleaseSourceTests(unittest.TestCase):
                 "path": ".gitea/workflows/release-images.yaml",
             },
             f"/repos/admin/akyldash/actions/runs/{CI_RUN}": {
-                "id": int(CI_RUN), "workflow_id": "ci.yaml", "name": "CI", "status": "success",
+                "id": int(CI_RUN), "workflow_id": "ci.yaml", "name": "CI", "status": "completed",
                 "conclusion": "success", "head_sha": COMMIT, "head_branch": "main", "event": "push",
             },
             "/repos/admin/akyldash/actions/workflows/ci.yaml": {

infra/deploy/test_release_workflow.py

@@ -43,6 +43,14 @@ def workflow_step_env(path, step_name, payload):
 
 
 class ReleaseWorkflowTests(unittest.TestCase):
+    def test_latest_main_lookup_uses_explicit_checkout(self):
+        with tempfile.TemporaryDirectory() as directory, patch.object(
+            release_handoff.subprocess, "run",
+            return_value=type("Result", (), {"stdout": "a" * 40 + "\trefs/heads/main\n"})(),
+        ) as run:
+            self.assertEqual(release_handoff.latest_main_sha(Path(directory)), "a" * 40)
+        self.assertEqual(run.call_args.args[0][:3], ["git", "-C", directory])
+
     def test_publisher_runs_only_after_successful_main_ci_and_binds_source_sha(self):
         workflow = (ROOT / ".gitea/workflows/release-images.yaml").read_text(encoding="utf-8")
         self.assertIn('workflows: ["CI"]', workflow)