"""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