diff --git a/modules/pacman/cmd/pacman-tools/keeper.go b/modules/pacman/cmd/pacman-tools/keeper.go index fe45082d..7c6eb981 100644 --- a/modules/pacman/cmd/pacman-tools/keeper.go +++ b/modules/pacman/cmd/pacman-tools/keeper.go @@ -152,9 +152,14 @@ type Hooks struct { // - 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, + root := k.Root + if root == nil { + root = newRootActs() + } + u := &Upgrades{Node: k.Node, Config: k.Config, Machine: k.Machine, Root: root, Events: k.Events, Installed: k.State(StateInstalled), Updates: k.State(StateUpdates), Runs: k.State(StateRuns), - Records: k.State(StateRecords), Rounds: k.State(StateRounds), Asks: k.State(StateAsks), AskRecord: k.AskRecord} + Records: k.State(StateRecords), Rounds: k.State(StateRounds), Asks: k.State(StateAsks), AskRecord: k.AskRecord, + Check: k.Check} k.upgrades = u const roundWake = 1 // the round's step is the second wake roundStep := func(k *Keeper, now time.Time) time.Time { @@ -261,6 +266,10 @@ type Keeper struct { 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) + // Root carries out the acts that need root; nil is the root helper's client (newRootActs). Check is the + // after-upgrade check; nil is the machine's (AfterCheck). A test sets both. + Root RootActs + Check func(run Run, record *Record) Check upgrades *Upgrades diff --git a/modules/pacman/cmd/pacman-tools/round_integration_test.go b/modules/pacman/cmd/pacman-tools/round_integration_test.go new file mode 100644 index 00000000..a6855d9b --- /dev/null +++ b/modules/pacman/cmd/pacman-tools/round_integration_test.go @@ -0,0 +1,180 @@ +//go:build integration + +package main + +// The round over a real bus (novox/hq ADR 0508 §5, §6; acceptance 1841/AC15): the keepers of two made-up nodes, +// each whole (its own bus account, its states, its watches) against a JetStream server of the test's own, with +// fake machines and a fake root helper each. The first node of the order opens the week's round and runs; the +// second starts only once the first's after-upgrade check passed; a failed check stops the round there. +// Compiled by the merge gate's `go vet -tags integration`; run by the merge check, never by hand. + +import ( + "context" + "encoding/json" + "strings" + "sync" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/process" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" +) + +// countingRoot is a node's root helper: it counts the transactions it was asked to start. +type countingRoot struct { + mu sync.Mutex + started []string +} + +func (r *countingRoot) Transaction(kind, run string, ignore []string) error { + r.mu.Lock() + defer r.mu.Unlock() + r.started = append(r.started, kind+" "+run) + return nil +} +func (r *countingRoot) Rollback(string, []RollbackItem) error { return nil } +func (r *countingRoot) Journal(string, int) ([]string, error) { return nil, nil } +func (r *countingRoot) Maps() ([]Mapped, error) { return []Mapped{}, nil } +func (r *countingRoot) count() int { + r.mu.Lock() + defer r.mu.Unlock() + return len(r.started) +} + +// roundNode is one made-up node's keeper on the test's bus. +type roundNode struct { + root *countingRoot + signals chan Signal +} + +// startRoundNode starts node's keeper; its machine's clock begins at the Saturday's opening, its transaction +// unit is gone (ended well) whenever asked, and its after-upgrade check passes unless passes says otherwise. +func startRoundNode(t *testing.T, ctx context.Context, url, node string, order []string, passes bool) *roundNode { + t.Helper() + db := &pacmanDB{sl: syncDBs, checkup: Ran{Status: 2}} + db.set("glibc 2.42-1\nzsh 5.9-1\n", "zsh\n") + base := db.runner() + run := func(c context.Context, name string, args ...string) Ran { + line := strings.TrimSpace(name + " " + strings.Join(args, " ")) + switch { + case strings.HasPrefix(line, "systemctl list-units --state=failed"), line == "dkms status": + return Ran{} + case strings.HasPrefix(line, "systemctl show "+unitPrefix): + return Ran{Stdout: "LoadState=not-found\n"} + } + return base(c, name, args...) + } + m := machine(run, 1000) + began := time.Now() + m.Now = func() time.Time { return t0.Add(time.Since(began)) } + m.ReadFile = func(path string) ([]byte, error) { + if path == OSRelease { + return []byte("7.2.8-arch1-2\n"), nil + } + return nil, errNoFile + } + n := &roundNode{root: &countingRoot{}, signals: make(chan Signal)} + k := &Keeper{Config: KeeperConfig{UpgradeOrder: order, RecordDays: 180, RoundStartMinutes: 120}, Machine: m, Clock: realClock{}, + Watcher: fakeWatcher{n.signals}, Logf: func(f string, a ...any) { t.Logf(node+": "+f, a...) }, MakeHooks: keeperHooks, + Root: n.root, Check: func(Run, *Record) Check { return Check{Passed: passes} }} + account := process.Account{URL: url, Node: node, Module: "pacman", User: node + ".pacman.keeper", Password: "pw"} + go func() { _ = runKeeper(ctx, account, k) }() + return n +} + +func roundBus(t *testing.T) (string, map[string]jetstream.KeyValue) { + t.Helper() + url := newTestBus(t) + nc, err := nats.Connect(url) + if err != nil { + t.Fatal(err) + } + t.Cleanup(nc.Close) + js, err := jetstream.New(nc) + if err != nil { + t.Fatal(err) + } + ctx := context.Background() + buckets := map[string]jetstream.KeyValue{} + for _, name := range States { + kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "pacman_" + name, History: 8}) + if err != nil { + t.Fatal(err) + } + buckets[name] = kv + } + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "pacman-events", Subjects: []string{"mesh.mod.pacman.event.>"}}); err != nil { + t.Fatal(err) + } + return url, buckets +} + +func eventually(t *testing.T, what string, within time.Duration, ok func() bool) { + t.Helper() + for deadline := time.Now().Add(within); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { + if ok() { + return + } + } + t.Fatalf("never: %s", what) +} + +func roundNodeOnBus(kv jetstream.KeyValue, round, node string) RoundNode { + var rn RoundNode + if e, err := kv.Get(context.Background(), RoundNodeKey(round, node)); err == nil { + _ = json.Unmarshal(e.Value(), &rn) + } + return rn +} + +func TestTheRoundGoesNodeByNodeOnARealBus(t *testing.T) { + url, buckets := roundBus(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + order := []string{"node-a", "node-b"} + a := startRoundNode(t, ctx, url, "node-a", order, true) + b := startRoundNode(t, ctx, url, "node-b", order, true) + round := RoundID(t0) + + eventually(t, "node-a opened the round and started its run", 20*time.Second, func() bool { + return a.root.count() == 1 && roundNodeOnBus(buckets[StateRounds], round, "node-a").Run != "" + }) + time.Sleep(time.Second) + if b.root.count() != 0 { + t.Fatalf("node-b started before node-a passed: %v", b.root.started) + } + // node-a's transaction ends: its lock is released, its run is finished and checked. + a.signals <- LockReleased + eventually(t, "node-b started once node-a passed", 20*time.Second, func() bool { + return b.root.count() == 1 && roundNodeOnBus(buckets[StateRounds], round, "node-b").Run != "" + }) + if a.root.count() != 1 { + t.Errorf("node-a ran again: %v", a.root.started) + } +} + +func TestAFailedCheckStopsTheRoundOnARealBus(t *testing.T) { + url, buckets := roundBus(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + order := []string{"node-a", "node-b"} + a := startRoundNode(t, ctx, url, "node-a", order, false) + b := startRoundNode(t, ctx, url, "node-b", order, true) + round := RoundID(t0) + eventually(t, "node-a started its run", 20*time.Second, func() bool { return a.root.count() == 1 }) + a.signals <- LockReleased + runA := "" + eventually(t, "node-a's run checked", 20*time.Second, func() bool { + runA = roundNodeOnBus(buckets[StateRounds], round, "node-a").Run + if runA == "" { + return false + } + e, err := buckets[StateRuns].Get(context.Background(), NodeKey("node-a", runA)) + return err == nil && strings.Contains(string(e.Value()), `"checked-after"`) + }) + time.Sleep(2 * time.Second) + if b.root.count() != 0 || roundNodeOnBus(buckets[StateRounds], round, "node-b").Run != "" { + t.Fatalf("node-b started after node-a's check failed: %v", b.root.started) + } +}