mesh-controller/internal/recordmirror/mirror_test.go

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 }, Holds: func() bool { return true }}
}

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 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.
func TestNatsAForeignWriteIsCorrectedOnTheNextPass(t *testing.T) {
	js, api := aMirrorBus(t)
	ctx := t.Context()
	events := listen(t, js.Conn(), "mesh.seat."+broker.ControllerSeat+".event."+Assignments.Event)
	w := aWriter(t, js, api, Assignments)
	store := map[string][]byte{"anchor.gitea": []byte(`{"a":1}`)}
	if _, err := w.Sync(ctx, store); err != nil {
		t.Fatal(err)
	}
	heardBefore := len(settle(events, 1))
	if _, err := w.KV.Put(ctx, "anchor.gitea", []byte(`{"a":9}`)); err != nil {
		t.Fatal(err)
	}
	r, err := w.Sync(ctx, store)
	if err != nil {
		t.Fatal(err)
	}
	if r != (Result{Written: 1}) {
		t.Errorf("a pass after a foreign write did %+v", r)
	}
	e, err := w.KV.Get(ctx, "anchor.gitea")
	if err != nil || string(e.Value()) != `{"a":1}` {
		t.Fatalf("the bucket still holds %v, %v", e, err)
	}
	if said := settle(events, heardBefore+1)[heardBefore:]; len(said) != 1 || said[0].Change != Changed {
		t.Errorf("the correction was said as %+v", said)
	}
	// And put right, it stays: the next pass writes nothing.
	if r, err := w.Sync(ctx, store); err != nil || r != (Result{}) {
		t.Errorf("a pass after the correction did %+v, %v", r, err)
	}
}

// **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.
func TestNatsAWriterWithoutTheLeaseWritesNothing(t *testing.T) {
	js, api := aMirrorBus(t)
	ctx := t.Context()
	held := aWriter(t, js, api, Assignments)
	if _, err := held.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":1}`)}); err != nil {
		t.Fatal(err)
	}
	before := writes(t, api, Assignments.Bucket)
	for _, holds := range []func() bool{nil, func() bool { return false }} {
		w := aWriter(t, js, api, Assignments)
		w.Holds = holds
		r, err := w.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":2}`), "anchor.forgejo": []byte(`{"a":1}`)})
		if !errors.Is(err, ErrNotHeld) || r != (Result{}) {
			t.Errorf("a writer without the lease did %+v, %v", r, err)
		}
		r, err = w.Sync(ctx, map[string][]byte{})
		if !errors.Is(err, ErrNotHeld) || r != (Result{}) {
			t.Errorf("a writer without the lease removed: %+v, %v", r, err)
		}
	}
	// And one that loses it mid-pass stops before its next write.
	asked := 0
	w := aWriter(t, js, api, Assignments)
	w.Holds = func() bool { asked++; return asked <= 2 }
	r, err := w.Sync(ctx, map[string][]byte{"anchor.a": []byte(`{}`), "anchor.b": []byte(`{}`), "anchor.gitea": []byte(`{"a":1}`)})
	if !errors.Is(err, ErrNotHeld) || r.Written != 1 {
		t.Errorf("a writer that lost the lease mid-pass did %+v, %v", r, err)
	}
	if got := writes(t, api, Assignments.Bucket) - before; got != 1 {
		t.Errorf("writers without the lease wrote the bucket %d times", got)
	}
}

// **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)
	}
}