jschoubben
1de9a903c8
The keeper reads pacman's database into installed.<node> and writes it only when it differs from the key, says each change as package.changed, and keeps the last value when the list is over the bus's cap, marking the node not current until it fits again; the health names the size. The loop runs over injected states, events, clock, watch and machine, and leaves hooks (start, after a transaction, state watches, wakes) for the upgrade parts of ADR 0508.
529 lines
17 KiB
Go
529 lines
17 KiB
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 names
|
|
// the hooks the keeper runs with; 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
|
|
}
|
|
|
|
// Emitter emits one event with its body; id is the event's id, the bus's dedup id, so emitting it again after a
|
|
// failure the bus did not report is dropped.
|
|
type Emitter func(event string, body any, id string) 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: the one place a part of the keeper is added (ADR 0508's
|
|
// records, runs, rounds and asks).
|
|
var keeperHooks = func() Hooks { return Hooks{} }
|
|
|
|
// Keeper is the module's process on one node.
|
|
type Keeper struct {
|
|
Node string
|
|
Config KeeperConfig
|
|
Machine *Machine
|
|
State func(name string) Store
|
|
Emit Emitter
|
|
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 Hooks
|
|
|
|
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 {
|
|
at := k.Clock.Now()
|
|
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.Clock.Now()
|
|
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()
|
|
}
|
|
}
|
|
}
|
|
|
|
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}
|
|
if err := k.Emit(EventPackageChanged, *ev, fmt.Sprintf("%s.%s.%d", StateInstalled, k.Node, rev)); 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
|
|
}
|
|
|
|
// processStore is a process state on the bus as the keeper's Store. Delete is the SDK's when it has one; until
|
|
// then a delete fails, named, and the not-current mark stays (said in the log), never silently.
|
|
type processStore struct{ process.State }
|
|
|
|
func (s processStore) Delete(key string) error {
|
|
if d, ok := any(s.State).(interface{ Delete(string) error }); ok {
|
|
return d.Delete(key)
|
|
}
|
|
return fmt.Errorf("the SDK's process state cannot delete a key yet, so %s stays", key)
|
|
}
|
|
|
|
// 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")) },
|
|
Hooks: 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 processStore{conn.State(name)} }
|
|
k.Emit = func(event string, body any, id string) error {
|
|
return conn.Emit(event, body, stdio.EmitOptions{MessageID: id})
|
|
}
|
|
return k.Run(ctx)
|
|
}
|