mesh-controller/internal/broker/streams_test.go

package broker

import (
	"errors"
	"strings"
	"testing"
)

type recorder struct {
	seen []Stream
	fail string
}

func (r *recorder) EnsureStream(s Stream) error {
	if s.Name == r.fail {
		return errors.New("refused")
	}
	r.seen = append(r.seen, s)
	return nil
}

// The controller asserts on every start, not only at genesis: a stream that was deleted, or a mesh
// raised from a backup, must converge rather than run without the guarantee its messages assume.
func TestAssertingTwiceIsTheSameAsOnce(t *testing.T) {
	a, b := &recorder{}, &recorder{}
	if err := AssertMeshStreams(a); err != nil {
		t.Fatal(err)
	}
	if err := AssertMeshStreams(a); err != nil {
		t.Fatal(err)
	}
	if err := AssertMeshStreams(b); err != nil {
		t.Fatal(err)
	}
	if len(a.seen) != 2*len(b.seen) {
		t.Fatalf("asserted %d then %d; assertion is not repeatable", len(a.seen), len(b.seen))
	}
}

func TestAFailedAssertionNamesItsStream(t *testing.T) {
	err := AssertMeshStreams(&recorder{fail: "NODES"})
	if err == nil || !strings.Contains(err.Error(), "NODES") {
		t.Fatalf("got %v, which does not say which stream failed", err)
	}
}

// Two streams matching one subject is accepted by NATS and stores the message twice under two
// retentions. Nothing reports that, so it is refused where the set is written.
func TestNoTwoStreamsClaimTheSameSubject(t *testing.T) {
	if clashes := Overlaps(); len(clashes) != 0 {
		t.Fatalf("overlapping subject filters: %v", clashes)
	}
}

// A heartbeat under mesh.control.> must not be persisted: a lost one is the next one, and a
// stream of them competes for retention with the messages that matter.
func TestHeartbeatsAreNotInTheControlStream(t *testing.T) {
	for _, s := range MeshStreams() {
		for _, subject := range s.Subjects {
			if subject == "mesh.control.>" || strings.Contains(subject, "alive") {
				t.Fatalf("stream %s claims %q, which captures heartbeats", s.Name, subject)
			}
		}
	}
}

// The reason the kind token exists: a filter over a module's whole namespace would persist every
// tool call in the mesh.
func TestTheEventsStreamDoesNotCaptureToolCalls(t *testing.T) {
	var events Stream
	for _, s := range MeshStreams() {
		if s.Name == "EVENTS" {
			events = s
		}
	}
	if len(events.Subjects) != 1 || events.Subjects[0] != "mesh.mod.*.event.>" {
		t.Fatalf("EVENTS filters on %v", events.Subjects)
	}
	// A tool subject the composer would actually produce must not match that filter.
	tool := "mesh.mod.billing.tool.status"
	if subjectMatches(events.Subjects[0], tool) {
		t.Fatalf("%q matches the events filter, so every tool call would be persisted", tool)
	}
	if !subjectMatches(events.Subjects[0], "mesh.mod.billing.event.order.placed") {
		t.Fatal("an event does not match the events filter")
	}
}

// subjectMatches is NATS subject matching, enough for these filters: `*` is one token, `>` is the
// rest.
func subjectMatches(filter, subject string) bool {
	f, s := strings.Split(filter, "."), strings.Split(subject, ".")
	for i, tok := range f {
		if tok == ">" {
			return i <= len(s)
		}
		if i >= len(s) {
			return false
		}
		if tok != "*" && tok != s[i] {
			return false
		}
	}
	return len(f) == len(s)
}

// Each relationship's retention is the thing that makes it what it is (design 29 §4).
func TestEachStreamCarriesTheRetentionItsShapeNeeds(t *testing.T) {
	want := map[string]Retention{
		"CONTROL": RetentionWorkQueue,
		"NODES":   RetentionLastPerSubject,
		"BUILDS":  RetentionWorkQueue,
		"EVENTS":  RetentionLimits,
	}
	got := map[string]Retention{}
	for _, s := range MeshStreams() {
		got[s.Name] = s.Retention
		if s.Why == "" {
			t.Errorf("stream %s says no reason it exists", s.Name)
		}
	}
	if len(got) != len(want) {
		t.Fatalf("the foundation set is %v", got)
	}
	for name, r := range want {
		if got[name] != r {
			t.Errorf("%s retains as %q, expected %q", name, got[name], r)
		}
	}
}