novox/mesh-catalog
18
0
Fork 0
Code Issues Pull Requests 4 Actions Packages Projects Releases Wiki Activity

issue 1841/S5: join the upgrade parts to the keeper's loop, so a node's log, records, unfinished runs and asks are taken up by its own keeper

Browse Source 
At its start the keeper reads the log into runs once, prunes old records and carries unfinished runs on;
after every released lock it carries them on again; it takes each ask at <node>.* and prunes records daily.
The root helper's client is a constructor (newRootActs) the helper replaces; until then every act is
refused, named.
This commit is contained in:
jschoubben
2026-10-22 14:39:59 +02:00
parent b7a4a1a6b0
commit d70135bf08
3 changed files with 150 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/pacman/cmd/pacman-tools/keeper.go
+68 -8
View File
@@ -33,8 +33,8 @@ package main
// 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.
// A hook's error is logged and the keeper goes on: a fault in one part never stops the view. keeperHooks builds
// the hooks the keeper runs with, once connected; it is the one place a part is added.
import (
"context"
@@ -140,9 +140,60 @@ type Hooks struct {
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{} }
// keeperHooks is the hooks the keeper runs with, built once the keeper is connected: the one place a part of
// the keeper is added. Today: the upgrades of ADR 0508 (upgrades.go, record.go, transaction.go), for this node.
//
// - Start: the package manager's log read into `runs` once (ImportLog), records past record-days pruned
// (PruneRecords), and every unfinished run carried on (Follow), a run a keeper that went away left included;
// - AfterTransaction: every unfinished run carried on (Follow): a released lock may be a run's transaction ending;
// - Changes: `asks` at `<node>.*`, each change handed to TakeAsk;
// - Wakes: PruneRecords once a day.
func keeperHooks(k *Keeper) Hooks {
u := &Upgrades{Node: k.Node, Config: k.Config, Machine: k.Machine, Root: newRootActs(), Events: k.Events,
Installed: k.State(StateInstalled), Updates: k.State(StateUpdates), Runs: k.State(StateRuns),
Records: k.State(StateRecords), Asks: k.State(StateAsks)}
return Hooks{
Start: []func(*Keeper) error{
func(k *Keeper) error {
n, err := u.ImportLog()
if n > 0 {
k.logf("[pacman] keeper: %d past upgrades read from the package manager's log into runs", n)
}
return err
},
func(*Keeper) error { _, err := u.PruneRecords(); return err },
func(*Keeper) error { return u.Follow() },
},
AfterTransaction: []func(*Keeper, Transaction) error{
func(*Keeper, Transaction) error { return u.Follow() },
},
Changes: []StateHook{{State: StateAsks, Key: k.Node + ".*", Handle: func(_ *Keeper, c stdio.StateChange) error {
return u.TakeAsk(c)
}}},
Wakes: []Wake{func(k *Keeper, now time.Time) time.Time {
if !now.Equal(k.started) {
if _, err := u.PruneRecords(); err != nil {
k.logf("[pacman] keeper: pruning records: %v", err)
}
}
return now.Add(UpdatesEvery)
}},
}
}
// newRootActs is the client of the module's root helper (ADR 0510), which carries out the acts that need root;
// the helper's socket client replaces it. Until then every act is refused, named.
var newRootActs = func() RootActs { return noRoot{} }
type noRoot struct{}
func (noRoot) Transaction(kind, run string, ignore []string) error {
return errors.New("the root helper's client is not built in yet: no transaction can start")
}
func (noRoot) Journal(unit string, n int) ([]string, error) {
return nil, errors.New("the root helper's client is not built in yet: no journal can be read")
}
// Keeper is the module's process on one node.
type Keeper struct {
@@ -161,7 +212,12 @@ type Keeper struct {
// DBChanged is when the local package database last changed, the time of a value read at the start.
DBChanged func() time.Time
Logf func(format string, a ...any)
// Hooks are the parts joining the loop; MakeHooks, when set, builds them once the keeper is connected
// (its State and Events set), replacing Hooks.
Hooks Hooks
MakeHooks func(k *Keeper) Hooks
started time.Time
nextUpdates time.Time
wakes []time.Time
@@ -180,7 +236,11 @@ type change struct {
// 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.MakeHooks != nil {
k.Hooks = k.MakeHooks(k)
}
k.started = k.Clock.Now()
at := k.started
if k.DBChanged != nil {
if t := k.DBChanged(); !t.IsZero() {
at = t
@@ -213,7 +273,7 @@ func (k *Keeper) Run(ctx context.Context) error {
return fmt.Errorf("watching %s %q: %w", h.State, h.Key, err)
}
}
now := k.Clock.Now()
now := k.started
k.wakes = make([]time.Time, len(k.Hooks.Wakes))
for i, w := range k.Hooks.Wakes {
k.wakes[i] = w(k, now)
@@ -480,7 +540,7 @@ func keeper() int {
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(),
MakeHooks: keeperHooks,
}
logf("[pacman] keeper of %s is starting as %s", account.Node, account.User)
if err := runKeeper(ctx, account, k); err != nil && ctx.Err() == nil {
modules/pacman/cmd/pacman-tools/keeper_hooks_test.go
+81
View File
@@ -0,0 +1,81 @@
package main
// The upgrade parts joined to the keeper's loop (keeperHooks): the log read into runs and old records pruned at
// the start, an ask for this node taken from `asks`, and the records pruned again a day later.
import (
"context"
"strings"
"testing"
"time"
)
func TestTheKeeperRunsTheUpgradePartsAtItsStartOnItsAsksAndDaily(t *testing.T) {
db := &pacmanDB{sl: syncDBs, checkup: Ran{Status: 2}}
db.set("glibc 2.42-1\n", "")
k, states, _, clock, _ := newTestKeeper(db)
log := capturedLog(t)
k.Machine.ReadFile = func(path string) ([]byte, error) {
if path == PacmanLog {
return []byte(log), nil
}
return nil, errNoFile
}
k.MakeHooks = keeperHooks
k.Logf = t.Logf
k.Config = KeeperConfig{RecordDays: 180}
records := states.store(StateRecords)
records.Put(NodeKey("node-a", "20250101T040000Z"), Record{Node: "node-a", Run: "20250101T040000Z"})
records.Put(NodeKey("node-b", "20250101T040000Z"), Record{Node: "node-b", Run: "20250101T040000Z"})
done := Run{Node: "node-a", Run: "20261003T040000Z", Kind: RunUpgrade, From: FromAsk, Ask: "ask-1",
Steps: []Step{{Step: StepRefused, At: t0, Why: "a test"}}}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
ended := make(chan error, 1)
go func() { ended <- k.Run(ctx) }()
until := func(what string, ok func() bool) {
t.Helper()
for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(time.Millisecond) {
if ok() {
return
}
}
t.Fatalf("never: %s", what)
}
until("the log read into runs", func() bool {
keys, _ := states.store(StateRuns).Keys()
return len(keys) > 1
})
until("the old record pruned at the start", func() bool {
e, _ := records.Get(NodeKey("node-a", "20250101T040000Z"))
return e == nil
})
if e, _ := records.Get(NodeKey("node-b", "20250101T040000Z")); e == nil {
t.Error("another node's record was pruned")
}
// a run an ask started, ended; its ask left behind by a keeper that went away.
states.store(StateRuns).Put(NodeKey("node-a", done.Run), done)
asks := states.store(StateAsks)
asks.Put(NodeKey("node-a", "ask-1"), Ask{ID: "ask-1", Node: "node-a", Act: ActUpgrade, At: t0})
until("the ask a run names deleted", func() bool {
e, _ := asks.Get(NodeKey("node-a", "ask-1"))
return e == nil
})
records.Put(NodeKey("node-a", "20250102T040000Z"), Record{Node: "node-a", Run: "20250102T040000Z"})
time.Sleep(20 * time.Millisecond)
if e, _ := records.Get(NodeKey("node-a", "20250102T040000Z")); e == nil {
t.Fatal("pruned before a day passed")
}
clock.Advance(UpdatesEvery)
until("the old record pruned a day later", func() bool {
e, _ := records.Get(NodeKey("node-a", "20250102T040000Z"))
return e == nil
})
cancel()
if err := <-ended; err == nil || !strings.Contains(err.Error(), "canceled") {
t.Errorf("Run: %v", err)
}
}
modules/pacman/cmd/pacman-tools/keeper_integration_test.go
+1 -1
View File
@@ -116,7 +116,7 @@ func TestTheKeeperPutsTheNodesPackagesOnARealBusAndAnotherInstanceReadsThem(t *t
db.set("glibc 2.42-1\nzsh 5.9-1\n", "zsh\n")
signals := make(chan Signal)
k := &Keeper{Config: KeeperConfig{}, Machine: machine(db.runner(), 1000), Clock: realClock{},
Watcher: fakeWatcher{signals}, Logf: t.Logf, Hooks: keeperHooks()}
Watcher: fakeWatcher{signals}, Logf: t.Logf, MakeHooks: keeperHooks}
runCtx, cancel := context.WithCancel(ctx)
done := make(chan error, 1)
go func() { done <- runKeeper(runCtx, account, k) }()