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

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
		}
	}
	check := u.Check
	if check == nil {
		check = u.AfterCheck
	}
	c := check(r.Run, rec)
	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
}

// transactionBlock is the run's transaction as pacman's log tells it: the first command of its kind pacman
// logged since the transaction was started; nil when the log holds none.
func (u *Upgrades) transactionBlock(kind string, since time.Time) *logBlock {
	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
		}
		return &b
	}
	return nil
}

// transactionChanges is what the run's transaction changed, read from its lines in pacman's log.
func (u *Upgrades) transactionChanges(kind string, since time.Time) *Changes {
	if b := u.transactionBlock(kind, since); b != nil {
		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
	case ActRollback:
		return u.takeRollback(c.Key, a)
	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)
	}
}