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

package main

// The keeper (novox/hq ADR 0507 §1-§4; ADR 0508): the module's process on every node, under an account of its
// own with a bus account of its own (ADR 0436 §1). It keeps the node's packages in the state `installed` and what
// an upgrade would change in `updates`, and says each change of the packages as `package.changed`. It reads the
// machine at its start and whenever the machine may have changed, never on a timer but the one daily read of
// `updates` ADR 0507 §4 allows:
//
//   - pacman's database lock released (an inotify watch on its directory: the lock file deleted) ends a
//     transaction, whoever ran it: `installed` and then `updates` are read again;
//   - the mirror list rewritten (reflector writes /etc/pacman.d/mirrorlist): `updates` is read again;
//   - a day after `updates` was last read: it is read again, since the repositories change what an upgrade
//     would do without any transaction on the node.
//
// A key is written only when what it holds changed (ADR 0289 §1). A value over the bus's cap is not written:
// the key keeps its last value and `<node>.not-current` says the size and since when, written once and deleted
// once the node's packages are current again (ADR 0507 §2).
//
// Everything the keeper touches is injected (its states, its events, the machine, the clock, the watch), so the
// loop is tested without a machine or a bus. One goroutine runs it all: the watch's signals, the clock, and every
// change a hook watches are handled one at a time, in the order they came.
//
// # Hooks (the upgrade parts of ADR 0508 join here)
//
// Hooks are where further work joins the loop, each run on the loop's goroutine:
//
//   - Start: once, after the first reads of `installed` and `updates`, before anything else;
//   - AfterTransaction: after every released lock, once `installed` and `updates` were read again, with what the
//     transaction changed (nil when nothing);
//   - Changes: each a state and a key pattern (`*` one name, `**` the rest) watched for the life of the keeper;
//     its Handle sees the current values first (Current set), then every put and delete in order;
//   - Wakes: each a function the keeper calls at its start and again at the time it last answered; it answers when
//     it next wants to be called (a zero time: not again). It is how a hook keeps a time (a round's opening, a
//     stall) without polling.
//
// A hook's error is logged and the keeper goes on: a fault in one part never stops the view. keeperHooks builds
// the hooks the keeper runs with, once connected; it is the one place a part is added.

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"os/signal"
	"path/filepath"
	"syscall"
	"time"

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

// ValueCap is the bus's cap on one value of a state (ADR 0201 §14): the mesh's bound, not the module's.
const ValueCap = 256 << 10

// Where pacman keeps its database and its lock, and where the mirror list is.
const (
	PacmanDBDir = "/var/lib/pacman"
	LockName    = "db.lck"
	MirrorDir   = "/etc/pacman.d"
	MirrorName  = "mirrorlist"
)

// UpdatesEvery is how long after the last read of `updates` it is read again with nothing else asking (ADR 0507
// §4: once a day).
const UpdatesEvery = 24 * time.Hour

// Store is one state as the keeper uses it: a process state on the bus (process.State), or a fake in a test.
type Store 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)
	Watch(key string, handle func(stdio.StateChange) error) error
}

// Clock is the time as the keeper sees it.
type Clock interface {
	Now() time.Time
	// After fires once, d from now.
	After(d time.Duration) <-chan time.Time
}

type realClock struct{}

func (realClock) Now() time.Time                         { return time.Now() }
func (realClock) After(d time.Duration) <-chan time.Time { return time.After(d) }

// Signal is what the machine's watch saw.
type Signal int

const (
	// LockReleased: pacman's database lock was deleted, a transaction (or a database sync) ended.
	LockReleased Signal = iota + 1
	// MirrorsRewritten: the mirror list was written or replaced.
	MirrorsRewritten
)

func (s Signal) String() string {
	switch s {
	case LockReleased:
		return "the lock released"
	case MirrorsRewritten:
		return "the mirror list rewritten"
	}
	return fmt.Sprintf("signal %d", int(s))
}

// Watcher is the machine's watch: its signals, until it is closed or fails (the channel closes).
type Watcher interface {
	Signals() <-chan Signal
	Close() error
}

// Transaction is a released lock as AfterTransaction sees it: when, and what it changed in `installed`
// (nil when it changed nothing, or the value could not be written).
type Transaction struct {
	At      time.Time
	Changed *PackageChanged
}

// StateHook is a watch on one of the module's states for the life of the keeper.
type StateHook struct {
	State  string
	Key    string
	Handle func(k *Keeper, c stdio.StateChange) error
}

// Wake is called at the keeper's start and then at the time it last answered; it answers when next.
type Wake func(k *Keeper, now time.Time) time.Time

// Hooks are the parts that join the keeper's loop (see the file's comment).
type Hooks struct {
	Start            []func(k *Keeper) error
	AfterTransaction []func(k *Keeper, t Transaction) error
	Changes          []StateHook
	Wakes            []Wake
}

// keeperHooks is the hooks the keeper runs with, built once the keeper is connected: the one place a part of
// the keeper is added. Today: the upgrades of ADR 0508 (upgrades.go, record.go, transaction.go, rounds.go), for
// this node.
//
//   - Start: the package manager's log read into `runs` once (ImportLog), records past record-days pruned
//     (PruneRecords), and every unfinished run carried on (Follow), a run a keeper that went away left included;
//   - AfterTransaction: every unfinished run carried on (Follow): a released lock may be a run's transaction ending;
//   - Changes: `asks` at `<node>.*`, each change handed to TakeAsk; `rounds` and `runs`, every change, to the
//     round's step (RoundStep): a node starts the moment the node before it passed;
//   - Wakes: PruneRecords once a day; the round's step at the time it answered (the Saturday's opening, a pass
//     whose answer is read again, a refused run tried again).
func keeperHooks(k *Keeper) Hooks {
	root := k.Root
	if root == nil {
		root = newRootActs()
	}
	u := &Upgrades{Node: k.Node, Config: k.Config, Machine: k.Machine, Root: root, Events: k.Events,
		Installed: k.State(StateInstalled), Updates: k.State(StateUpdates), Runs: k.State(StateRuns),
		Records: k.State(StateRecords), Rounds: k.State(StateRounds), Asks: k.State(StateAsks), AskRecord: k.AskRecord,
		Check: k.Check}
	k.upgrades = u
	const roundWake = 1 // the round's step is the second wake
	roundStep := func(k *Keeper, now time.Time) time.Time {
		next, err := u.RoundStep()
		if err != nil {
			k.logf("[pacman] keeper: the round: %v", err)
		}
		return next
	}
	// onRoundChange takes the round's step on a change of `rounds` or `runs`, and keeps the time it answers. The
	// current values handed at the watch's start are not each a step: the wake at the start takes one.
	onRoundChange := func(k *Keeper, c stdio.StateChange) error {
		if c.Current {
			return nil
		}
		k.wakeAt(roundWake, roundStep(k, k.Clock.Now()))
		return nil
	}
	return Hooks{
		Start: []func(*Keeper) error{
			func(k *Keeper) error {
				n, err := u.ImportLog()
				if n > 0 {
					k.logf("[pacman] keeper: %d past upgrades read from the package manager's log into runs", n)
				}
				return err
			},
			func(*Keeper) error { _, err := u.PruneRecords(); return err },
			func(*Keeper) error { return u.Follow() },
		},
		AfterTransaction: []func(*Keeper, Transaction) error{
			func(*Keeper, Transaction) error { return u.Follow() },
		},
		Changes: []StateHook{
			{State: StateAsks, Key: k.Node + ".*", Handle: func(_ *Keeper, c stdio.StateChange) error {
				return u.TakeAsk(c)
			}},
			{State: StateRounds, Key: "**", Handle: onRoundChange},
			{State: StateRuns, Key: "**", Handle: onRoundChange},
		},
		Wakes: []Wake{
			func(k *Keeper, now time.Time) time.Time {
				if !now.Equal(k.started) {
					if _, err := u.PruneRecords(); err != nil {
						k.logf("[pacman] keeper: pruning records: %v", err)
					}
				}
				return now.Add(UpdatesEvery)
			},
			roundStep,
		},
	}
}

// newRootActs is the client of the module's root helper (ADR 0510), which carries out the acts that need root,
// at the socket the manifest names to the keeper; with none named, every act is refused, named.
var newRootActs = func() RootActs {
	if sock := os.Getenv(EnvRootSocket); sock != "" {
		return rootClient{Socket: sock}
	}
	return noRoot{}
}

type noRoot struct{}

const noRootSaid = "the keeper was started without " + EnvRootSocket + ", the root helper's socket"

func (noRoot) Transaction(kind, run string, ignore []string) error {
	return errors.New(noRootSaid + ": no transaction can start")
}

func (noRoot) Journal(unit string, n int) ([]string, error) {
	return nil, errors.New(noRootSaid + ": no journal can be read")
}

func (noRoot) Rollback(run string, plan []RollbackItem) error {
	return errors.New(noRootSaid + ": no rollback can start")
}

func (noRoot) Maps() ([]Mapped, error) {
	return nil, errors.New(noRootSaid + ": no process's maps can be read")
}

// Keeper is the module's process on one node.
type Keeper struct {
	Node    string
	Config  KeeperConfig
	Machine *Machine
	State   func(name string) Store
	// Events emits the keeper's events; package.changed carries the id `installed.<node>.<revision>`, the bus's
	// dedup id, so the same change said twice is dropped.
	Events  EventSink
	Clock   Clock
	Watcher Watcher
	// Locked says pacman's lock is held now: a released lock followed at once by another transaction is read
	// at that one's end.
	Locked func() bool
	// DBChanged is when the local package database last changed, the time of a value read at the start.
	DBChanged func() time.Time
	Logf      func(format string, a ...any)
	// Hooks are the parts joining the loop; MakeHooks, when set, builds them once the keeper is connected
	// (its State and Events set), replacing Hooks.
	Hooks     Hooks
	MakeHooks func(k *Keeper) Hooks
	// AskRecord reads the operator channel's record of one of the module's asks (process.Conn.Ask).
	AskRecord func(id string) (*stdio.StateEntry, error)
	// Root carries out the acts that need root; nil is the root helper's client (newRootActs). Check is the
	// after-upgrade check; nil is the machine's (AfterCheck). A test sets both.
	Root  RootActs
	Check func(run Run, record *Record) Check

	upgrades *Upgrades

	started time.Time

	nextUpdates time.Time
	wakes       []time.Time
}

func (k *Keeper) logf(format string, a ...any) {
	if k.Logf != nil {
		k.Logf(format, a...)
	}
}

type change struct {
	hook StateHook
	c    stdio.StateChange
}

// Run reads the machine into the bus and then follows it until ctx ends or the watch fails.
func (k *Keeper) Run(ctx context.Context) error {
	if k.MakeHooks != nil {
		k.Hooks = k.MakeHooks(k)
	}
	k.started = k.Clock.Now()
	at := k.started
	if k.DBChanged != nil {
		if t := k.DBChanged(); !t.IsZero() {
			at = t
		}
	}
	k.refreshInstalled(at)
	k.refreshUpdates()
	for _, h := range k.Hooks.Start {
		if err := h(k); err != nil {
			k.logf("[pacman] keeper: a start hook: %v", err)
		}
	}
	changes := make(chan change, 256)
	for _, h := range k.Hooks.Changes {
		h := h
		err := k.State(h.State).Watch(h.Key, func(c stdio.StateChange) error {
			if c.Current {
				// the current values are handed over inside Watch, on this goroutine.
				k.handle(h, c)
				return nil
			}
			select {
			case changes <- change{h, c}:
				return nil
			case <-ctx.Done():
				return ctx.Err()
			}
		})
		if err != nil {
			return fmt.Errorf("watching %s %q: %w", h.State, h.Key, err)
		}
	}
	now := k.started
	k.wakes = make([]time.Time, len(k.Hooks.Wakes))
	for i, w := range k.Hooks.Wakes {
		k.wakes[i] = w(k, now)
	}
	signals := k.Watcher.Signals()
	for {
		now = k.Clock.Now()
		next := k.nextUpdates
		for _, w := range k.wakes {
			if !w.IsZero() && w.Before(next) {
				next = w
			}
		}
		timer := k.Clock.After(next.Sub(now))
		select {
		case <-ctx.Done():
			return ctx.Err()
		case s, open := <-signals:
			if !open {
				return errors.New("the machine's watch ended")
			}
			k.signal(s)
		case c := <-changes:
			k.handle(c.hook, c.c)
		case <-timer:
			k.due()
		}
	}
}

// wakeAt sets when wake i is called next, from a change a hook handled; before the wakes are first called
// (the current values handed at a watch's start) there is nothing to set.
func (k *Keeper) wakeAt(i int, at time.Time) {
	if i < len(k.wakes) {
		k.wakes[i] = at
	}
}

func (k *Keeper) handle(h StateHook, c stdio.StateChange) {
	if err := h.Handle(k, c); err != nil {
		k.logf("[pacman] keeper: handling %s %s: %v", h.State, c.Key, err)
	}
}

// signal acts on what the machine's watch saw.
func (k *Keeper) signal(s Signal) {
	switch s {
	case LockReleased:
		if k.Locked != nil && k.Locked() {
			return
		}
		at := k.Clock.Now()
		changed := k.refreshInstalled(at)
		k.refreshUpdates()
		for _, h := range k.Hooks.AfterTransaction {
			if err := h(k, Transaction{At: at, Changed: changed}); err != nil {
				k.logf("[pacman] keeper: after a transaction: %v", err)
			}
		}
	case MirrorsRewritten:
		k.refreshUpdates()
	}
}

// due runs what the clock has reached: the daily read of `updates`, and every wake whose time came.
func (k *Keeper) due() {
	now := k.Clock.Now()
	if !now.Before(k.nextUpdates) {
		k.refreshUpdates()
	}
	for i, w := range k.wakes {
		if !w.IsZero() && !now.Before(w) {
			k.wakes[i] = k.Hooks.Wakes[i](k, now)
		}
	}
}

// refreshInstalled reads the node's packages and writes `installed.<node>` when they differ from the key,
// emitting `package.changed` with what changed. It answers that change, nil when nothing was written.
func (k *Keeper) refreshInstalled(at time.Time) *PackageChanged {
	pkgs, skipped, err := k.Machine.ReadInstalled()
	if err != nil {
		k.logf("[pacman] keeper: reading the installed packages: %v", err)
		return nil
	}
	if skipped > 0 {
		k.logf("[pacman] keeper: %d lines of pacman's list were not a package and were left out", skipped)
	}
	st := k.State(StateInstalled)
	var before InstalledValue
	entry, err := st.Get(k.Node)
	if err != nil {
		k.logf("[pacman] keeper: reading installed.%s: %v", k.Node, err)
		return nil
	}
	if entry != nil {
		if err := json.Unmarshal(entry.Value, &before); err != nil {
			k.logf("[pacman] keeper: installed.%s holds no package list, so it is written whole: %v", k.Node, err)
			entry = nil
		}
	}
	if entry != nil && samePackages(before.Packages, pkgs) {
		k.current()
		return nil
	}
	value := InstalledValue{At: at.UTC(), Packages: pkgs}
	data, err := json.Marshal(value)
	if err != nil {
		k.logf("[pacman] keeper: %v", err)
		return nil
	}
	if len(data) > ValueCap {
		k.notCurrent(len(data), at)
		return nil
	}
	rev, err := st.Put(k.Node, value)
	if err != nil {
		k.logf("[pacman] keeper: writing installed.%s: %v", k.Node, err)
		return nil
	}
	k.current()
	c := DiffInstalled(before.Packages, pkgs)
	if c.Empty() {
		return nil
	}
	ev := &PackageChanged{Node: k.Node, At: at.UTC(), Added: c.Added, Removed: c.Removed, Changed: c.Changed}
	id := fmt.Sprintf("%s.%s.%d", StateInstalled, k.Node, rev)
	if err := k.Events.Emit(EventPackageChanged, *ev, stdio.EmitOptions{MessageID: id}); err != nil {
		k.logf("[pacman] keeper: emitting %s for installed.%s at revision %d: %v", EventPackageChanged, k.Node, rev, err)
	}
	return ev
}

func samePackages(a, b []InstalledPackage) bool {
	if len(a) != len(b) {
		return false
	}
	for i := range a {
		if a[i] != b[i] {
			return false
		}
	}
	return true
}

// notCurrent marks the node's packages as not current, once: the first size and moment stay until it is current
// again.
func (k *Keeper) notCurrent(size int, at time.Time) {
	st := k.State(StateInstalled)
	key := NotCurrentKey(k.Node)
	k.logf("[pacman] keeper: the node's packages come to %d bytes, over the bus's cap of %d: installed.%s keeps its last value", size, ValueCap, k.Node)
	if e, err := st.Get(key); err != nil || e != nil {
		if err != nil {
			k.logf("[pacman] keeper: reading %s: %v", key, err)
		}
		return
	}
	if _, err := st.Put(key, NotCurrent{Bytes: size, Since: at.UTC()}); err != nil {
		k.logf("[pacman] keeper: writing %s: %v", key, err)
	}
}

// current deletes the not-current mark, when there is one: the key holds the node's packages again.
func (k *Keeper) current() {
	st := k.State(StateInstalled)
	key := NotCurrentKey(k.Node)
	e, err := st.Get(key)
	if err != nil {
		k.logf("[pacman] keeper: reading %s: %v", key, err)
		return
	}
	if e == nil {
		return
	}
	if err := st.Delete(key); err != nil {
		k.logf("[pacman] keeper: deleting %s: %v", key, err)
	}
}

// refreshUpdates reads what a full upgrade would change and writes `updates.<node>` when the list or reachable
// changed. A failed read keeps the last list and says reachable false, never an empty list in its place.
func (k *Keeper) refreshUpdates() {
	k.nextUpdates = k.Clock.Now().Add(UpdatesEvery)
	list, err := k.Machine.UpdateList()
	st := k.State(StateUpdates)
	entry, gerr := st.Get(k.Node)
	if gerr != nil {
		k.logf("[pacman] keeper: reading updates.%s: %v", k.Node, gerr)
		return
	}
	var before UpdatesValue
	if entry != nil {
		if uerr := json.Unmarshal(entry.Value, &before); uerr != nil {
			entry = nil
		}
	}
	next := UpdatesValue{Reachable: true, Packages: list}
	if err != nil {
		k.logf("[pacman] keeper: reading what an upgrade would change: %v", err)
		next = UpdatesValue{Reachable: false, Packages: before.Packages}
	}
	if next.Packages == nil {
		next.Packages = []VersionChange{}
	}
	if entry != nil && sameUpdates(before, next) {
		return
	}
	if _, err := st.Put(k.Node, next); err != nil {
		k.logf("[pacman] keeper: writing updates.%s: %v", k.Node, err)
	}
}

func sameUpdates(a, b UpdatesValue) bool {
	if a.Reachable != b.Reachable || len(a.Packages) != len(b.Packages) {
		return false
	}
	for i := range a.Packages {
		if a.Packages[i] != b.Packages[i] {
			return false
		}
	}
	return true
}

// The keeper as a process.

// ReadKeeperConfig reads the keeper's config file, which the mesh writes from the module's config values.
func ReadKeeperConfig(path string) (KeeperConfig, error) {
	var c KeeperConfig
	if path == "" {
		return c, fmt.Errorf("the keeper was started without %s, its config file", EnvConfig)
	}
	data, err := os.ReadFile(path)
	if err != nil {
		return c, fmt.Errorf("reading the keeper's config: %w", err)
	}
	if err := json.Unmarshal(data, &c); err != nil {
		return c, fmt.Errorf("the keeper's config at %s is not its JSON: %w", path, err)
	}
	return c, nil
}

// keeper is `pacman-tools keeper`: it reads its bus account and its config, watches the machine, connects and
// runs until stopped. It refuses to start naming what is missing, and never logs a password.
func keeper() int {
	logf := func(f string, a ...any) { fmt.Fprintf(os.Stderr, f+"\n", a...) }
	brokerFile := os.Getenv(EnvBroker)
	if brokerFile == "" {
		logf("[pacman] keeper was started without %s, its bus account's file", EnvBroker)
		return 2
	}
	account, err := process.ReadAccount(brokerFile)
	if err != nil {
		logf("[pacman] keeper: %v", err)
		return 1
	}
	cfg, err := ReadKeeperConfig(os.Getenv(EnvConfig))
	if err != nil {
		logf("[pacman] keeper: %v", err)
		return 1
	}
	// the watch starts before the first read, so a transaction between the two is not missed.
	w, err := WatchMachine(PacmanDBDir, LockName, MirrorDir, MirrorName)
	if err != nil {
		logf("[pacman] keeper: watching the machine: %v", err)
		return 1
	}
	defer w.Close()
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()
	k := &Keeper{
		Node: account.Node, Config: cfg, Machine: ThisMachine(), Clock: realClock{}, Watcher: w, Logf: logf,
		Locked:    func() bool { _, err := os.Stat(filepath.Join(PacmanDBDir, LockName)); return err == nil },
		DBChanged: func() time.Time { return modified(filepath.Join(PacmanDBDir, "local")) },
		MakeHooks: keeperHooks,
	}
	logf("[pacman] keeper of %s is starting as %s", account.Node, account.User)
	if err := runKeeper(ctx, account, k); err != nil && ctx.Err() == nil {
		logf("[pacman] keeper stopped: %v", err)
		return 1
	}
	return 0
}

func modified(path string) time.Time {
	if fi, err := os.Stat(path); err == nil {
		return fi.ModTime()
	}
	return time.Time{}
}

// runKeeper connects as the keeper's own account and runs k on that connection until ctx ends. Apart from keeper
// so a test can drive it over a bus of its own.
func runKeeper(ctx context.Context, account process.Account, k *Keeper) error {
	conn, err := process.Connect(account)
	if err != nil {
		return fmt.Errorf("the bus cannot be reached: %w", err)
	}
	defer conn.Close()
	if k.Node == "" {
		k.Node = account.Node
	}
	k.State = func(name string) Store { return conn.State(name) }
	k.Events = conn
	k.AskRecord = func(id string) (*stdio.StateEntry, error) { return conn.Ask(MessengerAsksBucket, id) }
	return k.Run(ctx)
}