From a8d63653fb19e52739953ec3a2137a5472aa2160 Mon Sep 17 00:00:00 2001 From: jochen Date: Sat, 24 Oct 2026 17:19:07 +0200 Subject: [PATCH] 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) --- .../composed_integration_test.go | 317 ++++++++++++++++++ cmd/mesh-controller/composed_keys.go | 2 +- 2 files changed, 318 insertions(+), 1 deletion(-) create mode 100644 cmd/mesh-controller/composed_integration_test.go diff --git a/cmd/mesh-controller/composed_integration_test.go b/cmd/mesh-controller/composed_integration_test.go new file mode 100644 index 00000000..ea84d260 --- /dev/null +++ b/cmd/mesh-controller/composed_integration_test.go @@ -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..` 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) +} diff --git a/cmd/mesh-controller/composed_keys.go b/cmd/mesh-controller/composed_keys.go index fb1f526f..3711ff8b 100644 --- a/cmd/mesh-controller/composed_keys.go +++ b/cmd/mesh-controller/composed_keys.go @@ -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) {