diff --git a/modules/pacman/cmd/pacman-tools/installed_health.go b/modules/pacman/cmd/pacman-tools/installed_health.go new file mode 100644 index 00000000..bf47c9a8 --- /dev/null +++ b/modules/pacman/cmd/pacman-tools/installed_health.go @@ -0,0 +1,32 @@ +package main + +// A node whose packages are over the bus's cap is unhealthy, naming the size (novox/hq ADR 0507 §2, acceptance +// 1840/AC3): `installed.` keeps its last value, which is no longer the node's, and the module's health says +// so until the keeper writes it again and deletes the mark. + +import ( + "encoding/json" + "fmt" +) + +// InstalledHealth is the health of this node's `installed` key: unhealthy while `.not-current` is there, +// with its size and since when; ok false when there is nothing to say (the key is current, or the state cannot +// be read here, which the keeper's own log says). +func (m *Machine) InstalledHealth() (Health, bool) { + if m.States == nil || m.Node == "" { + return Health{}, false + } + e, err := m.States(StateInstalled).Get(NotCurrentKey(m.Node)) + if err != nil { + return Health{Why: "the node's packages on the bus could not be read: " + err.Error()}, true + } + if e == nil { + return Health{}, false + } + var nc NotCurrent + if err := json.Unmarshal(e.Value, &nc); err != nil { + return Health{Why: fmt.Sprintf("installed.%s is marked not current, and its mark is not readable: %v", m.Node, err)}, true + } + return Health{Why: fmt.Sprintf("the node's packages come to %d bytes, over the bus's cap of %d on a value, since %s: "+ + "installed.%s keeps an older list", nc.Bytes, ValueCap, nc.Since.Format("2006-01-02 15:04 MST"), m.Node)}, true +} diff --git a/modules/pacman/cmd/pacman-tools/installed_health_test.go b/modules/pacman/cmd/pacman-tools/installed_health_test.go new file mode 100644 index 00000000..5b3f559f --- /dev/null +++ b/modules/pacman/cmd/pacman-tools/installed_health_test.go @@ -0,0 +1,45 @@ +package main + +import ( + "strings" + "testing" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// fakeReader reads a fakeStore as a verb does. +type fakeReader struct{ *fakeStore } + +func (r fakeReader) History(key string) ([]stdio.StateRevision, error) { + e, _ := r.Get(key) + if e == nil { + return []stdio.StateRevision{}, nil + } + return []stdio.StateRevision{{Key: key, Value: e.Value, Revision: e.Revision, Op: "put", At: r.at[key]}}, nil +} + +// readerOf is a machine's States over fake states. +func readerOf(states *fakeStates) func(string) StateReader { + return func(name string) StateReader { return fakeReader{states.store(name)} } +} + +// 1840/AC3: the health is unhealthy naming the size while the node is marked not current, and says nothing of +// it once the mark is gone. +func TestTheHealthNamesTheSizeWhileTheNodeIsNotCurrent(t *testing.T) { + states := &fakeStates{} + m := machine(nil, 1000) + m.Node, m.States = "node-a", readerOf(states) + if _, said := m.InstalledHealth(); said { + t.Fatal("a current node's health said something") + } + states.store(StateInstalled).Put("node-a.not-current", NotCurrent{Bytes: 300000, Since: t0}) + states.store(StateInstalled).Put("node-b.not-current", NotCurrent{Bytes: 1, Since: t0}) + h, said := m.InstalledHealth() + if !said || h.Healthy || !strings.Contains(h.Why, "300000 bytes") || !strings.Contains(h.Why, "2026-10-24 04:00") { + t.Fatalf("%+v", h) + } + states.store(StateInstalled).Delete("node-a.not-current") + if _, said := m.InstalledHealth(); said { + t.Fatal("the mark is gone and the health still says it") + } +} diff --git a/modules/pacman/cmd/pacman-tools/keeper.go b/modules/pacman/cmd/pacman-tools/keeper.go new file mode 100644 index 00000000..ede386f3 --- /dev/null +++ b/modules/pacman/cmd/pacman-tools/keeper.go @@ -0,0 +1,525 @@ +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 `.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 +} + +// 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 + // Events emits the keeper's events; package.changed carries the id `installed..`, 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 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.` 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.` 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.Events = conn + return k.Run(ctx) +} diff --git a/modules/pacman/cmd/pacman-tools/keeper_test.go b/modules/pacman/cmd/pacman-tools/keeper_test.go new file mode 100644 index 00000000..21ddf094 --- /dev/null +++ b/modules/pacman/cmd/pacman-tools/keeper_test.go @@ -0,0 +1,630 @@ +package main + +// The keeper's start and its lock watch (novox/hq ADR 0507 §2, §3; acceptance 1840/AC1-AC3), over a fake state, +// a fake machine, a fake clock and a fake watch. + +import ( + "context" + "encoding/json" + "fmt" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// fakeStore is one state in memory, recording every write. +type fakeStore struct { + mu sync.Mutex + name string + values map[string]json.RawMessage + revs map[string]uint64 + at map[string]time.Time + rev uint64 + writes []string // "put ", "delete " + watchers []fakeWatch + now func() time.Time +} + +type fakeWatch struct { + key string + handle func(stdio.StateChange) error +} + +func newFakeStore(name string) *fakeStore { + return &fakeStore{name: name, values: map[string]json.RawMessage{}, revs: map[string]uint64{}, at: map[string]time.Time{}} +} + +func (s *fakeStore) Get(key string) (*stdio.StateEntry, error) { + s.mu.Lock() + defer s.mu.Unlock() + v, ok := s.values[key] + if !ok { + return nil, nil + } + return &stdio.StateEntry{Key: key, Value: v, Revision: s.revs[key]}, nil +} + +func (s *fakeStore) write(key string, value any, op string) (uint64, error) { + data, err := json.Marshal(value) + if err != nil { + return 0, err + } + s.mu.Lock() + s.rev++ + s.values[key], s.revs[key], s.at[key] = data, s.rev, s.clock() + s.writes = append(s.writes, op+" "+key) + rev, ws := s.rev, append([]fakeWatch(nil), s.watchers...) + s.mu.Unlock() + for _, w := range ws { + if matchKey(w.key, key) { + w.handle(stdio.StateChange{State: s.name, Key: key, Op: "put", Value: data, Revision: rev}) + } + } + return rev, nil +} + +// clock is when a write lands: the store's own time, which a test sets. +func (s *fakeStore) clock() time.Time { + if s.now != nil { + return s.now() + } + return t0 +} + +func (s *fakeStore) Put(key string, value any) (uint64, error) { return s.write(key, value, "put") } + +func (s *fakeStore) Create(key string, value any) (uint64, error) { + if e, _ := s.Get(key); e != nil { + return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged) + } + return s.write(key, value, "create") +} + +func (s *fakeStore) Update(key string, value any, revision uint64) (uint64, error) { + if e, _ := s.Get(key); e == nil || e.Revision != revision { + return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged) + } + return s.write(key, value, "update") +} + +func (s *fakeStore) Delete(key string) error { + s.mu.Lock() + delete(s.values, key) + delete(s.revs, key) + s.rev++ + s.writes = append(s.writes, "delete "+key) + rev, ws := s.rev, append([]fakeWatch(nil), s.watchers...) + s.mu.Unlock() + for _, w := range ws { + if matchKey(w.key, key) { + w.handle(stdio.StateChange{State: s.name, Key: key, Op: "delete", Revision: rev}) + } + } + return nil +} + +func (s *fakeStore) Keys() ([]string, error) { + s.mu.Lock() + defer s.mu.Unlock() + keys := []string{} + for k := range s.values { + keys = append(keys, k) + } + sortStrings(keys) + return keys, nil +} + +func (s *fakeStore) Watch(key string, handle func(stdio.StateChange) error) error { + keys, _ := s.Keys() + for _, k := range keys { + if !matchKey(key, k) { + continue + } + e, _ := s.Get(k) + if err := handle(stdio.StateChange{State: s.name, Key: k, Op: "put", Value: e.Value, Revision: e.Revision, Current: true}); err != nil { + return err + } + } + s.mu.Lock() + s.watchers = append(s.watchers, fakeWatch{key, handle}) + s.mu.Unlock() + return nil +} + +// Writes is what was written, in order, and forgets it. +func (s *fakeStore) Writes() []string { + s.mu.Lock() + defer s.mu.Unlock() + w := s.writes + s.writes = nil + return w +} + +// matchKey narrows as the bus does: `*` one name, `**` the rest, empty every key. +func matchKey(pattern, key string) bool { + if pattern == "" { + return true + } + p, k := strings.Split(pattern, "."), strings.Split(key, ".") + for i := range p { + if p[i] == "**" { + return len(k) > i + } + if i >= len(k) || (p[i] != "*" && p[i] != k[i]) { + return false + } + } + return len(p) == len(k) +} + +func sortStrings(s []string) { + for i := 1; i < len(s); i++ { + for j := i; j > 0 && s[j] < s[j-1]; j-- { + s[j], s[j-1] = s[j-1], s[j] + } + } +} + +// fakeStates is the module's states in memory. +type fakeStates struct { + mu sync.Mutex + states map[string]*fakeStore +} + +func (f *fakeStates) State(name string) Store { return f.store(name) } + +func (f *fakeStates) store(name string) *fakeStore { + f.mu.Lock() + defer f.mu.Unlock() + if f.states == nil { + f.states = map[string]*fakeStore{} + } + if f.states[name] == nil { + f.states[name] = newFakeStore(name) + } + return f.states[name] +} + +// emitted is the events a keeper emitted. +type emitted struct { + mu sync.Mutex + events []emittedEvent +} + +type emittedEvent struct { + name string + body any + id string +} + +func (e *emitted) Emit(name string, body any, opts ...stdio.EmitOptions) error { + e.mu.Lock() + defer e.mu.Unlock() + id := "" + for _, o := range opts { + id = o.MessageID + } + e.events = append(e.events, emittedEvent{name, body, id}) + return nil +} + +func (e *emitted) Take() []emittedEvent { + e.mu.Lock() + defer e.mu.Unlock() + out := e.events + e.events = nil + return out +} + +// fakeClock moves only when the test moves it; After fires once the clock passes it. +type fakeClock struct { + mu sync.Mutex + now time.Time + timers []fakeTimer +} + +type fakeTimer struct { + at time.Time + c chan time.Time +} + +func (c *fakeClock) Now() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + return c.now +} + +func (c *fakeClock) After(d time.Duration) <-chan time.Time { + c.mu.Lock() + defer c.mu.Unlock() + t := fakeTimer{c.now.Add(d), make(chan time.Time, 1)} + if d <= 0 { + t.c <- c.now + return t.c + } + c.timers = append(c.timers, t) + return t.c +} + +func (c *fakeClock) Advance(d time.Duration) { + c.mu.Lock() + defer c.mu.Unlock() + c.now = c.now.Add(d) + kept := c.timers[:0] + for _, t := range c.timers { + if !t.at.After(c.now) { + t.c <- c.now + continue + } + kept = append(kept, t) + } + c.timers = kept +} + +// fakeWatcher hands the test's signals to the keeper. +type fakeWatcher struct{ c chan Signal } + +func (w fakeWatcher) Signals() <-chan Signal { return w.c } +func (w fakeWatcher) Close() error { return nil } + +// pacmanDB is a node's package database as pacman answers it: the test changes it between reads. +type pacmanDB struct { + mu sync.Mutex + q, qe string + sl string + checkup Ran + commands []string +} + +func (d *pacmanDB) set(q, qe string) { + d.mu.Lock() + defer d.mu.Unlock() + d.q, d.qe = q, qe +} + +func (d *pacmanDB) setUpdates(r Ran) { + d.mu.Lock() + defer d.mu.Unlock() + d.checkup = r +} + +func (d *pacmanDB) runner() Runner { + return func(_ context.Context, name string, args ...string) Ran { + d.mu.Lock() + defer d.mu.Unlock() + line := strings.TrimSpace(name + " " + strings.Join(args, " ")) + d.commands = append(d.commands, line) + switch line { + case "pacman -Q": + return Ran{Stdout: d.q} + case "pacman -Qeq": + return Ran{Stdout: d.qe} + case "pacman -Sl": + return Ran{Stdout: d.sl} + case "checkupdates": + return d.checkup + } + return Ran{Status: 99, Stderr: "unexpected command: " + line} + } +} + +var syncDBs = "core glibc 2.42-1 [installed]\ncore linux 6.17.1-1 [installed]\nextra zsh 5.9-2 [installed: 5.9-1]\n" + + "extra zsh-completions 0.35-1\nextra tmux 3.5-1\nextra-testing zsh 5.9-3\n" + +var t0 = time.Date(2026, 10, 24, 4, 0, 0, 0, time.UTC) + +func newTestKeeper(db *pacmanDB) (*Keeper, *fakeStates, *emitted, *fakeClock, fakeWatcher) { + states, ev, clock := &fakeStates{}, &emitted{}, &fakeClock{now: t0} + w := fakeWatcher{make(chan Signal)} + m := machine(db.runner(), 1000) + k := &Keeper{Node: "node-a", Machine: m, State: states.State, Events: ev, Clock: clock, Watcher: w, + Logf: func(string, ...any) {}} + return k, states, ev, clock, w +} + +func installedOf(t *testing.T, s *fakeStore, key string) InstalledValue { + t.Helper() + e, _ := s.Get(key) + if e == nil { + t.Fatalf("no %s", key) + } + var v InstalledValue + if err := json.Unmarshal(e.Value, &v); err != nil { + t.Fatal(err) + } + return v +} + +// 1840/AC1: at the start every package is written with its version, its reason and its repository or foreign, +// and written again only when it differs from the key. +func TestAtItsStartTheKeeperWritesEveryPackageOnlyWhenItDiffersFromTheKey(t *testing.T) { + db := &pacmanDB{sl: syncDBs, checkup: Ran{Status: 2}} + db.set("glibc 2.42-1\nlinux 6.17.1-1\nyay 12.0-1\nzsh 5.9-1\n", "linux\nyay\nzsh\n") + k, states, ev, _, _ := newTestKeeper(db) + k.DBChanged = func() time.Time { return t0.Add(-time.Hour) } + k.refreshInstalled(k.DBChanged()) + inst := states.store(StateInstalled) + if w := inst.Writes(); strings.Join(w, ",") != "put node-a" { + t.Fatalf("writes: %v", w) + } + v := installedOf(t, inst, "node-a") + want := []InstalledPackage{ + {"glibc", "2.42-1", false, "core"}, {"linux", "6.17.1-1", true, "core"}, + {"yay", "12.0-1", true, RepoForeign}, {"zsh", "5.9-1", true, "extra"}, + } + if !samePackages(v.Packages, want) || !v.At.Equal(t0.Add(-time.Hour)) { + t.Fatalf("installed.node-a: %+v", v) + } + if e := ev.Take(); len(e) != 1 || len(e[0].body.(PackageChanged).Added) != 4 { + t.Fatalf("the first write adds every package: %+v", e) + } + // a second start over the same database writes nothing and says nothing, whatever the time. + k.refreshInstalled(t0) + if w := inst.Writes(); len(w) != 0 { + t.Errorf("the same packages were written again: %v", w) + } + if e := ev.Take(); len(e) != 0 { + t.Errorf("the same packages were said again: %+v", e) + } +} + +// 1840/AC2: a transaction that adds, removes and changes packages is one package.changed naming the node and +// exactly those; one that changes nothing writes and emits nothing; a lock held again waits for its release. +func TestAReleasedLockSaysExactlyWhatTheTransactionChangedOnce(t *testing.T) { + db := &pacmanDB{sl: syncDBs, checkup: Ran{Status: 2}} + db.set("glibc 2.42-1\nlinux 6.17.1-1\nzsh 5.9-1\n", "linux\nzsh\n") + k, states, ev, clock, _ := newTestKeeper(db) + k.refreshInstalled(t0) + ev.Take() + states.store(StateInstalled).Writes() + + held := false + k.Locked = func() bool { return held } + db.set("glibc 2.42-1\nlinux 6.17.2-1\ntmux 3.5-1\n", "linux\ntmux\n") + held = true + k.signal(LockReleased) + if w := states.store(StateInstalled).Writes(); len(w) != 0 { + t.Fatalf("read while the lock was held again: %v", w) + } + held = false + clock.Advance(time.Minute) + k.signal(LockReleased) + if w := states.store(StateInstalled).Writes(); strings.Join(w, ",") != "put node-a" { + t.Fatalf("writes: %v", w) + } + e := ev.Take() + if len(e) != 1 || e[0].name != EventPackageChanged { + t.Fatalf("events: %+v", e) + } + got := e[0].body.(PackageChanged) + if got.Node != "node-a" || !got.At.Equal(t0.Add(time.Minute)) || + fmt.Sprint(got.Added) != "[{tmux 3.5-1}]" || fmt.Sprint(got.Removed) != "[{zsh 5.9-1}]" || + fmt.Sprint(got.Changed) != "[{linux 6.17.1-1 6.17.2-1}]" { + t.Fatalf("package.changed: %+v", got) + } + if !strings.HasPrefix(e[0].id, "installed.node-a.") { + t.Errorf("the event's id is the key's revision, so a second emit is dropped: %q", e[0].id) + } + // a transaction that changed nothing (a database sync alone). + k.signal(LockReleased) + if w := states.store(StateInstalled).Writes(); len(w) != 0 { + t.Errorf("nothing changed and the key was written: %v", w) + } + if e := ev.Take(); len(e) != 0 { + t.Errorf("nothing changed and it was said: %+v", e) + } + // a change of reason alone rewrites the key and is no package.changed. + db.set("glibc 2.42-1\nlinux 6.17.2-1\ntmux 3.5-1\n", "glibc\nlinux\ntmux\n") + k.signal(LockReleased) + if w := states.store(StateInstalled).Writes(); len(w) != 1 { + t.Errorf("a changed reason is a changed key: %v", w) + } + if e := ev.Take(); len(e) != 0 { + t.Errorf("a changed reason is no change of a version: %+v", e) + } +} + +// bigDB is a package database whose value is over the bus's cap. +func bigDB(n int) string { + var b strings.Builder + for i := 0; i < n; i++ { + fmt.Fprintf(&b, "package-with-a-long-name-%05d 1.0.0-1\n", i) + } + return b.String() +} + +// 1840/AC3: a value over the bus's cap is not written: the key keeps its last value and `.not-current` +// holds the size and since when, written once; the next write that fits deletes it. +func TestAValueOverTheCapKeepsTheLastAndMarksTheNodeNotCurrent(t *testing.T) { + db := &pacmanDB{sl: syncDBs, checkup: Ran{Status: 2}} + db.set("glibc 2.42-1\n", "") + k, states, ev, clock, _ := newTestKeeper(db) + k.refreshInstalled(t0) + inst := states.store(StateInstalled) + inst.Writes() + ev.Take() + + db.set(bigDB(5000), "") + clock.Advance(time.Minute) + k.signal(LockReleased) + if w := inst.Writes(); strings.Join(w, ",") != "put node-a.not-current" { + t.Fatalf("writes: %v", w) + } + if v := installedOf(t, inst, "node-a"); len(v.Packages) != 1 { + t.Fatalf("the last value is kept: %d packages", len(v.Packages)) + } + if e := ev.Take(); len(e) != 0 { + t.Errorf("nothing was written and it was said: %+v", e) + } + e, _ := inst.Get("node-a.not-current") + var nc NotCurrent + json.Unmarshal(e.Value, &nc) + if nc.Bytes <= ValueCap || !nc.Since.Equal(t0.Add(time.Minute)) { + t.Fatalf("not-current: %+v", nc) + } + // still over: the mark is written once. + db.set(bigDB(5001), "") + clock.Advance(time.Minute) + k.signal(LockReleased) + if w := inst.Writes(); len(w) != 0 { + t.Errorf("the mark was written again: %v", w) + } + // back under the cap: written, and the mark deleted. + db.set("glibc 2.42-1\nzsh 5.9-1\n", "") + k.signal(LockReleased) + if w := inst.Writes(); strings.Join(w, ",") != "put node-a,delete node-a.not-current" { + t.Errorf("writes: %v", w) + } +} + +// The mark goes too when the node's packages come back to what the key holds: the key is current again. +func TestTheMarkGoesWhenThePackagesComeBackToTheKey(t *testing.T) { + db := &pacmanDB{sl: syncDBs, checkup: Ran{Status: 2}} + db.set("glibc 2.42-1\n", "") + k, states, _, _, _ := newTestKeeper(db) + k.refreshInstalled(t0) + db.set(bigDB(5000), "") + k.signal(LockReleased) + inst := states.store(StateInstalled) + inst.Writes() + db.set("glibc 2.42-1\n", "") + k.signal(LockReleased) + if w := inst.Writes(); strings.Join(w, ",") != "delete node-a.not-current" { + t.Errorf("writes: %v", w) + } +} + +// The loop: a released lock is read, a hook runs after it with what it changed, a watched state's change reaches +// its hook, and a wake comes at the time it asked. +func TestTheLoopRunsTheHooksOnItsOneGoroutine(t *testing.T) { + db := &pacmanDB{sl: syncDBs, checkup: Ran{Status: 2}} + db.set("glibc 2.42-1\n", "") + k, states, _, clock, w := newTestKeeper(db) + var mu sync.Mutex + var seen []string + say := func(s string) { mu.Lock(); seen = append(seen, s); mu.Unlock() } + k.Hooks = Hooks{ + Start: []func(*Keeper) error{func(*Keeper) error { say("start"); return nil }}, + AfterTransaction: []func(*Keeper, Transaction) error{func(_ *Keeper, tr Transaction) error { + say(fmt.Sprintf("transaction changed=%v", tr.Changed != nil)) + return nil + }}, + Changes: []StateHook{{State: StateAsks, Key: "node-a.*", Handle: func(_ *Keeper, c stdio.StateChange) error { + say(fmt.Sprintf("%s %s current=%v", c.Op, c.Key, c.Current)) + return nil + }}}, + Wakes: []Wake{func(_ *Keeper, now time.Time) time.Time { + if now.Equal(t0) { + return t0.Add(time.Hour) + } + say("wake " + now.Sub(t0).String()) + return time.Time{} + }}, + } + states.store(StateAsks).Put("node-a.ask-1", Ask{ID: "ask-1"}) + states.store(StateAsks).Put("node-b.ask-2", Ask{ID: "ask-2"}) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- k.Run(ctx) }() + waitFor(t, &mu, &seen, "start") + db.set("glibc 2.42-1\nzsh 5.9-1\n", "") + w.c <- LockReleased + waitFor(t, &mu, &seen, "transaction changed=true") + states.store(StateAsks).Put("node-a.ask-3", Ask{ID: "ask-3"}) + waitFor(t, &mu, &seen, "put node-a.ask-3 current=false") + clock.Advance(time.Hour) + waitFor(t, &mu, &seen, "wake 1h0m0s") + cancel() + if err := <-done; err != context.Canceled { + t.Errorf("Run ended with %v", err) + } + mu.Lock() + defer mu.Unlock() + if strings.Join(seen, "|") != "put node-a.ask-1 current=true|start|transaction changed=true|put node-a.ask-3 current=false|wake 1h0m0s" && + strings.Join(seen, "|") != "start|put node-a.ask-1 current=true|transaction changed=true|put node-a.ask-3 current=false|wake 1h0m0s" { + t.Errorf("seen: %v", seen) + } +} + +func waitFor(t *testing.T, mu *sync.Mutex, seen *[]string, want string) { + t.Helper() + for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(time.Millisecond) { + mu.Lock() + for _, s := range *seen { + if s == want { + mu.Unlock() + return + } + } + mu.Unlock() + } + t.Fatalf("never saw %q", want) +} + +// The watch ending ends the keeper, loudly. +func TestTheKeeperStopsWhenTheWatchEnds(t *testing.T) { + db := &pacmanDB{sl: syncDBs, checkup: Ran{Status: 2}} + k, _, _, _, w := newTestKeeper(db) + close(w.c) + if err := k.Run(context.Background()); err == nil || !strings.Contains(err.Error(), "watch ended") { + t.Errorf("Run: %v", err) + } +} + +// The machine's watch: the lock deleted and the mirror list written or moved in are each a signal; anything else +// in those directories is none. +func TestTheWatchSeesTheLockReleasedAndTheMirrorListRewritten(t *testing.T) { + lockDir, mirrorDir := t.TempDir(), t.TempDir() + w, err := WatchMachine(lockDir, LockName, mirrorDir, MirrorName) + if err != nil { + t.Fatal(err) + } + next := func() Signal { + select { + case s := <-w.Signals(): + return s + case <-time.After(200 * time.Millisecond): + return 0 + } + } + lock := filepath.Join(lockDir, LockName) + os.WriteFile(lock, nil, 0o644) + os.WriteFile(filepath.Join(lockDir, "other"), nil, 0o644) + os.Remove(filepath.Join(lockDir, "other")) + if s := next(); s != 0 { + t.Fatalf("a lock taken, and another file, are no signal: %v", s) + } + os.Remove(lock) + if s := next(); s != LockReleased { + t.Fatalf("the lock deleted: %v", s) + } + os.WriteFile(filepath.Join(mirrorDir, MirrorName), []byte("Server = x\n"), 0o644) + if s := next(); s != MirrorsRewritten { + t.Fatalf("the list written: %v", s) + } + tmp := filepath.Join(mirrorDir, "mirrorlist.new") + os.WriteFile(tmp, []byte("Server = y\n"), 0o644) + if s := next(); s != 0 { + t.Fatalf("another file written: %v", s) + } + os.Rename(tmp, filepath.Join(mirrorDir, MirrorName)) + if s := next(); s != MirrorsRewritten { + t.Fatalf("the list moved in: %v", s) + } + w.Close() + for deadline := time.After(2 * time.Second); ; { + select { + case _, open := <-w.Signals(): + if !open { + return + } + case <-deadline: + t.Fatal("closing the watch did not end its signals") + } + } +} diff --git a/modules/pacman/cmd/pacman-tools/machine.go b/modules/pacman/cmd/pacman-tools/machine.go index 691a19d9..adc85428 100644 --- a/modules/pacman/cmd/pacman-tools/machine.go +++ b/modules/pacman/cmd/pacman-tools/machine.go @@ -22,6 +22,8 @@ import ( "os/exec" "strings" "time" + + stdio "git.novox.be/novox/mesh-sdk/go" ) // Ran is what one command did: its output, its exit status, and why it never ran to an answer. @@ -105,6 +107,17 @@ type Machine struct { ReadFile func(path string) ([]byte, error) Now func() time.Time Sleep func(time.Duration) + // Node is the node this bundle runs on (the runtime's MESH_NODE), and States reads the module's states and + // those it reads (the manifest's `reads`), for the verbs that answer from the bus (novox/hq ADR 0507 §6). + Node string + States func(name string) StateReader +} + +// StateReader is a state as a verb reads it: stdio.State, or a fake in a test. +type StateReader interface { + Get(key string) (*stdio.StateEntry, error) + Keys() ([]string, error) + History(key string) ([]stdio.StateRevision, error) } // ThisMachine is the machine the runtime launched this bundle on. @@ -117,7 +130,8 @@ func ThisMachine() *Machine { if account == "" { account = user } - return &Machine{Run: ExecRunner, UID: os.Getuid(), User: user, Account: account, ReadFile: os.ReadFile, Now: time.Now, Sleep: time.Sleep} + return &Machine{Run: ExecRunner, UID: os.Getuid(), User: user, Account: account, ReadFile: os.ReadFile, Now: time.Now, Sleep: time.Sleep, + Node: os.Getenv("MESH_NODE"), States: func(name string) StateReader { return stdio.State(name) }} } // Out runs a command that only reads, and fails with what went wrong named. diff --git a/modules/pacman/cmd/pacman-tools/main.go b/modules/pacman/cmd/pacman-tools/main.go index fe082aaf..02ca23a1 100644 --- a/modules/pacman/cmd/pacman-tools/main.go +++ b/modules/pacman/cmd/pacman-tools/main.go @@ -20,6 +20,10 @@ const binaryName = "pacman-tools" func bg() context.Context { return context.Background() } func main() { + // `pacman-tools keeper` is the module's process (keeper.go); without a word it serves the tools. + if len(os.Args) > 1 && os.Args[1] == "keeper" { + os.Exit(keeper()) + } // An empty name serves as the module the runtime names (MESH_SERVED_MODULE): pacman. if err := stdio.Serve("", tools(ThisMachine())); err != nil { fmt.Fprintln(os.Stderr, err) @@ -174,7 +178,12 @@ func tools(m *Machine) []stdio.Tool { Description: "Whether the weekly full upgrade (Saturday 04:00 local time, ADR 0477) last passed, as the module's health check: unhealthy " + "with the exit and the last error of the journal when it failed, healthy while it runs, after it passed and before it ever ran. (r)", Input: schema(map[string]any{}), - Run: func(map[string]any) (any, error) { return m.UpgradeHealth(), nil }, + Run: func(map[string]any) (any, error) { + if h, said := m.InstalledHealth(); said { + return h, nil + } + return m.UpgradeHealth(), nil + }, }, { Name: "pacman_upgrades", diff --git a/modules/pacman/cmd/pacman-tools/packages.go b/modules/pacman/cmd/pacman-tools/packages.go new file mode 100644 index 00000000..4c52b9e7 --- /dev/null +++ b/modules/pacman/cmd/pacman-tools/packages.go @@ -0,0 +1,144 @@ +package main + +// What the keeper reads from the machine into the bus (novox/hq ADR 0507 §2, §4, §7): the local package +// database (`pacman -Q`, `pacman -Qeq`) with each package's repository from the sync databases (`pacman -Sl`), +// and what a full upgrade would change (`checkupdates`). Every line is read against the shape of a package name +// and a version, and a line that is not one is left out, so nothing but names, versions and repositories ever +// reaches a value on the bus (ADR 0507 §1): never a command line, an environment or a message pacman printed. +// None of these commands names a package, so none names a foreign one either. + +import ( + "fmt" + "regexp" + "sort" + "strings" +) + +var ( + // packageName is what pacman allows in a package's name (makepkg's lint: alphanumerics and @._+-). + packageName = regexp.MustCompile(`^[a-z0-9@._+][a-z0-9@._+-]*$`) + // packageVersion is an epoch, a version and a release: alphanumerics and ._+:~-. + packageVersion = regexp.MustCompile(`^[A-Za-z0-9._+~][A-Za-z0-9._+:~-]*$`) + // repoName is a repository's section name in pacman.conf. + repoName = regexp.MustCompile(`^[A-Za-z0-9_-][A-Za-z0-9._-]*$`) +) + +// nameVersion reads a ` ` line, refusing anything else. +func nameVersion(l string) (string, string, bool) { + f := strings.Fields(l) + if len(f) != 2 || !packageName.MatchString(f[0]) || !packageVersion.MatchString(f[1]) { + return "", "", false + } + return f[0], f[1], true +} + +// ReadInstalled is the node's installed packages, sorted by name, each with its version, whether it was installed +// explicitly, and its repository or RepoForeign; and how many lines of pacman's output were not a package and +// were left out. +func (m *Machine) ReadInstalled() ([]InstalledPackage, int, error) { + all, err := m.Out("pacman", "-Q") + if err != nil { + return nil, 0, err + } + explicit, err := m.query("-Qeq") + if err != nil { + return nil, 0, err + } + synced, err := m.Out("pacman", "-Sl") + if err != nil { + return nil, 0, err + } + skipped := 0 + ex := map[string]bool{} + for _, l := range lines(explicit) { + if n := strings.TrimSpace(l); packageName.MatchString(n) { + ex[n] = true + } + } + // A package in two repositories is the first's: `pacman -Sl` lists them in pacman.conf's order. + repo := map[string]string{} + for _, l := range lines(synced) { + f := strings.Fields(l) + if len(f) < 3 || !repoName.MatchString(f[0]) || !packageName.MatchString(f[1]) { + continue + } + if _, seen := repo[f[1]]; !seen { + repo[f[1]] = f[0] + } + } + pkgs := []InstalledPackage{} + for _, l := range lines(all) { + name, version, ok := nameVersion(l) + if !ok { + skipped++ + continue + } + p := InstalledPackage{Name: name, Version: version, Explicit: ex[name], Repo: repo[name]} + if p.Repo == "" { + p.Repo = RepoForeign + } + pkgs = append(pkgs, p) + } + sort.Slice(pkgs, func(i, j int) bool { return pkgs[i].Name < pkgs[j].Name }) + return pkgs, skipped, nil +} + +var updateLine = regexp.MustCompile(`^(\S+) (\S+) -> (\S+)$`) + +// UpdateList is what a full upgrade would change, sorted by name, from checkupdates: a copy of the sync +// databases refreshed apart from the machine's own, so asking never makes a partial upgrade possible. Its +// status 2 is nothing to do; any other failure is an error, never an empty list. +func (m *Machine) UpdateList() ([]VersionChange, error) { + r := m.Run(bg(), "checkupdates") + switch { + case r.Err == "ENOENT": + return nil, fmt.Errorf("checkupdates is not installed: it comes with pacman-contrib, which this module declares") + case r.Err == "" && r.Status == 2: + return []VersionChange{}, nil + case r.Err != "" || r.Status != 0: + return nil, failure("checkupdates", "checkupdates", r) + } + ups := []VersionChange{} + for _, l := range lines(r.Stdout) { + u := updateLine.FindStringSubmatch(strings.TrimSpace(l)) + if u == nil || !packageName.MatchString(u[1]) || !packageVersion.MatchString(u[2]) || !packageVersion.MatchString(u[3]) { + continue + } + ups = append(ups, VersionChange{Name: u[1], From: u[2], To: u[3]}) + } + sort.Slice(ups, func(i, j int) bool { return ups[i].Name < ups[j].Name }) + return ups, nil +} + +// DiffInstalled is what changed from one package list to the next, each list sorted by name: a package added, +// removed, or at another version. A change of reason or repository alone is no change of a version. +func DiffInstalled(before, after []InstalledPackage) Changes { + c := Changes{Added: []PackageVersion{}, Removed: []PackageVersion{}, Changed: []VersionChange{}} + was := map[string]string{} + for _, p := range before { + was[p.Name] = p.Version + } + is := map[string]bool{} + for _, p := range after { + is[p.Name] = true + v, ok := was[p.Name] + switch { + case !ok: + c.Added = append(c.Added, PackageVersion{p.Name, p.Version}) + case v != p.Version: + c.Changed = append(c.Changed, VersionChange{p.Name, v, p.Version}) + } + } + for _, p := range before { + if !is[p.Name] { + c.Removed = append(c.Removed, PackageVersion{p.Name, p.Version}) + } + } + 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 }) + return c +} + +// Empty says a change changed nothing. +func (c Changes) Empty() bool { return len(c.Added)+len(c.Removed)+len(c.Changed) == 0 } diff --git a/modules/pacman/cmd/pacman-tools/query.go b/modules/pacman/cmd/pacman-tools/query.go index 881263ae..4c8a4f14 100644 --- a/modules/pacman/cmd/pacman-tools/query.go +++ b/modules/pacman/cmd/pacman-tools/query.go @@ -314,25 +314,15 @@ type Update struct { To string `json:"to"` } -var updateLine = regexp.MustCompile(`^(\S+) (\S+) -> (\S+)`) - -// Updates is what a full upgrade would change, from checkupdates: a copy of the sync databases -// refreshed apart from the machine's own, so asking never makes a partial upgrade possible. +// Updates is what a full upgrade would change, from checkupdates (UpdateList). func (m *Machine) Updates() (map[string]any, error) { - r := m.Run(bg(), "checkupdates") - switch { - case r.Err == "ENOENT": - return nil, fmt.Errorf("checkupdates is not installed: it comes with pacman-contrib, which this module declares") - case r.Err == "" && r.Status == 2: - return map[string]any{"count": 0, "updates": []Update{}}, nil - case r.Err != "" || r.Status != 0: - return nil, failure("checkupdates", "checkupdates", r) + list, err := m.UpdateList() + if err != nil { + return nil, err } ups := []Update{} - for _, l := range lines(r.Stdout) { - if u := updateLine.FindStringSubmatch(strings.TrimSpace(l)); u != nil { - ups = append(ups, Update{u[1], u[2], u[3]}) - } + for _, u := range list { + ups = append(ups, Update{u.Name, u.From, u.To}) } return map[string]any{"count": len(ups), "updates": ups}, nil } diff --git a/modules/pacman/cmd/pacman-tools/updates_test.go b/modules/pacman/cmd/pacman-tools/updates_test.go new file mode 100644 index 00000000..6dc0e220 --- /dev/null +++ b/modules/pacman/cmd/pacman-tools/updates_test.go @@ -0,0 +1,147 @@ +package main + +// What an upgrade would change (novox/hq ADR 0507 §4; acceptance 1840/AC4): read at the keeper's start, after a +// transaction, after the mirror list is rewritten and a day after the last read; written only when the list or +// reachable changed; a failed read says reachable false and keeps the last list. + +import ( + "context" + "encoding/json" + "fmt" + "strings" + "sync" + "testing" + "time" +) + +func updatesOf(t *testing.T, s *fakeStore) UpdatesValue { + t.Helper() + e, _ := s.Get("node-a") + if e == nil { + t.Fatal("no updates.node-a") + } + var v UpdatesValue + if err := json.Unmarshal(e.Value, &v); err != nil { + t.Fatal(err) + } + return v +} + +func (d *pacmanDB) checkupdatesRuns() int { + d.mu.Lock() + defer d.mu.Unlock() + n := 0 + for _, c := range d.commands { + if c == "checkupdates" { + n++ + } + } + return n +} + +func TestUpdatesAreWrittenOnlyWhenTheListOrReachableChanged(t *testing.T) { + db := &pacmanDB{sl: syncDBs, checkup: Ran{Stdout: "zsh 5.9-1 -> 5.9-2\nlinux 6.17.1-1 -> 6.17.2-1\n"}} + db.set("glibc 2.42-1\n", "") + k, states, _, _, _ := newTestKeeper(db) + up := states.store(StateUpdates) + k.refreshUpdates() + if w := up.Writes(); strings.Join(w, ",") != "put node-a" { + t.Fatalf("writes: %v", w) + } + if v := updatesOf(t, up); !v.Reachable || fmt.Sprint(v.Packages) != "[{linux 6.17.1-1 6.17.2-1} {zsh 5.9-1 5.9-2}]" { + t.Fatalf("%+v", v) + } + // the same answer: nothing written. + k.refreshUpdates() + if w := up.Writes(); len(w) != 0 { + t.Fatalf("the same list was written again: %v", w) + } + // the mirrors cannot be reached: reachable false, the last list kept. + db.setUpdates(Ran{Status: 1, Stderr: "==> ERROR: Cannot fetch updates\n"}) + k.refreshUpdates() + if w := up.Writes(); len(w) != 1 { + t.Fatalf("writes: %v", w) + } + if v := updatesOf(t, up); v.Reachable || len(v.Packages) != 2 { + t.Fatalf("a failed read keeps the list and says reachable false: %+v", v) + } + k.refreshUpdates() + if w := up.Writes(); len(w) != 0 { + t.Fatalf("still unreachable, and written again: %v", w) + } + // reached again with nothing to do (status 2): an empty list, reachable. + db.setUpdates(Ran{Status: 2}) + k.refreshUpdates() + if v := updatesOf(t, up); !v.Reachable || v.Packages == nil || len(v.Packages) != 0 { + t.Fatalf("%+v", v) + } +} + +func TestAFirstReadThatFailsSaysUnreachableNeverAnEmptyListAsAnAnswer(t *testing.T) { + db := &pacmanDB{sl: syncDBs, checkup: Ran{Status: 1, Stderr: "==> ERROR: Cannot fetch updates\n"}} + k, states, _, _, _ := newTestKeeper(db) + k.refreshUpdates() + if v := updatesOf(t, states.store(StateUpdates)); v.Reachable { + t.Fatalf("%+v", v) + } + raw, _ := states.store(StateUpdates).Get("node-a") + if string(raw.Value) != `{"reachable":false,"packages":[]}` { + t.Errorf("%s", raw.Value) + } +} + +// Read at the start, after a transaction, after the mirror list is rewritten, and a day after the last read +// with nothing else asking; never in between. +func TestUpdatesAreReadAtStartAfterATransactionAfterTheMirrorsAndDaily(t *testing.T) { + db := &pacmanDB{sl: syncDBs, checkup: Ran{Status: 2}} + db.set("glibc 2.42-1\n", "") + k, states, _, clock, w := newTestKeeper(db) + var mu sync.Mutex + var seen []string + k.Hooks.AfterTransaction = []func(*Keeper, Transaction) error{func(*Keeper, Transaction) error { + mu.Lock() + seen = append(seen, "transaction") + mu.Unlock() + return nil + }} + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { done <- k.Run(ctx) }() + reads := func(want int) { + t.Helper() + for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(time.Millisecond) { + if db.checkupdatesRuns() == want { + return + } + } + t.Fatalf("checkupdates ran %d times, want %d", db.checkupdatesRuns(), want) + } + reads(1) + db.setUpdates(Ran{Stdout: "zsh 5.9-1 -> 5.9-2\n"}) + clock.Advance(time.Hour) + w.c <- MirrorsRewritten + reads(2) + db.setUpdates(Ran{Status: 2}) + w.c <- LockReleased + reads(3) + waitFor(t, &mu, &seen, "transaction") + clock.Advance(UpdatesEvery - time.Minute) + time.Sleep(50 * time.Millisecond) + if n := db.checkupdatesRuns(); n != 3 { + t.Fatalf("read before a day passed: %d", n) + } + db.setUpdates(Ran{Stdout: "glibc 2.42-1 -> 2.42-2\n"}) + clock.Advance(time.Minute) + reads(4) + for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(time.Millisecond) { + if e, _ := states.store(StateUpdates).Get("node-a"); e != nil && strings.Contains(string(e.Value), "glibc") { + break + } + } + if w := states.store(StateUpdates).Writes(); len(w) != 4 { + t.Errorf("each read that changed the list wrote it, and only those: %v", w) + } + cancel() + <-done +} diff --git a/modules/pacman/cmd/pacman-tools/watch.go b/modules/pacman/cmd/pacman-tools/watch.go new file mode 100644 index 00000000..e66f69ae --- /dev/null +++ b/modules/pacman/cmd/pacman-tools/watch.go @@ -0,0 +1,129 @@ +package main + +// The machine's watch (novox/hq ADR 0507 §2, §4): inotify on the directory of pacman's database lock, for the +// lock file deleted (a transaction ended), and on the mirror list's directory, for the list written or moved +// into place (reflector rewrote it). A directory is watched, not the file: the lock comes and goes, and a list +// replaced by a rename is a new file. Nothing is polled. + +import ( + "errors" + "fmt" + "os" + "sync" + "unsafe" + + "golang.org/x/sys/unix" +) + +type inotify struct { + f *os.File + lockWD int + mirWD int + lock string + mirrors string + out chan Signal + // pending holds a signal of each kind until the keeper takes it, so a burst of one kind is one signal and + // never crowds out the other. + mu sync.Mutex + pending map[Signal]bool + notify chan struct{} +} + +// WatchMachine watches lockDir for lockName deleted and mirrorDir for mirrorName written or moved in. +func WatchMachine(lockDir, lockName, mirrorDir, mirrorName string) (Watcher, error) { + fd, err := unix.InotifyInit1(unix.IN_CLOEXEC | unix.IN_NONBLOCK) + if err != nil { + return nil, fmt.Errorf("inotify: %w", err) + } + w := &inotify{lock: lockName, mirrors: mirrorName, out: make(chan Signal), pending: map[Signal]bool{}, + notify: make(chan struct{}, 1)} + if w.lockWD, err = unix.InotifyAddWatch(fd, lockDir, unix.IN_DELETE|unix.IN_MOVED_FROM); err != nil { + unix.Close(fd) + return nil, fmt.Errorf("watching %s: %w", lockDir, err) + } + if w.mirWD, err = unix.InotifyAddWatch(fd, mirrorDir, unix.IN_CLOSE_WRITE|unix.IN_MOVED_TO); err != nil { + unix.Close(fd) + return nil, fmt.Errorf("watching %s: %w", mirrorDir, err) + } + // a non-blocking descriptor as an os.File is read through the runtime's poller, so Close ends a read. + w.f = os.NewFile(uintptr(fd), "inotify") + go w.read() + go w.forward() + return w, nil +} + +func (w *inotify) Signals() <-chan Signal { return w.out } + +func (w *inotify) Close() error { return w.f.Close() } + +func (w *inotify) read() { + defer close(w.notify) + buf := make([]byte, 64*(unix.SizeofInotifyEvent+unix.NAME_MAX+1)) + for { + n, err := w.f.Read(buf) + if err != nil { + if !errors.Is(err, os.ErrClosed) { + fmt.Fprintf(os.Stderr, "[pacman] keeper: the machine's watch failed: %v\n", err) + } + return + } + for off := 0; off+unix.SizeofInotifyEvent <= n; { + ev := (*unix.InotifyEvent)(unsafe.Pointer(&buf[off])) + nameAt := off + unix.SizeofInotifyEvent + end := nameAt + int(ev.Len) + if end > n { + break + } + name := cstring(buf[nameAt:end]) + off = end + if ev.Mask&unix.IN_Q_OVERFLOW != 0 { + // events were lost: anything may have happened, so both are read again. + w.send(LockReleased) + w.send(MirrorsRewritten) + continue + } + switch { + case int(ev.Wd) == w.lockWD && name == w.lock: + w.send(LockReleased) + case int(ev.Wd) == w.mirWD && name == w.mirrors: + w.send(MirrorsRewritten) + } + } + } +} + +// send marks a signal pending; one of the same kind already waiting covers it. +func (w *inotify) send(s Signal) { + w.mu.Lock() + w.pending[s] = true + w.mu.Unlock() + select { + case w.notify <- struct{}{}: + default: + } +} + +// forward hands the pending signals to the keeper, in a fixed order, until the watch ends. +func (w *inotify) forward() { + defer close(w.out) + for range w.notify { + for _, s := range []Signal{LockReleased, MirrorsRewritten} { + w.mu.Lock() + due := w.pending[s] + delete(w.pending, s) + w.mu.Unlock() + if due { + w.out <- s + } + } + } +} + +func cstring(b []byte) string { + for i, c := range b { + if c == 0 { + return string(b[:i]) + } + } + return string(b) +} diff --git a/modules/pacman/go.mod b/modules/pacman/go.mod index b6e447fc..1fac27ba 100644 --- a/modules/pacman/go.mod +++ b/modules/pacman/go.mod @@ -3,3 +3,12 @@ module pacman go 1.26.0 require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022121659-076e1f7947db + +require ( + github.com/klauspost/compress v1.20.0 // indirect + github.com/nats-io/nats.go v1.54.0 // indirect + github.com/nats-io/nkeys v0.4.16 // indirect + github.com/nats-io/nuid v1.0.1 // indirect + golang.org/x/crypto v0.57.0 // indirect + golang.org/x/sys v0.48.0 +) diff --git a/modules/pacman/go.sum b/modules/pacman/go.sum index ed7b73f7..27e3d79b 100644 --- a/modules/pacman/go.sum +++ b/modules/pacman/go.sum @@ -1,2 +1,26 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022121659-076e1f7947db h1:WiBM7p87qsINU+0ipvchRL7ZUFSJHGLG41ZA7vMSR18= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022121659-076e1f7947db/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= +github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= +github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= +github.com/google/go-tpm v0.9.8/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY= +github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA= +github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= +github.com/minio/highwayhash v1.0.4 h1:asJizugGgchQod2ja9NJlGOWq4s7KsAWr5XUc9Clgl4= +github.com/minio/highwayhash v1.0.4/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ= +github.com/nats-io/jwt/v2 v2.8.1 h1:V0xpGuD/N8Mi+fQNDynXohVvp7ZztevW5io8CUWlPmU= +github.com/nats-io/jwt/v2 v2.8.1/go.mod h1:nWnOEEiVMiKHQpnAy4eXlizVEtSfzacZ1Q43LIRavZg= +github.com/nats-io/nats-server/v2 v2.11.17 h1:GKEghcFK6A+aFx11Yf1LjgLC3txAwvyhnYzhBIQZA8I= +github.com/nats-io/nats-server/v2 v2.11.17/go.mod h1:B1sFVz4StNosQ903ak4N1G01Fl/9f8e06mXpFIE2K24= +github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA= +github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k= +github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg= +github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs= +github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= +github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= +golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= +golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA= +golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= +golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= +golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= +golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=