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

package main

// Upgrades as runs on the bus (novox/hq ADR 0508 §1): every full upgrade of a node, and every rollback, is a run
// in the state `runs` at `<node>.<run>`, written at each of its steps and at no other time, and each step is also
// an event. The keeper's loop calls what is here; everything it reaches is behind an interface (a state, the
// events, the root acts) or the Machine's injected runner and files, so it is tested without a bus or a machine.

import (
	"encoding/json"
	"errors"
	"fmt"
	"regexp"
	"sort"
	"strings"
	"time"

	stdio "git.novox.be/novox/mesh-sdk/go"
)

// Bucket is one of the module's states as a writer sees it: a process's state, or a tools bundle's
// (stdio.KeptState). A write that lost a race is stdio.ErrChanged.
type Bucket interface {
	Get(key string) (*stdio.StateEntry, error)
	Create(key string, value any) (uint64, error)
	Update(key string, value any, revision uint64) (uint64, error)
	Put(key string, value any) (uint64, error)
	Delete(key string) error
	Keys() ([]string, error)
}

// EventSink emits the keeper's events: the process's connection (process.Conn).
type EventSink interface {
	Emit(eventType string, body any, opts ...stdio.EmitOptions) error
}

// RootActs are the acts that need root, carried out by the module's root helper (ADR 0510): the keeper never
// runs them itself.
type RootActs interface {
	// Transaction starts the run's transaction as a unit of the service manager, named TransactionUnit(kind,
	// run), leaving out the packages ignore names, and answers once it started.
	Transaction(kind, run string, ignore []string) error
	// Rollback starts a rollback's transaction as a unit of the service manager, named TransactionUnit(RunRollback,
	// run), carrying out the plan in the request (ADR 0510: the plan travels with the verb), and answers once it
	// started.
	Rollback(run string, plan []RollbackItem) error
	// Journal is the last n lines a transaction unit logged.
	Journal(unit string, n int) ([]string, error)
	// Maps is every process still mapping a replaced file, by its unit or container and its executable.
	Maps() ([]Mapped, error)
}

// TransactionUnit is the unit a run's transaction runs as: named by the run, so a keeper started again finds it.
func TransactionUnit(kind, run string) string {
	return unitPrefix + kind + "-" + strings.ToLower(run)
}

// Upgrades is one node's upgrades: what the keeper's loop calls for its own node.
type Upgrades struct {
	Node    string
	Config  KeeperConfig
	Machine *Machine
	Root    RootActs
	Events  EventSink
	// The module's states.
	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
}

// openRun is a run being written, with the revision of its key.
type openRun struct {
	Run
	rev uint64
}

// step writes the run with one more step: created at its first step, then only ever updated from the revision
// this writer wrote, so a second writer is refused rather than overwritten.
func (u *Upgrades) step(r *openRun, s Step) error {
	r.Steps = append(r.Steps, s)
	key := NodeKey(r.Node, r.Run.Run)
	var err error
	if r.rev == 0 {
		r.rev, err = u.Runs.Create(key, r.Run)
	} else {
		r.rev, err = u.Runs.Update(key, r.Run, r.rev)
	}
	if err != nil {
		r.Steps = r.Steps[:len(r.Steps)-1]
		return fmt.Errorf("writing step %s of run %s: %w", s.Step, key, err)
	}
	return nil
}

// emit says a step as its event; an event that is not sent is said in the error, the step stays written.
func (u *Upgrades) emit(event string, body any) error {
	if u.Events == nil {
		return nil
	}
	if err := u.Events.Emit(event, body); err != nil {
		return fmt.Errorf("emitting %s: %w", event, err)
	}
	return nil
}

// The package manager's log, read as blocks: each begins at a `Running '<command>'` line (or a full upgrade
// starting with none before it) and holds the transaction it ran and the packages it changed.
var (
	runningLine     = regexp.MustCompile(`^\[([^\]]+)\] \[PACMAN\] Running '(.*)'$`)
	transactionLine = regexp.MustCompile(`^\[([^\]]+)\] \[ALPM\] transaction (started|completed|failed|interrupted)$`)
	pacnewLine      = regexp.MustCompile(`^\[[^\]]+\] \[ALPM\] warning: \S+ installed as (\S+\.pacnew)$`)
	pacsaveLine     = regexp.MustCompile(`^\[[^\]]+\] \[ALPM\] warning: \S+ saved as (\S+\.pacsave)$`)
)

// logBlock is one command pacman ran, as its log tells it.
type logBlock struct {
	Start   time.Time
	Command string
	Full    bool      // it started a full system upgrade
	FullAt  time.Time // when
	Began   bool      // a transaction started
	End     string    // completed, failed or interrupted; empty when the log holds no end
	EndAt   time.Time
	Changes Changes
	// The .pacnew and .pacsave files the transaction left, by path, in the order logged.
	Pacnew, Pacsave []string
}

// logBlocks reads pacman's log into the commands it ran, oldest first.
func logBlocks(text string) []logBlock {
	var blocks []logBlock
	var cur *logBlock
	open := func(at time.Time, command string) {
		blocks = append(blocks, logBlock{Start: at, Command: command})
		cur = &blocks[len(blocks)-1]
	}
	for _, l := range strings.Split(text, "\n") {
		l = strings.TrimRight(l, "\r")
		if m := runningLine.FindStringSubmatch(l); m != nil {
			if t, ok := logTime(m[1]); ok {
				open(t, m[2])
			}
			continue
		}
		if m := fullUpdate.FindStringSubmatch(l); m != nil {
			t, ok := logTime(m[1])
			if !ok {
				continue
			}
			if cur == nil || cur.Full || cur.Began {
				open(t, "")
			}
			cur.Full, cur.FullAt = true, t
			continue
		}
		if cur == nil {
			continue
		}
		if m := transactionLine.FindStringSubmatch(l); m != nil {
			t, _ := logTime(m[1])
			if m[2] == "started" {
				cur.Began = true
			} else if cur.Began && cur.End == "" {
				cur.End, cur.EndAt = m[2], t
			}
			continue
		}
		if cur.Began && cur.End == "" {
			if m := logLine.FindStringSubmatch(l); m != nil {
				addChange(&cur.Changes, m[2], m[3], m[4])
			} else if m := pacnewLine.FindStringSubmatch(l); m != nil {
				cur.Pacnew = append(cur.Pacnew, m[1])
			} else if m := pacsaveLine.FindStringSubmatch(l); m != nil {
				cur.Pacsave = append(cur.Pacsave, m[1])
			}
		}
	}
	for i := range blocks {
		sortChanges(&blocks[i].Changes)
	}
	return blocks
}

func addChange(c *Changes, action, name, version string) {
	switch action {
	case "installed":
		c.Added = append(c.Added, PackageVersion{Name: name, Version: version})
	case "removed":
		c.Removed = append(c.Removed, PackageVersion{Name: name, Version: version})
	case "upgraded", "downgraded":
		from, to, _ := strings.Cut(version, " -> ")
		c.Changed = append(c.Changed, VersionChange{Name: name, From: from, To: to})
	}
}

func sortChanges(c *Changes) {
	if c.Added == nil {
		c.Added = []PackageVersion{}
	}
	if c.Removed == nil {
		c.Removed = []PackageVersion{}
	}
	if c.Changed == nil {
		c.Changed = []VersionChange{}
	}
	sort.Slice(c.Added, func(i, j int) bool { return c.Added[i].Name < c.Added[j].Name })
	sort.Slice(c.Removed, func(i, j int) bool { return c.Removed[i].Name < c.Removed[j].Name })
	sort.Slice(c.Changed, func(i, j int) bool { return c.Changed[i].Name < c.Changed[j].Name })
}

// noTransaction is why a full upgrade the log tells of has no transaction: the log does not say which.
const noTransaction = "the log holds no transaction for this upgrade: there was nothing to do, or it ended before changing anything"

// PastUpgrades is every full upgrade pacman's log tells of, as runs read from the log (ADR 0508 §1): started
// when the upgrade started, finished or failed when its transaction ended, with what it changed. One whose
// transaction has no end in the log failed, so said.
func PastUpgrades(node, text string) []Run {
	var runs []Run
	for _, b := range logBlocks(text) {
		if !b.Full {
			continue
		}
		r := Run{Node: node, Run: RunID(b.FullAt), Kind: RunUpgrade, From: FromLog,
			Steps: []Step{{Step: StepStarted, At: b.FullAt}}}
		changes := b.Changes
		switch {
		case !b.Began:
			r.Steps = append(r.Steps, Step{Step: StepFinished, At: b.FullAt, Why: noTransaction, Changes: &changes})
		case b.End == "completed":
			r.Steps = append(r.Steps, Step{Step: StepFinished, At: b.EndAt, Changes: &changes})
		case b.End == "":
			r.Steps = append(r.Steps, Step{Step: StepFailed, At: b.FullAt, Error: "the log holds no end of this transaction", Changes: &changes})
		default:
			r.Steps = append(r.Steps, Step{Step: StepFailed, At: b.EndAt, Error: "transaction " + b.End, Changes: &changes})
		}
		runs = append(runs, r)
	}
	return runs
}

// ImportLog reads the node's past full upgrades from pacman's log into `runs` once: when the node has no run
// yet. It answers how many it wrote. A run read from the log is history and emits nothing.
func (u *Upgrades) ImportLog() (int, error) {
	keys, err := u.Runs.Keys()
	if err != nil {
		return 0, fmt.Errorf("listing runs: %w", err)
	}
	for _, k := range keys {
		if strings.HasPrefix(k, u.Node+".") {
			return 0, nil
		}
	}
	text, err := u.Machine.ReadFile(PacmanLog)
	if err != nil {
		return 0, fmt.Errorf("reading %s: %w", PacmanLog, err)
	}
	n := 0
	for _, r := range PastUpgrades(u.Node, string(text)) {
		if _, err := u.Runs.Create(NodeKey(u.Node, r.Run), r); err != nil {
			if errors.Is(err, stdio.ErrChanged) {
				continue // two upgrades started in the same second: the first is kept
			}
			return n, fmt.Errorf("writing run %s: %w", r.Run, err)
		}
		n++
	}
	return n, nil
}

// runTime is when a run began, from its id.
func runTime(id string) (time.Time, bool) {
	t, err := time.Parse("20060102T150405Z", id)
	return t, err == nil
}

// RunSummary is one run as pacman_upgrades lists it.
type RunSummary struct {
	Node    string     `json:"node"`
	Run     string     `json:"run"`
	Kind    string     `json:"kind"`
	From    string     `json:"from"`
	Round   string     `json:"round,omitempty"`
	Began   time.Time  `json:"began"`
	Outcome string     `json:"outcome"`
	Ended   *time.Time `json:"ended,omitempty"`
	Error   string     `json:"error,omitempty"`
	Held    int        `json:"held_kernels,omitempty"`
	Checked string     `json:"check,omitempty"`
}

// summary is where a run stands: its outcome is its last step that ends it (refused, finished, failed), else
// running.
func summary(r Run) RunSummary {
	s := RunSummary{Node: r.Node, Run: r.Run, Kind: r.Kind, From: r.From, Round: r.Round, Outcome: "running"}
	if t, ok := runTime(r.Run); ok {
		s.Began = t
	}
	for _, st := range r.Steps {
		switch st.Step {
		case StepRefused, StepFinished, StepFailed:
			at := st.At
			s.Outcome, s.Ended = st.Step, &at
			s.Error = st.Error
			if st.Step == StepRefused {
				s.Error = st.Why
			}
		case StepCheckedBefore:
			s.Held = len(st.Held)
		case StepCheckedAfter:
			s.Checked = "failed"
			if st.Check != nil && st.Check.Passed {
				s.Checked = "passed"
			}
		}
	}
	return s
}

// ListUpgrades answers pacman_upgrades from `runs`: one run whole, or the runs of every node or one, between
// two days, newest first, at most limit of them.
func ListUpgrades(runs Bucket, node, from, until, run string, limit int) (map[string]any, error) {
	if run != "" {
		if node == "" {
			return nil, fmt.Errorf("a run is named with its node")
		}
		e, err := runs.Get(NodeKey(node, run))
		if err != nil {
			return nil, err
		}
		if e == nil {
			return nil, fmt.Errorf("no run %s of %s", run, node)
		}
		var r Run
		if err := jsonValue(e.Value, &r); err != nil {
			return nil, err
		}
		return map[string]any{"run": r, "summary": summary(r)}, nil
	}
	day := func(s string, end bool) (time.Time, error) {
		if s == "" {
			return time.Time{}, nil
		}
		t, err := time.ParseInLocation("2006-01-02", s, time.Local)
		if err != nil {
			return t, fmt.Errorf("%q is not a day as YYYY-MM-DD", s)
		}
		if end {
			t = t.AddDate(0, 0, 1)
		}
		return t, nil
	}
	since, err := day(from, false)
	if err != nil {
		return nil, err
	}
	before, err := day(until, true)
	if err != nil {
		return nil, err
	}
	keys, err := runs.Keys()
	if err != nil {
		return nil, err
	}
	var listed []RunSummary
	for _, k := range keys {
		n, id, ok := strings.Cut(k, ".")
		if !ok || node != "" && n != node {
			continue
		}
		t, ok := runTime(id)
		if !ok || t.Before(since) || !before.IsZero() && !t.Before(before) {
			continue
		}
		e, err := runs.Get(k)
		if err != nil {
			return nil, err
		}
		if e == nil {
			continue
		}
		var r Run
		if err := jsonValue(e.Value, &r); err != nil {
			return nil, fmt.Errorf("run %s: %w", k, err)
		}
		listed = append(listed, summary(r))
	}
	sort.Slice(listed, func(i, j int) bool {
		if !listed[i].Began.Equal(listed[j].Began) {
			return listed[i].Began.After(listed[j].Began)
		}
		return listed[i].Node < listed[j].Node
	})
	truncated := len(listed) > limit
	if truncated {
		listed = listed[:limit]
	}
	if listed == nil {
		listed = []RunSummary{}
	}
	return map[string]any{"count": len(listed), "runs": listed, "truncated": truncated}, nil
}

// jsonValue reads a state's value into its shape.
func jsonValue(raw []byte, v any) error { return json.Unmarshal(raw, v) }

// upgradeState is a state of this module as the tools bundle reads it; a test replaces it.
var upgradeState = func(name string) Bucket { return stdio.State(name) }