Raise every bus user's slow consumer and closed connection as S9, from the bus module's events (hq issue 270, ADR 0357)

The controller heard a slow consumer only on its own connection: the server
says another user's only in its record of closed connections. The bus module
now says that record as events; the watchdogs read the last hour back on every
tick and raise one condition per user and kind, naming the module and machine.
The module check lets the bus module emit, since its tool runner publishes.

internal/link/advisories.go

@@ -24,9 +24,10 @@ import (
 // which consumer of whose, which subject — for the watchdog to say as a condition, as the refused
 // reply of issue 265 is said today.
 //
-// What the bus says about **other** principals' connections — a module's slow consumer, a module
-// refused a subject — the server publishes only to a system account, which the mesh's bus does not
-// have; that half is not heard yet (to-be 45 S9, recorded as deferred).
+// What the bus does to **other** principals' connections the server does not publish in the mesh's
+// account. A module's slow consumer it closes and records, and the bus module says that record on the
+// bus (busfaults.go, novox/hq ADR 0357). A module refused a subject is told and written in the server's
+// log, and nowhere else: not heard yet (novox/hq issue 702).
 
 // The advisory kinds, as conditions name them.
 const (
@@ -48,6 +49,11 @@ type Advisory struct {
 	Stream, Consumer string
 	// Token is the condition key's last part, when it is not Kind.
 	Token string
+	// User is the bus user whose connection it is about, when it is about one (novox/hq ADR 0357): the
+	// controller's own, or another user's the bus module said.
+	User string
+	// Reason is the server's reason for closing that connection, as it gave it.
+	Reason string
 	// Said is the newest saying, in the mesh's words.
 	Said        string
 	First, Last time.Time

internal/link/busfaults.go

@@ -0,0 +1,213 @@
+package link
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"sort"
+	"time"
+
+	"github.com/nats-io/nats.go"
+
+	"github.com/novox/mesh-controller/internal/broker"
+)
+
+// What the bus did to other users' connections, as the bus module says it (novox/hq ADR 0357, issue 270).
+//
+// **The server says another user's slow consumer only in its record of closed connections**, which the
+// bus module reads on its own machine and says as events: `bus.slow-consumer` for a connection the server
+// closed as a slow consumer, `bus.connection-closed` for one it closed for breaking the bus's rules. They
+// are read back from EVENTS on every watchdog tick for the hour S9 says an advisory, rather than taken by
+// the controller's durable consumer: that one takes a message at a time while a merge builds for
+// minutes, and read back, a controller that restarts says the last hour again without keeping anything.
+//
+// **Said at least once, counted once.** The module says the last hour again whenever it starts; a closure
+// is its user, its connection id and when it opened, and is counted once however often it is said.
+
+// BusModule is the module whose events say what the bus did to its users' connections.
+const BusModule = "nats"
+
+// The bus module's events, by their local names (its manifest's `emits`).
+const (
+	BusSlowConsumerEvent     = "bus.slow-consumer"
+	BusConnectionClosedEvent = "bus.connection-closed"
+)
+
+// AdvisoryConnectionClosed is a connection the bus closed for breaking its rules, as conditions name it.
+const AdvisoryConnectionClosed = "connection-closed"
+
+// BusClosure is one closed connection as the bus module says it: the user, never a credential.
+type BusClosure struct {
+	User    string    `json:"user"`
+	Name    string    `json:"name,omitempty"`
+	CID     uint64    `json:"cid"`
+	Reason  string    `json:"reason"`
+	Opened  time.Time `json:"opened"`
+	Closed  time.Time `json:"closed"`
+	Pending int64     `json:"pending_bytes,omitempty"`
+}
+
+func (c BusClosure) identity() string {
+	return fmt.Sprintf("%s/%d/%s", c.User, c.CID, c.Opened.UTC().Format(time.RFC3339Nano))
+}
+
+// BusClosures are the advisories the bus module's closures make, closed at or after since: one per user
+// and kind, each closure counted once, the newest said.
+func BusClosures(closures []BusClosure, kinds []string, since time.Time) []Advisory {
+	type entry struct {
+		a      *Advisory
+		seen   map[string]bool
+		newest BusClosure
+	}
+	by := map[string]*entry{}
+	var order []string
+	for i, c := range closures {
+		if c.User == "" || c.Closed.Before(since) {
+			continue
+		}
+		kind := kinds[i]
+		key := kind + "/" + c.User
+		e := by[key]
+		if e == nil {
+			e = &entry{a: &Advisory{Kind: kind, ID: c.User, User: c.User}, seen: map[string]bool{}}
+			by[key] = e
+			order = append(order, key)
+		}
+		if e.seen[c.identity()] {
+			continue
+		}
+		e.seen[c.identity()] = true
+		e.a.Count++
+		if e.a.First.IsZero() || c.Closed.Before(e.a.First) {
+			e.a.First = c.Closed
+		}
+		if !c.Closed.Before(e.a.Last) {
+			e.a.Last, e.newest = c.Closed, c
+		}
+	}
+	sort.Strings(order)
+	out := make([]Advisory, 0, len(order))
+	for _, key := range order {
+		e := by[key]
+		c := e.newest
+		name := ""
+		if c.Name != "" && c.Name != c.User {
+			name = fmt.Sprintf(" (connection %q)", c.Name)
+		}
+		pending := ""
+		if c.Pending > 0 {
+			pending = fmt.Sprintf(", %d bytes still to be written to it", c.Pending)
+		}
+		switch e.a.Kind {
+		case AdvisorySlowConsumer:
+			e.a.Said = fmt.Sprintf("the bus closed %s's connection %d%s as a slow consumer: %s%s", c.User, c.CID,
+				name, c.Reason, pending)
+		default:
+			e.a.Said = fmt.Sprintf("the bus closed %s's connection %d%s: %s", c.User, c.CID, name, c.Reason)
+		}
+		e.a.Reason = c.Reason
+		out = append(out, *e.a)
+	}
+	return out
+}
+
+// busEventKind is the advisory kind of one of the bus module's events, by its subject.
+func busEventKind(subject string) (string, bool) {
+	switch subject {
+	case EventSubject(BusModule, BusSlowConsumerEvent):
+		return AdvisorySlowConsumer, true
+	case EventSubject(BusModule, BusConnectionClosedEvent):
+		return AdvisoryConnectionClosed, true
+	}
+	return "", false
+}
+
+// BusFaults reads back what the bus module said of its users' connections since a moment, from the events
+// stream, on a consumer of its own that acknowledges nothing (as PowerStates reads), and says it as
+// advisories. Nothing to read is no advisory, and costs two direct reads: the newest of each event.
+func (s *Server) BusFaults(ctx context.Context, since time.Time) ([]Advisory, error) {
+	if s.js == nil {
+		return nil, errors.New("this control plane is not on the bus, so it cannot read what the bus module said")
+	}
+	js := s.js.Context()
+	recent := false
+	for _, event := range []string{BusSlowConsumerEvent, BusConnectionClosedEvent} {
+		last, err := js.GetLastMsg(broker.EventsStream, EventSubject(BusModule, event), nats.Context(ctx))
+		if errors.Is(err, nats.ErrMsgNotFound) {
+			continue
+		}
+		if err != nil {
+			return nil, fmt.Errorf("reading what the bus module said of its users' connections: %w", err)
+		}
+		if !last.Time.Before(since) {
+			recent = true
+		}
+	}
+	if !recent {
+		return nil, nil
+	}
+	sub, err := js.SubscribeSync(EventSubject(BusModule, "bus.>"), nats.OrderedConsumer(), nats.StartTime(since))
+	if err != nil {
+		return nil, fmt.Errorf("reading what the bus module said of its users' connections: %w", err)
+	}
+	defer func() { _ = sub.Unsubscribe() }()
+	var closures []BusClosure
+	var kinds []string
+	for {
+		wait, cancel := context.WithTimeout(ctx, readQuiet)
+		msg, err := sub.NextMsgWithContext(wait)
+		cancel()
+		if err != nil {
+			if ctx.Err() != nil {
+				return nil, ctx.Err()
+			}
+			break
+		}
+		if kind, ok := busEventKind(msg.Subject); ok {
+			var c BusClosure
+			if json.Unmarshal(msg.Data, &c) == nil && c.User != "" && !c.Closed.IsZero() {
+				closures, kinds = append(closures, c), append(kinds, kind)
+			}
+		}
+		if meta, err := msg.Metadata(); err != nil || meta.NumPending == 0 {
+			break
+		}
+	}
+	return BusClosures(closures, kinds, since), nil
+}
+
+// MergeAdvisories is two lists of advisories as one: the same kind about the same thing is one advisory,
+// counting both, said by the newer. The controller's own slow consumer as its client library sees it and
+// as the server closed it are the same user's (`controller`).
+func MergeAdvisories(a, b []Advisory) []Advisory {
+	out := append([]Advisory(nil), a...)
+	at := map[string]int{}
+	for i, x := range out {
+		at[x.Kind+"/"+x.ID+"/"+x.Token] = i
+	}
+	for _, y := range b {
+		key := y.Kind + "/" + y.ID + "/" + y.Token
+		i, had := at[key]
+		if !had {
+			at[key] = len(out)
+			out = append(out, y)
+			continue
+		}
+		x := &out[i]
+		x.Count += y.Count
+		if y.First.Before(x.First) {
+			x.First = y.First
+		}
+		if y.Last.After(x.Last) {
+			x.Last, x.Said = y.Last, y.Said
+			if y.Reason != "" {
+				x.Reason = y.Reason
+			}
+		}
+		if x.User == "" {
+			x.User = y.User
+		}
+	}
+	return out
+}

internal/link/busfaults_test.go

@@ -0,0 +1,100 @@
+package link
+
+import (
+	"encoding/json"
+	"strings"
+	"testing"
+	"time"
+
+	"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 != 2 || !got[0].Last.Equal(at.Add(time.Minute)) || got[0].User != "controller" {
+		t.Fatalf("merged as %+v", got)
+	}
+}