issue 1840/S6: keep the node's packages in installed, read at the keeper's start and at every released lock
The keeper reads pacman's database into installed.<node> and writes it only when it differs from the key, says each change as package.changed, and keeps the last value when the list is over the bus's cap, marking the node not current until it fits again; the health names the size. The loop runs over injected states, events, clock, watch and machine, and leaves hooks (start, after a transaction, state watches, wakes) for the upgrade parts of ADR 0508.
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,528 @@
|
||||
package main
|
||||
|
||||
// The keeper (novox/hq ADR 0507 §1-§4; ADR 0508): the module's process on every node, under an account of its
|
||||
// own with a bus account of its own (ADR 0436 §1). It keeps the node's packages in the state `installed` and what
|
||||
// an upgrade would change in `updates`, and says each change of the packages as `package.changed`. It reads the
|
||||
// machine at its start and whenever the machine may have changed, never on a timer but the one daily read of
|
||||
// `updates` ADR 0507 §4 allows:
|
||||
//
|
||||
// - pacman's database lock released (an inotify watch on its directory: the lock file deleted) ends a
|
||||
// transaction, whoever ran it: `installed` and then `updates` are read again;
|
||||
// - the mirror list rewritten (reflector writes /etc/pacman.d/mirrorlist): `updates` is read again;
|
||||
// - a day after `updates` was last read: it is read again, since the repositories change what an upgrade
|
||||
// would do without any transaction on the node.
|
||||
//
|
||||
// A key is written only when what it holds changed (ADR 0289 §1). A value over the bus's cap is not written:
|
||||
// the key keeps its last value and `<node>.not-current` says the size and since when, written once and deleted
|
||||
// once the node's packages are current again (ADR 0507 §2).
|
||||
//
|
||||
// Everything the keeper touches is injected (its states, its events, the machine, the clock, the watch), so the
|
||||
// loop is tested without a machine or a bus. One goroutine runs it all: the watch's signals, the clock, and every
|
||||
// change a hook watches are handled one at a time, in the order they came.
|
||||
//
|
||||
// # Hooks (the upgrade parts of ADR 0508 join here)
|
||||
//
|
||||
// Hooks are where further work joins the loop, each run on the loop's goroutine:
|
||||
//
|
||||
// - Start: once, after the first reads of `installed` and `updates`, before anything else;
|
||||
// - AfterTransaction: after every released lock, once `installed` and `updates` were read again, with what the
|
||||
// transaction changed (nil when nothing);
|
||||
// - Changes: each a state and a key pattern (`*` one name, `**` the rest) watched for the life of the keeper;
|
||||
// its Handle sees the current values first (Current set), then every put and delete in order;
|
||||
// - Wakes: each a function the keeper calls at its start and again at the time it last answered; it answers when
|
||||
// it next wants to be called (a zero time: not again). It is how a hook keeps a time (a round's opening, a
|
||||
// stall) without polling.
|
||||
//
|
||||
// A hook's error is logged and the keeper goes on: a fault in one part never stops the view. keeperHooks names
|
||||
// the hooks the keeper runs with; it is the one place a part is added.
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"os/signal"
|
||||
"path/filepath"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
)
|
||||
|
||||
// ValueCap is the bus's cap on one value of a state (ADR 0201 §14): the mesh's bound, not the module's.
|
||||
const ValueCap = 256 << 10
|
||||
|
||||
// Where pacman keeps its database and its lock, and where the mirror list is.
|
||||
const (
|
||||
PacmanDBDir = "/var/lib/pacman"
|
||||
LockName = "db.lck"
|
||||
MirrorDir = "/etc/pacman.d"
|
||||
MirrorName = "mirrorlist"
|
||||
)
|
||||
|
||||
// UpdatesEvery is how long after the last read of `updates` it is read again with nothing else asking (ADR 0507
|
||||
// §4: once a day).
|
||||
const UpdatesEvery = 24 * time.Hour
|
||||
|
||||
// Store is one state as the keeper uses it: a process state on the bus (process.State), or a fake in a test.
|
||||
type Store interface {
|
||||
Get(key string) (*stdio.StateEntry, error)
|
||||
Create(key string, value any) (uint64, error)
|
||||
Update(key string, value any, revision uint64) (uint64, error)
|
||||
Put(key string, value any) (uint64, error)
|
||||
Delete(key string) error
|
||||
Keys() ([]string, error)
|
||||
Watch(key string, handle func(stdio.StateChange) error) error
|
||||
}
|
||||
|
||||
// Emitter emits one event with its body; id is the event's id, the bus's dedup id, so emitting it again after a
|
||||
// failure the bus did not report is dropped.
|
||||
type Emitter func(event string, body any, id string) error
|
||||
|
||||
// Clock is the time as the keeper sees it.
|
||||
type Clock interface {
|
||||
Now() time.Time
|
||||
// After fires once, d from now.
|
||||
After(d time.Duration) <-chan time.Time
|
||||
}
|
||||
|
||||
type realClock struct{}
|
||||
|
||||
func (realClock) Now() time.Time { return time.Now() }
|
||||
func (realClock) After(d time.Duration) <-chan time.Time { return time.After(d) }
|
||||
|
||||
// Signal is what the machine's watch saw.
|
||||
type Signal int
|
||||
|
||||
const (
|
||||
// LockReleased: pacman's database lock was deleted, a transaction (or a database sync) ended.
|
||||
LockReleased Signal = iota + 1
|
||||
// MirrorsRewritten: the mirror list was written or replaced.
|
||||
MirrorsRewritten
|
||||
)
|
||||
|
||||
func (s Signal) String() string {
|
||||
switch s {
|
||||
case LockReleased:
|
||||
return "the lock released"
|
||||
case MirrorsRewritten:
|
||||
return "the mirror list rewritten"
|
||||
}
|
||||
return fmt.Sprintf("signal %d", int(s))
|
||||
}
|
||||
|
||||
// Watcher is the machine's watch: its signals, until it is closed or fails (the channel closes).
|
||||
type Watcher interface {
|
||||
Signals() <-chan Signal
|
||||
Close() error
|
||||
}
|
||||
|
||||
// Transaction is a released lock as AfterTransaction sees it: when, and what it changed in `installed`
|
||||
// (nil when it changed nothing, or the value could not be written).
|
||||
type Transaction struct {
|
||||
At time.Time
|
||||
Changed *PackageChanged
|
||||
}
|
||||
|
||||
// StateHook is a watch on one of the module's states for the life of the keeper.
|
||||
type StateHook struct {
|
||||
State string
|
||||
Key string
|
||||
Handle func(k *Keeper, c stdio.StateChange) error
|
||||
}
|
||||
|
||||
// Wake is called at the keeper's start and then at the time it last answered; it answers when next.
|
||||
type Wake func(k *Keeper, now time.Time) time.Time
|
||||
|
||||
// Hooks are the parts that join the keeper's loop (see the file's comment).
|
||||
type Hooks struct {
|
||||
Start []func(k *Keeper) error
|
||||
AfterTransaction []func(k *Keeper, t Transaction) error
|
||||
Changes []StateHook
|
||||
Wakes []Wake
|
||||
}
|
||||
|
||||
// keeperHooks is the hooks the keeper runs with: the one place a part of the keeper is added (ADR 0508's
|
||||
// records, runs, rounds and asks).
|
||||
var keeperHooks = func() Hooks { return Hooks{} }
|
||||
|
||||
// Keeper is the module's process on one node.
|
||||
type Keeper struct {
|
||||
Node string
|
||||
Config KeeperConfig
|
||||
Machine *Machine
|
||||
State func(name string) Store
|
||||
Emit Emitter
|
||||
Clock Clock
|
||||
Watcher Watcher
|
||||
// Locked says pacman's lock is held now: a released lock followed at once by another transaction is read
|
||||
// at that one's end.
|
||||
Locked func() bool
|
||||
// DBChanged is when the local package database last changed, the time of a value read at the start.
|
||||
DBChanged func() time.Time
|
||||
Logf func(format string, a ...any)
|
||||
Hooks Hooks
|
||||
|
||||
nextUpdates time.Time
|
||||
wakes []time.Time
|
||||
}
|
||||
|
||||
func (k *Keeper) logf(format string, a ...any) {
|
||||
if k.Logf != nil {
|
||||
k.Logf(format, a...)
|
||||
}
|
||||
}
|
||||
|
||||
type change struct {
|
||||
hook StateHook
|
||||
c stdio.StateChange
|
||||
}
|
||||
|
||||
// Run reads the machine into the bus and then follows it until ctx ends or the watch fails.
|
||||
func (k *Keeper) Run(ctx context.Context) error {
|
||||
at := k.Clock.Now()
|
||||
if k.DBChanged != nil {
|
||||
if t := k.DBChanged(); !t.IsZero() {
|
||||
at = t
|
||||
}
|
||||
}
|
||||
k.refreshInstalled(at)
|
||||
k.refreshUpdates()
|
||||
for _, h := range k.Hooks.Start {
|
||||
if err := h(k); err != nil {
|
||||
k.logf("[pacman] keeper: a start hook: %v", err)
|
||||
}
|
||||
}
|
||||
changes := make(chan change, 256)
|
||||
for _, h := range k.Hooks.Changes {
|
||||
h := h
|
||||
err := k.State(h.State).Watch(h.Key, func(c stdio.StateChange) error {
|
||||
if c.Current {
|
||||
// the current values are handed over inside Watch, on this goroutine.
|
||||
k.handle(h, c)
|
||||
return nil
|
||||
}
|
||||
select {
|
||||
case changes <- change{h, c}:
|
||||
return nil
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("watching %s %q: %w", h.State, h.Key, err)
|
||||
}
|
||||
}
|
||||
now := k.Clock.Now()
|
||||
k.wakes = make([]time.Time, len(k.Hooks.Wakes))
|
||||
for i, w := range k.Hooks.Wakes {
|
||||
k.wakes[i] = w(k, now)
|
||||
}
|
||||
signals := k.Watcher.Signals()
|
||||
for {
|
||||
now = k.Clock.Now()
|
||||
next := k.nextUpdates
|
||||
for _, w := range k.wakes {
|
||||
if !w.IsZero() && w.Before(next) {
|
||||
next = w
|
||||
}
|
||||
}
|
||||
timer := k.Clock.After(next.Sub(now))
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case s, open := <-signals:
|
||||
if !open {
|
||||
return errors.New("the machine's watch ended")
|
||||
}
|
||||
k.signal(s)
|
||||
case c := <-changes:
|
||||
k.handle(c.hook, c.c)
|
||||
case <-timer:
|
||||
k.due()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (k *Keeper) handle(h StateHook, c stdio.StateChange) {
|
||||
if err := h.Handle(k, c); err != nil {
|
||||
k.logf("[pacman] keeper: handling %s %s: %v", h.State, c.Key, err)
|
||||
}
|
||||
}
|
||||
|
||||
// signal acts on what the machine's watch saw.
|
||||
func (k *Keeper) signal(s Signal) {
|
||||
switch s {
|
||||
case LockReleased:
|
||||
if k.Locked != nil && k.Locked() {
|
||||
return
|
||||
}
|
||||
at := k.Clock.Now()
|
||||
changed := k.refreshInstalled(at)
|
||||
k.refreshUpdates()
|
||||
for _, h := range k.Hooks.AfterTransaction {
|
||||
if err := h(k, Transaction{At: at, Changed: changed}); err != nil {
|
||||
k.logf("[pacman] keeper: after a transaction: %v", err)
|
||||
}
|
||||
}
|
||||
case MirrorsRewritten:
|
||||
k.refreshUpdates()
|
||||
}
|
||||
}
|
||||
|
||||
// due runs what the clock has reached: the daily read of `updates`, and every wake whose time came.
|
||||
func (k *Keeper) due() {
|
||||
now := k.Clock.Now()
|
||||
if !now.Before(k.nextUpdates) {
|
||||
k.refreshUpdates()
|
||||
}
|
||||
for i, w := range k.wakes {
|
||||
if !w.IsZero() && !now.Before(w) {
|
||||
k.wakes[i] = k.Hooks.Wakes[i](k, now)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// refreshInstalled reads the node's packages and writes `installed.<node>` when they differ from the key,
|
||||
// emitting `package.changed` with what changed. It answers that change, nil when nothing was written.
|
||||
func (k *Keeper) refreshInstalled(at time.Time) *PackageChanged {
|
||||
pkgs, skipped, err := k.Machine.ReadInstalled()
|
||||
if err != nil {
|
||||
k.logf("[pacman] keeper: reading the installed packages: %v", err)
|
||||
return nil
|
||||
}
|
||||
if skipped > 0 {
|
||||
k.logf("[pacman] keeper: %d lines of pacman's list were not a package and were left out", skipped)
|
||||
}
|
||||
st := k.State(StateInstalled)
|
||||
var before InstalledValue
|
||||
entry, err := st.Get(k.Node)
|
||||
if err != nil {
|
||||
k.logf("[pacman] keeper: reading installed.%s: %v", k.Node, err)
|
||||
return nil
|
||||
}
|
||||
if entry != nil {
|
||||
if err := json.Unmarshal(entry.Value, &before); err != nil {
|
||||
k.logf("[pacman] keeper: installed.%s holds no package list, so it is written whole: %v", k.Node, err)
|
||||
entry = nil
|
||||
}
|
||||
}
|
||||
if entry != nil && samePackages(before.Packages, pkgs) {
|
||||
k.current()
|
||||
return nil
|
||||
}
|
||||
value := InstalledValue{At: at.UTC(), Packages: pkgs}
|
||||
data, err := json.Marshal(value)
|
||||
if err != nil {
|
||||
k.logf("[pacman] keeper: %v", err)
|
||||
return nil
|
||||
}
|
||||
if len(data) > ValueCap {
|
||||
k.notCurrent(len(data), at)
|
||||
return nil
|
||||
}
|
||||
rev, err := st.Put(k.Node, value)
|
||||
if err != nil {
|
||||
k.logf("[pacman] keeper: writing installed.%s: %v", k.Node, err)
|
||||
return nil
|
||||
}
|
||||
k.current()
|
||||
c := DiffInstalled(before.Packages, pkgs)
|
||||
if c.Empty() {
|
||||
return nil
|
||||
}
|
||||
ev := &PackageChanged{Node: k.Node, At: at.UTC(), Added: c.Added, Removed: c.Removed, Changed: c.Changed}
|
||||
if err := k.Emit(EventPackageChanged, *ev, fmt.Sprintf("%s.%s.%d", StateInstalled, k.Node, rev)); err != nil {
|
||||
k.logf("[pacman] keeper: emitting %s for installed.%s at revision %d: %v", EventPackageChanged, k.Node, rev, err)
|
||||
}
|
||||
return ev
|
||||
}
|
||||
|
||||
func samePackages(a, b []InstalledPackage) bool {
|
||||
if len(a) != len(b) {
|
||||
return false
|
||||
}
|
||||
for i := range a {
|
||||
if a[i] != b[i] {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// notCurrent marks the node's packages as not current, once: the first size and moment stay until it is current
|
||||
// again.
|
||||
func (k *Keeper) notCurrent(size int, at time.Time) {
|
||||
st := k.State(StateInstalled)
|
||||
key := NotCurrentKey(k.Node)
|
||||
k.logf("[pacman] keeper: the node's packages come to %d bytes, over the bus's cap of %d: installed.%s keeps its last value", size, ValueCap, k.Node)
|
||||
if e, err := st.Get(key); err != nil || e != nil {
|
||||
if err != nil {
|
||||
k.logf("[pacman] keeper: reading %s: %v", key, err)
|
||||
}
|
||||
return
|
||||
}
|
||||
if _, err := st.Put(key, NotCurrent{Bytes: size, Since: at.UTC()}); err != nil {
|
||||
k.logf("[pacman] keeper: writing %s: %v", key, err)
|
||||
}
|
||||
}
|
||||
|
||||
// current deletes the not-current mark, when there is one: the key holds the node's packages again.
|
||||
func (k *Keeper) current() {
|
||||
st := k.State(StateInstalled)
|
||||
key := NotCurrentKey(k.Node)
|
||||
e, err := st.Get(key)
|
||||
if err != nil {
|
||||
k.logf("[pacman] keeper: reading %s: %v", key, err)
|
||||
return
|
||||
}
|
||||
if e == nil {
|
||||
return
|
||||
}
|
||||
if err := st.Delete(key); err != nil {
|
||||
k.logf("[pacman] keeper: deleting %s: %v", key, err)
|
||||
}
|
||||
}
|
||||
|
||||
// refreshUpdates reads what a full upgrade would change and writes `updates.<node>` when the list or reachable
|
||||
// changed. A failed read keeps the last list and says reachable false, never an empty list in its place.
|
||||
func (k *Keeper) refreshUpdates() {
|
||||
k.nextUpdates = k.Clock.Now().Add(UpdatesEvery)
|
||||
list, err := k.Machine.UpdateList()
|
||||
st := k.State(StateUpdates)
|
||||
entry, gerr := st.Get(k.Node)
|
||||
if gerr != nil {
|
||||
k.logf("[pacman] keeper: reading updates.%s: %v", k.Node, gerr)
|
||||
return
|
||||
}
|
||||
var before UpdatesValue
|
||||
if entry != nil {
|
||||
if uerr := json.Unmarshal(entry.Value, &before); uerr != nil {
|
||||
entry = nil
|
||||
}
|
||||
}
|
||||
next := UpdatesValue{Reachable: true, Packages: list}
|
||||
if err != nil {
|
||||
k.logf("[pacman] keeper: reading what an upgrade would change: %v", err)
|
||||
next = UpdatesValue{Reachable: false, Packages: before.Packages}
|
||||
}
|
||||
if next.Packages == nil {
|
||||
next.Packages = []VersionChange{}
|
||||
}
|
||||
if entry != nil && sameUpdates(before, next) {
|
||||
return
|
||||
}
|
||||
if _, err := st.Put(k.Node, next); err != nil {
|
||||
k.logf("[pacman] keeper: writing updates.%s: %v", k.Node, err)
|
||||
}
|
||||
}
|
||||
|
||||
func sameUpdates(a, b UpdatesValue) bool {
|
||||
if a.Reachable != b.Reachable || len(a.Packages) != len(b.Packages) {
|
||||
return false
|
||||
}
|
||||
for i := range a.Packages {
|
||||
if a.Packages[i] != b.Packages[i] {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// The keeper as a process.
|
||||
|
||||
// ReadKeeperConfig reads the keeper's config file, which the mesh writes from the module's config values.
|
||||
func ReadKeeperConfig(path string) (KeeperConfig, error) {
|
||||
var c KeeperConfig
|
||||
if path == "" {
|
||||
return c, fmt.Errorf("the keeper was started without %s, its config file", EnvConfig)
|
||||
}
|
||||
data, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return c, fmt.Errorf("reading the keeper's config: %w", err)
|
||||
}
|
||||
if err := json.Unmarshal(data, &c); err != nil {
|
||||
return c, fmt.Errorf("the keeper's config at %s is not its JSON: %w", path, err)
|
||||
}
|
||||
return c, nil
|
||||
}
|
||||
|
||||
// processStore is a process state on the bus as the keeper's Store. Delete is the SDK's when it has one; until
|
||||
// then a delete fails, named, and the not-current mark stays (said in the log), never silently.
|
||||
type processStore struct{ process.State }
|
||||
|
||||
func (s processStore) Delete(key string) error {
|
||||
if d, ok := any(s.State).(interface{ Delete(string) error }); ok {
|
||||
return d.Delete(key)
|
||||
}
|
||||
return fmt.Errorf("the SDK's process state cannot delete a key yet, so %s stays", key)
|
||||
}
|
||||
|
||||
// keeper is `pacman-tools keeper`: it reads its bus account and its config, watches the machine, connects and
|
||||
// runs until stopped. It refuses to start naming what is missing, and never logs a password.
|
||||
func keeper() int {
|
||||
logf := func(f string, a ...any) { fmt.Fprintf(os.Stderr, f+"\n", a...) }
|
||||
brokerFile := os.Getenv(EnvBroker)
|
||||
if brokerFile == "" {
|
||||
logf("[pacman] keeper was started without %s, its bus account's file", EnvBroker)
|
||||
return 2
|
||||
}
|
||||
account, err := process.ReadAccount(brokerFile)
|
||||
if err != nil {
|
||||
logf("[pacman] keeper: %v", err)
|
||||
return 1
|
||||
}
|
||||
cfg, err := ReadKeeperConfig(os.Getenv(EnvConfig))
|
||||
if err != nil {
|
||||
logf("[pacman] keeper: %v", err)
|
||||
return 1
|
||||
}
|
||||
// the watch starts before the first read, so a transaction between the two is not missed.
|
||||
w, err := WatchMachine(PacmanDBDir, LockName, MirrorDir, MirrorName)
|
||||
if err != nil {
|
||||
logf("[pacman] keeper: watching the machine: %v", err)
|
||||
return 1
|
||||
}
|
||||
defer w.Close()
|
||||
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
||||
defer stop()
|
||||
k := &Keeper{
|
||||
Node: account.Node, Config: cfg, Machine: ThisMachine(), Clock: realClock{}, Watcher: w, Logf: logf,
|
||||
Locked: func() bool { _, err := os.Stat(filepath.Join(PacmanDBDir, LockName)); return err == nil },
|
||||
DBChanged: func() time.Time { return modified(filepath.Join(PacmanDBDir, "local")) },
|
||||
Hooks: keeperHooks(),
|
||||
}
|
||||
logf("[pacman] keeper of %s is starting as %s", account.Node, account.User)
|
||||
if err := runKeeper(ctx, account, k); err != nil && ctx.Err() == nil {
|
||||
logf("[pacman] keeper stopped: %v", err)
|
||||
return 1
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
func modified(path string) time.Time {
|
||||
if fi, err := os.Stat(path); err == nil {
|
||||
return fi.ModTime()
|
||||
}
|
||||
return time.Time{}
|
||||
}
|
||||
|
||||
// runKeeper connects as the keeper's own account and runs k on that connection until ctx ends. Apart from keeper
|
||||
// so a test can drive it over a bus of its own.
|
||||
func runKeeper(ctx context.Context, account process.Account, k *Keeper) error {
|
||||
conn, err := process.Connect(account)
|
||||
if err != nil {
|
||||
return fmt.Errorf("the bus cannot be reached: %w", err)
|
||||
}
|
||||
defer conn.Close()
|
||||
if k.Node == "" {
|
||||
k.Node = account.Node
|
||||
}
|
||||
k.State = func(name string) Store { return processStore{conn.State(name)} }
|
||||
k.Emit = func(event string, body any, id string) error {
|
||||
return conn.Emit(event, body, stdio.EmitOptions{MessageID: id})
|
||||
}
|
||||
return k.Run(ctx)
|
||||
}
|
||||
@@ -0,0 +1,626 @@
|
||||
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, id string) error {
|
||||
e.mu.Lock()
|
||||
defer e.mu.Unlock()
|
||||
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, Emit: ev.Emit, 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)
|
||||
@@ -157,7 +161,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_orphans",
|
||||
|
||||
@@ -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,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