diff --git a/cmd/mesh-controller/bus_users_closed_test.go b/cmd/mesh-controller/bus_users_closed_test.go index 689bd9f8..25bd26b5 100644 --- a/cmd/mesh-controller/bus_users_closed_test.go +++ b/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) + } +} diff --git a/cmd/mesh-controller/plain_words.go b/cmd/mesh-controller/plain_words.go index da0511e5..4347d1b2 100644 --- a/cmd/mesh-controller/plain_words.go +++ b/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 " + diff --git a/cmd/mesh-controller/signals.go b/cmd/mesh-controller/signals.go index 5603533d..c75dfbe3 100644 --- a/cmd/mesh-controller/signals.go +++ b/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" diff --git a/cmd/mesh-controller/watchdogs.go b/cmd/mesh-controller/watchdogs.go index a8d2f793..2c1140d7 100644 --- a/cmd/mesh-controller/watchdogs.go +++ b/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()) diff --git a/internal/catalogue/bus_snapshot_test.go b/internal/catalogue/bus_snapshot_test.go index e9904350..57384996 100644 --- a/internal/catalogue/bus_snapshot_test.go +++ b/internal/catalogue/bus_snapshot_test.go @@ -30,6 +30,15 @@ func TestTheBussAccountSaysNothingOnTheBus(t *testing.T) { if _, err := ParseManifest(bundled); err != nil { t.Fatalf("a bus module whose tools bundle declares what it emits was refused: %v", err) } + // And as built: the resolved manifest registration reads keeps the bundles its build made and drops the + // build, and must be registrable just the same. + built := []byte(`{"module":"bus","version":"1","provides":[{"name":"mesh-bus","scope":"mesh"}], + "claims":[{"name":"mesh-broker","scope":"mesh"}],"own-secrets":{"broker":"/run/broker"}, + "emits":["bus.slow-consumer"],"tools":["bus_streams"], + "bundles":[{"name":"tools","source":"bundles/bus/tools","digest":"sha256:00","language":"go"}]}`) + if _, err := ParseManifest(built); err != nil { + t.Fatalf("the built bus module, its bundle made, is not registrable: %v", err) + } quiet := []byte(`{"module":"bus","version":"1","provides":[{"name":"mesh-bus","scope":"mesh"}], "claims":[{"name":"mesh-broker","scope":"mesh"}],"own-secrets":{"broker":"/run/broker"}, diff --git a/internal/catalogue/manifest.go b/internal/catalogue/manifest.go index 21ec6363..a4ea5073 100644 --- a/internal/catalogue/manifest.go +++ b/internal/catalogue/manifest.go @@ -2903,8 +2903,13 @@ func whileStoppedProblems(m Manifest, r map[string]any, scheduled bool) []string return problems } -// shipsABundle says the module builds a tools bundle, which the node's runtime carries and speaks for. +// shipsABundle says the module builds a tools bundle, which the node's runtime carries and speaks for: as +// written, by its build's artifacts; as built and registered, by the bundles its build made — the resolved +// manifest keeps those and drops the build (the merge gate found nats not registrable on that, review 274). func (m Manifest) shipsABundle() bool { + if len(m.Bundles) > 0 { + return true + } if m.Build == nil { return false } diff --git a/internal/link/busfaults.go b/internal/link/busfaults.go index ff4eb64a..fe4910b7 100644 --- a/internal/link/busfaults.go +++ b/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 +} diff --git a/internal/link/busfaults_test.go b/internal/link/busfaults_test.go index d1eda4ef..806c5939 100644 --- a/internal/link/busfaults_test.go +++ b/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) + } +}