Prove the compositions on a real bus and store: a done key composed and granted, a change rewriting only what reads it, a fresh start composing every node and deleting the plans' buckets once (issue 1945, AC2, AC3, AC6, AC14)

2026-10-24 17:19:07 +02:00
parent aaadb5695c
commit a8d63653fb
2 changed files with 318 additions and 1 deletions
@@ -0,0 +1,317 @@
+//go:build integration
+
+package main
+
+import (
+	"context"
+	"crypto/ed25519"
+	"encoding/json"
+	"errors"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+
+	"github.com/novox/mesh-controller/internal/broker"
+	"github.com/novox/mesh-controller/internal/identity"
+	"github.com/novox/mesh-controller/internal/link"
+	"github.com/novox/mesh-controller/internal/testbus"
+)
+
+// A mesh that composes (novox/hq ADR 0539): the real store (theCatalogue: sudo from the catalogue, mesh-host from the
+// node-engine's repository, the private network on anchor and laptop), the controller's signing key in its own
+// identity store, and a real bus server this process composes onto, its lease taken by the act as a command's is. Every
+// subject published on the bus is heard, so a test can say nothing went to a node's declaration subject.
+type composingMesh struct {
+	open  *stores
+	js    jetstream.JetStream
+	conn  *nats.Conn
+	mu    sync.Mutex
+	heard []string
+}
+
+func aComposingMesh(t *testing.T) *composingMesh {
+	t.Helper()
+	open := theCatalogue(t)
+	ident := identity.ForTest(t)
+	if _, err := ident.Establish(t.Context()); err != nil {
+		t.Fatal(err)
+	}
+	s := testbus.Start(t)
+	t.Setenv(broker.NATSVar, s.ClientURL())
+	was := theLease
+	theLease = &actor{}
+	t.Cleanup(func() { theLease.release(); theLease = was })
+	conn, err := nats.Connect(s.ClientURL())
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(conn.Close)
+	js, err := jetstream.New(conn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	m := &composingMesh{open: open, js: js, conn: conn}
+	sub, err := conn.Subscribe(">", func(msg *nats.Msg) {
+		m.mu.Lock()
+		m.heard = append(m.heard, msg.Subject)
+		m.mu.Unlock()
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = sub.Unsubscribe() })
+	legacy, err := conn.JetStream()
+	if err != nil {
+		t.Fatal(err)
+	}
+	// The controller's seat events, kept as the mesh's events stream keeps them.
+	if _, err := legacy.AddStream(&nats.StreamConfig{Name: "CONTROLLER_EVENTS", Subjects: []string{"mesh.seat.mesh-controller.event.>"}}); err != nil {
+		t.Fatal(err)
+	}
+	wasEvents := checkEvents
+	checkEvents = link.OverNATS{Conn: conn, JS: legacy}
+	t.Cleanup(func() { checkEvents = wasEvents })
+	withTrunkList(t, true)
+	return m
+}
+
+// noDeclarationSent fails the test if anything was published on a node's declaration subject.
+func (m *composingMesh) noDeclarationSent(t *testing.T) {
+	t.Helper()
+	_ = m.conn.Flush()
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	for _, s := range m.heard {
+		if strings.HasPrefix(s, "mesh.node.") && strings.HasSuffix(s, ".declare") {
+			t.Fatalf("published on %s: nothing is sent to a node", s)
+		}
+	}
+}
+
+// composed reads a composition and its revision from the node's bucket; revision 0 when there is none.
+func (m *composingMesh) composed(t *testing.T, node, key string) (sdkdelivery.Composed, uint64) {
+	t.Helper()
+	bucket, _ := sdkdelivery.ComposedBucketFor(node)
+	kv, err := m.js.KeyValue(t.Context(), bucket)
+	if errors.Is(err, jetstream.ErrBucketNotFound) {
+		return sdkdelivery.Composed{}, 0
+	}
+	if err != nil {
+		t.Fatal(err)
+	}
+	e, err := kv.Get(t.Context(), key)
+	if errors.Is(err, jetstream.ErrKeyNotFound) {
+		return sdkdelivery.Composed{}, 0
+	}
+	if err != nil {
+		t.Fatal(err)
+	}
+	c, err := sdkdelivery.DecodeComposed(e.Value())
+	if err != nil {
+		t.Fatal(err)
+	}
+	return c, e.Revision()
+}
+
+// keysOf are the keys of a node's composition bucket.
+func (m *composingMesh) keysOf(t *testing.T, node string) map[string]uint64 {
+	t.Helper()
+	bucket, _ := sdkdelivery.ComposedBucketFor(node)
+	kv, err := m.js.KeyValue(t.Context(), bucket)
+	if err != nil {
+		t.Fatal(err)
+	}
+	lister, err := kv.ListKeys(t.Context())
+	if err != nil {
+		t.Fatal(err)
+	}
+	out := map[string]uint64{}
+	for k := range lister.Keys() {
+		e, err := kv.Get(t.Context(), k)
+		if err != nil {
+			t.Fatal(err)
+		}
+		out[k] = e.Revision()
+	}
+	return out
+}
+
+// sudoAt takes sudo in from a done key of fp at commit and assigns it to anchor: what a merged change of it is.
+func (m *composingMesh) sudoAt(t *testing.T, fp, commit string) registerKeys {
+	t.Helper()
+	keys, trunk := aKeyFor("sudo", "2-"+commit[:4], fp, "novox/mesh-catalog", "modules/sudo", commit)
+	return trunkKeys{keyBuckets: keys, trunk: trunk}
+}
+
+// trunkKeys are a done key and the indexer's trunk value naming it.
+type trunkKeys struct {
+	keyBuckets
+	trunk sdkdelivery.Module
+}
+
+func (k trunkKeys) ModuleAt(context.Context, string, string) (sdkdelivery.Module, error) { return k.trunk, nil }
+
+// 1945/AC3 (and AC1, AC2's half for a new fingerprint): a done key of a new fingerprint of an assigned module is taken
+// in, its composition is written on the node it is assigned to, signed by the controller's key, with its digest beside
+// it, and `grants.<module>.<fp>` is written: both before any want can name it. Nothing is published on a node's
+// declaration subject.
+func TestADoneKeyOfAnAssignedModuleIsComposedAndGranted(t *testing.T) {
+	m := aComposingMesh(t)
+	ctx := t.Context()
+	if _, err := m.open.inventory.Assign(ctx, "anchor", "sudo"); err != nil {
+		t.Fatal(err)
+	}
+	fp := fpOf('a')
+	said, err := liveDoneKeySteps(m.open, m.sudoAt(t, fp, "c0ffee0123456789")).take(ctx, "sudo")
+	if err != nil {
+		t.Fatalf("the done key's take-in: %v", err)
+	}
+	if !strings.Contains(said, "registered from the builder's done key") {
+		t.Errorf("said %q", said)
+	}
+	key, _ := sdkdelivery.ComposedKey("anchor", "sudo", fp)
+	c, rev := m.composed(t, "anchor", key)
+	if rev == 0 || c.FP != fp || len(c.Resources) == 0 {
+		t.Fatalf("sudo's composition on anchor at %s: revision %d, %+v", fp, rev, c)
+	}
+	ident, _ := m.open.Identity(ctx)
+	active, _ := ident.Active(ctx)
+	digest, err := sdkdelivery.VerifyComposed(c, []ed25519.PublicKey{ed25519.PublicKey(active.Public)})
+	if err != nil {
+		t.Fatalf("the composition is not signed by the controller's key: %v", err)
+	}
+	digests, err := m.js.KeyValue(ctx, sdkdelivery.ControllerComposedBucket)
+	if err != nil {
+		t.Fatal(err)
+	}
+	e, err := digests.Get(ctx, key)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if v, err := sdkdelivery.DecodeComposedDigest(e.Value()); err != nil || v.Digest != digest {
+		t.Errorf("the digest beside it is %+v (%v), want %s", v, err, digest)
+	}
+	grants, err := m.js.KeyValue(ctx, broker.GrantsBucket)
+	if err != nil {
+		t.Fatal(err)
+	}
+	grantKey, _ := sdkdelivery.ControllerGrantKey("sudo", fp)
+	if _, err := grants.Get(ctx, grantKey); err != nil {
+		t.Errorf("grants.%s is not written: %v", grantKey, err)
+	}
+	m.noDeclarationSent(t)
+}
+
+// 1945/AC2: an input one module's composition reads changes, and that act writes a new revision of that module's
+// composition alone: every other composition of the node keeps its revision, and nothing is sent. The input here is
+// the module's newest done key, which only its own composition reads; a rotated secret reaches composeModules the
+// same way (rotate.go), through the same writer.
+func TestAChangeWritesANewRevisionOfOnlyTheCompositionsThatReadIt(t *testing.T) {
+	m := aComposingMesh(t)
+	ctx := t.Context()
+	if _, err := m.open.inventory.Assign(ctx, "anchor", "sudo"); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := liveDoneKeySteps(m.open, m.sudoAt(t, fpOf('a'), "c0ffee0123456789")).take(ctx, "sudo"); err != nil {
+		t.Fatal(err)
+	}
+	before := m.keysOf(t, "anchor")
+	if _, err := composeModules(ctx, m.open, []string{"anchor"}); err != nil {
+		t.Fatal(err)
+	}
+	if again := m.keysOf(t, "anchor"); !sameRevisions(before, again) {
+		t.Fatalf("composing again with nothing changed wrote %v (was %v)", again, before)
+	}
+	if _, err := liveDoneKeySteps(m.open, m.sudoAt(t, fpOf('b'), "d00d0123456789ab")).take(ctx, "sudo"); err != nil {
+		t.Fatal(err)
+	}
+	after := m.keysOf(t, "anchor")
+	newKey, _ := sdkdelivery.ComposedKey("anchor", "sudo", fpOf('b'))
+	for k, rev := range after {
+		if k == newKey {
+			continue
+		}
+		if before[k] != rev {
+			t.Errorf("%s moved from revision %d to %d: only sudo's composition reads its new build", k, before[k], rev)
+		}
+	}
+	if after[newKey] == 0 {
+		t.Errorf("sudo's new composition is not written: %v", after)
+	}
+	m.noDeclarationSent(t)
+}
+
+func sameRevisions(a, b map[string]uint64) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	for k, v := range a {
+		if b[k] != v {
+			return false
+		}
+	}
+	return true
+}
+
+// 1945/AC6 and AC14: after a fresh start every node has its own composition and every assigned module its composition;
+// the first start deletes the merge window's, the silent sends' and the building plans' buckets and says each once as
+// the controller's event bucket-deleted; a later start says nothing.
+func TestAFreshStartComposesEveryNodeAndDeletesThePlansBucketsOnce(t *testing.T) {
+	m := aComposingMesh(t)
+	ctx := t.Context()
+	for _, b := range broker.PlanStateBuckets {
+		if _, err := m.js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: b}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	var said []BucketDeleted
+	var mu sync.Mutex
+	sub, err := m.conn.Subscribe(link.SeatEventSubject(link.MeshControllerSeat, "bucket-deleted"), func(msg *nats.Msg) {
+		var d BucketDeleted
+		if json.Unmarshal(msg.Data, &d) == nil {
+			mu.Lock()
+			said = append(said, d)
+			mu.Unlock()
+		}
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer sub.Unsubscribe()
+
+	composeAtStart(ctx, m.open)
+	for _, node := range []string{"anchor", "laptop"} {
+		own, _ := sdkdelivery.ComposedNodeKey(node)
+		if _, rev := m.composed(t, node, own); rev == 0 {
+			t.Errorf("%s has no composition of its own after the start", node)
+		}
+	}
+	for _, b := range broker.PlanStateBuckets {
+		if _, err := m.js.KeyValue(ctx, b); !errors.Is(err, jetstream.ErrBucketNotFound) {
+			t.Errorf("%s is still there after the first start: %v", b, err)
+		}
+	}
+	_ = m.conn.Flush()
+	time.Sleep(50 * time.Millisecond)
+	mu.Lock()
+	if len(said) != len(broker.PlanStateBuckets) {
+		t.Errorf("the first start said %d deletions, want %d: %+v", len(said), len(broker.PlanStateBuckets), said)
+	}
+	said = nil
+	mu.Unlock()
+
+	composeAtStart(ctx, m.open)
+	_ = m.conn.Flush()
+	time.Sleep(50 * time.Millisecond)
+	mu.Lock()
+	defer mu.Unlock()
+	if len(said) != 0 {
+		t.Errorf("a later start said %+v", said)
+	}
+	m.noDeclarationSent(t)
+}
@@ -137,7 +137,7 @@ func takeDoneKey(ctx context.Context, open *stores, module string) {
 }

 // liveDoneKeySteps are the steps on the controller's own stores and the bus.
-func liveDoneKeySteps(open *stores, keys builderkeys.Reader) doneKeySteps {
+func liveDoneKeySteps(open *stores, keys registerKeys) doneKeySteps {
 	inv := open.inventory
 	return doneKeySteps{
 		held: func(ctx context.Context, module string) (bool, error) {