issue 1841/S12: prove the round on a real bus with two made-up nodes, so the second node is seen to start only after the first passed and never after a failed check
This commit is contained in:
@@ -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
|
||||
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user