#!/usr/bin/env python3 """Plan, register, apply and recover a full versioned deployment ZIP.""" import argparse import contextlib import copy import fcntl import json import os from pathlib import Path import stat import sys import tempfile import uuid from bundle_format import (APP_SERVICES, INFRA_SERVICES, OPTIONAL_INFRA_SERVICES, STATE, BundleError, canonical, checked_path, digest, extract_bundle, file_record, plan_files, read_json, write_json) from update_runtime import Runtime def atomic_file(source, target, mode): target.parent.mkdir(parents=True, exist_ok=True) fd, temporary = tempfile.mkstemp(prefix=".bundle-", dir=target.parent) try: with os.fdopen(fd, "wb") as stream: stream.write(source.read_bytes()) os.fchmod(stream.fileno(), mode) stream.flush() os.fsync(stream.fileno()) os.replace(temporary, target) fd = os.open(target.parent, os.O_RDONLY | os.O_DIRECTORY) try: os.fsync(fd) finally: os.close(fd) finally: if os.path.exists(temporary): os.unlink(temporary) class Updater: def __init__(self, root, mode, runtime=None, timeout=300): self.root = Path(root).resolve() self.state = self.root / STATE if self.state.is_symlink(): raise BundleError("update state must not be a symlink") self.mode = mode self.runtime = runtime or Runtime(self.root, mode, Path(__file__).parent, timeout) @contextlib.contextmanager def lock(self): self.state.mkdir(mode=0o700, exist_ok=True) self.state.chmod(0o700) with open(self.state / "lock", "a") as stream: try: fcntl.flock(stream, fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError as exc: raise BundleError("another updater holds the installation lock") from exc yield def installed(self): if not (self.state / "installed.json").exists(): raise BundleError("unregistered installation: register a trusted matching baseline ZIP first") current = read_json(self.state / "installed.json") if (current["root"] != str(self.root) or current["mode"] != self.mode or current["project"] != self.runtime.project): raise BundleError("installation path, deployment mode or Compose project changed") return current def pending(self): path = self.state / "pending.json" return read_json(path) if path.exists() else None def require_idle(self): if self.pending(): raise BundleError("unfinished update exists; inspect status and run recover before applying another ZIP") def register(self, manifest, apply=False): self.require_idle() if (self.state / "installed.json").exists(): raise BundleError("installation is already registered") for name, entry in manifest["files"].items(): path = checked_path(self.root, name) if not path.exists() or file_record(path) != entry: raise BundleError("baseline does not match installed managed file: " + name) for key in manifest["required_env"]: if not self.runtime.env.get(key): raise BundleError("missing required .env key: " + key) plan = self.runtime.plan(self.root) snapshot = self.runtime.snapshot(plan) actual = {x["image"] for x in snapshot.values() if x["service"] in APP_SERVICES} if actual != {manifest["application"]["id"]}: raise BundleError("baseline app image differs from running installation") self.runtime.check_schema(plan, manifest) current = {"manifest": manifest, "root": str(self.root), "mode": self.mode, "project": self.runtime.project, "applied_tasks": [], "plan": plan, "env_sha256": digest((self.root / ".env").read_bytes())} # Registration does not assume historical data tasks ran: new releases replay their # declared cumulative idempotent tasks. No unmanaged file is adopted or overwritten. if apply: write_json(self.state / "installed.json", current) return {"action": "register" if apply else "register-plan", "release": manifest["release"], "managed_files": len(manifest["files"])} def preflight(self, target, source): self.require_idle() current = self.installed() before = current["manifest"] files = plan_files(self.root, before, target) if target["release"] == before["release"]: if canonical(target) != canonical(before): raise BundleError("same release identifier has different content") return {"noop": True, "release": target["release"]} if (before["commit"] not in target["ancestors"] and before["commit"] != target["commit"] or before["contract"] != target["contract"]): raise BundleError("release lineage/contract does not permit this upgrade (including skipped versions)") previous_tasks = {t["id"]: t for t in before["data_updates"]} target_tasks = {t["id"]: t for t in target["data_updates"]} if any(target_tasks.get(k) != v for k, v in previous_tasks.items()): raise BundleError("release data tasks must be cumulative and existing task IDs immutable") missing = [k for k in target["required_env"] if not self.runtime.env.get(k)] if missing: raise BundleError("add missing required values to .env, then retry: " + ", ".join(missing)) old_plan = copy.deepcopy(current["plan"]) app = target["application"] new_plan = self.runtime.plan(source, app["ref"] if app["action"] == "replace" else old_plan["services"]["web"]["image"]) if old_plan["volumes"] != new_plan["volumes"] or old_plan["networks"] != new_plan["networks"]: raise BundleError("data volumes/network topology changed; separate operator procedure required") for service in INFRA_SERVICES + OPTIONAL_INFRA_SERVICES: if old_plan["services"].get(service) != new_plan["services"].get(service): raise BundleError("infrastructure change excluded from ordinary updates: " + service) for service in OPTIONAL_INFRA_SERVICES: if service in new_plan["services"] and service not in target["infrastructure"]: raise BundleError("release must declare enabled infrastructure compatibility: " + service) for service, ref in target["infrastructure"].items(): if service in OPTIONAL_INFRA_SERVICES and service not in new_plan["services"]: continue if new_plan["services"][service]["image"] != ref: raise BundleError("release infrastructure compatibility requirement differs: " + service) if old_plan["services"]["nginx"]["image"] != new_plan["services"]["nginx"]["image"]: raise BundleError("nginx image change requires the infrastructure upgrade procedure") snapshot = self.runtime.snapshot(old_plan) current_ids = {v["image"] for v in snapshot.values() if v["service"] in APP_SERVICES} if app["action"] == "keep" and not current_ids <= set(app["compatible_ids"]): raise BundleError("running app image is not compatible with this file-only release") if app["action"] == "replace" and self.runtime.image_id(app["ref"]) != app["id"]: raise BundleError("load the release app tar; local image ID does not match the manifest") self.runtime.validate_stop_signals(new_plan) self.runtime.check_schema(old_plan, target) # must finish before any file/service mutation changed = set(files["change"] + files["delete"] + files["add"]) if any(p.startswith("conf/postgres/") for p in changed): raise BundleError("PostgreSQL initialization config changed; separate infrastructure procedure required") if "litellm" in new_plan["services"] and any(p.startswith("conf/litellm/") for p in changed): raise BundleError("LiteLLM config changed; separate infrastructure procedure required") affected = set() if current["env_sha256"] != digest((self.root / ".env").read_bytes()): affected.update(APP_SERVICES) for service in APP_SERVICES + ("nginx",): old, new = old_plan["services"][service], new_plan["services"][service] if old != new: affected.add(service) if service in APP_SERVICES and app["action"] == "replace" and current_ids != {app["id"]}: affected.add(service) for spec in (old, new): for i, arg in enumerate(spec["opts"]): if arg == "-v": path = Path(spec["opts"][i + 1].split(":", 1)[0]) if any(path == self.root / p or path in (self.root / p).parents for p in changed): affected.add(service) tasks = [t for t in target["data_updates"] if t["id"] not in current["applied_tasks"]] app_change = bool(set(APP_SERVICES) & affected) or bool(tasks) if app_change: queues = self.runtime.rq(old_plan, "status") if any(q["suspended"] for q in queues.values()): raise BundleError("RQ is already suspended; finish the existing maintenance first") images = {s: self.runtime.image_id(new_plan["services"][s]["image"]) for s in affected} old_images = {v["service"]: v["image"] for v in snapshot.values()} old_plan["runtime_environment"] = {v["service"]: v["environment"] for v in snapshot.values() if v["service"] in APP_SERVICES and "environment" in v} if "nginx" in affected: self.runtime.validate_nginx(new_plan, source, target) return {"noop": False, "release": target["release"], "files": files, "services": sorted(affected), "app_change": app_change, "tasks": tasks, "old_plan": old_plan, "new_plan": new_plan, "images": images, "old_images": old_images, "previous": current} def save(self, journal, phase=None): if phase: journal["phase"] = phase write_json(self.state / "pending.json", journal) write_json(Path(journal["directory"]) / "journal.json", journal) print("update phase: " + journal["phase"], flush=True) def track_stops(self, journal): def record(token): sent = journal.setdefault("stop_signals", []) if token in sent: return False sent.append(token) self.save(journal) return True self.runtime.stop_record = record def protected_fingerprint(self): return {name: digest((self.root / name).read_bytes()) if (self.root / name).exists() else None for name in (".env", "compose.patches.yml", "compose.ports.yml")} def assert_protected(self, journal): if journal["protected"] != self.protected_fingerprint(): raise BundleError("site environment/overlays changed during transaction; manual recovery required") def apply_files(self, target, source, files): for name in files["add"] + files["change"]: atomic_file(Path(source) / name, checked_path(self.root, name), target["files"][name]["mode"]) for name in files["delete"]: checked_path(self.root, name).unlink() def set_image(self, ref): path = self.state / "images.env" temporary = path.with_suffix(".tmp") temporary.write_text("APP_IMAGE=" + ref + "\n") temporary.chmod(0o600) os.replace(temporary, path) def apply(self, target, source, credentials, drain_timeout): plan = self.preflight(target, source) if plan["noop"]: return plan needs_services = plan["services"] or plan["app_change"] if needs_services and not credentials: raise BundleError("service update needs --credentials FILE for private login verification") directory = self.state / "transactions" / uuid.uuid4().hex directory.mkdir(parents=True, mode=0o700) backup = directory / "files" for name, entry in plan["previous"]["manifest"]["files"].items(): dest = backup / name atomic_file(checked_path(self.root, name), dest, entry["mode"]) # Preserve a runnable updater even if a process dies while replacing bin/. helpers = ("update-bundle.sh", "update_bundle.py", "update_runtime.py", "bundle_format.py", "_compose_to_plan.py", "_miniyaml.py", "checks/update_rq.py", "checks/update_login.py", "checks/update_storage.py", "checks/processing.py") for name in helpers: path = Path(__file__).parent / name atomic_file(path, directory / "updater" / name, stat.S_IMODE(path.stat().st_mode)) journal = {**plan, "directory": str(directory), "target": target, "phase": "backed_up", "protected": self.protected_fingerprint(), "files_started": False, "edge_stopped": False, "suspended": False, "scheduler_paused": False, "apps_stop_requested": False, "tasks_started": [], "tasks_done": [], "static_volume": None, "drain_timeout": drain_timeout, "old_image_env": (self.state / "images.env").read_text() if (self.state / "images.env").exists() else None} self.save(journal) self.track_stops(journal) try: old = plan["old_plan"] if needs_services: journal["edge_stopped"] = True self.save(journal, "closing_edge") self.runtime.signal_stop(old, ["nginx"]) if plan["app_change"]: journal["scheduler_paused"] = True self.save(journal, "pausing_scheduler") self.runtime.pause_scheduler(old) journal["suspended"] = True self.save(journal, "draining_rq") self.runtime.rq(old, "suspend") self.runtime.drain(old, drain_timeout) journal["apps_stop_requested"] = True self.save(journal, "stopping_apps") self.runtime.signal_stop(old, ["scheduler", "worker-default", "worker-high", "worker-log", "web"]) if "web" in plan["services"]: volume = self.runtime.static_volume(old) self.runtime.static_backup(volume, directory / "static.tar", plan["old_images"]["nginx"]) journal["static_volume"] = volume self.save(journal, "static_backed_up") self.assert_protected(journal) # Recheck files and schema after draining to close the preflight/apply gap. plan_files(self.root, plan["previous"]["manifest"], target) self.runtime.check_schema(old, target) journal["files_started"] = True self.save(journal, "applying_files") self.apply_files(target, source, plan["files"]) self.set_image(plan["new_plan"]["services"]["web"]["image"]) self.save(journal, "recreating_services") if plan["services"]: self.runtime.remove_excess(plan["old_plan"], plan["new_plan"], plan["services"]) self.runtime.recreate(plan["new_plan"], plan["services"], plan["images"], directory) if plan["app_change"]: self.runtime.start(plan["new_plan"], [s for s in APP_SERVICES if s != "scheduler"]) for task in plan["tasks"]: journal["tasks_started"].append(task["id"]) self.save(journal, "data_update") self.runtime.data_update(plan["new_plan"], task) journal["tasks_done"].append(task["id"]) self.save(journal) self.save(journal, "verifying") self.runtime.check_schema(plan["new_plan"], target) self.runtime.check_mounts(plan["new_plan"], target, [s for s in plan["services"] if s not in ("nginx", "scheduler")]) if needs_services: self.runtime.private_probe(plan["new_plan"], target, credentials, directory) current = {**plan["previous"], "manifest": target, "plan": plan["new_plan"], "env_sha256": digest((self.root / ".env").read_bytes()), "applied_tasks": sorted(set(plan["previous"]["applied_tasks"] + journal["tasks_done"]))} write_json(self.state / "installed.json", current) self.save(journal, "verified") self.finish_open(journal) return {"release": target["release"], "result": "applied", "services": plan["services"]} except Exception as exc: journal["error"] = str(exc) if isinstance(exc, BundleError) else type(exc).__name__ if journal["phase"] in ("verified", "opening", "opening_failed"): self.save(journal, "opening_failed") else: self.save(journal, "failed") try: self.rollback(journal, credentials) except Exception as recovery_error: journal["recovery_error"] = str(recovery_error) if isinstance(recovery_error, BundleError) else type(recovery_error).__name__ self.save(journal, "recovery_required") raise BundleError("update failed: " + journal["error"] + "; inspect status before retrying") from exc def finish_open(self, journal): self.save(journal, "opening") self.assert_protected(journal) plan = journal["new_plan"] if journal["suspended"]: self.runtime.rq(plan, "resume") if journal["app_change"]: self.runtime.start(plan, ["scheduler"]) if journal["edge_stopped"]: self.runtime.start(plan, ["nginx"]) self.runtime.check_mounts(plan, journal["target"], ["nginx"]) self.save(journal, "complete") (self.state / "pending.json").unlink() def rollback(self, journal, credentials): self.assert_protected(journal) old, new = journal["old_plan"], journal["new_plan"] previous = journal["previous"] self.runtime.check_schema(old, previous["manifest"]) unsafe = [t["id"] for t in journal["tasks"] if t["id"] in journal["tasks_started"] and not t["backward_compatible"]] if unsafe: raise BundleError("data update crossed rollback boundary: " + ", ".join(unsafe)) self.save(journal, "rolling_back") if journal["files_started"]: if journal["app_change"]: self.runtime.signal_stop(new, list(APP_SERVICES)) self.runtime.signal_stop(new, ["nginx"] if journal["edge_stopped"] else []) for name, record in previous["manifest"]["files"].items(): atomic_file(Path(journal["directory"]) / "files" / name, checked_path(self.root, name), record["mode"]) for name in journal["files"]["add"]: checked_path(self.root, name).unlink(missing_ok=True) image_env = self.state / "images.env" if journal["old_image_env"] is None: image_env.unlink(missing_ok=True) else: image_env.write_text(journal["old_image_env"]) image_env.chmod(0o600) if journal["services"]: self.runtime.remove_excess(new, old, journal["services"]) self.runtime.recreate(old, journal["services"], journal["old_images"], journal["directory"]) elif journal["apps_stop_requested"]: # A timed-out TERM must never be repeated; wait for the original shutdown. self.runtime.wait_stopped(old, list(APP_SERVICES)) if journal["app_change"] and journal["apps_stop_requested"]: self.runtime.start(old, [s for s in APP_SERVICES if s != "scheduler"]) if journal["static_volume"]: self.runtime.static_restore(journal["static_volume"], Path(journal["directory"]) / "static.tar", journal["old_images"]["nginx"]) if journal["files_started"] and (journal["services"] or journal["app_change"]): self.runtime.private_probe(old, previous["manifest"], credentials, journal["directory"]) if journal["suspended"]: self.runtime.rq(old, "resume") if journal["scheduler_paused"] or journal["apps_stop_requested"]: self.runtime.start(old, ["scheduler"]) if journal["edge_stopped"]: if not journal["files_started"]: self.runtime.wait_stopped(old, ["nginx"]) self.runtime.start(old, ["nginx"]) write_json(self.state / "installed.json", previous) self.save(journal, "rolled_back") (self.state / "pending.json").unlink() def recover(self, credentials): journal = self.pending() if not journal: return {"result": "no pending update"} directory = Path(journal["directory"]) if directory.parent != self.state / "transactions" or directory.is_symlink(): raise BundleError("invalid transaction path") self.installed() self.track_stops(journal) if journal["phase"] in ("verified", "opening", "opening_failed", "complete"): plan_files(self.root, journal["target"], journal["target"]) self.runtime.check_schema(journal["new_plan"], journal["target"]) self.finish_open(journal) else: if journal["files_started"] and (journal["services"] or journal["app_change"]) and not credentials: raise BundleError("recovery needs the verification credentials") self.rollback(journal, credentials) return {"result": "recovered"} def main(): p = argparse.ArgumentParser(description=__doc__) p.add_argument("--install-dir", type=Path, default=Path(__file__).resolve().parent.parent) p.add_argument("--mode", choices=("docker-compose", "docker-run", "podman-run"), required=True) p.add_argument("--credentials", type=Path) p.add_argument("--drain-timeout", type=int, default=300) p.add_argument("--service-timeout", type=int, default=300) sub = p.add_subparsers(dest="command", required=True) for cmd in ("plan", "apply", "register"): c = sub.add_parser(cmd) c.add_argument("zip", type=Path) c.add_argument("--sha256", help="trusted out-of-band ZIP checksum") if cmd == "register": c.add_argument("--apply", action="store_true", help="record ownership; default only checks") sub.add_parser("status") sub.add_parser("recover") args = p.parse_args() if args.drain_timeout < 1 or args.service_timeout < 1: p.error("timeouts must be positive") updater = Updater(args.install_dir, args.mode, timeout=args.service_timeout) credentials = None if args.credentials: if (args.credentials.is_symlink() or not stat.S_ISREG(args.credentials.stat().st_mode) or args.credentials.stat().st_mode & 0o077): raise BundleError("credentials file must be a regular private file (0600)") credentials = read_json(args.credentials) if not credentials.get("username") or not credentials.get("password"): raise BundleError("credentials need username/password") if args.command == "status": pending = updater.pending() current = updater.installed() result = {"installed": current["manifest"]["release"], "phase": pending["phase"] if pending else "idle", "error": pending.get("error") if pending else None, "recovery_error": pending.get("recovery_error") if pending else None, "next": "recover" if pending else "plan/apply"} elif args.command == "recover": with updater.lock(): result = updater.recover(credentials) else: if args.sha256 and digest(args.zip.read_bytes()) != args.sha256: raise BundleError("ZIP checksum does not match trusted checksum") with tempfile.TemporaryDirectory(prefix="pi-bundle-") as temporary: manifest = extract_bundle(args.zip, temporary) if args.command == "plan": plan = updater.preflight(manifest, temporary) result = {k: v for k, v in plan.items() if k in ("noop", "release", "files", "services", "tasks", "app_change")} elif args.command == "register" and not args.apply: result = updater.register(manifest) else: with updater.lock(): result = updater.register(manifest, args.apply) if args.command == "register" else updater.apply(manifest, temporary, credentials, args.drain_timeout) print(json.dumps(result, ensure_ascii=False, indent=2)) if __name__ == "__main__": try: main() except (BundleError, OSError, ValueError) as exc: print("UPDATE BLOCKED: " + str(exc), file=sys.stderr) sys.exit(1)