mesh-catalog/modules/pacman/cmd/pacman-tools/rounds.go

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...)}
			// The mirrors are ranked inside the write that opens the round, before its key, which names the list's
			// revision (ADR 0585 §2): every node of the round upgrades from that list.
			if e, err := u.Rounds.Get(id); err != nil {
				errs = append(errs, fmt.Errorf("reading round %s: %w", id, err))
			} else if e == nil {
				round.Mirrors, round.MirrorsNote = u.mirrorsForRound(id)
				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
}