Carry out an upgrade as an ask its node's keeper takes, through the root helper, and follow a run to its end after a restart, so a hand upgrade is a recorded, checked run like any other (issue 1841/S5)

This commit is contained in:
2026-10-22 14:35:19 +02:00
parent f40fa6e8e0
commit 95e69276c3
5 changed files with 811 additions and 73 deletions
+3 -22
View File
@@ -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) {
+22 -5
View File
@@ -133,11 +133,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) {
@@ -149,7 +152,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
},
},
{
+2 -46
View File
@@ -1,7 +1,6 @@
package main
import (
"errors"
"strings"
"testing"
"time"
@@ -146,60 +145,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")
}
}
@@ -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)
}
}
@@ -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")
}