Merge branch 'feat/upgrade-safety' into p2

modules/pacman/cmd/pacman-tools/acts.go

@@ -55,30 +55,11 @@ func (m *Machine) journal(unit string, n int) []string {
 	return lines(out)
 }
 
-// UpgradeArgs is the full system upgrade, in one place: pacman_upgrade starts pacman with these arguments as
-// a transient unit (issue 1574). The weekly timer that ran the same is gone: a round of runs replaces it
-// (ADR 0508 §5).
+// UpgradeArgs is the full system upgrade, in one place: the root helper starts a run's transaction with these
+// arguments (ADR 0510), and the packages a run holds back are added to them as --ignore.
 var UpgradeArgs = []string{"-Syu", "--noconfirm"}
 
-// Upgrade starts a full system upgrade as a transient unit and answers at once, with the
-// distribution's news since the last upgrade; or, given a unit it started, answers how it went.
-func (m *Machine) Upgrade(unit string, n int) (map[string]any, error) {
-	if unit != "" {
-		return m.UpgradeStatus(unit, n)
-	}
-	news := m.News("")
-	started, _, err := m.transaction("upgrade", false, UpgradeArgs...)
-	if err != nil {
-		return nil, err
-	}
-	return map[string]any{
-		"started": started,
-		"follow":  "call pacman_upgrade with this unit to read how it goes",
-		"news":    news,
-	}, nil
-}
-
-var unitName = regexp.MustCompile(`^` + unitPrefix + `[a-z-]+-[0-9]+$`)
+var unitName = regexp.MustCompile(`^` + unitPrefix + `[a-z-]+-[0-9a-z]+$`)
 
 // UpgradeStatus is a transaction unit's state and the tail of what it logged.
 func (m *Machine) UpgradeStatus(unit string, n int) (map[string]any, error) {

modules/pacman/cmd/pacman-tools/dkms.go

@@ -15,10 +15,10 @@ import (
 )
 
 var (
-	vmlinuzPath   = regexp.MustCompile(`^/?usr/lib/modules/([^/]+)/vmlinuz$`)
-	linuxMaximum  = regexp.MustCompile(`(?m)^Linux-Maximum:\s*(\S+)\s*$`)
+	vmlinuzPath    = regexp.MustCompile(`^/?usr/lib/modules/([^/]+)/vmlinuz$`)
+	linuxMaximum   = regexp.MustCompile(`(?m)^Linux-Maximum:\s*(\S+)\s*$`)
 	buildExclusive = regexp.MustCompile(`(?m)^BUILD_EXCLUSIVE_KERNEL=(?:"([^"]*)"|'([^']*)'|(\S+))\s*$`)
-	leadingNumber = regexp.MustCompile(`^[0-9]+(\.[0-9]+)*`)
+	leadingNumber  = regexp.MustCompile(`^[0-9]+(\.[0-9]+)*`)
 )
 
 // kernelRelease is the release an installed kernel package's image is for, and whether the package is a kernel.

modules/pacman/cmd/pacman-tools/main.go

@@ -189,11 +189,14 @@ func tools(m *Machine) []stdio.Tool {
 		},
 		{
 			Name: "pacman_upgrade",
-			Description: "Start a full system upgrade (pacman -Syu --noconfirm, through sudo -n) as a transient unit of its own that outlives the call, " +
-				"answering at once with the unit and the distribution's news since the last upgrade — read the news first. Given that unit, " +
-				"answer whether it is running, finished and succeeded, with the tail of its log.",
+			Description: "Ask for a full system upgrade of a node (this one unless named): the ask is written to the bus, where that node's keeper " +
+				"takes it and runs it as a run: the package record, the kernel check, the transaction as a unit of its own, and the after-upgrade " +
+				"check (ADR 0508 §8). Answers at once with the ask and the distribution's news since the last upgrade — read the news first; " +
+				"pacman_upgrades with the node shows the run that names the ask. Given a transaction unit, answers whether it is running, " +
+				"finished and succeeded, with the tail of its log.",
 			Input: schema(map[string]any{
-				"unit":  map[string]any{"type": "string", "description": "a unit this tool started, to read how it goes (optional)"},
+				"node":  map[string]any{"type": "string", "description": "the node to upgrade (default: this one)"},
+				"unit":  map[string]any{"type": "string", "description": "a transaction unit, to read how it goes (optional)"},
 				"lines": limitArg(60, 400),
 			}),
 			Run: func(args map[string]any) (any, error) {
@@ -205,7 +208,21 @@ func tools(m *Machine) []stdio.Tool {
 				if err != nil {
 					return nil, err
 				}
-				return m.Upgrade(unit, n)
+				if unit != "" {
+					return m.UpgradeStatus(unit, n)
+				}
+				node, err := text(args, "node", false)
+				if err != nil {
+					return nil, err
+				}
+				if node == "" {
+					node = thisNode()
+				}
+				a, err := AskUpgrade(upgradeState(StateAsks), node, m.Now())
+				if err != nil {
+					return nil, err
+				}
+				return map[string]any{"asked": a, "follow": "pacman_upgrades with node " + node + " lists the run that names this ask", "news": m.News("")}, nil
 			},
 		},
 		{

modules/pacman/cmd/pacman-tools/pacman_test.go

@@ -1,7 +1,6 @@
 package main
 
 import (
-	"errors"
 	"strings"
 	"testing"
 	"time"
@@ -135,60 +134,17 @@ func lockless(m *Machine) *Machine {
 	return m
 }
 
-func TestAnUpgradeRunsAsAUnitOfItsOwnThroughSudoAndBringsTheNews(t *testing.T) {
-	fetch = func(string) ([]byte, error) { return nil, errors.New("no network") }
-	var calls []call
-	m := lockless(machine(fake(func(c call) Ran { return Ran{} }, &calls), 1000))
-	r, err := m.Upgrade("", 60)
-	if err != nil {
-		t.Fatal(err)
-	}
-	unit := r["started"].(string)
-	if unit != "mesh-pacman-upgrade-1791115200" {
-		t.Fatalf("unit: %s", unit)
-	}
-	last := calls[len(calls)-1]
-	want := "sudo -n systemd-run --unit=" + unit
-	if !strings.HasPrefix(last.String(), want) || !strings.HasSuffix(last.String(), "--quiet pacman -Syu --noconfirm") || strings.Contains(last.String(), "--wait") {
-		t.Fatalf("started as: %s", last)
-	}
-	news := r["news"].(map[string]any)
-	if news["reachable"] != false || !strings.Contains(news["error"].(string), "no network") {
-		t.Fatalf("no network is an answer: %v", news)
-	}
-}
-
-func TestAnUpgradeIsRefusedWhileTheDatabaseIsLocked(t *testing.T) {
-	fetch = func(string) ([]byte, error) { return nil, errors.New("offline") }
-	var calls []call
-	m := machine(fake(func(c call) Ran { return Ran{} }, &calls), 1000)
-	m.ReadFile = func(p string) ([]byte, error) {
-		if p == DBLock {
-			return []byte{}, nil
-		}
-		return nil, errNoFile
-	}
-	if _, err := m.Upgrade("", 60); err == nil || !strings.Contains(err.Error(), "another pacman holds") {
-		t.Fatalf("%v", err)
-	}
-	for _, c := range calls {
-		if c.name == "sudo" {
-			t.Fatal("started while locked")
-		}
-	}
-}
-
 func TestAnUpgradesUnitIsReadBack(t *testing.T) {
 	unit := "mesh-pacman-upgrade-1791115200"
 	m := machine(byLine(map[string]Ran{
 		"systemctl show " + unit + " --no-pager --property=LoadState --property=ActiveState --property=SubState --property=Result --property=ExecMainStatus": {Stdout: "LoadState=loaded\nActiveState=failed\nSubState=failed\nResult=exit-code\nExecMainStatus=1\n"},
 		"sudo -n journalctl --no-pager -o cat -n 60 -u " + unit: {Stdout: "error: failed to commit transaction (conflicting files)\n"},
 	}, nil), 1000)
-	r, err := m.Upgrade(unit, 60)
+	r, err := m.UpgradeStatus(unit, 60)
 	if err != nil || r["finished"] != true || r["succeeded"] != false || r["exit_status"] != "1" || len(r["log"].([]string)) != 1 {
 		t.Fatalf("%v %v", r, err)
 	}
-	if _, err := m.Upgrade("sshd.service", 60); err == nil {
+	if _, err := m.UpgradeStatus("sshd.service", 60); err == nil {
 		t.Fatal("a unit the tools did not start was read")
 	}
 }

modules/pacman/cmd/pacman-tools/transaction.go

@@ -0,0 +1,379 @@
+package main
+
+// An upgrade's transaction through an ask (novox/hq ADR 0508 §1, §8): a hand upgrade writes an ask in `asks` at
+// `<node>.<id>`; that node's keeper alone takes it and runs the steps: the record, the DKMS gate, the
+// transaction as a unit of the service manager started by the root helper (ADR 0510), its end, and the
+// after-upgrade check, each written to the run and emitted. A keeper started again finds an unfinished run in
+// `runs` and follows it to its end.
+
+import (
+	"crypto/rand"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"os"
+	"strconv"
+	"strings"
+	"time"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+)
+
+// journalLines is how much of a failed transaction's journal is read for its last error.
+const journalLines = 40
+
+// thisNode is the node the tools bundle runs on, as the runtime names it; a test replaces it.
+var thisNode = func() string { return os.Getenv("MESH_NODE") }
+
+// AskID names an ask by the moment it was asked and a few random digits, so two asks in one second differ.
+func AskID(at time.Time) string {
+	b := make([]byte, 3)
+	_, _ = rand.Read(b)
+	return RunID(at) + "-" + hex.EncodeToString(b)
+}
+
+// AskUpgrade writes an ask for a full upgrade of a node, for that node's keeper to take: what pacman_upgrade
+// does (ADR 0508 §8).
+func AskUpgrade(asks Bucket, node string, now time.Time) (Ask, error) {
+	if node == "" {
+		return Ask{}, errors.New("no node to ask: the runtime names none (MESH_NODE), and none was given")
+	}
+	a := Ask{ID: AskID(now), Node: node, Act: ActUpgrade, At: now.UTC()}
+	if _, err := asks.Create(NodeKey(node, a.ID), a); err != nil {
+		return Ask{}, fmt.Errorf("writing the ask: %w", err)
+	}
+	return a, nil
+}
+
+// lastStep is a run's last step, or none.
+func lastStep(r Run) string {
+	if len(r.Steps) == 0 {
+		return ""
+	}
+	return r.Steps[len(r.Steps)-1].Step
+}
+
+// ended says whether a run has nothing more to do.
+func ended(r Run) bool {
+	switch lastStep(r) {
+	case StepRefused, StepCheckedAfter:
+		return true
+	}
+	return false
+}
+
+// ownRuns are the node's runs, oldest first, each with the revision of its key.
+func (u *Upgrades) ownRuns() ([]openRun, error) {
+	keys, err := u.Runs.Keys()
+	if err != nil {
+		return nil, fmt.Errorf("listing runs: %w", err)
+	}
+	var out []openRun
+	for _, k := range keys {
+		if !strings.HasPrefix(k, u.Node+".") {
+			continue
+		}
+		e, err := u.Runs.Get(k)
+		if err != nil {
+			return nil, err
+		}
+		if e == nil {
+			continue
+		}
+		var r Run
+		if err := json.Unmarshal(e.Value, &r); err != nil {
+			return nil, fmt.Errorf("run %s: %w", k, err)
+		}
+		out = append(out, openRun{Run: r, rev: e.Revision})
+	}
+	return out, nil
+}
+
+// refuse ends a run that does not start, saying why.
+func (u *Upgrades) refuse(r *openRun, why string) error {
+	if err := u.step(r, Step{Step: StepRefused, At: u.Machine.Now().UTC(), Why: why}); err != nil {
+		return err
+	}
+	return u.emit(EventFailed, UpgradeFailed{Node: r.Node, Run: r.Run.Run, Step: StepRefused, Error: why})
+}
+
+// StartUpgrade runs a full upgrade's steps up to its transaction's start: the record, the DKMS gate and the
+// transaction's unit. The rest follows when the transaction ends (Follow). A run that cannot start is
+// written as refused, saying why; only a run that could not be written at all is an error.
+func (u *Upgrades) StartUpgrade(from, round, ask string) (Run, error) {
+	now := u.Machine.Now().UTC()
+	r := &openRun{Run: Run{Node: u.Node, Run: RunID(now), Kind: RunUpgrade, From: from, Round: round, Ask: ask}}
+	err := u.startUpgrade(r)
+	return r.Run, err
+}
+
+func (u *Upgrades) startUpgrade(r *openRun) error {
+	runs, err := u.ownRuns()
+	if err != nil {
+		return err
+	}
+	for _, o := range runs {
+		if o.Run.Run == r.Run.Run {
+			return fmt.Errorf("run %s of %s is there already: one run a second", o.Run.Run, u.Node)
+		}
+		if !ended(o.Run) {
+			return u.refuse(r, fmt.Sprintf("run %s is still going (its last step is %s): one run at a time", o.Run.Run, lastStep(o.Run)))
+		}
+	}
+	if u.Machine.locked() {
+		return u.refuse(r, "another pacman holds "+DBLock+": a transaction is running, or one was killed and left its lock")
+	}
+	rec, why := u.TakeRecord(r.Run.Run)
+	if why != "" {
+		return u.refuse(r, why)
+	}
+	if err := u.step(r, Step{Step: StepRecorded, At: u.Machine.Now().UTC()}); err != nil {
+		return err
+	}
+	var errs []error
+	if err := u.emit(EventRecorded, UpgradeRecorded{Node: u.Node, Run: r.Run.Run, Packages: len(rec.Installed.Packages), Kernel: rec.Kernel}); err != nil {
+		errs = append(errs, err)
+	}
+	held, ignore, why := u.Gate()
+	if why != "" {
+		return errors.Join(append(errs, u.refuse(r, why))...)
+	}
+	if err := u.step(r, Step{Step: StepCheckedBefore, At: u.Machine.Now().UTC(), Held: held}); err != nil {
+		return errors.Join(append(errs, err)...)
+	}
+	if len(held) > 0 {
+		if err := u.emit(EventKernelHeld, KernelHeld{Node: u.Node, Run: r.Run.Run, Held: held}); err != nil {
+			errs = append(errs, err)
+		}
+	}
+	return errors.Join(append(errs, u.start(r, ignore))...)
+}
+
+// start starts the run's transaction through the root helper and writes it started, or failed when it could
+// not start, followed by the after-upgrade check.
+func (u *Upgrades) start(r *openRun, ignore []string) error {
+	at := u.Machine.Now().UTC()
+	unit := TransactionUnit(r.Kind, r.Run.Run)
+	if err := u.Root.Transaction(r.Kind, r.Run.Run, ignore); err != nil {
+		return u.end(r, Step{Step: StepFailed, At: u.Machine.Now().UTC(), Error: "the transaction could not be started: " + err.Error()})
+	}
+	if err := u.step(r, Step{Step: StepStarted, At: at, Unit: unit, Ignored: ignore}); err != nil {
+		return err
+	}
+	if r.Kind == RunRollback {
+		return u.emit(EventRollbackStarted, RollbackStarted{Node: u.Node, Run: r.Run.Run, Of: r.Of, Unit: unit})
+	}
+	return u.emit(EventStarted, UpgradeStarted{Node: u.Node, Run: r.Run.Run, From: r.From, Round: r.Round, Unit: unit, Ignored: ignore})
+}
+
+// end writes how the transaction ended, emits it, and runs the after-upgrade check.
+func (u *Upgrades) end(r *openRun, s Step) error {
+	if err := u.step(r, s); err != nil {
+		return err
+	}
+	var errs []error
+	exit := 0
+	if s.Exit != nil {
+		exit = *s.Exit
+	}
+	switch {
+	case r.Kind == RunRollback:
+		errs = append(errs, u.emit(EventRollbackFinished, RollbackFinished{Node: u.Node, Run: r.Run.Run, Of: r.Of, Exit: exit, Error: s.Error}))
+	case s.Step == StepFinished:
+		changes := Changes{}
+		if s.Changes != nil {
+			changes = *s.Changes
+		}
+		errs = append(errs, u.emit(EventFinished, UpgradeFinished{Node: u.Node, Run: r.Run.Run, Changes: changes}))
+	default:
+		errs = append(errs, u.emit(EventFailed, UpgradeFailed{Node: u.Node, Run: r.Run.Run, Step: StepFailed, Exit: s.Exit, Error: s.Error}))
+	}
+	return errors.Join(append(errs, u.checkAfter(r))...)
+}
+
+// checkAfter writes the after-upgrade check (ADR 0508 §4) and emits it.
+func (u *Upgrades) checkAfter(r *openRun) error {
+	var rec *Record
+	key := NodeKey(u.Node, r.Run.Run)
+	if r.Kind == RunRollback {
+		key = NodeKey(u.Node, r.Of)
+	}
+	if e, err := u.Records.Get(key); err == nil && e != nil {
+		var got Record
+		if json.Unmarshal(e.Value, &got) == nil {
+			rec = &got
+		}
+	}
+	var c Check
+	if u.Check != nil {
+		c = u.Check(r.Run, rec)
+	} else {
+		c = Check{Passed: transactionSucceeded(r.Run)}
+	}
+	if err := u.step(r, Step{Step: StepCheckedAfter, At: u.Machine.Now().UTC(), Check: &c}); err != nil {
+		return err
+	}
+	return u.emit(EventChecked, UpgradeChecked{Node: u.Node, Run: r.Run.Run, Kind: r.Kind, Passed: c.Passed})
+}
+
+// transactionSucceeded says whether a run's transaction finished.
+func transactionSucceeded(r Run) bool {
+	for _, s := range r.Steps {
+		if s.Step == StepFinished {
+			return true
+		}
+	}
+	return false
+}
+
+// unitEnd reads a transaction unit's state: whether it ended, and its exit status. A transient unit that ended
+// well is let go by the service manager (not found); one that failed stays, failed.
+func (u *Upgrades) unitEnd(unit string) (done bool, exit int, err error) {
+	p, err := u.Machine.unitProps(unit, "LoadState", "ActiveState", "Result", "ExecMainStatus")
+	if err != nil {
+		return false, 0, err
+	}
+	switch {
+	case p["LoadState"] == "not-found":
+		return true, 0, nil
+	case p["ActiveState"] == "active" || p["ActiveState"] == "activating" || p["ActiveState"] == "deactivating" || p["ActiveState"] == "reloading":
+		return false, 0, nil
+	}
+	exit, _ = strconv.Atoi(p["ExecMainStatus"])
+	if exit == 0 && p["Result"] != "success" {
+		exit = 1
+	}
+	return true, exit, nil
+}
+
+// transactionChanges is what the run's transaction changed, read from its lines in pacman's log: the first
+// command of its kind pacman logged since the transaction was started.
+func (u *Upgrades) transactionChanges(kind string, since time.Time) *Changes {
+	text, err := u.Machine.ReadFile(PacmanLog)
+	if err != nil {
+		return nil
+	}
+	since = since.Truncate(time.Second)
+	for _, b := range logBlocks(string(text)) {
+		if b.Start.Before(since) {
+			continue
+		}
+		if kind == RunUpgrade && !b.Full || kind == RunRollback && !strings.Contains(b.Command, " -U") {
+			continue
+		}
+		c := b.Changes
+		return &c
+	}
+	return nil
+}
+
+// follow writes the end of a run whose transaction has ended, and its check; a transaction still running is
+// left to the next call.
+func (u *Upgrades) follow(r *openRun) error {
+	started := r.Steps[len(r.Steps)-1]
+	done, exit, err := u.unitEnd(started.Unit)
+	if err != nil {
+		return fmt.Errorf("reading %s: %w", started.Unit, err)
+	}
+	if !done {
+		return nil
+	}
+	s := Step{Step: StepFinished, At: u.Machine.Now().UTC(), Exit: &exit, Changes: u.transactionChanges(r.Kind, started.At)}
+	if exit != 0 {
+		s.Step = StepFailed
+		lines, err := u.Root.Journal(started.Unit, journalLines)
+		if err != nil {
+			s.Error = "(its journal could not be read: " + err.Error() + ")"
+		} else {
+			s.Error = lastError(lines)
+		}
+	}
+	return u.end(r, s)
+}
+
+// Follow carries every unfinished run of the node on as far as it can go now: a transaction that ended is
+// written finished or failed and checked; a run stopped between its steps by a keeper that went away is taken
+// up where it stopped. The keeper calls it at its start and whenever pacman's database lock is released.
+func (u *Upgrades) Follow() error {
+	runs, err := u.ownRuns()
+	if err != nil {
+		return err
+	}
+	var errs []error
+	for i := range runs {
+		r := &runs[i]
+		if ended(r.Run) || r.From == FromLog {
+			continue
+		}
+		switch lastStep(r.Run) {
+		case StepStarted:
+			errs = append(errs, u.follow(r))
+		case StepFinished, StepFailed:
+			errs = append(errs, u.checkAfter(r))
+		case StepRecorded, StepCheckedBefore, "":
+			errs = append(errs, u.resumeBeforeStart(r))
+		}
+	}
+	return errors.Join(errs...)
+}
+
+// resumeBeforeStart takes up a run whose keeper went away before it wrote the transaction started: when the
+// run's unit is there, or pacman's log holds an upgrade since the run's last step, the transaction did start
+// and is followed; otherwise nothing started, and the run is refused, so said.
+func (u *Upgrades) resumeBeforeStart(r *openRun) error {
+	unit := TransactionUnit(r.Kind, r.Run.Run)
+	since := time.Time{}
+	if n := len(r.Steps); n > 0 {
+		since = r.Steps[n-1].At
+	}
+	p, err := u.Machine.unitProps(unit, "LoadState")
+	if err != nil {
+		return fmt.Errorf("reading %s: %w", unit, err)
+	}
+	if p["LoadState"] == "not-found" && u.transactionChanges(r.Kind, since) == nil {
+		return u.refuse(r, "the keeper stopped before it started the transaction, and nothing was started")
+	}
+	if err := u.step(r, Step{Step: StepStarted, At: since, Unit: unit}); err != nil {
+		return err
+	}
+	return u.follow(r)
+}
+
+// TakeAsk takes one change of `asks`: an ask for this node, not taken yet, is carried out and then deleted. An
+// ask for another node is that node's keeper's. An ask a run already names was taken by a keeper that went
+// away before deleting it, and is deleted. The keeper watches `asks` at `<node>.*` and hands each change here.
+func (u *Upgrades) TakeAsk(c stdio.StateChange) error {
+	if c.Op != "put" || !strings.HasPrefix(c.Key, u.Node+".") {
+		return nil
+	}
+	var a Ask
+	if err := json.Unmarshal(c.Value, &a); err != nil {
+		return fmt.Errorf("ask %s is not readable: %w", c.Key, err)
+	}
+	if a.Node != u.Node {
+		return fmt.Errorf("ask %s names node %q, not this one", c.Key, a.Node)
+	}
+	runs, err := u.ownRuns()
+	if err != nil {
+		return err
+	}
+	for _, r := range runs {
+		if r.Ask == a.ID {
+			return u.Asks.Delete(c.Key)
+		}
+	}
+	switch a.Act {
+	case ActUpgrade:
+		r, err := u.StartUpgrade(FromAsk, "", a.ID)
+		if len(r.Steps) == 0 {
+			return err // no run names the ask yet: it stays, and is taken again
+		}
+		if derr := u.Asks.Delete(c.Key); derr != nil {
+			err = errors.Join(err, fmt.Errorf("deleting the ask %s: %w", c.Key, derr))
+		}
+		return err
+	default:
+		return fmt.Errorf("ask %s asks %q, which this keeper does not carry out yet", c.Key, a.Act)
+	}
+}

modules/pacman/cmd/pacman-tools/transaction_test.go

@@ -0,0 +1,405 @@
+package main
+
+// An upgrade through an ask, and a keeper started again following its run (novox/hq ADR 0508 §1, §8;
+// acceptance 1841/AC1 and 1841/AC9).
+
+import (
+	"encoding/json"
+	"errors"
+	"strings"
+	"testing"
+	"time"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+)
+
+// fakeRoot is the root helper: it records the transactions it was asked to start.
+type fakeRoot struct {
+	started []string
+	fail    error
+	journal []string
+}
+
+func (f *fakeRoot) Transaction(kind, run string, ignore []string) error {
+	if f.fail != nil {
+		return f.fail
+	}
+	f.started = append(f.started, kind+" "+run+" ignore="+strings.Join(ignore, ","))
+	return nil
+}
+
+func (f *fakeRoot) Journal(unit string, n int) ([]string, error) { return f.journal, nil }
+
+// keeperBed is one node's keeper over states in memory and a machine of fakes: a unit's state, the lock and
+// pacman's log are set by the test as the transaction goes.
+type keeperBed struct {
+	u                                       *Upgrades
+	root                                    *fakeRoot
+	events                                  *eventLog
+	installed, updates, runs, records, asks *memBucket
+	now                                     time.Time
+	unitState                               string // systemctl show's answer for the run's unit
+	locked                                  bool
+	log                                     string
+}
+
+const upgradeLog = "[2026-10-04T12:00:01+0000] [PACMAN] Running 'pacman -Syu --noconfirm --ignore linux,linux-headers'\n" +
+	"[2026-10-04T12:00:02+0000] [PACMAN] starting full system upgrade\n" +
+	"[2026-10-04T12:00:09+0000] [ALPM] transaction started\n" +
+	"[2026-10-04T12:00:10+0000] [ALPM] upgraded bash (5.3.15-1 -> 5.3.20-1)\n" +
+	"[2026-10-04T12:00:10+0000] [ALPM] installed made-up-library (1.0-1)\n" +
+	"[2026-10-04T12:00:11+0000] [ALPM] transaction completed\n"
+
+func newKeeperBed(t *testing.T) *keeperBed {
+	k := &keeperBed{root: &fakeRoot{}, events: &eventLog{}, installed: newBucket(), updates: newBucket(), runs: newBucket(),
+		records: newBucket(), asks: newBucket(), now: time.Date(2026, 10, 4, 12, 0, 0, 0, time.UTC), unitState: "LoadState=not-found\n"}
+	if _, err := k.installed.Put("node-a", nodeAPackages); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := k.updates.Put("node-a", UpdatesValue{Reachable: true, Packages: []VersionChange{
+		{Name: "bash", From: "5.3.15-1", To: "5.3.20-1"},
+		{Name: "linux", From: "7.2.8.arch1-2", To: "7.3.1.arch1-1"},
+		{Name: "linux-headers", From: "7.2.8.arch1-2", To: "7.3.1.arch1-1"},
+	}}); err != nil {
+		t.Fatal(err)
+	}
+	sources := capturedZFSSource(t)
+	files := func(paths ...string) Ran { return Ran{Stdout: strings.Join(paths, "\n") + "\n"} }
+	m := machine(fake(func(c call) Ran {
+		line := c.String()
+		switch {
+		case line == "systemctl list-units --state=failed --plain --no-legend --no-pager":
+			return Ran{}
+		case line == "dkms status":
+			return Ran{Stdout: zfsStatus}
+		case line == "pacman -Qlq -- linux":
+			return files("/usr/lib/modules/7.2.8-arch1-2/vmlinuz")
+		case line == "pacman -Qlq -- linux-headers":
+			return files("/usr/lib/modules/7.2.8-arch1-2/build/Makefile")
+		case strings.HasPrefix(line, "systemctl show "+unitPrefix):
+			return Ran{Stdout: k.unitState}
+		}
+		return Ran{Status: 99, Stderr: "unexpected command: " + line}
+	}, nil), 1000)
+	m.Now = func() time.Time { return k.now }
+	m.ReadFile = func(path string) ([]byte, error) {
+		switch {
+		case path == OSRelease:
+			return []byte("7.2.8-arch1-2\n"), nil
+		case path == DBLock && k.locked:
+			return []byte{}, nil
+		case path == PacmanLog:
+			return []byte(k.log), nil
+		}
+		if s, ok := sources[path]; ok {
+			return []byte(s), nil
+		}
+		return nil, errNoFile
+	}
+	k.u = k.keeper(m)
+	return k
+}
+
+// keeper is a keeper for node-a over the bed's states: a second call is the keeper started again.
+func (k *keeperBed) keeper(m *Machine) *Upgrades {
+	return &Upgrades{Node: "node-a", Machine: m, Root: k.root, Events: k.events, Installed: k.installed, Updates: k.updates,
+		Runs: k.runs, Records: k.records, Asks: k.asks}
+}
+
+// ask writes an ask for node and hands its change to the keeper, as the watch of `asks` would.
+func (k *keeperBed) ask(t *testing.T, node, act string) (string, error) {
+	t.Helper()
+	a := Ask{ID: RunID(k.now) + "-abc123", Node: node, Act: act, At: k.now}
+	key := NodeKey(node, a.ID)
+	rev, err := k.asks.Create(key, a)
+	if err != nil {
+		t.Fatal(err)
+	}
+	raw, _ := json.Marshal(a)
+	return key, k.u.TakeAsk(stdio.StateChange{State: StateAsks, Key: key, Op: "put", Value: raw, Revision: rev})
+}
+
+func (k *keeperBed) run(t *testing.T) Run {
+	t.Helper()
+	keys, _ := k.runs.Keys()
+	if len(keys) != 1 {
+		t.Fatalf("runs: %v", keys)
+	}
+	var r Run
+	k.runs.value(keys[0], &r)
+	return r
+}
+
+func steps(r Run) string {
+	var s []string
+	for _, st := range r.Steps {
+		s = append(s, st.Step)
+	}
+	return strings.Join(s, " ")
+}
+
+func TestAnAskedUpgradeIsARunWrittenAndEmittedAtEachStep(t *testing.T) {
+	k := newKeeperBed(t)
+	key, err := k.ask(t, "node-a", ActUpgrade)
+	if err != nil {
+		t.Fatal(err)
+	}
+	r := k.run(t)
+	if r.Run != "20261004T120000Z" || r.From != FromAsk || r.Ask != "20261004T120000Z-abc123" || r.Kind != RunUpgrade {
+		t.Errorf("the run: %+v", r)
+	}
+	if steps(r) != "recorded checked-before started" {
+		t.Fatalf("steps before the transaction ends: %s", steps(r))
+	}
+	if h := r.Steps[1].Held; len(h) != 1 || h[0].Package != "linux" {
+		t.Errorf("the gate's step names the held kernel: %+v", h)
+	}
+	if s := r.Steps[2]; s.Unit != "mesh-pacman-upgrade-20261004t120000z" || strings.Join(s.Ignored, ",") != "linux,linux-headers" {
+		t.Errorf("started: %+v", s)
+	}
+	if strings.Join(k.root.started, ";") != "upgrade 20261004T120000Z ignore=linux,linux-headers" {
+		t.Errorf("the root helper was asked: %v", k.root.started)
+	}
+	if _, there := k.asks.values[key]; there {
+		t.Error("the ask is deleted once the run names it")
+	}
+	var rec Record
+	if !k.records.value("node-a.20261004T120000Z", &rec) {
+		t.Error("no record was taken")
+	}
+
+	// The transaction ends: the lock is released and the keeper follows the run.
+	k.unitState, k.log, k.now = "LoadState=not-found\n", upgradeLog, k.now.Add(15*time.Second)
+	if err := k.u.Follow(); err != nil {
+		t.Fatal(err)
+	}
+	r = k.run(t)
+	if steps(r) != "recorded checked-before started finished checked-after" {
+		t.Fatalf("steps: %s", steps(r))
+	}
+	end := r.Steps[3]
+	if end.Exit == nil || *end.Exit != 0 || end.Changes == nil || len(end.Changes.Changed) != 1 || end.Changes.Changed[0].Name != "bash" ||
+		len(end.Changes.Added) != 1 {
+		t.Errorf("finished: %+v %+v", end, end.Changes)
+	}
+	if c := r.Steps[4].Check; c == nil || !c.Passed {
+		t.Errorf("the check: %+v", c)
+	}
+	if got := strings.Join(k.events.names(), " "); got != "upgrade.recorded upgrade.kernel-held upgrade.started upgrade.finished upgrade.checked" {
+		t.Errorf("events: %s", got)
+	}
+	if w := k.runs.writesTo("node-a."); len(w) != 5 || w[0] != "create node-a.20261004T120000Z" {
+		t.Errorf("each step written once: %v", w)
+	}
+	if err := k.u.Follow(); err != nil || len(k.runs.writesTo("node-a.")) != 5 {
+		t.Errorf("an ended run is not written again: %v", err)
+	}
+}
+
+func TestAFailedTransactionIsWrittenWithItsExitAndLastErrorAndChecked(t *testing.T) {
+	k := newKeeperBed(t)
+	if _, err := k.ask(t, "node-a", ActUpgrade); err != nil {
+		t.Fatal(err)
+	}
+	k.unitState = "LoadState=loaded\nActiveState=failed\nResult=exit-code\nExecMainStatus=1\n"
+	k.root.journal = []string{"checking for file conflicts...", "error: failed to commit transaction (conflicting files)", "Errors occurred, no packages were upgraded."}
+	if err := k.u.Follow(); err != nil {
+		t.Fatal(err)
+	}
+	r := k.run(t)
+	end := r.Steps[3]
+	if steps(r) != "recorded checked-before started failed checked-after" || *end.Exit != 1 ||
+		end.Error != "error: failed to commit transaction (conflicting files)" || r.Steps[4].Check.Passed {
+		t.Errorf("%s %+v %+v", steps(r), end, r.Steps[4].Check)
+	}
+	if got := k.events.names(); got[len(got)-2] != EventFailed || got[len(got)-1] != EventChecked {
+		t.Errorf("events: %v", got)
+	}
+}
+
+func TestATransactionThatCannotStartFailsAndIsChecked(t *testing.T) {
+	k := newKeeperBed(t)
+	k.root.fail = errors.New("the root helper's socket is not there")
+	if _, err := k.ask(t, "node-a", ActUpgrade); err != nil {
+		t.Fatal(err)
+	}
+	r := k.run(t)
+	if steps(r) != "recorded checked-before failed checked-after" || !strings.Contains(r.Steps[2].Error, "socket is not there") {
+		t.Errorf("%s %+v", steps(r), r.Steps[2])
+	}
+}
+
+func TestAnAskIsTakenByItsNodesKeeperAlone(t *testing.T) {
+	k := newKeeperBed(t)
+	key, err := k.ask(t, "node-b", ActUpgrade)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if keys, _ := k.runs.Keys(); len(keys) != 0 || len(k.root.started) != 0 {
+		t.Errorf("node-a's keeper took node-b's ask: %v %v", keys, k.root.started)
+	}
+	if _, there := k.asks.values[key]; !there {
+		t.Error("another node's ask is left for its keeper")
+	}
+	if err := k.u.TakeAsk(stdio.StateChange{State: StateAsks, Key: "node-a.x", Op: "delete"}); err != nil {
+		t.Errorf("a delete is no ask: %v", err)
+	}
+}
+
+func TestAnAskARunAlreadyNamesIsDeletedAndNotRunAgain(t *testing.T) {
+	k := newKeeperBed(t)
+	done := Run{Node: "node-a", Run: "20261004T115900Z", Kind: RunUpgrade, From: FromAsk, Ask: RunID(k.now) + "-abc123",
+		Steps: []Step{{Step: StepRefused, At: k.now, Why: "x"}}}
+	if _, err := k.runs.Create(NodeKey("node-a", done.Run), done); err != nil {
+		t.Fatal(err)
+	}
+	key, err := k.ask(t, "node-a", ActUpgrade)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if keys, _ := k.runs.Keys(); len(keys) != 1 || len(k.root.started) != 0 {
+		t.Errorf("taken twice: %v", keys)
+	}
+	if _, there := k.asks.values[key]; there {
+		t.Error("the ask a run names is deleted")
+	}
+}
+
+func TestARunDoesNotStartWhileTheDatabaseIsLockedOrAnotherRunGoesOn(t *testing.T) {
+	k := newKeeperBed(t)
+	k.locked = true
+	if _, err := k.ask(t, "node-a", ActUpgrade); err != nil {
+		t.Fatal(err)
+	}
+	r := k.run(t)
+	if steps(r) != "refused" || !strings.Contains(r.Steps[0].Why, DBLock) || len(k.root.started) != 0 {
+		t.Errorf("locked: %s %+v", steps(r), r.Steps)
+	}
+	if e := k.events.sent; len(e) != 1 || e[0].Name != EventFailed || e[0].Body.(UpgradeFailed).Step != StepRefused {
+		t.Errorf("a refused run is said: %+v", e)
+	}
+
+	k = newKeeperBed(t)
+	going := Run{Node: "node-a", Run: "20261004T110000Z", Kind: RunUpgrade, From: FromRound, Steps: []Step{{Step: StepStarted, At: k.now}}}
+	if _, err := k.runs.Create(NodeKey("node-a", going.Run), going); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := k.u.StartUpgrade(FromAsk, "", "a1"); err != nil {
+		t.Fatal(err)
+	}
+	var r2 Run
+	k.runs.value("node-a.20261004T120000Z", &r2)
+	if steps(r2) != "refused" || !strings.Contains(r2.Steps[0].Why, "20261004T110000Z is still going") {
+		t.Errorf("another run going: %+v", r2)
+	}
+}
+
+func TestAKeeperStartedAgainFollowsItsRunToItsEnd(t *testing.T) {
+	k := newKeeperBed(t)
+	if _, err := k.ask(t, "node-a", ActUpgrade); err != nil {
+		t.Fatal(err)
+	}
+	// The keeper goes away while the transaction runs; a new one starts and finds it still running.
+	again := k.keeper(k.u.Machine)
+	k.unitState = "LoadState=loaded\nActiveState=active\nResult=success\nExecMainStatus=0\n"
+	before := len(k.runs.writes)
+	if err := again.Follow(); err != nil || len(k.runs.writes) != before {
+		t.Fatalf("a running transaction is left running: %v, %d writes", err, len(k.runs.writes)-before)
+	}
+	k.unitState, k.log = "LoadState=not-found\n", upgradeLog
+	if err := again.Follow(); err != nil {
+		t.Fatal(err)
+	}
+	if r := k.run(t); steps(r) != "recorded checked-before started finished checked-after" {
+		t.Errorf("followed to its end: %s", steps(r))
+	}
+}
+
+func TestAKeeperStartedAgainTakesUpARunStoppedBetweenItsSteps(t *testing.T) {
+	// Stopped after the gate, before the transaction started: nothing started, so the run is refused.
+	k := newKeeperBed(t)
+	stopped := Run{Node: "node-a", Run: "20261004T115900Z", Kind: RunUpgrade, From: FromAsk,
+		Steps: []Step{{Step: StepRecorded, At: k.now}, {Step: StepCheckedBefore, At: k.now}}}
+	if _, err := k.runs.Create(NodeKey("node-a", stopped.Run), stopped); err != nil {
+		t.Fatal(err)
+	}
+	if err := k.u.Follow(); err != nil {
+		t.Fatal(err)
+	}
+	if r := k.run(t); steps(r) != "recorded checked-before refused" || !strings.Contains(r.Steps[2].Why, "nothing was started") {
+		t.Errorf("%s %+v", steps(r), r.Steps)
+	}
+
+	// Stopped after the transaction started and before it was written: its log says it ran.
+	k = newKeeperBed(t)
+	if _, err := k.runs.Create(NodeKey("node-a", stopped.Run), stopped); err != nil {
+		t.Fatal(err)
+	}
+	k.log = upgradeLog
+	if err := k.u.Follow(); err != nil {
+		t.Fatal(err)
+	}
+	if r := k.run(t); steps(r) != "recorded checked-before started finished checked-after" || r.Steps[2].Unit != "mesh-pacman-upgrade-20261004t115900z" {
+		t.Errorf("%s %+v", steps(r), r.Steps)
+	}
+
+	// Stopped after the transaction ended and before it was checked.
+	k = newKeeperBed(t)
+	zero := 0
+	unchecked := Run{Node: "node-a", Run: "20261004T115900Z", Kind: RunUpgrade, From: FromAsk,
+		Steps: []Step{{Step: StepRecorded, At: k.now}, {Step: StepCheckedBefore, At: k.now}, {Step: StepStarted, At: k.now}, {Step: StepFinished, At: k.now, Exit: &zero}}}
+	if _, err := k.runs.Create(NodeKey("node-a", unchecked.Run), unchecked); err != nil {
+		t.Fatal(err)
+	}
+	if err := k.u.Follow(); err != nil {
+		t.Fatal(err)
+	}
+	if r := k.run(t); steps(r) != "recorded checked-before started finished checked-after" {
+		t.Errorf("%s", steps(r))
+	}
+}
+
+func TestPacmanUpgradeWritesAnAskForTheNodeAndStartsNothing(t *testing.T) {
+	asks := newBucket()
+	savedState, savedNode, savedFetch := upgradeState, thisNode, fetch
+	upgradeState = func(name string) Bucket {
+		if name != StateAsks {
+			t.Fatalf("wrote %s", name)
+		}
+		return asks
+	}
+	thisNode = func() string { return "node-a" }
+	fetch = func(string) ([]byte, error) { return nil, errors.New("offline") }
+	defer func() { upgradeState, thisNode, fetch = savedState, savedNode, savedFetch }()
+	var calls []call
+	m := machine(fake(func(c call) Ran { return Ran{} }, &calls), 1000)
+	for _, tool := range tools(m) {
+		if tool.Name != "pacman_upgrade" {
+			continue
+		}
+		out, err := tool.Run(map[string]any{})
+		if err != nil {
+			t.Fatal(err)
+		}
+		a := out.(map[string]any)["asked"].(Ask)
+		if a.Node != "node-a" || a.Act != ActUpgrade || !strings.HasPrefix(a.ID, "20261004T120000Z-") {
+			t.Errorf("ask: %+v", a)
+		}
+		if keys, _ := asks.Keys(); len(keys) != 1 || keys[0] != NodeKey("node-a", a.ID) {
+			t.Errorf("asks: %v", keys)
+		}
+		for _, c := range calls {
+			if c.name == "sudo" || c.name == "systemd-run" {
+				t.Errorf("the tool acted itself: %s", c)
+			}
+		}
+		if _, err := tool.Run(map[string]any{"node": "node-b"}); err != nil {
+			t.Fatal(err)
+		}
+		if keys, _ := asks.Keys(); len(keys) != 2 || !strings.HasPrefix(keys[1], "node-b.") {
+			t.Errorf("another node's ask: %v", keys)
+		}
+		return
+	}
+	t.Fatal("no pacman_upgrade")
+}