from contextlib import contextmanager from datetime import datetime, timezone import fcntl import hashlib import json import os from pathlib import Path, PurePosixPath import re import tempfile import time import uuid from transport import OperationsError, snapshot_id WRITERS = ["nginx", "app", "queue", "scheduler", "reverb"] def private_directory(path): path = Path(path) if path.is_symlink(): raise OperationsError("Private workspaces must not be symbolic links") path.mkdir(parents=True, exist_ok=True, mode=0o700) if path.stat().st_mode & 0o077: raise OperationsError("Private workspaces must have mode 0700") return path.resolve() def atomic_json(path, values): with tempfile.NamedTemporaryFile(mode="w", dir=path.parent, prefix=".state-", delete=False) as output: json.dump(values, output) temporary = output.name os.chmod(temporary, 0o600) os.replace(temporary, path) class BackupJob: def __init__(self, stack, repository, workspace): self.stack = stack self.repository = repository self.workspace = private_directory(workspace) self.journal = self.workspace / "maintenance.json" self.status = self.workspace / "status.json" @contextmanager def lock(self): descriptor = os.open(self.workspace / "operations.lock", os.O_RDWR | os.O_CREAT, 0o600) try: try: fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError: raise OperationsError("Another operation owns this workspace") from None yield finally: os.close(descriptor) def fingerprint(self, model): values = {name: model["services"][name]["image"] for name in WRITERS} mounts = {name: {key: model["services"].get(name, {}).get(key) for key in ["image", "volumes", "secrets", "configs"]} for name in [*WRITERS, "mysql", "redis", "meilisearch"]} environment = model["services"]["app"].get("environment", {}) domains = {name: environment.get(name) for name in ["APP_ENV", "APP_URL", "WEB_DOMAIN", "ADMIN_DOMAIN", "API_DOMAIN", "DB_DATABASE"]} database = model["services"]["mysql"].get("environment", {}).get("MYSQL_DATABASE") return hashlib.sha256(json.dumps([self.stack.project, str(self.stack.env_file), values, mounts, domains, database], sort_keys=True).encode()).hexdigest() def record(self, success, resumed, identity=None, duration=0, last_success=None): previous = json.loads(self.status.read_text()) if self.status.is_file() else {} values = {"last_attempt": int(time.time()), "success": bool(success), "resumed": bool(resumed), "duration": duration, "last_success": last_success if last_success is not None else int(time.time()) if success else previous.get("last_success", 0), "snapshot": identity if success else previous.get("snapshot")} atomic_json(self.status, values) metrics = self.workspace / "metrics" metrics.mkdir(mode=0o755, exist_ok=True) metrics.chmod(0o755) lines = [f'post76_backup_success{{project="{self.stack.project}"}} {int(success)}', f'post76_backup_resume_success{{project="{self.stack.project}"}} {int(resumed)}', f'post76_backup_last_success_timestamp_seconds{{project="{self.stack.project}"}} {values["last_success"]}', f'post76_backup_duration_seconds{{project="{self.stack.project}"}} {duration}', f'post76_backup_maintenance_active{{project="{self.stack.project}"}} {int(self.journal.exists())}'] temporary = metrics / ".backup.prom.tmp" temporary.write_text("\n".join(lines) + "\n") temporary.chmod(0o644) os.replace(temporary, metrics / "backup.prom") def backup(self, capture, quiesce=False): if not quiesce: raise OperationsError("A consistent capture requires the explicit --quiesce maintenance-window option") if self.journal.exists(): raise OperationsError("Interrupted maintenance requires recover-services before another backup") capture = private_directory(capture) if any(capture.iterdir()): raise OperationsError("Capture directory must be empty") started = time.monotonic() services = [] identity = None success = False resumed = True owned_files = [] cleanup_failed = False try: model = self.stack.model() self.repository.snapshots(["post76", "project:" + self.stack.project]) state = self.stack.state() for name in ["mysql", "redis", "meilisearch"]: if state.get(name, {}).get("State") != "running" or state[name].get("Health") != "healthy": raise OperationsError("Healthy database, Redis and search services are required") if any(name not in state for name in WRITERS): raise OperationsError("The application stack is incomplete") services = [name for name in WRITERS if state[name]["State"] == "running"] atomic_json(self.journal, {"services": services, "fingerprint": self.fingerprint(model), "started": int(time.time())}) resumed = False self.record(False, False) self.stack.stop(services) dump = capture / "database.sql" owned_files.extend([dump, capture / "manifest.json"]) self.stack.dump(dump) self.stack.snapshot_redis() with dump.open("rb") as source: digest = hashlib.file_digest(source, "sha256").hexdigest() metadata = {"format": 1, "project": self.stack.project, "created_at": datetime.now(timezone.utc).isoformat(), "roots": self.repository.roots, "database_sha256": digest, "database_bytes": dump.stat().st_size, "application_image": model["services"]["app"]["image"], "mysql_image": model["services"]["mysql"]["image"], "database": model["services"]["mysql"].get("environment", {}).get("MYSQL_DATABASE"), "legacy_files": "captured" if "legacy_attachments" in self.repository.roots else "external-dependency"} atomic_json(capture / "manifest.json", metadata) tags = ["post76", "format:1", "project:" + self.stack.project, "capture:" + uuid.uuid4().hex] candidate = self.repository.capture(capture, tags) identity = self.repository.verify_tag(candidate) success = True finally: for file in owned_files: try: if file.is_file() and not file.is_symlink(): file.unlink() except OSError: cleanup_failed = True if self.journal.exists() and services: try: self.stack.resume(services) resumed = True except Exception: resumed = False elif self.journal.exists(): resumed = True if resumed and self.journal.exists(): self.journal.unlink() self.record(success, resumed, identity, round(time.monotonic() - started, 3)) if not resumed: raise OperationsError("Application services did not resume; run recover-services with the same configuration") if cleanup_failed: raise OperationsError("Application services resumed, but private capture files require manual cleanup") return identity def recover_services(self): if not self.journal.is_file(): raise OperationsError("No interrupted maintenance is recorded") journal = json.loads(self.journal.read_text()) services = journal.get("services") if not isinstance(services, list) or any(name not in WRITERS for name in services): raise OperationsError("Invalid maintenance service list") if journal.get("fingerprint") != self.fingerprint(self.stack.model()): raise OperationsError("The recorded deployment configuration has changed") self.stack.resume(services) self.journal.unlink() previous = json.loads(self.status.read_text()) if self.status.is_file() else {} self.record(previous.get("success", False), True, previous.get("snapshot"), previous.get("duration", 0), previous.get("last_success", 0)) def restore(self, identity, destination): identity = snapshot_id(identity) destination = Path(destination) if destination.exists() or destination.is_symlink(): raise OperationsError("Restore destination must not exist") rows = self.repository.snapshots(["post76", "format:1", "project:" + self.stack.project, "verified"]) if not any(row.get("id") == identity for row in rows): raise OperationsError("The snapshot is not a verified capture for this project") metadata = self.repository.manifest(identity) if metadata.get("format") != 1 or metadata.get("project") != self.stack.project or metadata.get("roots") != self.repository.roots: raise OperationsError("Snapshot manifest does not match the expected project and source roots") if type(metadata.get("database_bytes")) is not int or metadata["database_bytes"] <= 0 \ or not re.fullmatch(r"[a-f0-9]{64}", str(metadata.get("database_sha256", ""))): raise OperationsError("Invalid database integrity metadata") root = PurePosixPath(metadata["roots"]["capture"]) if not root.is_absolute() or ".." in root.parts: raise OperationsError("Invalid capture path in manifest") for node in self.repository.nodes(identity): if node.get("struct_type") == "node" and node.get("type") not in ["file", "dir"]: raise OperationsError("Restore accepts only regular files and directories") private_directory(destination) data = private_directory(destination / "data") self.stack.mounts(self.stack.environment["POST76_BACKUP_CAPTURE_DIR"], data) self.repository.restore(identity, data) dump = data / str(root).lstrip("/") / "database.sql" if dump.is_symlink() or not dump.is_file() or dump.stat().st_size != metadata["database_bytes"]: raise OperationsError("Restored database size does not match the capture") with dump.open("rb") as source: digest = hashlib.file_digest(source, "sha256").hexdigest() if digest != metadata["database_sha256"]: raise OperationsError("Restored database checksum does not match the capture") redis = data / metadata["roots"]["redis"].lstrip("/") if redis.is_symlink() or not redis.is_file() or redis.stat().st_size == 0: raise OperationsError("Restored queue snapshot is missing") atomic_json(destination / "restore-result.json", {"snapshot": identity, "database_verified": True, "redis_snapshot_restored": True, "legacy_files": metadata.get("legacy_files"), "production_database_modified": False}) return data