From d70135bf08a990e6de421489e8745877cdd3b820 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 22 Oct 2026 14:39:59 +0200 Subject: [PATCH] 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 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 .* and prunes records daily. The root helper's client is a constructor (newRootActs) the helper replaces; until then every act is refused, named. --- modules/pacman/cmd/pacman-tools/keeper.go | 76 +++++++++++++++-- .../cmd/pacman-tools/keeper_hooks_test.go | 81 +++++++++++++++++++ .../pacman-tools/keeper_integration_test.go | 2 +- 3 files changed, 150 insertions(+), 9 deletions(-) create mode 100644 modules/pacman/cmd/pacman-tools/keeper_hooks_test.go diff --git a/modules/pacman/cmd/pacman-tools/keeper.go b/modules/pacman/cmd/pacman-tools/keeper.go index 16a8871d..92c1ee44 100644 --- a/modules/pacman/cmd/pacman-tools/keeper.go +++ b/modules/pacman/cmd/pacman-tools/keeper.go @@ -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 `.*`, 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 { diff --git a/modules/pacman/cmd/pacman-tools/keeper_hooks_test.go b/modules/pacman/cmd/pacman-tools/keeper_hooks_test.go new file mode 100644 index 00000000..91a71ade --- /dev/null +++ b/modules/pacman/cmd/pacman-tools/keeper_hooks_test.go @@ -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) + } +} diff --git a/modules/pacman/cmd/pacman-tools/keeper_integration_test.go b/modules/pacman/cmd/pacman-tools/keeper_integration_test.go index f65d3ca8..4cb3c2f2 100644 --- a/modules/pacman/cmd/pacman-tools/keeper_integration_test.go +++ b/modules/pacman/cmd/pacman-tools/keeper_integration_test.go @@ -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) }()