Merge branch 'feat/upgrade-safety' into p1
This commit is contained in:
@@ -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.<node>` 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 `<node>.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
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
@@ -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 `<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
|
||||
}
|
||||
|
||||
// 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.<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 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}
|
||||
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
|
||||
}
|
||||
|
||||
// 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)
|
||||
}
|
||||
@@ -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 <key>", "delete <key>"
|
||||
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 `<node>.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")
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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.
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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 `<name> <version>` 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 }
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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
|
||||
)
|
||||
|
||||
@@ -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=
|
||||
|
||||
Reference in New Issue
Block a user