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 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user