"""Run the update, emitting newline-delimited JSON on stdout.
The front end is a separate process, so the two halves need a protocol. This is
it: one JSON object per line, nothing else.
**stdout is JSON only.** Anything a human would want to read goes to stderr or
into the per-phase log file. A stray print on stdout is a parse failure in the Go
side, and this codebase has already been bitten once by a tool that printed an
error and still exited zero (see flatpak.py).
Events:
{"event":"start", "phases":[...], "total":N}
{"event":"phase_start","phase":"repo","command":"pacman -Syu","total":N}
{"event":"build", "packages":[{"name":..,"phase":..,"elapsed":..,"jobs":..}],
"download_bytes":N,"download_rate":F,"done":N}
{"event":"finished", "name":..,"elapsed":F,"state":"done","version":..,
"shared":bool}
{"event":"failed", "name":..,"reason":..}
{"event":"recovery", "label":..,"argv":[..],"note":..}
{"event":"phase_end", "phase":"repo","returncode":0,"elapsed":F}
{"event":"error", "message":.., "details":[..]}
{"event":"done", "returncode":0,"elapsed":F}
"""
from __future__ import annotations
import contextlib
import json
import os
import signal
import subprocess
import sys
import time
from pathlib import Path
from .live import BuildObserver, build_env, spawn_kwargs, terminate
POLL_HZ = 5.0
#: a child that survived SIGKILL: the phase did not finish and did not stop
EXIT_STILL_RUNNING = 125
#: How much of the progress bar a package that actually builds is worth,
#: against one that only downloads and installs. A build runs for minutes and a
#: download for about a second, so without this the bar raced to nearly full
#: during the repository phase -- every package there finishes in no time --
#: and the estimate that follows from it was meaningless.
BUILD_WEIGHT = 60
PLAIN_WEIGHT = 1
class SkipRequest:
"""A request to abandon the package in progress and go on to the next.
Set from a signal handler, which may only touch simple state, and taken by
the supervise loop between polls. One package per invocation is what makes
this possible at all: inside a single transaction there is nothing to skip
without unpicking what pacman has already done.
"""
def __init__(self) -> None:
self._asked = False
def ask(self) -> None:
self._asked = True
def take(self) -> bool:
asked, self._asked = self._asked, False
return asked
#: the process-wide request; a signal handler has nowhere else to put it
SKIP = SkipRequest()
def request_skip(*_args: object) -> None:
"""Handler for the signal the dashboard sends when `s` is pressed."""
SKIP.ask()
def listen_for_skip() -> None:
"""Route the skip signal to the request.
Installed only where something can act on it, so an ordinary CLI run keeps
the default disposition and an unexpected SIGUSR1 still stops it.
"""
with contextlib.suppress(ValueError, OSError):
signal.signal(signal.SIGUSR1, request_skip)
class Emitter:
"""Writes one JSON object per line and flushes, so the reader sees it live.
A dead reader must never interrupt a running transaction. If the front end
exits, this process's stdout pipe breaks; letting that exception escape would
unwind the phase loop into the `finally` that terminates pacman, killing an
upgrade halfway through. That is precisely how a system ends up in a
half-upgraded state, so a broken pipe silences the reporting and nothing else.
"""
def __init__(self, stream=None) -> None:
self.stream = stream or sys.stdout
self.alive = True
def _die(self) -> None:
"""Stop reporting and make the stream safe to write to again.
Redirecting the descriptor to /dev/null is the documented way to handle
this: without it the interpreter prints its own BrokenPipeError noise on
exit, from a place no handler of ours can reach.
"""
self.alive = False
try:
fileno = self.stream.fileno()
except (AttributeError, OSError, ValueError):
return
try:
devnull = os.open(os.devnull, os.O_WRONLY)
os.dup2(devnull, fileno)
os.close(devnull)
except OSError:
pass
def __call__(self, event: str, **fields) -> None:
if not self.alive:
return
payload = {"event": event}
payload.update(fields)
try:
self.stream.write(json.dumps(payload, separators=(",", ":")) + "\n")
self.stream.flush()
except (BrokenPipeError, ValueError, OSError):
self._die()
def human(self, message: str) -> None:
"""Anything a person should read. Never stdout."""
try:
print(message, file=sys.stderr, flush=True)
except (BrokenPipeError, ValueError, OSError):
pass
class StreamingRunner:
"""Runs one phase while emitting build state, without rendering anything."""
#: most lines of a command's output to send in one tick. A cap rather than
#: a window: the interface keeps what it has been sent, so nothing is lost
#: except a burst larger than this between two polls.
TAIL_LINES = 200
def __init__(
self,
observer: BuildObserver,
emit: Emitter,
*,
poll_hz: float = POLL_HZ,
workers: int = 1,
nproc: int | None = None,
) -> None:
self.observer = observer
self.emit = emit
self.poll_hz = poll_hz
self.workers = workers
self.nproc = nproc
self._pid: int | None = None
self._log: Path | None = None
self._log_seen: Path | None = None
self._log_offset = 0
#: set when the run was asked to stop. The interrupt is caught around
#: the child so that the child can be shut down cleanly, which means
#: the caller has to be told: catching it here otherwise looked like a
#: phase that had simply finished, and a phase made of many
#: invocations carried on to the next one.
self.interrupted = False
self.skipped = False
def run(
self,
argv: list[str],
phase: str,
log_path: Path,
*,
total: int | None = None,
since: float | None = None,
) -> int:
"""`since` is when the whole run began, for a cumulative elapsed.
A phase's own duration is meaningless once several invocations make up
one phase: reporting it made the clock read 19s while a package had
been building for nearly two minutes.
"""
log_path.parent.mkdir(parents=True, exist_ok=True)
self._log = log_path
# Per package, not per runner. One runner makes every invocation in a
# phase, so a flag left set made a single skip look like a skip of
# every package that followed it. `interrupted` is deliberately not
# reset: once the run has been asked to stop, it has stopped.
self.skipped = False
started = time.monotonic()
began = since if since is not None else started
count = total if total is not None else self.observer.total
self.emit(
"phase_start", phase=phase, command=" ".join(argv), total=count, log=str(log_path)
)
with open(log_path, "wb") as log:
proc = subprocess.Popen( # noqa: S603 - argv is built by the caller
argv, **spawn_kwargs(build_env(self.workers, self.nproc), log)
)
self._pid = proc.pid
try:
while proc.poll() is None:
if SKIP.take():
# the caller decides what a skip means; the runner
# only stops this child and says that it did
self.skipped = True
terminate(proc)
break
self._tick(proc.pid, phase)
time.sleep(1.0 / self.poll_hz)
except KeyboardInterrupt:
self.interrupted = True
# No human line here: the caller reports the stop as a proper
# event, and a human line is echoed back into the interface as a
# second, identical warning.
terminate(proc)
finally:
if proc.poll() is None:
terminate(proc)
self._pid = None
# drain whatever landed between the last tick and exit, so a package
# that finished at the very end still reaches the feed
self._tick(None, phase)
elapsed = time.monotonic() - began
# A child that outlived SIGKILL is not a success. `returncode or 0`
# turned None into 0, so the run reported the phase as done and started
# the next one beside a live pacman transaction.
code = proc.returncode
if code is None:
self.interrupted = True
code = EXIT_STILL_RUNNING
self.emit("phase_end", phase=phase, returncode=code, elapsed=round(elapsed, 2))
return code
def _new_log_lines(self) -> list[str]:
"""Lines the running command printed since the last tick.
Sending the *last* few lines each time lost anything a burst produced
between two polls, and meant the interface could never show more than a
window of the output. Tracking the read offset instead makes the stream
whole, so it can be read back in full.
"""
if self._log is None:
return []
if self._log != self._log_seen:
# a new package has its own log; start from the beginning of it
self._log_seen = self._log
self._log_offset = 0
try:
if self._log.stat().st_size < self._log_offset:
self._log_offset = 0 # truncated or replaced
with open(self._log, "rb") as handle:
handle.seek(self._log_offset)
chunk = handle.read(1 << 20)
except OSError:
return []
# Work in bytes: a partial final line has to be left for the next tick,
# and re-decoding with replacement would not tell us how many there were.
cut = chunk.rfind(b"\n")
if cut < 0:
return []
text = chunk[: cut + 1].decode("utf-8", errors="replace")
self._log_offset += cut + 1
lines = [line.rstrip() for line in text.splitlines() if line.strip()]
return [line[:400] for line in lines[-self.TAIL_LINES :]]
def _tick(self, pid: int | None, phase: str) -> None:
finished = self.observer.poll(pid)
for entry in finished:
# the weight travels with the result, so the consumer can weigh
# progress by work rather than by package
self.emit(
"finished",
name=entry.name,
elapsed=round(entry.elapsed, 2),
state=entry.state,
version=entry.version,
shared=entry.shared,
)
states = [
{
"name": state.package,
"phase": state.phase,
"elapsed": round(state.elapsed, 2),
"jobs": state.jobs,
"workers": getattr(state, "workers", ""),
}
for state in self.observer.states.values()
]
done = sum(1 for e in self.observer.feed if e.state != "fail")
self.emit(
"build",
phase=phase,
packages=states,
download_bytes=self.observer.download_bytes,
download_rate=round(self.observer.download_rate, 1),
done=done,
total=self.observer.total,
log_lines=self._new_log_lines(),
)
# --------------------------------------------------------------------------
# phase planning
# --------------------------------------------------------------------------
def repo_command(held: list[str], *, refresh: bool = True) -> list[str]:
"""`-Syu` with `--ignore`, which is exactly the IgnorePkg mechanism.
Holding a package back makes the run a partial upgrade, which Arch does not
support. It is still offered because it is occasionally what you want, but
the caller is expected to have said so out loud first.
"""
# `-y` refreshes the databases first. The plan is built from the databases
# on disk, so refreshing here installs from a set the reader never saw --
# the same option has to decide it in both places.
argv = ["sudo", "pacman", "-Syu" if refresh else "-Su", "--noconfirm"]
if held:
argv.append("--ignore=" + ",".join(sorted(held)))
return argv
def aur_command(targets: list[str]) -> list[str]:
"""An explicit target list rather than `-Sua`.
`-Sua` updates everything and offers no way to leave one package out, so
skipping a package requires naming the ones to include.
`--nobatchinstall` is deliberate, and is not paru's default. By default paru
builds every package and then hands the whole set to pacman as a *single*
transaction, which makes one bad package poison the entire update: a single
unresolved conflict meant none of twenty-seven packages installed. paru's
documented alternative is to install each package as soon as it is built, so
a failure costs one package instead of all of them.
`--failfast` is deliberately absent too: paru's default is to carry on after
a build fails, which is the behaviour wanted here.
"""
return ["paru", "-S", "--needed", "--noconfirm", "--nobatchinstall", *sorted(targets)]
def aur_command_interactive(targets: list[str]) -> list[str]:
"""The same target list, but with the prompts left switched on.
Some questions have no correct automatic answer -- "replace the installed
scenefx0.5 with scenefx?" is a decision, not a default. Those can only be
put to a human, which needs the real terminal the interface normally keeps
to itself.
"""
return ["paru", "-S", "--needed", "--nobatchinstall", *sorted(targets)]
def flatpak_command() -> list[str]:
return ["flatpak", "update", "--noninteractive", "-y"]
def installed_progress(updates) -> int:
"""How many planned updates have actually landed.
The repo phase has no build graph, so progress has to be measured: a package
is done when its new version directory exists in the local database.
"""
local = Path("/var/lib/pacman/local")
done = 0
for update in updates:
if (local / f"{update.name}-{update.new}").is_dir():
done += 1
return done