mesh-controller/internal/link/busfaults_test.go

package link

import (
	"encoding/json"
	"strings"
	"testing"
	"time"

	"github.com/nats-io/nats.go"

	"github.com/novox/mesh-controller/internal/broker"
	"github.com/novox/mesh-controller/internal/testbus"
)

// What the bus module says of its users' connections, read back as advisories (novox/hq ADR 0357, issue
// 270): one per user and kind, each closure counted once however often it is said.
//
//	go test ./internal/link/ -run TestNatsBusFaults   (each test on a bus of its own: internal/testbus)

func aBusWithEvents(t *testing.T) *Server {
	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)
	}
	return ConnectNats(js, nil, nil)
}

func sayClosure(t *testing.T, s *Server, event string, c BusClosure) {
	t.Helper()
	body, _ := json.Marshal(c)
	if _, err := s.js.Context().Publish(EventSubject(BusModule, event), body); err != nil {
		t.Fatal(err)
	}
}

func TestNatsBusFaultsAreOnePerUserAndKindEachClosureCountedOnce(t *testing.T) {
	s := aBusWithEvents(t)
	now := time.Now().UTC()
	slow := BusClosure{User: "laptop.claude-code", Name: "laptop.claude-code", CID: 10, Reason: "Slow Consumer (Write Deadline)",
		Opened: now.Add(-20 * time.Minute), Closed: now.Add(-2 * time.Minute), Pending: 334000}
	sayClosure(t, s, BusSlowConsumerEvent, slow)
	sayClosure(t, s, BusSlowConsumerEvent, slow) // said again: the module started again
	again := slow
	again.CID, again.Opened, again.Closed = 31, now.Add(-90*time.Second), now.Add(-time.Minute)
	sayClosure(t, s, BusSlowConsumerEvent, again)
	sayClosure(t, s, BusConnectionClosedEvent, BusClosure{User: "anchor.desk", CID: 7,
		Reason: "Maximum Message Payload Exceeded", Opened: now.Add(-time.Hour / 2), Closed: now.Add(-30 * time.Second)})
	// Older than the hour: S9 no longer says it.
	sayClosure(t, s, BusSlowConsumerEvent, BusClosure{User: "anchor.media", CID: 3, Reason: "Slow Consumer (Pending Bytes)",
		Opened: now.Add(-3 * time.Hour), Closed: now.Add(-2 * time.Hour)})

	got, err := s.BusFaults(t.Context(), now.Add(-time.Hour))
	if err != nil {
		t.Fatal(err)
	}
	if len(got) != 2 {
		t.Fatalf("read %d advisories, want one for each user within the hour: %+v", len(got), got)
	}
	for _, a := range got {
		switch a.User {
		case "laptop.claude-code":
			if a.Kind != AdvisorySlowConsumer || a.ID != "laptop.claude-code" || a.Count != 2 ||
				!a.Last.Equal(again.Closed) || !a.First.Equal(slow.Closed) || !strings.Contains(a.Said, "connection 31") {
				t.Errorf("the slow consumer read as %+v: two closures, the one said twice counted once", a)
			}
		case "anchor.desk":
			if a.Kind != AdvisoryConnectionClosed || a.Count != 1 || a.Reason != "Maximum Message Payload Exceeded" {
				t.Errorf("the closed connection read as %+v", a)
			}
		default:
			t.Errorf("an advisory for %s", a.User)
		}
	}
}

func TestNatsBusFaultsReadNothingWhenTheBusModuleSaidNothing(t *testing.T) {
	s := aBusWithEvents(t)
	started := time.Now()
	got, err := s.BusFaults(t.Context(), time.Now().Add(-time.Hour))
	if err != nil || len(got) != 0 {
		t.Fatalf("read %v, %v from a stream the bus module never wrote", got, err)
	}
	if took := time.Since(started); took > time.Second {
		t.Fatalf("reading nothing took %s: a tick must not wait on an empty stream", took)
	}
}

func TestTheControllersOwnSlowConsumerAndItsClosedConnectionAreOneAdvisory(t *testing.T) {
	at := time.Date(2026, 10, 14, 12, 0, 0, 0, time.UTC)
	own := []Advisory{{Kind: AdvisorySlowConsumer, ID: "controller", Said: "the controller fell behind", First: at, Last: at, Count: 1}}
	closed := BusClosures([]BusClosure{{User: "controller", CID: 5, Reason: "Slow Consumer (Pending Bytes)",
		Opened: at.Add(-time.Hour), Closed: at.Add(time.Minute)}}, []string{AdvisorySlowConsumer}, at.Add(-time.Hour))
	got := MergeAdvisories(own, closed)
	if len(got) != 1 || got[0].Count != 1 || !got[0].Last.Equal(at.Add(time.Minute)) || got[0].User != "controller" {
		t.Fatalf("merged as %+v: one fall behind, told to the client and closed by the server, is counted once", got)
	}
}

// **A busy hour is read up to a bound, and says it was cut** (review of #326): a storm of closures must not
// spend a watchdog tick, and a count cut short is a floor, said as one.
func TestNatsBusFaultsReadAtMostTheBoundAndSaySo(t *testing.T) {
	s := aBusWithEvents(t)
	was := MaxBusEventsRead
	MaxBusEventsRead = 2
	t.Cleanup(func() { MaxBusEventsRead = was })
	now := time.Now().UTC()
	for i := range 3 {
		sayClosure(t, s, BusSlowConsumerEvent, BusClosure{User: "laptop.claude-code", CID: uint64(10 + i),
			Reason: "Slow Consumer (Pending Bytes)", Opened: now.Add(-time.Hour / 2), Closed: now.Add(-time.Duration(3-i) * time.Minute)})
	}
	got, err := s.BusFaults(t.Context(), now.Add(-time.Hour))
	if err != nil {
		t.Fatal(err)
	}
	if len(got) != 1 || got[0].Count != 2 || !strings.Contains(got[0].Said, "at least: the read stopped at 2") {
		t.Fatalf("a read past its bound read as %+v", got)
	}
}

// **The bus module's watch is heard by its heartbeat** (review 274): the newest is read directly, with the
// machine the runtime stamped on it; none at all is a bus module older than its watch.
func TestNatsTheBusWatchsNewestHeartbeatIsRead(t *testing.T) {
	s := aBusWithEvents(t)
	if _, heard, err := s.BusWatchLast(t.Context()); err != nil || heard {
		t.Fatalf("a stream with no heartbeat read as heard %v, %v", heard, err)
	}
	at := time.Now().UTC().Truncate(time.Second)
	for i, healthy := range []bool{true, false} {
		msg := nats.NewMsg(EventSubject(BusModule, BusWatchEvent))
		msg.Header.Set("x-node", "anchor")
		msg.Data, _ = json.Marshal(map[string]any{"at": at.Add(time.Duration(i) * time.Second), "healthy": healthy,
			"why": "the watch of slow consumers last read the bus's closed connections 95s ago"})
		if _, err := s.js.Context().PublishMsg(msg); err != nil {
			t.Fatal(err)
		}
	}
	b, heard, err := s.BusWatchLast(t.Context())
	if err != nil || !heard || b.Healthy || b.Machine != "anchor" || !b.At.Equal(at.Add(time.Second)) ||
		!strings.Contains(b.Why, "95s ago") {
		t.Fatalf("read %+v, %v, %v", b, heard, err)
	}
}