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

2026-10-22 17:35:36 +02:00
parent 28ef6e9e27
commit 276422a85b
2 changed files with 146 additions and 14 deletions
@@ -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 `<node>.*`, each change handed to TakeAsk;
-//   - Wakes: PruneRecords once a day.
+//   - Changes: `asks` at `<node>.*`, 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)
 }
@@ -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))
+	}
+}