Raise the bus module's watch when it falls silent or says it cannot see (hq issue 270, review 274)

A dead watch of slow consumers reads as none. The watch now beats after every
look; S9 reads the newest heartbeat and raises bus-watch-dead when it is older
than two minutes or says the watch is not healthy, on the bus's machine. A
stream with no heartbeat at all is a bus module older than its watch.

cmd/mesh-controller/bus_users_closed_test.go

@@ -90,3 +90,34 @@ func TestABusModuleReadThatFailsKeepsTheControllersOwnSignals(t *testing.T) {
 		}
 	}
 }
+
+// **A dead watch is heard** (review 274): the bus module's watch silent for longer than three looks, or saying
+// it cannot see, is a condition on the bus's machine — never read as no slow consumers. A stream with no
+// heartbeat at all is a bus module older than its watch, and says nothing.
+func TestTheBusWatchSilentOrBlindIsAConditionAndNoHeartbeatAtAllIsNone(t *testing.T) {
+	now := time.Date(2026, 10, 14, 12, 0, 0, 0, time.UTC)
+	look := func(b link.BusWatchBeat, heard bool) []conditions.Observation {
+		var out []conditions.Observation
+		for _, o := range watchAdvisories(&signalFacts{now: now, busWatch: b, busWatchHeard: heard}) {
+			if o.Kind == kindBusWatchDead {
+				out = append(out, o)
+			}
+		}
+		return out
+	}
+	if got := look(link.BusWatchBeat{}, false); len(got) != 0 {
+		t.Fatalf("no heartbeat ever raised %+v", got)
+	}
+	if got := look(link.BusWatchBeat{At: now.Add(-30 * time.Second), Healthy: true, Machine: "anchor"}, true); len(got) != 0 {
+		t.Fatalf("a healthy watch heard 30s ago raised %+v", got)
+	}
+	silent := look(link.BusWatchBeat{At: now.Add(-5 * time.Minute), Healthy: true, Machine: "anchor"}, true)
+	if len(silent) != 1 || silent[0].Machine != "anchor" || !strings.Contains(silent[0].Summary, "said nothing") {
+		t.Fatalf("a watch silent for 5 minutes raised %+v", silent)
+	}
+	blind := look(link.BusWatchBeat{At: now.Add(-10 * time.Second), Healthy: false, Machine: "anchor",
+		Why: "the watch of slow consumers last read the bus's closed connections 95s ago: connection refused"}, true)
+	if len(blind) != 1 || !strings.Contains(blind[0].Said, "connection refused") {
+		t.Fatalf("a watch that says it cannot see raised %+v", blind)
+	}
+}

cmd/mesh-controller/plain_words.go

@@ -620,6 +620,12 @@ var plainWordings = map[string]func(conditions.Observation) words{
 			Explanation: "The bus reports a listener too slow to keep up, so messages to it are late.",
 			Resolved:    "The listener keeps up again"}
 	}),
+	kindBusWatchDead: worded(func(o conditions.Observation) words {
+		return words{Headline: "The bus's watch of slow clients is down",
+			Explanation: "The part of the bus that notices clients too slow to keep up has stopped, or cannot see, so " +
+				"such a client would go unnoticed. It is part of the bus's tools on " + machineOr(o, "the bus's machine") + ".",
+			Resolved: "Resolved: the bus's watch of slow clients works again"}
+	}),
 	kindBusFaultsUnread: worded(func(o conditions.Observation) words {
 		return words{Headline: "Slow clients of the bus cannot be checked",
 			Explanation: "The controller could not read what the bus said about its clients' connections, so a " +

cmd/mesh-controller/signals.go

@@ -589,9 +589,43 @@ func watchAdvisories(f *signalFacts) []conditions.Observation {
 	if f.busFaultsErr != nil {
 		out = append(out, busFaultsUnread(f.busFaultsErr))
 	}
+	if o, ok := busWatchDead(f); ok {
+		out = append(out, o)
+	}
 	return append(out, watchDeadLetters(f)...)
 }
 
+// kindBusWatchDead is the bus module's watch not heard, or saying it cannot see (novox/hq ADR 0357, review
+// 274): while it lasts, a slow consumer would read as none.
+const kindBusWatchDead = "bus-watch-dead"
+
+// busWatchStale is how long the watch may be silent: three of its looks, and one more for the tick.
+const busWatchStale = 2 * time.Minute
+
+// busWatchDead says the watch silent past busWatchStale, or saying it is not healthy. Nothing while the stream
+// holds no heartbeat at all: a bus module older than its watch says nothing either way.
+func busWatchDead(f *signalFacts) (conditions.Observation, bool) {
+	if !f.busWatchHeard {
+		return conditions.Observation{}, false
+	}
+	b := f.busWatch
+	o := conditions.Observation{Scope: conditions.ScopeBus, ID: "bus-module", Kind: kindBusWatchDead,
+		Machine: b.Machine, Severity: conditions.Warning}
+	switch {
+	case f.now.Sub(b.At) > busWatchStale:
+		o.Summary = fmt.Sprintf("the bus module's watch of slow consumers has said nothing for %s, so a client "+
+			"the bus cut off would not be said", ago(f.now.Sub(b.At)))
+		o.Said = "last heartbeat " + b.At.UTC().Format(time.RFC3339)
+		return o, true
+	case !b.Healthy:
+		o.Summary = "the bus module's watch of slow consumers says it cannot see, so a client the bus cut off " +
+			"would not be said"
+		o.Said = firstLine(b.Why)
+		return o, true
+	}
+	return conditions.Observation{}, false
+}
+
 // kindBusFaultsUnread is the bus module's events not read back (novox/hq ADR 0357): every other user's
 // slow consumer is unheard while it lasts, and the controller's own signals are still said.
 const kindBusFaultsUnread = "bus-faults-unread"

cmd/mesh-controller/watchdogs.go

@@ -94,6 +94,9 @@ type signalFacts struct {
 	advisoriesErr     error
 	// busFaultsErr is why the bus module's events could not be read this tick (novox/hq ADR 0357).
 	busFaultsErr error
+	// busWatch is the bus module's watch's newest heartbeat, when the stream holds one (review 274).
+	busWatch      link.BusWatchBeat
+	busWatchHeard bool
 
 	selfCheck selfCheckFacts
 
@@ -368,6 +371,9 @@ func (w *watchdogs) gather(ctx context.Context) *signalFacts {
 		if faults, f.busFaultsErr = w.server.BusFaults(ctx, now.Add(-advisoryQuiet)); f.busFaultsErr == nil {
 			f.advisories = link.MergeAdvisories(f.advisories, faults)
 		}
+		if f.busFaultsErr == nil {
+			f.busWatch, f.busWatchHeard, f.busFaultsErr = w.server.BusWatchLast(ctx)
+		}
 	}
 	if f.advisoriesErr == nil && w.js != nil {
 		f.deadLetters, f.advisoriesErr = link.HeldDeadLetters(w.js.Context())

internal/link/busfaults.go

@@ -32,6 +32,9 @@ const BusModule = "nats"
 const (
 	BusSlowConsumerEvent     = "bus.slow-consumer"
 	BusConnectionClosedEvent = "bus.connection-closed"
+	// BusWatchEvent is the bus module's watch saying it is there after every look, and whether it is healthy
+	// (novox/hq ADR 0357, review 274): a watch that stops saying it is a watch nobody would otherwise hear.
+	BusWatchEvent = "bus.watch"
 )
 
 // AdvisoryConnectionClosed is a connection the bus closed for breaking its rules, as conditions name it.
@@ -233,3 +236,33 @@ func MergeAdvisories(a, b []Advisory) []Advisory {
 	}
 	return out
 }
+
+// BusWatchBeat is the bus module's watch's newest heartbeat.
+type BusWatchBeat struct {
+	At      time.Time `json:"at"`
+	Healthy bool      `json:"healthy"`
+	Why     string    `json:"why,omitempty"`
+	// Machine is the machine the runtime stamped on it (`x-node`).
+	Machine string `json:"-"`
+}
+
+// BusWatchLast is the newest heartbeat of the bus module's watch, read directly from the events stream; false
+// when the stream holds none — a bus module older than its watch, which says nothing either way.
+func (s *Server) BusWatchLast(ctx context.Context) (BusWatchBeat, bool, error) {
+	if s.js == nil {
+		return BusWatchBeat{}, false, errors.New("this control plane is not on the bus, so it cannot read the bus module's watch")
+	}
+	msg, err := s.js.Context().GetLastMsg(broker.EventsStream, EventSubject(BusModule, BusWatchEvent), nats.Context(ctx))
+	if errors.Is(err, nats.ErrMsgNotFound) {
+		return BusWatchBeat{}, false, nil
+	}
+	if err != nil {
+		return BusWatchBeat{}, false, fmt.Errorf("reading the bus module's watch's heartbeat: %w", err)
+	}
+	var b BusWatchBeat
+	if err := json.Unmarshal(msg.Data, &b); err != nil || b.At.IsZero() {
+		b = BusWatchBeat{At: msg.Time, Why: "its heartbeat could not be read"}
+	}
+	b.Machine = msg.Header.Get("x-node")
+	return b, true, nil
+}

internal/link/busfaults_test.go

@@ -6,6 +6,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/nats-io/nats.go"
+
 	"github.com/novox/mesh-controller/internal/broker"
 	"github.com/novox/mesh-controller/internal/testbus"
 )
@@ -119,3 +121,27 @@ func TestNatsBusFaultsReadAtMostTheBoundAndSaySo(t *testing.T) {
 		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)
+	}
+}