From 276422a85b1c3e3ca0fe3a546150581064fa84ae Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 22 Oct 2026 17:35:36 +0200 Subject: [PATCH] issue 1841/S8: join the round to the keeper's loop, so a node starts the moment the node before it passed and a pass is answered from the operator's channel --- modules/pacman/cmd/pacman-tools/keeper.go | 69 +++++++++++--- .../cmd/pacman-tools/keeper_hooks_test.go | 91 +++++++++++++++++++ 2 files changed, 146 insertions(+), 14 deletions(-) diff --git a/modules/pacman/cmd/pacman-tools/keeper.go b/modules/pacman/cmd/pacman-tools/keeper.go index 85e21c47..fe45082d 100644 --- a/modules/pacman/cmd/pacman-tools/keeper.go +++ b/modules/pacman/cmd/pacman-tools/keeper.go @@ -141,17 +141,38 @@ type Hooks struct { } // 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. +// the keeper is added. Today: the upgrades of ADR 0508 (upgrades.go, record.go, transaction.go, rounds.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. +// - Changes: `asks` at `.*`, each change handed to TakeAsk; `rounds` and `runs`, every change, to the +// round's step (RoundStep): a node starts the moment the node before it passed; +// - Wakes: PruneRecords once a day; the round's step at the time it answered (the Saturday's opening, a pass +// whose answer is read again, a refused run tried again). 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)} + Records: k.State(StateRecords), Rounds: k.State(StateRounds), Asks: k.State(StateAsks), AskRecord: k.AskRecord} + k.upgrades = u + const roundWake = 1 // the round's step is the second wake + roundStep := func(k *Keeper, now time.Time) time.Time { + next, err := u.RoundStep() + if err != nil { + k.logf("[pacman] keeper: the round: %v", err) + } + return next + } + // onRoundChange takes the round's step on a change of `rounds` or `runs`, and keeps the time it answers. The + // current values handed at the watch's start are not each a step: the wake at the start takes one. + onRoundChange := func(k *Keeper, c stdio.StateChange) error { + if c.Current { + return nil + } + k.wakeAt(roundWake, roundStep(k, k.Clock.Now())) + return nil + } return Hooks{ Start: []func(*Keeper) error{ func(k *Keeper) error { @@ -167,17 +188,24 @@ func keeperHooks(k *Keeper) Hooks { 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) + Changes: []StateHook{ + {State: StateAsks, Key: k.Node + ".*", Handle: func(_ *Keeper, c stdio.StateChange) error { + return u.TakeAsk(c) + }}, + {State: StateRounds, Key: "**", Handle: onRoundChange}, + {State: StateRuns, Key: "**", Handle: onRoundChange}, + }, + 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) - }}, + return now.Add(UpdatesEvery) + }, + roundStep, + }, } } @@ -231,6 +259,10 @@ type Keeper struct { // (its State and Events set), replacing Hooks. Hooks Hooks MakeHooks func(k *Keeper) Hooks + // AskRecord reads the operator channel's record of one of the module's asks (process.Conn.Ask). + AskRecord func(id string) (*stdio.StateEntry, error) + + upgrades *Upgrades started time.Time @@ -319,6 +351,14 @@ func (k *Keeper) Run(ctx context.Context) error { } } +// wakeAt sets when wake i is called next, from a change a hook handled; before the wakes are first called +// (the current values handed at a watch's start) there is nothing to set. +func (k *Keeper) wakeAt(i int, at time.Time) { + if i < len(k.wakes) { + k.wakes[i] = at + } +} + 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) @@ -585,5 +625,6 @@ func runKeeper(ctx context.Context, account process.Account, k *Keeper) error { } k.State = func(name string) Store { return conn.State(name) } k.Events = conn + k.AskRecord = func(id string) (*stdio.StateEntry, error) { return conn.Ask(MessengerAsksBucket, id) } return k.Run(ctx) } diff --git a/modules/pacman/cmd/pacman-tools/keeper_hooks_test.go b/modules/pacman/cmd/pacman-tools/keeper_hooks_test.go index 91a71ade..a2f8c13f 100644 --- a/modules/pacman/cmd/pacman-tools/keeper_hooks_test.go +++ b/modules/pacman/cmd/pacman-tools/keeper_hooks_test.go @@ -5,9 +5,12 @@ package main import ( "context" + "encoding/json" "strings" "testing" "time" + + stdio "git.novox.be/novox/mesh-sdk/go" ) func TestTheKeeperRunsTheUpgradePartsAtItsStartOnItsAsksAndDaily(t *testing.T) { @@ -79,3 +82,91 @@ func TestTheKeeperRunsTheUpgradePartsAtItsStartOnItsAsksAndDaily(t *testing.T) { t.Errorf("Run: %v", err) } } + +// roundKeeper is a keeper of node in a round of the made-up nodes order, its machine's time the keeper's clock. +func roundKeeper(t *testing.T, node string, order []string) (*Keeper, *fakeStates, *fakeClock) { + t.Helper() + db := &pacmanDB{sl: syncDBs, checkup: Ran{Status: 2}} + db.set("glibc 2.42-1\n", "") + k, states, _, clock, _ := newTestKeeper(db) + k.Node = node + k.Machine.Now = clock.Now + k.MakeHooks = keeperHooks + k.Logf = t.Logf + k.Config = KeeperConfig{RecordDays: 180, RoundStartMinutes: 120, UpgradeOrder: order} + k.AskRecord = func(string) (*stdio.StateEntry, error) { return nil, nil } + return k, states, clock +} + +func until(t *testing.T, 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) +} + +func roundNodeOf(s *fakeStore, round, node string) RoundNode { + var rn RoundNode + if e, _ := s.Get(RoundNodeKey(round, node)); e != nil { + _ = json.Unmarshal(e.Value, &rn) + } + return rn +} + +// 1841/AC6 through the keeper's loop: the first of the order opens the week's round at its start on Saturday +// 04:00 and starts its run in it; the round's parts get the operator channel's reader and the rounds state. +func TestTheFirstNodesKeeperOpensTheRoundAndTakesItsTurn(t *testing.T) { + k, states, _ := roundKeeper(t, "node-a", []string{"node-a", "node-b"}) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go func() { _ = k.Run(ctx) }() + rounds := states.store(StateRounds) + round := RoundID(t0) + until(t, "the round opened", func() bool { e, _ := rounds.Get(round); return e != nil }) + until(t, "node-a's run named in the round", func() bool { return roundNodeOf(rounds, round, "node-a").Run != "" }) + if rn := roundNodeOf(rounds, round, "node-b"); rn.Run != "" { + t.Errorf("node-b started in node-a's keeper: %+v", rn) + } +} + +// 1841/AC6 through the keeper's loop: a node waits until the node before it passed its check, and starts on +// the change of `runs` that says so, with no wait for a timer. +func TestTheNextNodesKeeperStartsWhenTheRunBeforeItPassed(t *testing.T) { + k, states, clock := roundKeeper(t, "node-b", []string{"node-a", "node-b"}) + round := RoundID(t0) + rounds, runs := states.store(StateRounds), states.store(StateRuns) + runA := RunID(t0) + rounds.Put(round, Round{Round: round, Opened: t0, OpenedBy: "node-a", Order: []string{"node-a", "node-b"}}) + rounds.Put(RoundNodeKey(round, "node-a"), RoundNode{Round: round, Node: "node-a", Run: runA}) + a := Run{Node: "node-a", Run: runA, Kind: RunUpgrade, From: FromRound, Round: round, + Steps: []Step{{Step: StepRecorded, At: t0}, {Step: StepStarted, At: t0}}} + runs.Put(NodeKey("node-a", runA), a) + clock.Advance(10 * time.Minute) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go func() { _ = k.Run(ctx) }() + time.Sleep(30 * time.Millisecond) + if rn := roundNodeOf(rounds, round, "node-b"); rn.Run != "" { + t.Fatalf("node-b started before node-a passed: %+v", rn) + } + a.Steps = append(a.Steps, Step{Step: StepFinished, At: clock.Now()}, Step{Step: StepCheckedAfter, At: clock.Now(), Check: &Check{Passed: true}}) + runs.Put(NodeKey("node-a", runA), a) + until(t, "node-b's run named in the round", func() bool { return roundNodeOf(rounds, round, "node-b").Run != "" }) +} + +// The pass of a node is answered from the operator channel's record: the keeper hands its reader on. +func TestTheKeeperHandsTheOperatorChannelsReaderToTheRound(t *testing.T) { + k, states, _ := roundKeeper(t, "node-a", nil) + k.State = states.State + h := keeperHooks(k) + if k.upgrades == nil || k.upgrades.AskRecord == nil || k.upgrades.Rounds == nil { + t.Fatalf("the round's parts lack the channel's reader or the rounds state: %+v", k.upgrades) + } + if len(h.Wakes) < 2 { + t.Fatalf("no wake for the round: %d wakes", len(h.Wakes)) + } +}