package recordmirror import ( "encoding/json" "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" ) // 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. // 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 }, Holds: func() bool { return true }} } // **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. // **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. // **A value written behind the writer's back is put right on the next pass** (ADR 0289 §10): a former lease // holder's write differs from the store, and the store's value is written again, said as one change. // **A writer without the lease writes nothing** (ADR 0289 §9, design 54 §2): not a new key, not a change, not // a removal, and it says so. // **A removal asks for the lease too** (ADR 0289 §9): a writer that held the lease when its pass began and lost // it before its first removal purges nothing, and says it does not hold the lease. // **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) } }