"""Run a build while watching it, without parsing its output. Two halves, deliberately separated: BuildObserver a state machine. Takes ticks, maintains the feed, knows what has finished and what failed. No terminal, no Rich, fully testable. LiveRunner the terminal. Owns the child process, the log file, the key handling and the redraw. Contains no build logic. The observer never reads the child's stdout to decide anything. State comes from /proc and from the filesystem (see procwatch), so a change in paru's or makepkg's output format cannot break it, and the raw log is kept intact for when something goes wrong. """ from __future__ import annotations import contextlib import errno import os import re import select import shutil import signal import subprocess import sys import time from dataclasses import dataclass, field from pathlib import Path from rich.console import Console, Group from rich.live import Live from rich.text import Text from . import procwatch from .buildgraph import Graph from .procwatch import BuildState, format_duration from .render import build_screen from .render.build import FeedEntry, Totals from .theme import Theme POLL_HZ = 5.0 #: a child that survived SIGKILL: the phase did not finish and did not stop EXIT_STILL_RUNNING = 125 GRACE_SIGINT = 5.0 GRACE_SIGTERM = 5.0 # -------------------------------------------------------------------------- # environment # -------------------------------------------------------------------------- #: A phase runs in its own process group so a Ctrl+C meant for the interface #: does not land on pacman mid-transaction -- but in the *same session*, so it #: keeps the controlling terminal. #: #: That second part is not incidental. `start_new_session=True` calls setsid(), #: which detaches the child from the controlling terminal; sudo keys its #: credential cache to that terminal, so every privileged phase then failed with #: "a terminal is required to read the password" no matter how recently the user #: had authenticated. Detaching the session broke the very thing the #: pre-authentication exists to set up. def spawn_kwargs(env: dict[str, str], log) -> dict: return { "stdout": log, "stderr": subprocess.STDOUT, "stdin": subprocess.DEVNULL, "env": env, "process_group": 0, } def can_sudo() -> bool: """Whether sudo can act without a prompt, under the phases' own conditions. Verified up front rather than discovered by the first phase. The repository phase runs first and its failure aborts everything after it, so a run that could never have worked should not start, stream a build screen, and then report an exit code. `-n` is deliberate: this asks "would sudo succeed silently", which is exactly the question the phases need answered. A prompt here would mean a prompt mid-transaction, which is what the check exists to prevent. """ try: with open(os.devnull, "wb") as null: proc = subprocess.run( # noqa: S603 - a fixed argv ["sudo", "-n", "true"], check=False, **spawn_kwargs(dict(os.environ), null) ) except (OSError, subprocess.SubprocessError): return False return proc.returncode == 0 def build_env( workers: int = 1, nproc: int | None = None, base: dict[str, str] | None = None ) -> dict[str, str]: """Explicitly set build parallelism rather than inheriting it by accident. Whether a build runs in parallel currently depends on how the tool was launched: a shell that exports MAKEFLAGS gives `-j15`, while a systemd timer or a GUI launcher gives nothing and make runs at `-j1` on a 16-core machine. With N workers the cores must be *divided*, not multiplied: `-j15` across 8 workers is 120 concurrent compilers on a machine with 16 cores. """ env = dict(os.environ if base is None else base) nproc = nproc or os.cpu_count() or 1 jobs = max(1, nproc // max(1, workers)) env["MAKEFLAGS"] = f"-j{jobs}" env["CARGO_BUILD_JOBS"] = str(jobs) env["NINJAFLAGS"] = f"-j{jobs}" return env def parallelism_label(workers: int = 1, nproc: int | None = None) -> str: nproc = nproc or os.cpu_count() or 1 jobs = max(1, nproc // max(1, workers)) if workers <= 1: return f"{nproc} cores · {jobs} jobs" return f"{nproc} cores · {workers} workers × {jobs} jobs" # -------------------------------------------------------------------------- # observation # -------------------------------------------------------------------------- class BuildObserver: """Tracks a build from the outside. One `poll()` per tick, no terminal.""" def __init__( self, graph: Graph, clone_dir: Path, *, total: int | None = None, cache_dir: Path = Path("/var/cache/pacman/pkg"), ) -> None: self.graph = graph self.clone_dir = clone_dir self.cache_dir = cache_dir self.total = total if total is not None else len(graph) self.started_at = time.monotonic() self.feed: list[FeedEntry] = [] self.completed: set[str] = set() self.running: set[str] = set() self.failed: set[str] = set() self.states: dict[str, BuildState] = {} self.download_bytes = 0 self.download_peak = 0 self.download_rate = 0.0 self._last_dl_at = time.monotonic() self._last_dl_bytes = 0 self._durations: dict[str, float] = {} self._watcher = procwatch.Watcher(clone_dir, known=set(graph.nodes) or None) # -- ticks ------------------------------------------------------------- def poll(self, root_pid: int | None) -> list[FeedEntry]: """Advance one tick. Returns the entries newly appended to the feed.""" # always consult the watcher, even when the tree is gone: that final # call is what turns "still running" into "finished" live = self._watcher.poll(root_pid) self.states = {state.package: state for state in live} self.running = set(self.states) # durations must be captured while the process is alive; once it exits # there is nothing left to measure for name, state in self.states.items(): self._durations[name] = max(self._durations.get(name, 0.0), state.elapsed) fresh = [ self._finish(name) for name in self._watcher.new_finishes() if name not in self.failed ] self._sample_downloads() return fresh def _finish(self, name: str) -> FeedEntry: self.completed.add(name) self.running.discard(name) node = self.graph.nodes.get(name) entry = FeedEntry( name=name, version=node.version if node else "", elapsed=self._durations.get(name, 0.0), state="done", shared=name in getattr(self.graph, "shared", set()), ) self.feed.append(entry) return entry def fail(self, name: str, note: str = "failed") -> FeedEntry: """Record a failure. Not the same as a crash: the build continues.""" self.failed.add(name) self.running.discard(name) node = self.graph.nodes.get(name) entry = FeedEntry( name=name, version=node.version if node else "", elapsed=self._durations.get(name, 0.0), state="fail", note=note, ) self.feed.append(entry) return entry def seed(self, entries: list[FeedEntry]) -> None: """Pre-populate, for resuming when some packages are already built.""" self.feed.extend(entries) self.completed |= {e.name for e in entries if e.state != "fail"} self.failed |= {e.name for e in entries if e.state == "fail"} # -- downloads --------------------------------------------------------- def _sample_downloads(self) -> None: now = time.monotonic() total, _count = procwatch.download_progress(self.cache_dir) self.download_bytes = total # `.part` files are renamed on completion, so the raw total can fall. # Track a high-water mark so the reported rate never goes negative. self.download_peak = max(self.download_peak, total) dt = now - self._last_dl_at if dt >= 0.5: delta = self.download_peak - self._last_dl_bytes if delta > 0: self.download_rate = delta / dt self._last_dl_at, self._last_dl_bytes = now, self.download_peak # -- derived ----------------------------------------------------------- @property def blocked(self) -> set[str]: return self.graph.blocked(self.failed) def totals(self) -> Totals: elapsed = time.monotonic() - self.started_at done = sum(1 for e in self.feed if e.state != "fail") remaining = max(0, self.total - done - len(self.running)) eta = (elapsed / done) * remaining if done else None return Totals( total=self.total, elapsed=elapsed, eta=eta, download_bytes=self.download_bytes, download_rate=self.download_rate, ) def render( self, *, theme: Theme, glyphs, width: int, dashboard: bool = True, parallelism: str = "" ) -> list[Text]: return build_screen( graph=self.graph, theme=theme, glyphs=glyphs, width=width, feed=self.feed, states=list(self.states.values()), completed=self.completed, running=self.running, failed=self.failed, totals=self.totals(), parallelism=parallelism, dashboard=dashboard, ) @staticmethod def feed_line(entry: FeedEntry) -> str: """One permanent line for ledger mode, so history is what scrolled past.""" mark = "x" if entry.state == "fail" else "+" version = f" {entry.version}" if entry.version else "" shared = " (shared)" if entry.shared else "" took = f" {format_duration(entry.elapsed)}" if entry.elapsed else "" return f"{mark} {entry.name}{version}{shared}{took}" # -------------------------------------------------------------------------- # terminal # -------------------------------------------------------------------------- @contextlib.contextmanager def cbreak(): """Put stdin in cbreak mode so single keys arrive without Enter. Yields False when stdin is not a terminal, in which case key handling is skipped rather than crashing under a pipe or a systemd unit. """ if not sys.stdin.isatty(): yield False return import termios import tty fd = sys.stdin.fileno() saved = termios.tcgetattr(fd) try: tty.setcbreak(fd) yield True finally: termios.tcsetattr(fd, termios.TCSADRAIN, saved) def read_keys() -> str: """Whatever is pending on stdin right now. Never blocks.""" try: ready, _, _ = select.select([sys.stdin], [], [], 0) except (OSError, ValueError): return "" if not ready: return "" try: return os.read(sys.stdin.fileno(), 32).decode(errors="replace") except OSError as exc: if exc.errno in (errno.EIO, errno.EBADF): return "" raise def terminate( proc: subprocess.Popen, *, grace: float = GRACE_SIGINT, term_grace: float = GRACE_SIGTERM ) -> bool: """Stop a child politely before escalating. True if it is gone. Returns whether the child actually died, because the caller used to do `returncode or 0`: a process stuck in D state, or on a hung mount, survives SIGKILL, keeps `returncode is None`, and was reported as a *successful* phase -- with a live pacman transaction still holding the database lock while the next phase started beside it. SIGINT goes to the child's process group because paru, makepkg and pacman all install handlers that unwind a half-finished transaction. SIGKILL would leave the package database in exactly the state we are trying to avoid, so it is only ever the last resort. """ if proc.poll() is not None: return True for sig, wait in ((signal.SIGINT, grace), (signal.SIGTERM, term_grace)): with contextlib.suppress(ProcessLookupError, PermissionError): os.killpg(os.getpgid(proc.pid), sig) try: proc.wait(timeout=wait) return True except subprocess.TimeoutExpired: continue with contextlib.suppress(ProcessLookupError, PermissionError): os.killpg(os.getpgid(proc.pid), signal.SIGKILL) with contextlib.suppress(subprocess.TimeoutExpired): proc.wait(timeout=5) return proc.poll() is not None @dataclass class RunResult: returncode: int log_path: Path elapsed: float feed: list[FeedEntry] = field(default_factory=list) interrupted: bool = False @property def ok(self) -> bool: return self.returncode == 0 and not self.interrupted #: pacman's own progress lines. #: #: The build phase needs no parsing: makepkg forks per package, so /proc says #: exactly what is running. pacman is a single process no matter how many #: packages it touches, so there is nothing for the watcher to attribute -- its #: output is the only per-package signal that exists. These two shapes are #: stable, machine-oriented and locale-independent. _P_DOWNLOADING = re.compile(r"^\s*(\S+?)-[^-]+-[^-]+-(?:x86_64|any|aarch64|[a-z0-9_]+) downloading") _P_CHANGING = re.compile(r"^(?:upgrading|installing|reinstalling|downgrading) ([^\s.]+)\.\.\.") _P_TOTAL_DOWNLOAD = re.compile(r"^Total Download Size:\s*([\d.]+)\s*(\w+)") _P_HOOK = re.compile(r"^\((\d+)/(\d+)\) (.+)\.\.\.") #: pacman prints sizes with either a short or a long unit suffix, depending on #: its build; both spellings mean the same thing. _UNITS = { "B": 1, "K": 1 << 10, "KB": 1 << 10, "KIB": 1 << 10, "M": 1 << 20, "MB": 1 << 20, "MIB": 1 << 20, "G": 1 << 30, "GB": 1 << 30, "GIB": 1 << 30, "T": 1 << 40, "TB": 1 << 40, "TIB": 1 << 40, } def _to_bytes(value: str, unit: str) -> int: """`474.12`, `MiB` -> 497,166,213. Unknown units are treated as bytes.""" try: return int(float(value) * _UNITS.get(unit.strip().upper(), 1)) except (TypeError, ValueError): return 0 class RepoObserver: """Progress for a pacman transaction, read from pacman's own output. Presents the same surface as `BuildObserver` so the same dashboard renders it. Without this the repository phase -- the long, network-bound half of an update -- showed a build screen with nothing on it. """ def __init__( self, packages: list[str], *, log_path: Path, cache_dir: Path = Path("/var/cache/pacman/pkg"), ) -> None: self.packages = sorted(packages) self.total = len(self.packages) self.log_path = Path(log_path) self.cache_dir = cache_dir self.feed: list[FeedEntry] = [] self.states: dict[str, BuildState] = {} self.completed: set[str] = set() self.failed: set[str] = set() self.download_bytes = 0 self.download_rate = 0.0 self.expected_bytes = 0 self.phase = "download" self._offset = 0 self._known_set = set(self.packages) self._reported = 0 self._started: dict[str, float] = {} self._at = time.monotonic() self._baseline = self._cache_bytes() self._last_sample = (self._at, self._baseline) self._hooked = 0 # -- reading the log --------------------------------------------------- def poll(self, root_pid: int | None = None) -> list[FeedEntry]: # noqa: ARG002 """Advance one tick. Returns entries newly appended to the feed.""" for line in self._new_lines(): self._absorb(line) self._sample() return self._drain_feed() def _new_lines(self) -> list[str]: try: with open(self.log_path, "rb") as handle: handle.seek(self._offset) chunk = handle.read() self._offset = handle.tell() except OSError: return [] if not chunk: return [] # Cut on the raw bytes. `cut` indexes bytes and `len(chunk)` counts # them; indexing the decoded string instead lets the offset land behind # the newline whenever the chunk holds multi-byte output, so the next # poll re-reads consumed content and emits a fabricated line. cut = chunk.rfind(b"\n") if cut < 0: self._offset -= len(chunk) return [] self._offset -= len(chunk) - cut - 1 text = chunk[: cut + 1].decode("utf-8", errors="replace") return [line.rstrip("\r") for line in text.splitlines() if line.strip()] def _absorb(self, line: str) -> None: if match := _P_TOTAL_DOWNLOAD.match(line): self.expected_bytes = _to_bytes(match.group(1), match.group(2)) return if match := _P_DOWNLOADING.match(line): # pacman names the next package only once the previous one has # landed, so seeing a new name is what finishes the old one. Without # this every download stayed "running" and they all completed at # once when the install phase began. self._settle(match.group(1)) self._begin(match.group(1), "download") return if match := _P_CHANGING.match(line): self.phase = "install" name = match.group(1) self._settle(name) self._begin(name, "installing") return if line.startswith(":: Running post-transaction hooks"): self.phase = "hooks" self._settle(None) return if match := _P_HOOK.match(line): self.phase = "hooks" self.states = { f"hook {match.group(1)}/{match.group(2)}": BuildState( package=f"hook {match.group(1)}/{match.group(2)}", phase=match.group(3)[:40], elapsed=time.monotonic() - self._at, ) } return def _begin(self, name: str, phase: str) -> None: # pacman names packages without the repo prefix; match on the bare name if name not in self._known(): return self._started.setdefault(name, time.monotonic()) self.states[name] = BuildState( package=name, phase=phase, elapsed=time.monotonic() - self._started[name] ) def _settle(self, keep: str | None) -> None: """Everything but `keep` has finished.""" for name in list(self.states): if name != keep and name in self._known(): self._finish(name) if keep is None: self.states = {} def _known(self) -> set[str]: return self._known_set def _finish(self, name: str) -> None: self.states.pop(name, None) if name in self.completed: return self.completed.add(name) self.feed.append( FeedEntry(name=name, elapsed=time.monotonic() - self._started.get(name, self._at)) ) def _drain_feed(self) -> list[FeedEntry]: fresh = self.feed[self._reported :] self._reported = len(self.feed) return fresh # -- numbers ----------------------------------------------------------- def _cache_bytes(self) -> int: total = 0 try: with os.scandir(self.cache_dir) as entries: for entry in entries: try: if entry.is_file(): total += entry.stat().st_size except OSError: continue except OSError: return 0 return total def _sample(self) -> None: now = time.monotonic() if now - self._at < 1.0: return self._at = now current = self._cache_bytes() self.download_bytes = max(0, current - self._baseline) last_at, last_bytes = self._last_sample elapsed = now - last_at if elapsed > 0: self.download_rate = max(0.0, (current - last_bytes) / elapsed) self._last_sample = (now, current) class LiveRunner: """Runs one command under the dashboard, or streams it plainly.""" def __init__( self, observer: BuildObserver, *, theme: Theme, glyphs, console: Console | None = None, dashboard: bool = True, workers: int = 1, nproc: int | None = None, ) -> None: self.observer = observer self.theme = theme self.glyphs = glyphs self.console = console or Console() self.dashboard = dashboard self.parallelism = parallelism_label(workers, nproc) self.workers = workers self.nproc = nproc self.log_path: Path | None = None self._pid: int | None = None self._printed = 0 # -- public ------------------------------------------------------------ def run( self, command: list[str], log_path: Path, env: dict[str, str] | None = None ) -> RunResult: self.log_path = log_path log_path.parent.mkdir(parents=True, exist_ok=True) started = time.monotonic() interrupted = False with open(log_path, "wb") as log, cbreak() as keys_on: proc = subprocess.Popen( # noqa: S603 - argv is built by the caller command, **spawn_kwargs(env or build_env(self.workers, self.nproc), log) ) self._pid = proc.pid try: self._supervise(proc, keys_on) except KeyboardInterrupt: interrupted = True self.console.print("\n[dim]interrupting build…[/dim]") finally: if proc.poll() is None: terminate(proc) self._pid = None elapsed = time.monotonic() - started self.observer.poll(None) # drain anything that landed at the end self._flush_feed() # and make sure it is on screen code = proc.returncode if code is None: interrupted = True code = EXIT_STILL_RUNNING return RunResult( returncode=code, log_path=log_path, elapsed=elapsed, feed=list(self.observer.feed), interrupted=interrupted, ) # -- internals --------------------------------------------------------- @property def _interactive(self) -> bool: return bool(sys.stdout.isatty() and getattr(self.console, "is_terminal", False)) def _supervise(self, proc: subprocess.Popen, keys_on: bool) -> None: """One loop drives both modes; the toggle just enters or leaves `Live`. Keeping a single loop is what makes `v` work in both directions without re-spawning anything or losing the feed: leaving the dashboard flushes the accumulated feed to permanent scrollback first. """ live: Live | None = None try: while proc.poll() is None: self.observer.poll(self._pid) want_live = self.dashboard and self._interactive if want_live and live is None: live = Live( self._renderable(), console=self.console, refresh_per_second=POLL_HZ, vertical_overflow="visible", ) live.start() elif not want_live and live is not None: live.stop() live = None self._flush_feed() if live is not None: live.update(self._renderable()) else: self._flush_feed() if keys_on: self._handle_keys(read_keys()) time.sleep(1.0 / POLL_HZ) finally: if live is not None: live.stop() def _handle_keys(self, keys: str) -> None: for key in keys: if key in ("v", " "): self.dashboard = not self.dashboard elif key == "l" and self.log_path: self.console.print(f"[dim]log:[/dim] {self.log_path}") elif key == "q": self.dashboard = False elif key == "\x03": raise KeyboardInterrupt def _renderable(self) -> Group: lines = self.observer.render( theme=self.theme, glyphs=self.glyphs, width=self._width(), dashboard=self.dashboard, parallelism=self.parallelism, ) return Group(*lines) def _flush_feed(self) -> None: """Print feed entries that have not been shown as permanent lines yet.""" pending = self.observer.feed[self._printed :] for entry in pending: self.console.print(self.observer.feed_line(entry), highlight=False) self._printed = len(self.observer.feed) @staticmethod def _width() -> int: return shutil.get_terminal_size((100, 40)).columns