"""Where a deployment keeps its releases, and which one is live. /opt/taxpilot-ai/ releases/ 1.0.0/ one extracted release, never modified after install 1.1.0/ shared/ .env configuration, which no package may contain current -> releases/1.1.0 state.json what is installed, what came before, what happened The point of the layout is that switching versions is one atomic operation and rolling back is the same operation in reverse. Nothing is overwritten in place, so a half-finished install cannot leave a running service with a mix of two releases — which is the failure mode of "extract over the top", and the reason the CMS installer has to take a full backup before it starts. Configuration lives in `shared/` and is never part of a package. That is what lets a rollback be a pointer change: the old release directory is still exactly as it was installed, and it is still pointing at the same .env it always used. """ from __future__ import annotations import contextlib import json import os from dataclasses import dataclass, field from datetime import UTC, datetime from pathlib import Path STATE_NAME = "state.json" CURRENT_NAME = "current" #: Written when a symlink cannot be created, holding the version name instead. #: #: Windows refuses symlinks without Developer Mode or elevation. The deployment #: target is Linux, but the tests run here, and install code that cannot be #: exercised on the machine it is written on is install code nobody has run. POINTER_NAME = "current.txt" class LayoutError(RuntimeError): """The deployment directory is not in a state that can be worked with.""" @dataclass(frozen=True, slots=True) class Event: """One thing that happened to this deployment.""" at: str action: str version: str outcome: str detail: str = "" def to_dict(self) -> dict: return { "at": self.at, "action": self.action, "version": self.version, "outcome": self.outcome, "detail": self.detail, } @classmethod def from_dict(cls, data: dict) -> Event: return cls( at=str(data.get("at") or ""), action=str(data.get("action") or ""), version=str(data.get("version") or ""), outcome=str(data.get("outcome") or ""), detail=str(data.get("detail") or ""), ) @dataclass class State: """The deployment's own record of itself. Kept beside the releases rather than in the database, deliberately: an installer has to work when the database is unreachable, and "which version is live" is the first thing anyone asks in exactly that situation. """ version: str | None = None previous: str | None = None history: list[Event] = field(default_factory=list) #: Newest first, and bounded. A deployment updated weekly for five years #: would otherwise grow a file nobody prunes. HISTORY_LIMIT = 50 def record(self, action: str, version: str, outcome: str, detail: str = "") -> None: self.history.insert( 0, Event( at=datetime.now(UTC).replace(microsecond=0).isoformat(), action=action, version=version, outcome=outcome, detail=detail, ), ) del self.history[self.HISTORY_LIMIT :] def to_dict(self) -> dict: return { "version": self.version, "previous": self.previous, "history": [e.to_dict() for e in self.history], } @classmethod def from_dict(cls, data: object) -> State: if not isinstance(data, dict): return cls() history = data.get("history") events = [Event.from_dict(e) for e in history if isinstance(e, dict)] if isinstance(history, list) else [] return cls( version=data.get("version") or None, previous=data.get("previous") or None, history=events, ) class Layout: """The deployment directory.""" def __init__(self, root: Path | str) -> None: self.root = Path(root) # ── Paths ───────────────────────────────────────────────────────────── @property def releases(self) -> Path: return self.root / "releases" @property def shared(self) -> Path: return self.root / "shared" @property def current(self) -> Path: return self.root / CURRENT_NAME @property def state_file(self) -> Path: return self.root / STATE_NAME def release(self, version: str) -> Path: return self.releases / version def prepare(self) -> None: """Create the directories. Safe to call on an existing deployment.""" self.releases.mkdir(parents=True, exist_ok=True) self.shared.mkdir(parents=True, exist_ok=True) def installed(self) -> list[str]: if not self.releases.is_dir(): return [] return sorted(p.name for p in self.releases.iterdir() if p.is_dir()) # ── The live release ────────────────────────────────────────────────── def current_version(self) -> str | None: """Which release is live, according to the pointer on disk. Read from the filesystem rather than from state.json, because the pointer is what actually runs. If the two disagree, the pointer is right and the state file is stale — and `status` says so rather than reporting the tidier of the two. """ pointer = self.root / POINTER_NAME if self.current.is_symlink(): target = Path(os.readlink(self.current)) return target.name or None if pointer.is_file(): name = pointer.read_text(encoding="utf-8").strip() return name or None return None def point_to(self, version: str) -> None: """Make `version` the live release, atomically. A symlink is replaced by creating a new one under a temporary name and renaming it over the old one — `os.replace` is atomic, so there is no instant where `current` does not exist. Deleting and recreating would leave exactly such a window, and a process starting in it would fail to find the application at all. """ target = self.release(version) if not target.is_dir(): raise LayoutError(f"Release {version} is not installed.") if self._link_to(target): # A deployment that gains symlink support should not keep reading a # stale pointer file that says something else. (self.root / POINTER_NAME).unlink(missing_ok=True) return # No symlink privilege — Windows without Developer Mode, or a filesystem # that does not support them. Fall back to a pointer file. # # Removing any existing symlink FIRST is not tidiness. current_version() # reads the symlink before the pointer file, because the symlink is what # a process actually follows — so leaving a stale one behind means the # switch silently does nothing, and the deployment reports the version # it used to be running. self._unlink_current() pointer = self.root / POINTER_NAME temporary = self.root / f".{POINTER_NAME}.incoming" temporary.write_text(version, encoding="utf-8") os.replace(temporary, pointer) def _link_to(self, target: Path) -> bool: """Point `current` at `target` with a symlink. False if not possible.""" temporary = self.root / f".{CURRENT_NAME}.incoming" try: if temporary.is_symlink() or temporary.exists(): temporary.unlink() os.symlink(target, temporary, target_is_directory=True) except (OSError, NotImplementedError): return False try: os.replace(temporary, self.current) return True except OSError: pass # Windows refuses to rename over an existing directory symlink # (WinError 5), where POSIX replaces it happily. Unlinking first # reopens the gap that `os.replace` exists to avoid, so it is done only # after the atomic path has actually failed — and only on a deployment # that already is not the production target. try: if self.current.is_symlink(): self.current.unlink() os.replace(temporary, self.current) return True except OSError: with contextlib.suppress(OSError): temporary.unlink() return False def _unlink_current(self) -> None: with contextlib.suppress(OSError): if self.current.is_symlink(): self.current.unlink() def live_path(self) -> Path | None: """The directory that is actually live, resolving whichever pointer.""" version = self.current_version() return self.release(version) if version else None # ── State ───────────────────────────────────────────────────────────── def state(self) -> State: if not self.state_file.is_file(): return State() try: return State.from_dict(json.loads(self.state_file.read_text(encoding="utf-8"))) except (OSError, json.JSONDecodeError): # A corrupted state file must not stop an install or a rollback. # It is a record, not the source of truth — the pointer is. return State() def save(self, state: State) -> None: temporary = self.state_file.with_suffix(".json.incoming") temporary.write_text(json.dumps(state.to_dict(), indent=2) + "\n", encoding="utf-8") os.replace(temporary, self.state_file)