novox/mesh-controller
18
0
Fork 0
Code Issues Pull Requests 4 Releases Wiki Activity

Prove the compositions through serve() whole, on a bus running the composed user list: the start, a done key, a real rotation (issue 1945; the review's G1)

Browse Source 
The tests no longer call the writer on a bus without permissions: the controller runs as it serves, the builder and
the indexer write their own keys as their own accounts, and the operator's rotate is the real verb, so a subject the
user list does not grant fails them as it would fail the mesh.
This commit is contained in:
jschoubben
2026-10-24 18:41:23 +02:00
parent 49fc8b6d7d
commit af6bcea41e
2 changed files with 374 additions and 232 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/mesh-controller/composed_integration_test.go
-232
View File
@@ -4,13 +4,9 @@ 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"
@@ -80,74 +76,6 @@ func aComposingMesh(t *testing.T) *composingMesh {
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
@@ -158,166 +86,6 @@ func (k trunkKeys) ModuleAt(context.Context, string, string) (sdkdelivery.Module
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)
}
// 1945/AC14: the declarations' stream stays while one node's have names another node-engine than the one registered,
// and goes, said once, when every node's names it (ADR 0539 clause 11).
func TestTheDeclarationsStreamGoesOnceEveryNodeRunsTheNewEngine(t *testing.T) {