From ea5cc6fc3536d10e1730900d87f04070f05a72b8 Mon Sep 17 00:00:00 2001 From: mesh-admin Date: Wed, 14 Oct 2026 00:32:29 +0000 Subject: [PATCH] Prove a record mirror against a real bus and the store An assignment and a registration each change exactly one key and raise exactly one event; the same records written again change nothing (hq issue 652, the package's done-when). Listed as a store test, so the merge check runs it. Co-Authored-By: Claude Opus 5.5 --- internal/recordmirror/mirror_store_test.go | 158 +++++++++++++ internal/recordmirror/mirror_test.go | 261 +++++++++++++++++++++ 2 files changed, 419 insertions(+) create mode 100644 internal/recordmirror/mirror_store_test.go create mode 100644 internal/recordmirror/mirror_test.go diff --git a/internal/recordmirror/mirror_store_test.go b/internal/recordmirror/mirror_store_test.go new file mode 100644 index 00000000..c0601a7c --- /dev/null +++ b/internal/recordmirror/mirror_store_test.go @@ -0,0 +1,158 @@ +package recordmirror + +import ( + "encoding/json" + "strings" + "testing" + + "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/catalogue" + "github.com/novox/mesh-controller/internal/inventory" +) + +// **Design 57's F3, end to end** (novox/hq issue 652): from the controller's store to a real bus. An +// assignment changes exactly one key of the assignments mirror and raises exactly one `assignment-changed`; +// a registration changes exactly one key of the declared-config mirror and raises exactly one +// `declared-config-changed`; and the same records written again change nothing and say nothing — a module +// registered again unchanged included, although the store's registration time moved. +func TestNatsAnAssignmentAndARegistrationEachChangeOneKeyAndRaiseOneEvent(t *testing.T) { + inv := inventory.ForTest(t) + ctx := t.Context() + js, api := aMirrorBus(t) + assignments := aWriter(t, js, api, Assignments) + declared := aWriter(t, js, api, DeclaredConfig) + heardAssignments := listen(t, js.Conn(), "mesh.seat."+broker.ControllerSeat+".event."+Assignments.Event) + heardDeclared := listen(t, js.Conn(), "mesh.seat."+broker.ControllerSeat+".event."+DeclaredConfig.Event) + + pass := func() (Result, Result) { + t.Helper() + a, err := inv.AssignmentsMirror(ctx) + if err != nil { + t.Fatal(err) + } + ra, err := assignments.Sync(ctx, a) + if err != nil { + t.Fatal(err) + } + d, err := inv.DeclaredConfigMirror(ctx) + if err != nil { + t.Fatal(err) + } + rd, err := declared.Sync(ctx, d) + if err != nil { + t.Fatal(err) + } + return ra, rd + } + + if _, err := inv.AddNode(ctx, "anchor"); err != nil { + t.Fatal(err) + } + ui := catalogue.Manifest{Module: "ui", Version: "1", Settings: map[string]catalogue.SettingDeclaration{ + "font-size": {Kind: catalogue.KindPreference, Default: 14, Why: "readable on the desktop"}}} + if err := inv.RegisterModule(ctx, ui, inventory.Source{}); err != nil { + t.Fatal(err) + } + pass() + assignmentWrites, declaredWrites := writes(t, api, Assignments.Bucket), writes(t, api, DeclaredConfig.Bucket) + assignmentsSaid, declaredSaid := len(settle(heardAssignments, 0)), len(settle(heardDeclared, 1)) + + // **A registration**: a module new to the catalogue. + if err := inv.RegisterModule(ctx, catalogue.Manifest{Module: "notes", Version: "1"}, inventory.Source{}); err != nil { + t.Fatal(err) + } + ra, rd := pass() + if ra != (Result{}) || rd != (Result{Written: 1}) { + t.Errorf("a registration wrote assignments %+v and declared config %+v", ra, rd) + } + if got := writes(t, api, DeclaredConfig.Bucket) - declaredWrites; got != 1 { + t.Errorf("a registration changed %d keys of the declared config", got) + } + if got := writes(t, api, Assignments.Bucket) - assignmentWrites; got != 0 { + t.Errorf("a registration changed %d keys of the assignments", got) + } + said := settle(heardDeclared, declaredSaid+1)[declaredSaid:] + if len(said) != 1 || said[0].Key != "notes" || said[0].Change != Added { + t.Errorf("a registration raised %+v", said) + } + declaredWrites, declaredSaid = writes(t, api, DeclaredConfig.Bucket), declaredSaid+len(said) + + // **An assignment.** + if _, err := inv.Assign(ctx, "anchor", "ui"); err != nil { + t.Fatal(err) + } + ra, rd = pass() + if ra != (Result{Written: 1}) || rd != (Result{}) { + t.Errorf("an assignment wrote assignments %+v and declared config %+v", ra, rd) + } + if got := writes(t, api, Assignments.Bucket) - assignmentWrites; got != 1 { + t.Errorf("an assignment changed %d keys of the assignments", got) + } + said = settle(heardAssignments, assignmentsSaid+1)[assignmentsSaid:] + if len(said) != 1 || said[0].Key != "anchor.ui" || said[0].Change != Added { + t.Errorf("an assignment raised %+v", said) + } + assignmentWrites, assignmentsSaid = writes(t, api, Assignments.Bucket), assignmentsSaid+len(said) + e, err := assignments.KV.Get(ctx, "anchor.ui") + if err != nil { + t.Fatal(err) + } + var value inventory.AssignmentMirrored + if err := json.Unmarshal(e.Value(), &value); err != nil || value.Node != "anchor" || value.Module != "ui" || + value.Assigned.IsZero() { + t.Errorf("the assignment is mirrored as %s (%v)", e.Value(), err) + } + + // **The same records written again change nothing**: an assignment made again, and a module registered + // again with the manifest it had. + if _, err := inv.Assign(ctx, "anchor", "ui"); err != nil { + t.Fatal(err) + } + if err := inv.RegisterModule(ctx, ui, inventory.Source{}); err != nil { + t.Fatal(err) + } + if ra, rd := pass(); ra != (Result{}) || rd != (Result{}) { + t.Errorf("unchanged records wrote assignments %+v and declared config %+v", ra, rd) + } + if writes(t, api, Assignments.Bucket) != assignmentWrites || writes(t, api, DeclaredConfig.Bucket) != declaredWrites { + t.Error("unchanged records changed a key") + } + if a, d := settle(heardAssignments, assignmentsSaid+1), settle(heardDeclared, declaredSaid+1); len(a) != assignmentsSaid || + len(d) != declaredSaid { + t.Errorf("unchanged records raised %+v and %+v", a[assignmentsSaid:], d[declaredSaid:]) + } + + // **A declaration changed by a registration**: one key, one event, naming the field that moved. + ui.Version = "2" + ui.Settings["font-size"] = catalogue.SettingDeclaration{Kind: catalogue.KindPreference, Default: 16, + Why: "readable on the desktop"} + if err := inv.RegisterModule(ctx, ui, inventory.Source{}); err != nil { + t.Fatal(err) + } + if ra, rd := pass(); ra != (Result{}) || rd != (Result{Written: 1}) { + t.Errorf("a changed declaration wrote assignments %+v and declared config %+v", ra, rd) + } + said = settle(heardDeclared, declaredSaid+1)[declaredSaid:] + if len(said) != 1 || said[0].Key != "ui" || said[0].Change != Changed || strings.Join(said[0].Fields, ",") != "settings" { + t.Errorf("a changed declaration raised %+v", said) + } + got, err := declared.KV.Get(ctx, "ui") + if err != nil { + t.Fatal(err) + } + if want := `{"module":"ui","settings":{"font-size":{"default":16,"kind":"preference","why":"readable on the desktop"}}}`; string(got.Value()) != want { + t.Errorf("the declaration is mirrored as %s, want %s", got.Value(), want) + } + + // **An unassignment** removes its key, said once. + if err := inv.Unassign(ctx, "anchor", "ui"); err != nil { + t.Fatal(err) + } + if ra, _ := pass(); ra != (Result{Removed: 1}) { + t.Errorf("an unassignment did %+v", ra) + } + said = settle(heardAssignments, assignmentsSaid+1)[assignmentsSaid:] + if len(said) != 1 || said[0].Key != "anchor.ui" || said[0].Change != Removed { + t.Errorf("an unassignment raised %+v", said) + } +} diff --git a/internal/recordmirror/mirror_test.go b/internal/recordmirror/mirror_test.go new file mode 100644 index 00000000..41f0c2ac --- /dev/null +++ b/internal/recordmirror/mirror_test.go @@ -0,0 +1,261 @@ +package recordmirror + +import ( + "encoding/json" + "errors" + "slices" + "sync" + "testing" + "time" + + "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/link" + "github.com/novox/mesh-controller/internal/testbus" +) + +// A record mirror against a real bus (novox/hq design 54 §2, §3; ADR 0289): a change of a record changes +// exactly one key and raises exactly one event; a pass over unchanged records writes nothing and says +// nothing; a record gone is a purge said as `removed`. + +// aMirrorBus is a bus of this test alone with the mesh's streams and the controller's buckets, as the +// controller asserts them at its start, and a counter of the writes made to one bucket. +func aMirrorBus(t *testing.T) (*broker.JetStream, jetstream.JetStream) { + t.Helper() + js, err := broker.Dial(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(js.Close) + if err := broker.AssertMeshStreams(js); err != nil { + t.Fatal(err) + } + if err := js.EnsureControllerBuckets(); err != nil { + t.Fatal(err) + } + api, err := jetstream.New(js.Conn()) + if err != nil { + t.Fatal(err) + } + return js, api +} + +// heard counts what is said on one subject, each body kept. +type heard struct { + mu sync.Mutex + bodies []Event +} + +func (h *heard) all() []Event { + h.mu.Lock() + defer h.mu.Unlock() + return append([]Event(nil), h.bodies...) +} + +func listen(t *testing.T, conn *nats.Conn, subject string) *heard { + t.Helper() + h := &heard{} + sub, err := conn.Subscribe(subject, func(m *nats.Msg) { + var e Event + if err := json.Unmarshal(m.Data, &e); err != nil { + t.Errorf("an event that is not one: %s", m.Data) + return + } + h.mu.Lock() + h.bodies = append(h.bodies, e) + h.mu.Unlock() + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = sub.Unsubscribe() }) + if err := conn.Flush(); err != nil { + t.Fatal(err) + } + return h +} + +// writes is how many messages the bucket's stream has taken: every put and every purge is one. +func writes(t *testing.T, api jetstream.JetStream, bucket string) uint64 { + t.Helper() + s, err := api.Stream(t.Context(), "KV_"+bucket) + if err != nil { + t.Fatal(err) + } + info, err := s.Info(t.Context()) + if err != nil { + t.Fatal(err) + } + return info.State.LastSeq +} + +// settle waits until n events are heard or a second passed, and answers what was heard. +func settle(h *heard, n int) []Event { + deadline := time.Now().Add(time.Second) + for time.Now().Before(deadline) && len(h.all()) < n { + time.Sleep(10 * time.Millisecond) + } + // And a moment more, so an event too many is heard too. + time.Sleep(100 * time.Millisecond) + return h.all() +} + +func aWriter(t *testing.T, js *broker.JetStream, api jetstream.JetStream, m Mirror) *Writer { + t.Helper() + kv, err := api.KeyValue(t.Context(), m.Bucket) + if err != nil { + t.Fatal(err) + } + return &Writer{Mirror: m, KV: kv, Announce: link.OverNATS{Conn: js.Conn(), JS: js.Context()}, + Epoch: func() uint64 { return 7 }} +} + +func TestNatsAMirrorChangesOneKeyAndRaisesOneEventPerChange(t *testing.T) { + for _, m := range []Mirror{Assignments, DeclaredConfig} { + t.Run(m.Name, func(t *testing.T) { + js, api := aMirrorBus(t) + events := listen(t, js.Conn(), "mesh.seat."+broker.ControllerSeat+".event."+m.Event) + w := aWriter(t, js, api, m) + ctx := t.Context() + + first := map[string][]byte{"anchor.gitea": []byte(`{"a":1}`)} + if _, err := w.Sync(ctx, first); err != nil { + t.Fatal(err) + } + before := writes(t, api, m.Bucket) + heardBefore := len(settle(events, 1)) + + // One record changes, one is added: two keys, two events, nothing else. + second := map[string][]byte{"anchor.gitea": []byte(`{"a":2}`), "anchor.forgejo": []byte(`{"a":1}`)} + r, err := w.Sync(ctx, second) + if err != nil { + t.Fatal(err) + } + if r.Written != 2 || r.Removed != 0 { + t.Errorf("a pass over one changed and one new record wrote %+v", r) + } + if got := writes(t, api, m.Bucket) - before; got != 2 { + t.Errorf("the bucket took %d writes for two changed records", got) + } + said := settle(events, heardBefore+2)[heardBefore:] + if len(said) != 2 { + t.Fatalf("two changes raised %d events: %+v", len(said), said) + } + changes := map[string]Event{} + for _, e := range said { + changes[e.Key] = e + } + if e := changes["anchor.gitea"]; e.Change != Changed || !slices.Equal(e.Fields, []string{"a"}) || + e.Mirror != m.Name || e.Epoch != 7 || e.Revision == 0 { + t.Errorf("the change of anchor.gitea was said as %+v", e) + } + if e := changes["anchor.forgejo"]; e.Change != Added { + t.Errorf("the new anchor.forgejo was said as %+v", e) + } + + // **An unchanged pass writes nothing**, and says nothing (ADR 0289 §1). + heardBefore = len(events.all()) + before = writes(t, api, m.Bucket) + r, err = w.Sync(ctx, second) + if err != nil { + t.Fatal(err) + } + if r != (Result{}) { + t.Errorf("a pass over unchanged records did %+v", r) + } + if got := writes(t, api, m.Bucket) - before; got != 0 { + t.Errorf("a pass over unchanged records wrote the bucket %d times", got) + } + if said := settle(events, heardBefore+1)[heardBefore:]; len(said) != 0 { + t.Errorf("a pass over unchanged records raised %+v", said) + } + + // **A record gone is a purge**, one write and one `removed`. + before = writes(t, api, m.Bucket) + r, err = w.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":2}`)}) + if err != nil { + t.Fatal(err) + } + if r.Removed != 1 || r.Written != 0 { + t.Errorf("a record gone did %+v", r) + } + if got := writes(t, api, m.Bucket) - before; got != 1 { + t.Errorf("a record gone wrote the bucket %d times", got) + } + if _, err := w.KV.Get(ctx, "anchor.forgejo"); !errors.Is(err, jetstream.ErrKeyNotFound) { + t.Errorf("the key of a record gone is still read: %v", err) + } + said = settle(events, heardBefore+1)[heardBefore:] + if len(said) != 1 || said[0].Change != Removed || said[0].Key != "anchor.forgejo" { + t.Errorf("a record gone was said as %+v", said) + } + }) + } +} + +// **A controller started again writes nothing for what the bucket already holds** (design 54 §2: the writer's +// memory is filled from the bucket), and a key nobody holds a record for is removed. +func TestNatsAWriterStartedAgainWritesOnlyWhatDiffers(t *testing.T) { + js, api := aMirrorBus(t) + ctx := t.Context() + first := aWriter(t, js, api, Assignments) + held := map[string][]byte{"anchor.gitea": []byte(`{"a":1}`), "anchor.forgejo": []byte(`{"a":1}`)} + if _, err := first.Sync(ctx, held); err != nil { + t.Fatal(err) + } + before := writes(t, api, Assignments.Bucket) + again := aWriter(t, js, api, Assignments) + r, err := again.Sync(ctx, held) + if err != nil { + t.Fatal(err) + } + if r != (Result{}) || writes(t, api, Assignments.Bucket) != before { + t.Errorf("a writer started again over unchanged records did %+v", r) + } +} + +// **Another writer in between is no lost value** (design 54 §2, compare-and-set): a write refused because the +// key moved reads it again, takes the revision when it already holds the value, and otherwise writes again. +func TestNatsAWriteRefusedByCompareAndSetReadsAgain(t *testing.T) { + js, api := aMirrorBus(t) + ctx := t.Context() + w := aWriter(t, js, api, Assignments) + if _, err := w.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":1}`)}); err != nil { + t.Fatal(err) + } + // Behind its back, as a former lease holder would. + if _, err := w.KV.Put(ctx, "anchor.gitea", []byte(`{"a":9}`)); err != nil { + t.Fatal(err) + } + r, err := w.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":2}`)}) + if err != nil { + t.Fatal(err) + } + if r.Written != 1 { + t.Errorf("a refused write was not made again: %+v", r) + } + e, err := w.KV.Get(ctx, "anchor.gitea") + if err != nil || string(e.Value()) != `{"a":2}` { + t.Fatalf("the bucket holds %v, %v", e, err) + } +} + +// **A key the bus cannot hold is refused alone**, and the others are written. +func TestAKeyTheBusCannotHoldIsRefusedAlone(t *testing.T) { + if validKey("anchor.gitea") != nil || validKey("bad key") == nil || validKey(".lead") == nil || + validKey("") == nil || validKey("trail.") == nil { + t.Error("the keys a bucket may hold are not told apart") + } +} + +func TestTheFieldsThatChangedAreNamed(t *testing.T) { + got := changedFields([]byte(`{"a":1,"b":{"x":1},"c":"same"}`), []byte(`{"a":2,"b":{"x":1},"d":true,"c":"same"}`)) + if !slices.Equal(got, []string{"a", "d"}) { + t.Errorf("the fields that changed are %v", got) + } + if got := changedFields(nil, []byte(`{"a":1}`)); got != nil { + t.Errorf("an added key names fields %v", got) + } +}