diff --git a/modules/pacman/cmd/pacman-tools/keeper_manifest_test.go b/modules/pacman/cmd/pacman-tools/keeper_manifest_test.go index caaff26f..ae7cdfec 100644 --- a/modules/pacman/cmd/pacman-tools/keeper_manifest_test.go +++ b/modules/pacman/cmd/pacman-tools/keeper_manifest_test.go @@ -109,8 +109,19 @@ func TestTheKeeperDeclaresTheModulesStatesAndEvents(t *testing.T) { t.Errorf("the keeper's events: %v, want %v", got, events) } for key := range bus { - if key != "state" && key != "emits" { - t.Errorf("the keeper's bus declares %s: it keeps its own states and emits its events, nothing else yet", key) + if key != "state" && key != "emits" && key != "asks" { + t.Errorf("the keeper's bus declares %s: it keeps its own states, emits its events and reads its asks' answers, nothing else", key) + } + } + // A pass is answered on the operator's channel: the tools ask, the keeper reads the channel's record of its own + // asks (ADR 0508 §6, ADR 0259, ADR 0436 §1). + if bus["asks"] != true || strings.Join(strs(raw["uses"]), " ") != "operator-channel" { + t.Errorf("the keeper reads its asks' answers and the module uses the operator's channel: %v %v", bus["asks"], raw["uses"]) + } + // The tools read round-start-minutes from the keeper's config file. + for _, a := range manifest(t).Build.Artifacts { + if env, _ := a["env"].(map[string]any); a["name"] == "tools" && env[EnvConfig] != "${dir:state}/keeper.json" { + t.Errorf("the tools bundle is not named the keeper's config file: %v", a["env"]) } } if got := strs(raw["state"]); !reflect.DeepEqual(got, States) { diff --git a/modules/pacman/cmd/pacman-tools/main.go b/modules/pacman/cmd/pacman-tools/main.go index 8c732a25..0483b46e 100644 --- a/modules/pacman/cmd/pacman-tools/main.go +++ b/modules/pacman/cmd/pacman-tools/main.go @@ -265,6 +265,37 @@ func tools(m *Machine) []stdio.Tool { return ListUpgrades(upgradeState(StateRuns), v[0], v[1], v[2], v[3], n) }, }, + { + Name: "pacman_round", + Description: "Where the week's upgrade rounds stand (ADR 0508 §5, §6): each round's order and opening, and every node's state in it: " + + "waiting (for the node named), turn, stalled (its turn came longer than round-start-minutes ago), running, passed, failed " + + "(the round stops there) or refused, with since when. With pass, a node and a reason, asks the operator on their channel to let " + + "the round go on past that node; it goes on only on the operator's Approve.", + Input: schema(map[string]any{ + "round": map[string]any{"type": "string", "description": "one round, as 2026-W43 (default: the newest)"}, + "pass": map[string]any{"type": "string", "description": "a node to ask the operator to pass"}, + "reason": map[string]any{"type": "string", "description": "why the node should be passed: required with pass, and shown to the operator"}, + "limit": limitArg(4, 52), + }), + Run: func(args map[string]any) (any, error) { + var v [3]string + for i, k := range []string{"round", "pass", "reason"} { + s, err := text(args, k, false) + if err != nil { + return nil, err + } + v[i] = s + } + if v[1] != "" { + return AskPass(upgradeState(StateRounds), upgradeState(StateRuns), upgradeState(StateAsks), v[0], v[1], v[2], m.Now()) + } + n, err := whole(args, "limit", 4, 1, 52) + if err != nil { + return nil, err + } + return ListRounds(upgradeState(StateRounds), upgradeState(StateRuns), v[0], m.Now(), roundStartMinutes(), n) + }, + }, { Name: "pacman_orphans", Description: "Packages installed as dependencies that nothing requires any more (pacman -Qdt), with versions.", diff --git a/modules/pacman/cmd/pacman-tools/rounds.go b/modules/pacman/cmd/pacman-tools/rounds.go new file mode 100644 index 00000000..dcb10823 --- /dev/null +++ b/modules/pacman/cmd/pacman-tools/rounds.go @@ -0,0 +1,565 @@ +package main + +// The week's round (novox/hq ADR 0508 §5, §6): the nodes upgrade one at a time in the order of the mesh-level key +// `upgrade-order`, each after the node before it passed its after-upgrade check. On Saturday at 04:00 local time +// the first node of the order opens the week's round in `rounds` (a key created once, so one round a week); every +// node's keeper watches `rounds` and `runs` and starts its own run when its turn came. A failed run or check +// stops the round at that node. Only the operator passes a node: `pacman_round` with pass and a reason asks the +// operator through the operator's channel (ADR 0259), and the keeper of the node waiting on it lets the round go +// on only on the operator's Approve, read from the channel's own record. With no order, no round opens. + +import ( + "encoding/json" + "errors" + "fmt" + "os" + "sort" + "strings" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/asks" +) + +// Module is this module's name: the asker of its asks on the operator's channel. +const Module = "pacman" + +// MessengerAsksBucket is the operator channel's records bucket, where the keeper reads the answer to a pass. +const MessengerAsksBucket = "messenger_asks" + +// The operator's answers to a pass, besides PassApproved and PassDeclined: an ask nobody answered in time. +const PassExpired = "expired" + +// passLasts is how long a pass waits for the operator's answer. +const passLasts = 12 * time.Hour + +// refusedRetry is how long after a run of the round was refused (it started nothing) the node tries again. +const refusedRetry = time.Hour + +// passCheckEvery is how often a keeper reads the channel's record of a pass that waits for its answer. +const passCheckEvery = time.Minute + +// defaultRoundStartMinutes is round-start-minutes' default in the manifest. +const defaultRoundStartMinutes = 120 + +// RoundOpening is the Saturday at 04:00 of the ISO week that holds at, in at's own time zone. +func RoundOpening(at time.Time) time.Time { + day := int(at.Weekday()+6) % 7 // Monday 0 … Sunday 6 + monday := time.Date(at.Year(), at.Month(), at.Day()-day, 0, 0, 0, 0, at.Location()) + return monday.AddDate(0, 0, 5).Add(4 * time.Hour) +} + +// PassOperatorAsk is the ask on the operator's channel for a pass: composed from the pass alone, so the tool that +// publishes it and the keeper that reads its answer compose the same ask, with the same digest. +func PassOperatorAsk(round, node, next, reason, id string, at time.Time) asks.Ask { + headline := "Upgrade round: pass " + node + if len([]rune(headline)) > asks.HeadlineLength { + headline = "Upgrade round: pass a node" + } + binds, _ := asks.ActDigest(passAct(round, node, reason, id)) + return asks.Ask{ + ID: id, + Headline: headline, + Explanation: fmt.Sprintf("The week's upgrade round %s stands at %s. Passing it lets %s and the nodes after it upgrade; "+ + "%s is not upgraded in this round.\nThe reason given: %s", round, node, next, node, reason), + Options: []asks.Option{ + {ID: "pass", Label: "Pass it", Does: "the round goes on past " + node, Level: asks.Approve, Binds: binds}, + {ID: "keep", Label: "Keep it waiting", Does: "the round stays at " + node, Level: asks.Acknowledge}, + }, + Who: asks.Operator, + Expires: at.UTC().Add(passLasts), + OnExpiry: "the round stays at " + node, + } +} + +func passAct(round, node, reason, id string) asks.Act { + return asks.Act{"verb": Module + ".pass", "round": round, "node": node, "reason": reason, "ask": id} +} + +// roundView is one round as the bus holds it: the round, each node's place, and each node's run in it. +type roundView struct { + Round Round + Nodes map[string]RoundNode + revs map[string]uint64 + Runs map[string]Run +} + +// latestRound is the newest round's id, or none. +func latestRound(rounds Bucket) (string, error) { + keys, err := rounds.Keys() + if err != nil { + return "", fmt.Errorf("listing rounds: %w", err) + } + latest := "" + for _, k := range keys { + if !strings.Contains(k, ".") && k > latest { + latest = k + } + } + return latest, nil +} + +func readRound(rounds, runs Bucket, id string) (*roundView, error) { + e, err := rounds.Get(id) + if err != nil || e == nil { + return nil, err + } + v := &roundView{Nodes: map[string]RoundNode{}, revs: map[string]uint64{}, Runs: map[string]Run{}} + if err := json.Unmarshal(e.Value, &v.Round); err != nil { + return nil, fmt.Errorf("round %s: %w", id, err) + } + for _, node := range v.Round.Order { + e, err := rounds.Get(RoundNodeKey(id, node)) + if err != nil { + return nil, err + } + if e == nil { + continue + } + var rn RoundNode + if err := json.Unmarshal(e.Value, &rn); err != nil { + return nil, fmt.Errorf("round %s, node %s: %w", id, node, err) + } + v.Nodes[node], v.revs[node] = rn, e.Revision + if rn.Run == "" { + continue + } + re, err := runs.Get(NodeKey(node, rn.Run)) + if err != nil { + return nil, err + } + if re != nil { + var r Run + if err := json.Unmarshal(re.Value, &r); err != nil { + return nil, fmt.Errorf("run %s of %s: %w", rn.Run, node, err) + } + v.Runs[node] = r + } + } + return v, nil +} + +// checkedAt is when a run's after-upgrade check passed, or zero. +func checkedAt(r Run) time.Time { + for _, s := range r.Steps { + if s.Step == StepCheckedAfter && s.Check != nil && s.Check.Passed { + return s.At + } + } + return time.Time{} +} + +// passedAt is when a node let the round go on: its check passed, or the operator approved its pass; zero when +// it has not. +func (v *roundView) passedAt(node string) time.Time { + if rn := v.Nodes[node]; rn.Pass != nil && rn.Pass.Answer == PassApproved && rn.Pass.Answered != nil { + return *rn.Pass.Answered + } + if r, ok := v.Runs[node]; ok { + return checkedAt(r) + } + return time.Time{} +} + +// turnSince is when a node's turn came: the round's opening for the first, the passing of the node before it +// for the rest; zero when it has not come. +func (v *roundView) turnSince(node string) time.Time { + for i, n := range v.Round.Order { + if n != node { + continue + } + if i == 0 { + return v.Round.Opened + } + return v.passedAt(v.Round.Order[i-1]) + } + return time.Time{} +} + +// next is the node after node in the order, or none. +func (v *roundView) next(node string) string { + for i, n := range v.Round.Order { + if n == node && i+1 < len(v.Round.Order) { + return v.Round.Order[i+1] + } + } + return "" +} + +// putRoundNode writes a node's place in a round from the revision read, so a pass and a run written by two +// keepers never overwrite each other: a lost race is read again and changed again. +func putRoundNode(rounds Bucket, round, node string, change func(*RoundNode)) error { + key := RoundNodeKey(round, node) + for attempt := 0; attempt < 5; attempt++ { + e, err := rounds.Get(key) + if err != nil { + return err + } + rn := RoundNode{Round: round, Node: node} + if e != nil { + if err := json.Unmarshal(e.Value, &rn); err != nil { + return fmt.Errorf("%s: %w", key, err) + } + change(&rn) + _, err = rounds.Update(key, rn, e.Revision) + } else { + change(&rn) + _, err = rounds.Create(key, rn) + } + if !errors.Is(err, stdio.ErrChanged) { + return err + } + } + return fmt.Errorf("%s was written by another keeper five times in a row", key) +} + +// RoundStep does what the node's keeper has to do for the rounds now, and answers when it should be called +// again (zero: only when `rounds` or `runs` change). The keeper calls it at its start, at the time it answered, +// and whenever `rounds` or `runs` change. The first node of the order opens the week's round once it is +// Saturday 04:00; a node whose turn came starts its run; a node waiting on a pass reads the operator's answer. +func (u *Upgrades) RoundStep() (time.Time, error) { + now := u.Machine.Now() + order := u.Config.UpgradeOrder + var next time.Time + soonest := func(t time.Time) { + if !t.IsZero() && (next.IsZero() || t.Before(next)) { + next = t + } + } + if len(order) == 0 { + return time.Time{}, nil // no order: no round (the health says so) + } + var errs []error + if order[0] == u.Node { + opening := RoundOpening(now) + if now.Before(opening) { + soonest(opening) + } else { + soonest(opening.AddDate(0, 0, 7)) + id := RoundID(now) + round := Round{Round: id, Opened: now.UTC(), OpenedBy: u.Node, Order: append([]string(nil), order...)} + if _, err := u.Rounds.Create(id, round); err != nil && !errors.Is(err, stdio.ErrChanged) { + errs = append(errs, fmt.Errorf("opening round %s: %w", id, err)) + } + } + } + id, err := latestRound(u.Rounds) + if err != nil || id == "" { + return next, errors.Join(append(errs, err)...) + } + v, err := readRound(u.Rounds, u.Runs, id) + if err != nil || v == nil { + return next, errors.Join(append(errs, err)...) + } + // A pass of the node before this one that waits for the operator's answer. + for i, node := range v.Round.Order { + if i+1 >= len(v.Round.Order) || v.Round.Order[i+1] != u.Node { + continue + } + if rn := v.Nodes[node]; rn.Pass != nil && rn.Pass.Answer == "" { + answered, err := u.readPass(v, node) + if err != nil { + errs = append(errs, err) + } + if !answered { + soonest(now.Add(passCheckEvery)) + } else if v, err = readRound(u.Rounds, u.Runs, id); err != nil { + return next, errors.Join(append(errs, err)...) + } + } + } + t, err := u.takeTurn(v, now) + soonest(t) + return next, errors.Join(append(errs, err)...) +} + +// takeTurn starts this node's run in the round when its turn came and it has none, and again an hour after a +// run of it was refused (it started nothing); it answers when to look again. +func (u *Upgrades) takeTurn(v *roundView, now time.Time) (time.Time, error) { + if !contains(v.Round.Order, u.Node) || v.turnSince(u.Node).IsZero() || !v.passedAt(u.Node).IsZero() { + return time.Time{}, nil + } + if rn := v.Nodes[u.Node]; rn.Pass != nil && rn.Pass.Answer == "" { + return time.Time{}, nil // the operator is asked to pass this node: it waits for the answer + } + own, err := u.ownRuns() + if err != nil { + return time.Time{}, err + } + var latest *Run + for i := range own { + if own[i].Round == v.Round.Round && (latest == nil || own[i].Run.Run > latest.Run) { + latest = &own[i].Run + } + } + if latest != nil { + if v.Nodes[u.Node].Run != latest.Run { + // a keeper that went away between starting its run and naming it in the round + if err := putRoundNode(u.Rounds, v.Round.Round, u.Node, func(rn *RoundNode) { rn.Run = latest.Run }); err != nil { + return time.Time{}, err + } + } + if lastStep(*latest) != StepRefused { + return time.Time{}, nil // running, or ended: a failure stops the round here + } + retry := latest.Steps[len(latest.Steps)-1].At.Add(refusedRetry) + if now.Before(retry) { + return retry, nil + } + } + r, err := u.StartUpgrade(FromRound, v.Round.Round, "") + if len(r.Steps) == 0 { + return time.Time{}, err + } + if perr := putRoundNode(u.Rounds, v.Round.Round, u.Node, func(rn *RoundNode) { rn.Run = r.Run }); perr != nil { + err = errors.Join(err, perr) + } + if lastStep(r) == StepRefused { + return r.Steps[len(r.Steps)-1].At.Add(refusedRetry), err + } + return time.Time{}, err +} + +// readPass reads the operator's answer to a pass of node from the channel's record, and writes it into the +// round once there is one. It answers whether the pass is answered. +func (u *Upgrades) readPass(v *roundView, node string) (bool, error) { + rn := v.Nodes[node] + p := rn.Pass + if u.AskRecord == nil { + return false, errors.New("no reader of the operator channel's records: a pass cannot be answered") + } + e, err := u.AskRecord(p.Ask) + if err != nil { + return false, fmt.Errorf("reading the operator's answer to the pass of %s: %w", node, err) + } + if e == nil { + return false, nil + } + var rec struct { + State string `json:"state"` + Warrant *asks.Warrant `json:"warrant"` + } + if err := json.Unmarshal(e.Value, &rec); err != nil { + return false, fmt.Errorf("the channel's record of the pass of %s: %w", node, err) + } + if rec.State == "" || rec.State == "open" || rec.Warrant == nil { + return false, nil + } + w := *rec.Warrant + answer := PassDeclined + switch { + case w.Outcome == asks.OutcomeExpired: + answer = PassExpired + case w.Outcome == asks.OutcomeChosen: + ask := PassOperatorAsk(v.Round.Round, node, v.next(node), p.Reason, p.Ask, p.Asked) + option, err := w.For(Module, ask) + if err == nil { + err = option.Performs(passAct(v.Round.Round, node, p.Reason, p.Ask)) + } + if err != nil { + return false, fmt.Errorf("the answer to the pass of %s is not the operator's to this pass: %w", node, err) + } + if option.ID == "pass" { + answer = PassApproved + } + } + at := u.Machine.Now().UTC() + return true, putRoundNode(u.Rounds, v.Round.Round, node, func(rn *RoundNode) { + if rn.Pass != nil && rn.Pass.Ask == p.Ask && rn.Pass.Answer == "" { + rn.Pass.Answer, rn.Pass.Answered = answer, &at + } + }) +} + +// takePass is a pass ask taken by the keeper of the node waiting on the passed node: the pass is written into +// the round, waiting for the operator's answer (the tool asked the operator). +func (u *Upgrades) takePass(a Ask) error { + v, err := readRound(u.Rounds, u.Runs, a.Round) + if err != nil { + return err + } + switch { + case v == nil: + return fmt.Errorf("ask %s passes %s in round %s, which is not open", a.ID, a.Pass, a.Round) + case v.next(a.Pass) != u.Node: + return fmt.Errorf("ask %s passes %s, on which %s does not wait", a.ID, a.Pass, u.Node) + } + return putRoundNode(u.Rounds, a.Round, a.Pass, func(rn *RoundNode) { + if rn.Pass == nil || rn.Pass.Ask != a.ID { // taken again after a restart: the answer already read stays + rn.Pass = &Pass{Reason: a.Reason, Ask: a.ID, Asked: a.At.UTC()} + } + }) +} + +// NodeInRound is where one node stands in a round, as pacman_round and the health say it. +type NodeInRound struct { + Node string `json:"node"` + State string `json:"state"` + Run string `json:"run,omitempty"` + Since *time.Time `json:"since,omitempty"` + WaitsFor string `json:"waits_for,omitempty"` + Pass *Pass `json:"pass,omitempty"` +} + +// The states of a node in a round. +const ( + InRoundWaiting = "waiting" // the node before it has not passed + InRoundTurn = "turn" // its turn came, its run has not started + InRoundStalled = "stalled" // its turn came longer than round-start-minutes ago, and it has not started + InRoundRunning = "running" // its run goes on + InRoundPassed = "passed" // its check passed, or the operator passed it + InRoundFailed = "failed" // its run failed or its check did not pass: the round stops here + InRoundRefused = "refused" // its run did not start (it is tried again) + InRoundOff = "off-order" // not in the round's order +) + +// RoundState is where every node of the order stands in a round at now. +func RoundState(v *roundView, now time.Time, startMinutes int) []NodeInRound { + if startMinutes < 1 { + startMinutes = defaultRoundStartMinutes + } + out := []NodeInRound{} + for i, node := range v.Round.Order { + s := NodeInRound{Node: node, Run: v.Nodes[node].Run, Pass: v.Nodes[node].Pass} + at := func(t time.Time) *time.Time { + if t.IsZero() { + return nil + } + t = t.UTC() + return &t + } + r, hasRun := v.Runs[node] + switch { + case !v.passedAt(node).IsZero(): + s.State, s.Since = InRoundPassed, at(v.passedAt(node)) + case hasRun && lastStep(r) == StepRefused: + s.State, s.Since = InRoundRefused, at(r.Steps[len(r.Steps)-1].At) + case hasRun && lastStep(r) == StepCheckedAfter: + s.State, s.Since = InRoundFailed, at(r.Steps[len(r.Steps)-1].At) + case hasRun: + s.State = InRoundRunning + if len(r.Steps) > 0 { + s.Since = at(r.Steps[0].At) + } + case !v.turnSince(node).IsZero(): + since := v.turnSince(node) + s.State, s.Since = InRoundTurn, at(since) + if now.Sub(since) > time.Duration(startMinutes)*time.Minute { + s.State = InRoundStalled + } + default: + s.State = InRoundWaiting + for j := i - 1; j >= 0; j-- { + if v.passedAt(v.Round.Order[j]).IsZero() { + s.WaitsFor = v.Round.Order[j] + } + } + if s.WaitsFor != "" { + // since the node it waits for stopped or its turn came + prev := out[indexOf(v.Round.Order, s.WaitsFor)] + s.Since = prev.Since + } + } + out = append(out, s) + } + return out +} + +func indexOf(list []string, s string) int { + for i, x := range list { + if x == s { + return i + } + } + return -1 +} + +// seatPublish publishes on a seat for the tools bundle; a test replaces it. +var seatPublish = stdio.SeatPublish + +// ListRounds answers pacman_round's read: the newest rounds, or one, each with every node's state. +func ListRounds(rounds, runs Bucket, round string, now time.Time, startMinutes, limit int) (map[string]any, error) { + keys, err := rounds.Keys() + if err != nil { + return nil, err + } + var ids []string + for _, k := range keys { + if !strings.Contains(k, ".") && (round == "" || k == round) { + ids = append(ids, k) + } + } + if round != "" && len(ids) == 0 { + return nil, fmt.Errorf("no round %s", round) + } + sort.Sort(sort.Reverse(sort.StringSlice(ids))) + if len(ids) > limit { + ids = ids[:limit] + } + listed := []map[string]any{} + for _, id := range ids { + v, err := readRound(rounds, runs, id) + if err != nil || v == nil { + return nil, err + } + listed = append(listed, map[string]any{"round": id, "opened": v.Round.Opened, "opened_by": v.Round.OpenedBy, + "order": v.Round.Order, "nodes": RoundState(v, now, startMinutes)}) + } + return map[string]any{"rounds": listed}, nil +} + +// AskPass asks the operator to let a round go on past a node: the ask on the operator's channel, and the ask in +// `asks` for the keeper of the node waiting on it, which reads the answer. A reason is required. +func AskPass(rounds, runs, asksState Bucket, round, node, reason string, now time.Time) (map[string]any, error) { + if strings.TrimSpace(reason) == "" { + return nil, errors.New("a pass needs a reason: the operator is asked to let the round go on past a node, and is told why") + } + if round == "" { + id, err := latestRound(rounds) + if err != nil { + return nil, err + } + round = id + } + v, err := readRound(rounds, runs, round) + if err != nil { + return nil, err + } + if v == nil { + return nil, fmt.Errorf("no round %s is open", round) + } + if !contains(v.Round.Order, node) { + return nil, fmt.Errorf("%s is not in round %s's order (%s)", node, round, strings.Join(v.Round.Order, ", ")) + } + if !v.passedAt(node).IsZero() { + return nil, fmt.Errorf("%s has passed already in round %s", node, round) + } + next := v.next(node) + if next == "" { + return nil, fmt.Errorf("%s is the last of the order: no node waits on it", node) + } + a := Ask{ID: AskID(now), Node: next, Act: ActPass, At: now.UTC(), Round: round, Pass: node, Reason: reason} + opAsk := PassOperatorAsk(round, node, next, reason, a.ID, a.At) + if err := opAsk.Check(now); err != nil { + return nil, err + } + if err := seatPublish(asks.AskSubject(Module), opAsk, opAsk.ID); err != nil { + return nil, fmt.Errorf("the operator could not be asked: %w; nothing was passed", err) + } + if _, err := asksState.Create(NodeKey(next, a.ID), a); err != nil { + return nil, fmt.Errorf("the operator was asked, and the ask for %s's keeper could not be written: %w; an Approve passes nothing", next, err) + } + return map[string]any{"asked": opAsk.ID, "round": round, "pass": node, "waiting": next, + "next": "the round goes on past " + node + " only when the operator approves on their channel; pacman_round says where it stands"}, nil +} + +// roundStartMinutes is round-start-minutes as the tools bundle reads it: from the keeper's config file, which +// the manifest names to the bundle too (EnvConfig); its default when the file cannot be read. +var roundStartMinutes = func() int { + raw, err := os.ReadFile(os.Getenv(EnvConfig)) + var c KeeperConfig + if err != nil || json.Unmarshal(raw, &c) != nil || c.RoundStartMinutes < 1 { + return defaultRoundStartMinutes + } + return c.RoundStartMinutes +} diff --git a/modules/pacman/cmd/pacman-tools/rounds_test.go b/modules/pacman/cmd/pacman-tools/rounds_test.go new file mode 100644 index 00000000..26a54748 --- /dev/null +++ b/modules/pacman/cmd/pacman-tools/rounds_test.go @@ -0,0 +1,305 @@ +package main + +// The week's round (novox/hq ADR 0508 §5, §6; acceptance 1841/AC6 and 1841/AC7), for three made-up nodes over +// states in memory. + +import ( + "encoding/json" + "strings" + "testing" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/asks" +) + +// roundBed is three keepers over one bed's states, each with the order node-a, node-b, node-c. +type roundBed struct { + *keeperBed + nodes map[string]*Upgrades +} + +func newRoundBed(t *testing.T, order ...string) *roundBed { + k := newKeeperBed(t) + k.now = time.Date(2026, 10, 24, 3, 0, 0, 0, time.UTC) // a Saturday + b := &roundBed{keeperBed: k, nodes: map[string]*Upgrades{}} + var updates UpdatesValue + k.updates.value("node-a", &updates) + for _, n := range []string{"node-a", "node-b", "node-c"} { + u := k.keeper(k.u.Machine) + u.Node, u.Config.UpgradeOrder, u.Config.RoundStartMinutes = n, order, 120 + if _, err := k.installed.Put(n, nodeAPackages); err != nil { + t.Fatal(err) + } + if _, err := k.updates.Put(n, updates); err != nil { + t.Fatal(err) + } + b.nodes[n] = u + } + return b +} + +func (b *roundBed) step(t *testing.T, node string) time.Time { + t.Helper() + next, err := b.nodes[node].RoundStep() + if err != nil { + t.Fatalf("%s: %v", node, err) + } + return next +} + +// end ends node's running transaction as unitState says and follows it. +func (b *roundBed) end(t *testing.T, node, unitState string) { + t.Helper() + b.unitState = unitState + if err := b.nodes[node].Follow(); err != nil { + t.Fatal(err) + } + b.now = b.now.Add(5 * time.Minute) +} + +func (b *roundBed) runsOf(node string) []Run { + keys, _ := b.runs.Keys() + var out []Run + for _, k := range keys { + if strings.HasPrefix(k, node+".") { + var r Run + b.runs.value(k, &r) + out = append(out, r) + } + } + return out +} + +func (b *roundBed) state(t *testing.T) map[string]NodeInRound { + t.Helper() + v, err := readRound(b.rounds, b.runs, "2026-W43") + if err != nil || v == nil { + t.Fatalf("round: %v", err) + } + out := map[string]NodeInRound{} + for _, s := range RoundState(v, b.now, 120) { + out[s.Node] = s + } + return out +} + +func TestTheRoundOpensSaturdayAtFourAndThePassedCheckGivesTheNextItsTurn(t *testing.T) { + if o := RoundOpening(time.Date(2026, 10, 22, 15, 0, 0, 0, time.UTC)); !o.Equal(time.Date(2026, 10, 24, 4, 0, 0, 0, time.UTC)) { + t.Fatalf("the opening of the week of Thursday 22 October: %s", o) + } + b := newRoundBed(t, "node-a", "node-b", "node-c") + if next := b.step(t, "node-a"); !next.Equal(time.Date(2026, 10, 24, 4, 0, 0, 0, time.UTC)) { + t.Errorf("before 04:00 the first node waits for 04:00: %s", next) + } + if keys, _ := b.rounds.Keys(); len(keys) != 0 { + t.Fatalf("a round before 04:00: %v", keys) + } + b.step(t, "node-b") + if keys, _ := b.rounds.Keys(); len(keys) != 0 { + t.Fatalf("only the first of the order opens the round: %v", keys) + } + + b.now = time.Date(2026, 10, 24, 4, 0, 0, 0, time.UTC) + if next := b.step(t, "node-a"); !next.Equal(time.Date(2026, 10, 31, 4, 0, 0, 0, time.UTC)) { + t.Errorf("after opening, next Saturday: %s", next) + } + var round Round + if !b.rounds.value("2026-W43", &round) || round.OpenedBy != "node-a" || strings.Join(round.Order, " ") != "node-a node-b node-c" { + t.Fatalf("the round: %+v", round) + } + if r := b.runsOf("node-a"); len(r) != 1 || r[0].Round != "2026-W43" || r[0].From != FromRound || lastStep(r[0]) != StepStarted { + t.Fatalf("the first node starts its run in the round: %+v", r) + } + b.now = b.now.Add(time.Minute) + b.step(t, "node-a") + if w := b.rounds.writesTo("2026-W43"); len(w) != 2 || len(b.runsOf("node-a")) != 1 { + t.Errorf("the round is created once and the run started once: %v", w) + } + b.step(t, "node-b") + if len(b.runsOf("node-b")) != 0 || b.state(t)["node-b"].State != InRoundWaiting { + t.Fatalf("node-b waits while node-a runs: %+v", b.state(t)["node-b"]) + } + + b.end(t, "node-a", "LoadState=not-found\n") + if s := b.state(t)["node-a"]; s.State != InRoundPassed { + t.Fatalf("node-a: %+v", s) + } + b.step(t, "node-c") + if len(b.runsOf("node-c")) != 0 { + t.Fatal("node-c started before node-b") + } + b.step(t, "node-b") + if r := b.runsOf("node-b"); len(r) != 1 || r[0].Round != "2026-W43" { + t.Fatalf("node-b's turn after node-a's check passed: %+v", r) + } + + // node-b's transaction fails: the round stops at node-b, and node-c waits. + b.end(t, "node-b", "LoadState=loaded\nActiveState=failed\nResult=exit-code\nExecMainStatus=1\n") + b.step(t, "node-c") + st := b.state(t) + if len(b.runsOf("node-c")) != 0 || st["node-b"].State != InRoundFailed || st["node-c"].State != InRoundWaiting || st["node-c"].WaitsFor != "node-b" { + t.Errorf("a failed check stops the round: %+v %+v", st["node-b"], st["node-c"]) + } +} + +func TestNoOrderNoRound(t *testing.T) { + b := newRoundBed(t) + b.now = time.Date(2026, 10, 24, 5, 0, 0, 0, time.UTC) + if next := b.step(t, "node-a"); !next.IsZero() { + t.Errorf("no wake without an order: %s", next) + } + if keys, _ := b.rounds.Keys(); len(keys) != 0 || len(b.runsOf("node-a")) != 0 { + t.Errorf("a round without an order: %v", keys) + } +} + +func TestANodeOffAtItsTurnRunsAtItsNextStartAndAStalledTurnIsSaid(t *testing.T) { + b := newRoundBed(t, "node-a", "node-b", "node-c") + b.now = time.Date(2026, 10, 24, 4, 0, 0, 0, time.UTC) + b.step(t, "node-a") + b.end(t, "node-a", "LoadState=not-found\n") + passed := b.now.Add(-5 * time.Minute) + b.now = passed.Add(121 * time.Minute) // node-b is off + st := b.state(t) + if st["node-b"].State != InRoundStalled || st["node-b"].Since == nil || !st["node-b"].Since.Equal(passed) { + t.Errorf("node-b's turn came at %s and it has not started: %+v", passed, st["node-b"]) + } + if st["node-c"].WaitsFor != "node-b" || st["node-c"].State != InRoundWaiting { + t.Errorf("node-c names the node it waits for: %+v", st["node-c"]) + } + b.step(t, "node-b") // node-b starts again + if len(b.runsOf("node-b")) != 1 { + t.Error("a node off at its turn runs at its next start") + } +} + +// warrant is the operator channel's record of an ask answered with option. +func warrant(a asks.Ask, option string, level asks.Level, at time.Time) json.RawMessage { + w := asks.Warrant{Ask: a.ID, Asker: Module, Outcome: asks.OutcomeChosen, Option: option, Level: level, + By: &asks.Person{Who: asks.Operator, Kind: "made-up-channel", Identity: "1", Verified: "user id"}, At: at, AskDigest: a.Digest()} + raw, _ := json.Marshal(map[string]any{"state": "chosen", "warrant": w}) + return raw +} + +func TestOnlyTheOperatorsApprovePassesANodeAndAPassNeedsAReason(t *testing.T) { + b := newRoundBed(t, "node-a", "node-b", "node-c") + b.now = time.Date(2026, 10, 24, 4, 0, 0, 0, time.UTC) + b.step(t, "node-a") + b.end(t, "node-a", "LoadState=not-found\n") + b.step(t, "node-b") + b.end(t, "node-b", "LoadState=loaded\nActiveState=failed\nResult=exit-code\nExecMainStatus=1\n") + + var published []asks.Ask + var subjects []string + saved := seatPublish + seatPublish = func(subject string, body any, id string) error { + subjects, published = append(subjects, subject), append(published, body.(asks.Ask)) + return nil + } + defer func() { seatPublish = saved }() + if _, err := AskPass(b.rounds, b.runs, b.asks, "", "node-b", " ", b.now); err == nil || len(published) != 0 { + t.Fatalf("a pass without a reason: %v", err) + } + out, err := AskPass(b.rounds, b.runs, b.asks, "", "node-b", "its disk is being replaced", b.now) + if err != nil { + t.Fatal(err) + } + if len(published) != 1 || subjects[0] != "mesh.seat.operator-channel.accept.ask.pacman" || out["waiting"] != "node-c" { + t.Fatalf("the operator is asked on their channel: %v %v", subjects, out) + } + opAsk := published[0] + if err := opAsk.Check(b.now); err != nil || !strings.Contains(opAsk.Explanation, "its disk is being replaced") { + t.Errorf("the ask: %v %+v", err, opAsk) + } + keys, _ := b.asks.Keys() + if len(keys) != 1 || keys[0] != NodeKey("node-c", opAsk.ID) { + t.Fatalf("the ask for node-c's keeper: %v", keys) + } + var a Ask + b.asks.value(keys[0], &a) + raw, _ := json.Marshal(a) + if err := b.nodes["node-c"].TakeAsk(stdio.StateChange{State: StateAsks, Key: keys[0], Op: "put", Value: raw}); err != nil { + t.Fatal(err) + } + if st := b.state(t)["node-b"]; st.Pass == nil || st.Pass.Reason != "its disk is being replaced" || st.Pass.Answer != "" { + t.Fatalf("the pass waits in the round: %+v", st.Pass) + } + + record := map[string]json.RawMessage{} + b.nodes["node-c"].AskRecord = func(id string) (*stdio.StateEntry, error) { + if v, ok := record[id]; ok { + return &stdio.StateEntry{Key: Module + "." + id, Value: v}, nil + } + return nil, nil + } + if next := b.step(t, "node-c"); len(b.runsOf("node-c")) != 0 || !next.Equal(b.now.Add(passCheckEvery)) { + t.Fatalf("no answer yet: node-c waits and looks again in a minute (%s)", next) + } + // An answer to another ask, or to this one as it was not shown, passes nothing. + other := opAsk + other.Explanation = "something else" + record[opAsk.ID] = warrant(other, "pass", asks.Approve, b.now) + if _, err := b.nodes["node-c"].RoundStep(); err == nil || len(b.runsOf("node-c")) != 0 { + t.Fatalf("a warrant for another ask: %v", err) + } + record[opAsk.ID] = warrant(opAsk, "pass", asks.Approve, b.now) + b.step(t, "node-c") + if st := b.state(t)["node-b"]; st.State != InRoundPassed || st.Pass.Answer != PassApproved { + t.Errorf("the operator's Approve passes node-b: %+v", st) + } + if r := b.runsOf("node-c"); len(r) != 1 || r[0].Round != "2026-W43" { + t.Errorf("node-c's turn after the pass: %+v", r) + } +} + +func TestAPassTheOperatorKeepsWaitingPassesNothing(t *testing.T) { + b := newRoundBed(t, "node-a", "node-b") + b.now = time.Date(2026, 10, 24, 4, 0, 0, 0, time.UTC) + b.step(t, "node-a") // node-a runs and never ends + saved := seatPublish + var opAsk asks.Ask + seatPublish = func(_ string, body any, _ string) error { opAsk = body.(asks.Ask); return nil } + defer func() { seatPublish = saved }() + if _, err := AskPass(b.rounds, b.runs, b.asks, "2026-W43", "node-a", "it hangs", b.now); err != nil { + t.Fatal(err) + } + keys, _ := b.asks.Keys() + var a Ask + b.asks.value(keys[0], &a) + raw, _ := json.Marshal(a) + if err := b.nodes["node-b"].TakeAsk(stdio.StateChange{State: StateAsks, Key: keys[0], Op: "put", Value: raw}); err != nil { + t.Fatal(err) + } + b.nodes["node-b"].AskRecord = func(string) (*stdio.StateEntry, error) { + return &stdio.StateEntry{Value: warrant(opAsk, "keep", asks.Acknowledge, b.now)}, nil + } + b.step(t, "node-b") + if st := b.state(t); st["node-a"].Pass.Answer != PassDeclined || len(b.runsOf("node-b")) != 0 { + t.Errorf("kept waiting: %+v", st["node-a"]) + } + if _, err := AskPass(b.rounds, b.runs, b.asks, "2026-W43", "node-b", "x", b.now); err == nil { + t.Error("the last of the order has no node waiting on it") + } + if _, err := AskPass(b.rounds, b.runs, b.asks, "2026-W43", "node-z", "x", b.now); err == nil { + t.Error("a node not in the order") + } +} + +func TestPacmanRoundReadsTheRounds(t *testing.T) { + b := newRoundBed(t, "node-a", "node-b") + b.now = time.Date(2026, 10, 24, 4, 0, 0, 0, time.UTC) + b.step(t, "node-a") + out, err := ListRounds(b.rounds, b.runs, "", b.now, 120, 4) + if err != nil { + t.Fatal(err) + } + listed := out["rounds"].([]map[string]any) + nodes := listed[0]["nodes"].([]NodeInRound) + if len(listed) != 1 || listed[0]["round"] != "2026-W43" || nodes[0].State != InRoundRunning || nodes[1].WaitsFor != "node-a" { + t.Errorf("%+v", listed) + } + if _, err := ListRounds(b.rounds, b.runs, "2026-W40", b.now, 120, 4); err == nil { + t.Error("a round that is not there") + } +} diff --git a/modules/pacman/cmd/pacman-tools/transaction.go b/modules/pacman/cmd/pacman-tools/transaction.go index ae94fee8..2ccaf0b7 100644 --- a/modules/pacman/cmd/pacman-tools/transaction.go +++ b/modules/pacman/cmd/pacman-tools/transaction.go @@ -380,6 +380,11 @@ func (u *Upgrades) TakeAsk(c stdio.StateChange) error { err = errors.Join(err, fmt.Errorf("deleting the ask %s: %w", c.Key, derr)) } return err + case ActPass: + if err := u.takePass(a); err != nil { + return err + } + return u.Asks.Delete(c.Key) 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 index ad2cc04a..89413b99 100644 --- a/modules/pacman/cmd/pacman-tools/transaction_test.go +++ b/modules/pacman/cmd/pacman-tools/transaction_test.go @@ -41,6 +41,7 @@ type keeperBed struct { root *fakeRoot events *eventLog installed, updates, runs, records, asks *memBucket + rounds *memBucket now time.Time unitState string // systemctl show's answer for the run's unit locked bool @@ -56,7 +57,7 @@ const upgradeLog = "[2026-10-04T12:00:01+0000] [PACMAN] Running 'pacman -Syu --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"} + records: newBucket(), asks: newBucket(), rounds: newBucket(), now: time.Date(2026, 10, 4, 12, 0, 0, 0, time.UTC), unitState: "LoadState=not-found\n"} saved := fetch fetch = func(string) ([]byte, error) { return nil, errors.New("offline") } t.Cleanup(func() { fetch = saved }) @@ -112,7 +113,7 @@ func newKeeperBed(t *testing.T) *keeperBed { // 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} + Runs: k.runs, Records: k.records, Rounds: k.rounds, Asks: k.asks} } // ask writes an ask for node and hands its change to the keeper, as the watch of `asks` would. diff --git a/modules/pacman/cmd/pacman-tools/upgrades.go b/modules/pacman/cmd/pacman-tools/upgrades.go index 567dacc5..e5542edb 100644 --- a/modules/pacman/cmd/pacman-tools/upgrades.go +++ b/modules/pacman/cmd/pacman-tools/upgrades.go @@ -58,7 +58,10 @@ type Upgrades struct { Root RootActs Events EventSink // The module's states. - Installed, Updates, Runs, Records, Asks Bucket + Installed, Updates, Runs, Records, Rounds, Asks Bucket + // AskRecord reads the operator channel's record of one of this module's asks (process.Conn.Ask on + // MessengerAsksBucket): how a pass is answered. + AskRecord func(id string) (*stdio.StateEntry, error) // Check is the after-upgrade check (ADR 0508 §4); nil is AfterCheck. Check func(run Run, record *Record) Check } diff --git a/modules/pacman/module.json b/modules/pacman/module.json index 2c9a6250..b8910957 100644 --- a/modules/pacman/module.json +++ b/modules/pacman/module.json @@ -14,6 +14,9 @@ "reads": [ "mesh-controller.declared-packages" ], + "uses": [ + "operator-channel" + ], "config": { "upgrade-order": { "type": "list", @@ -73,6 +76,7 @@ "pacman_upgrade", "pacman_upgrade_health", "pacman_upgrades", + "pacman_round", "pacman_orphans", "pacman_remove_orphans", "pacman_cache", @@ -195,7 +199,8 @@ "upgrade.checked", "rollback.started", "rollback.finished" - ] + ], + "asks": true }, "health": { "kind": "tool", @@ -218,7 +223,10 @@ "binary": "pacman-tools", "loads": [ "pacman-tools" - ] + ], + "env": { + "PACMAN_KEEPER_CONFIG": "${dir:state}/keeper.json" + } }, { "name": "files",