diff --git a/modules/pacman/cmd/pacman-tools/acts.go b/modules/pacman/cmd/pacman-tools/acts.go index c53b68b9..ef3682ef 100644 --- a/modules/pacman/cmd/pacman-tools/acts.go +++ b/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) { diff --git a/modules/pacman/cmd/pacman-tools/dkms.go b/modules/pacman/cmd/pacman-tools/dkms.go index 202bd655..11f324a6 100644 --- a/modules/pacman/cmd/pacman-tools/dkms.go +++ b/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. diff --git a/modules/pacman/cmd/pacman-tools/main.go b/modules/pacman/cmd/pacman-tools/main.go index f6ce4d70..8c732a25 100644 --- a/modules/pacman/cmd/pacman-tools/main.go +++ b/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 }, }, { diff --git a/modules/pacman/cmd/pacman-tools/pacman_test.go b/modules/pacman/cmd/pacman-tools/pacman_test.go index d6ca6b64..f8b34b4d 100644 --- a/modules/pacman/cmd/pacman-tools/pacman_test.go +++ b/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") } } diff --git a/modules/pacman/cmd/pacman-tools/transaction.go b/modules/pacman/cmd/pacman-tools/transaction.go new file mode 100644 index 00000000..837163c8 --- /dev/null +++ b/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 +// `.`; 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 `.*` 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) + } +} diff --git a/modules/pacman/cmd/pacman-tools/transaction_test.go b/modules/pacman/cmd/pacman-tools/transaction_test.go new file mode 100644 index 00000000..e256b52c --- /dev/null +++ b/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") +}