issue 1840/S10: run the keeper whole against a real bus, so its key, its event and another instance's view are proved together

Written and compiled under the integration tag; the merge check runs it.
2026-10-22 14:37:45 +02:00
parent 9651a8b2da
commit b7f3664839
3 changed files with 230 additions and 2 deletions
@@ -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")
+	}
+}
@@ -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
@@ -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=