diff --git a/modules/pacman/cmd/pacman-tools/keeper_integration_test.go b/modules/pacman/cmd/pacman-tools/keeper_integration_test.go new file mode 100644 index 00000000..f65d3ca8 --- /dev/null +++ b/modules/pacman/cmd/pacman-tools/keeper_integration_test.go @@ -0,0 +1,216 @@ +//go:build integration + +package main + +// The keeper whole over a real bus (novox/hq ADR 0507 §2, §3, §6; acceptance 1840/AC10; the shape of brother's +// device_integration_test.go): a JetStream server of the test's own, the module's state buckets and its event +// stream, and a fake machine. At its start and after a released lock the node's key and package.changed are on +// the bus, and pacman_packages on another instance reads them from there. Compiled by the merge gate's +// `go vet -tags integration`; run by the merge check, never by hand. + +import ( + "context" + "encoding/json" + "errors" + "strings" + "testing" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/process" + "github.com/nats-io/nats-server/v2/server" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" +) + +func newTestBus(t *testing.T) string { + t.Helper() + opts := &server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, JetStream: true, StoreDir: t.TempDir(), + NoLog: true, NoSigs: true} + s, err := server.NewServer(opts) + if err != nil { + t.Fatalf("a bus for this test could not be made: %v", err) + } + go s.Start() + if !s.ReadyForConnections(30 * time.Second) { + s.Shutdown() + t.Fatal("a bus for this test did not come up within 30s") + } + t.Cleanup(func() { s.Shutdown(); s.WaitForShutdown() }) + return s.ClientURL() +} + +// kvReader reads a bucket as a tools bundle's state does (stdio.State): what another instance's verbs see. +type kvReader struct{ kv jetstream.KeyValue } + +func (r kvReader) Get(key string) (*stdio.StateEntry, error) { + e, err := r.kv.Get(context.Background(), key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return nil, nil + } + if err != nil { + return nil, err + } + return &stdio.StateEntry{Key: key, Value: json.RawMessage(e.Value()), Revision: e.Revision()}, nil +} + +func (r kvReader) Keys() ([]string, error) { + keys, err := r.kv.Keys(context.Background()) + if errors.Is(err, jetstream.ErrNoKeysFound) { + return []string{}, nil + } + return keys, err +} + +func (r kvReader) History(key string) ([]stdio.StateRevision, error) { + h, err := r.kv.History(context.Background(), key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return []stdio.StateRevision{}, nil + } + if err != nil { + return nil, err + } + out := []stdio.StateRevision{} + for _, e := range h { + op := "put" + if e.Operation() != jetstream.KeyValuePut { + op = "delete" + } + out = append(out, stdio.StateRevision{Key: key, Value: e.Value(), Revision: e.Revision(), Op: op, At: e.Created()}) + } + return out, nil +} + +func TestTheKeeperPutsTheNodesPackagesOnARealBusAndAnotherInstanceReadsThem(t *testing.T) { + url := newTestBus(t) + ctx := context.Background() + nc, err := nats.Connect(url) + if err != nil { + t.Fatal(err) + } + defer nc.Close() + js, err := jetstream.New(nc) + if err != nil { + t.Fatal(err) + } + account := process.Account{URL: url, Node: "node-a", Module: "pacman", User: "node-a.pacman.keeper", Password: "pw"} + buckets := map[string]jetstream.KeyValue{} + for _, name := range States { + kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: account.Module + "_" + name, History: 8}) + if err != nil { + t.Fatal(err) + } + buckets[name] = kv + } + declared, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "mesh-controller_declared-packages"}) + if err != nil { + t.Fatal(err) + } + declared.Put(ctx, "node-a", []byte(`{"node":"node-a","packages":[{"package":"zsh","modules":["shell"],"absent":false}]}`)) + stream, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "pacman-events", Subjects: []string{"mesh.mod.pacman.event.>"}}) + if err != nil { + t.Fatal(err) + } + + db := &pacmanDB{sl: syncDBs, checkup: Ran{Stdout: "zsh 5.9-1 -> 5.9-2\n"}} + 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()} + runCtx, cancel := context.WithCancel(ctx) + done := make(chan error, 1) + go func() { done <- runKeeper(runCtx, account, k) }() + + waitKey := func(kv jetstream.KeyValue, key, holds string) string { + t.Helper() + for deadline := time.Now().Add(15 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { + if e, err := kv.Get(ctx, key); err == nil && strings.Contains(string(e.Value()), holds) { + return string(e.Value()) + } + } + t.Fatalf("%s never held %q", key, holds) + return "" + } + waitKey(buckets[StateInstalled], "node-a", `"zsh"`) + waitKey(buckets[StateUpdates], "node-a", `"reachable":true`) + + // a transaction: tmux added, zsh upgraded; the lock released. + db.set("glibc 2.42-1\ntmux 3.5-1\nzsh 5.9-2\n", "tmux\nzsh\n") + db.setUpdates(Ran{Status: 2}) + signals <- LockReleased + waitKey(buckets[StateInstalled], "node-a", `"tmux"`) + waitKey(buckets[StateUpdates], "node-a", `"packages":[]`) + + var changes []PackageChanged + for deadline := time.Now().Add(15 * time.Second); len(changes) < 2 && time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { + changes = nil + info, err := stream.Info(ctx) + if err != nil { + t.Fatal(err) + } + for seq := uint64(1); seq <= info.State.LastSeq; seq++ { + msg, err := stream.GetMsg(ctx, seq) + if err != nil || msg.Subject != "mesh.mod.pacman.event."+EventPackageChanged { + continue + } + var pc PackageChanged + if err := json.Unmarshal(msg.Data, &pc); err != nil { + t.Fatal(err) + } + changes = append(changes, pc) + } + } + if len(changes) != 2 { + t.Fatalf("package.changed at the start and after the transaction: %+v", changes) + } + last := changes[1] + if last.Node != "node-a" || len(last.Added) != 1 || last.Added[0].Name != "tmux" || len(last.Changed) != 1 || + last.Changed[0] != (VersionChange{"zsh", "5.9-1", "5.9-2"}) || len(last.Removed) != 0 { + t.Errorf("the transaction's package.changed: %+v", last) + } + + // another instance's verbs read what the keeper wrote, asking no machine. + other := machine(func(_ context.Context, name string, args ...string) Ran { + t.Errorf("another instance asked a machine: %s %v", name, args) + return Ran{Status: 99} + }, 1000) + other.Node = "node-b" + other.States = func(name string) StateReader { + if name == DeclaredPackagesState { + return kvReader{declared} + } + return kvReader{buckets[name]} + } + r, err := other.Packages("zsh", 500) + if err != nil { + t.Fatal(err) + } + on := r["nodes"].([]PackageOnNode) + if len(on) != 1 || on[0].Node != "node-a" || on[0].Version != "5.9-2" || strings.Join(on[0].DeclaredBy, ",") != "shell" { + t.Errorf("pacman_packages zsh from another instance: %+v", on) + } + r, err = other.Packages("", 500) + if err != nil { + t.Fatal(err) + } + if v := r["nodes"].([]NodeView); len(v) != 1 || v[0].Packages != 3 || len(v[0].Undeclared) != 1 || v[0].Undeclared[0].Name != "tmux" { + t.Errorf("the view from another instance: %+v", v) + } + u, err := other.UpdatesFor("node-a") + if err != nil { + t.Fatal(err) + } + if v := u["nodes"].([]UpdatesView); len(v) != 1 || !v[0].Reachable || v[0].Count != 0 || v[0].Changed == nil { + t.Errorf("pacman_updates from another instance: %+v", v) + } + + cancel() + select { + case err := <-done: + if !errors.Is(err, context.Canceled) { + t.Errorf("runKeeper ended with %v, want its context's end", err) + } + case <-time.After(10 * time.Second): + t.Fatal("runKeeper did not stop once its context was cancelled") + } +} diff --git a/modules/pacman/go.mod b/modules/pacman/go.mod index 1fac27ba..2d6f5744 100644 --- a/modules/pacman/go.mod +++ b/modules/pacman/go.mod @@ -2,11 +2,22 @@ module pacman go 1.26.0 -require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022121659-076e1f7947db +require ( + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022121659-076e1f7947db + github.com/nats-io/nats-server/v2 v2.11.17 +) + +require ( + github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect + github.com/google/go-tpm v0.9.8 // indirect + github.com/minio/highwayhash v1.0.4 // indirect + github.com/nats-io/jwt/v2 v2.8.1 // indirect + golang.org/x/time v0.15.0 // indirect +) require ( github.com/klauspost/compress v1.20.0 // indirect - github.com/nats-io/nats.go v1.54.0 // indirect + github.com/nats-io/nats.go v1.54.0 github.com/nats-io/nkeys v0.4.16 // indirect github.com/nats-io/nuid v1.0.1 // indirect golang.org/x/crypto v0.57.0 // indirect diff --git a/modules/pacman/go.sum b/modules/pacman/go.sum index 27e3d79b..050b4546 100644 --- a/modules/pacman/go.sum +++ b/modules/pacman/go.sum @@ -20,6 +20,7 @@ github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA= +golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=