From fd9308d3287ee7f70bc94acc3a01e39b97b22efa Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 14 Oct 2026 08:41:47 +0200 Subject: [PATCH 1/2] 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. --- cmd/mesh-controller/bus_users_closed_test.go | 69 ++++++ cmd/mesh-controller/plain_words.go | 6 + cmd/mesh-controller/signals.go | 45 +++- cmd/mesh-controller/watchdogs.go | 8 + internal/broker/bus_module_emits_test.go | 34 +++ internal/catalogue/bus_snapshot_test.go | 10 + internal/catalogue/manifest.go | 20 +- internal/link/advisories.go | 12 +- internal/link/busfaults.go | 213 +++++++++++++++++++ internal/link/busfaults_test.go | 100 +++++++++ 10 files changed, 511 insertions(+), 6 deletions(-) create mode 100644 cmd/mesh-controller/bus_users_closed_test.go create mode 100644 internal/broker/bus_module_emits_test.go create mode 100644 internal/link/busfaults.go create mode 100644 internal/link/busfaults_test.go diff --git a/cmd/mesh-controller/bus_users_closed_test.go b/cmd/mesh-controller/bus_users_closed_test.go new file mode 100644 index 00000000..10db7397 --- /dev/null +++ b/cmd/mesh-controller/bus_users_closed_test.go @@ -0,0 +1,69 @@ +package main + +import ( + "strings" + "testing" + "time" + + "github.com/novox/mesh-controller/internal/conditions" + "github.com/novox/mesh-controller/internal/link" +) + +// **Every user's slow consumer is S9's, not the controller's alone** (novox/hq ADR 0357, issue 270). On +// 2026-10-07 the bus closed two clients on the laptop as slow consumers and nothing raised a condition: +// the controller heard only its own connection. Said by the bus module and read back, a user the bus +// closes is one condition naming the module and its machine, raised at the first look and cleared an hour +// after the last closure. +func TestAUserTheBusClosedAsASlowConsumerIsOneConditionNamingItsModuleAndMachine(t *testing.T) { + closed := time.Date(2026, 10, 14, 12, 0, 0, 0, time.UTC) + now := closed.Add(20 * time.Second) + store := conditions.NewInMemory() + k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store, + Teller: &conditions.Told{}, Now: func() time.Time { return now }}) + t.Cleanup(func() { k.Close(t.Context()) }) + + advisories := link.BusClosures([]link.BusClosure{ + {User: "laptop.claude-code", CID: 10, Reason: "Slow Consumer (Write Deadline)", Opened: closed.Add(-time.Hour), + Closed: closed.Add(-10 * time.Minute)}, + {User: "laptop.claude-code", CID: 44, Reason: "Slow Consumer (Write Deadline)", Opened: closed.Add(-time.Minute), + Closed: closed}, + {User: "anchor.desk", CID: 7, Reason: "Maximum Message Payload Exceeded", Opened: closed.Add(-time.Hour), + Closed: closed}, + }, []string{link.AdvisorySlowConsumer, link.AdvisorySlowConsumer, link.AdvisoryConnectionClosed}, closed.Add(-time.Hour)) + look := func() { + t.Helper() + f := &signalFacts{now: now, host: "anchor", advisories: advisories} + if err := k.Reconcile(t.Context(), "S9", watchAdvisories(f)); err != nil { + t.Fatal(err) + } + } + look() + + slow, found, err := conditions.ReadOne(t.Context(), store, "bus.laptop.claude-code.slow-consumer") + if err != nil || !found { + t.Fatalf("no slow-consumer condition for the user: %v %v", found, err) + } + if slow.Subject.Machine != "laptop" || slow.Observations != 2 || !slow.LastObserved.Equal(closed) { + t.Fatalf("the slow consumer is said on %q, %d closure(s), last %s: %+v", slow.Subject.Machine, slow.Observations, + slow.LastObserved, slow) + } + if !strings.Contains(slow.Summary, "claude-code on laptop") || strings.Contains(slow.Summary, "laptop.claude-code") { + t.Fatalf("the summary does not name the module in words, or carries the dotted user: %q", slow.Summary) + } + if !strings.Contains(slow.Headline, "Claude-code on laptop") { + t.Fatalf("the headline is %q", slow.Headline) + } + cut, found, err := conditions.ReadOne(t.Context(), store, "bus.anchor.desk.connection-closed") + if err != nil || !found || cut.Subject.Machine != "anchor" || !strings.Contains(cut.Summary, "maximum message payload exceeded") { + t.Fatalf("the user cut off for breaking the bus's rules: %v %v %+v", found, err, cut) + } + + // An hour after the newest closure without another, both clear. + now = closed.Add(time.Hour + time.Minute) + look() + for _, key := range []string{"bus.laptop.claude-code.slow-consumer", "bus.anchor.desk.connection-closed"} { + if _, found, _ := k.Get(t.Context(), key); found { + t.Fatalf("%s is still open an hour after the last closure", key) + } + } +} diff --git a/cmd/mesh-controller/plain_words.go b/cmd/mesh-controller/plain_words.go index 7d60b16a..0e16f0cc 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"} }), + "connection-closed": worded(func(o conditions.Observation) words { + return words{Headline: "The bus cut a client off", + Explanation: "The bus closed a client's connection because it broke the bus's rules, so what it was " + + "sending was lost.", + Resolved: "Resolved: the client keeps to the bus's rules"} + }), kindBusReconnects: worded(func(o conditions.Observation) words { return words{Headline: "A client keeps losing the bus", Explanation: "One of the mesh's clients lost its connection to the bus again and again in the last hour. " + diff --git a/cmd/mesh-controller/signals.go b/cmd/mesh-controller/signals.go index e2cbc387..1d5f257c 100644 --- a/cmd/mesh-controller/signals.go +++ b/cmd/mesh-controller/signals.go @@ -153,11 +153,15 @@ var signalsTable = []signalRow{ return newestOf(f.standings, func(c conditions.Condition) time.Time { return c.LastObserved }) }}, {Row: "S9", Signal: "bus advisories: maximum deliveries, consumer deleted; the controller's own slow " + - "consumer and refused subjects", Emitter: "bus server's advisory subjects; the controller's connection", + "consumer and refused subjects; every user's connection the server closed as a slow consumer or for " + + "breaking the bus's rules (ADR 0357)", + Emitter: "bus server's advisory subjects; the controller's connection; the bus module's watch of its " + + "server (bus.slow-consumer, bus.connection-closed)", Trigger: "any", Bound: "any occurrence; clears after an hour without another, a deleted consumer " + "once it exists again or the mesh no longer expects it, and a message given up on once DEAD_LETTERS " + "no longer holds it (novox/hq issue 330)", - Kind: "slow-consumer, max-deliveries, refused, consumer-lost", Severity: conditions.Warning, Phase: 1, + Kind: "slow-consumer, connection-closed, max-deliveries, refused, consumer-lost", Severity: conditions.Warning, + Phase: 1, needs: func(f *signalFacts) error { return f.advisoriesErr }, watch: watchAdvisories, newest: func(f *signalFacts) time.Time { return newestOf(f.advisories, func(a link.Advisory) time.Time { return a.Last }) @@ -569,6 +573,9 @@ func watchAdvisories(f *signalFacts) []conditions.Observation { // The advisory is remembered and read again on every look for an hour: the condition counts // what the bus said and when, not the looks (novox/hq issue 402). Happened: a.Last, Times: a.Count} + if a.User != "" && a.Reason != "" { + busUserClosed(&o, a, f.host) + } if a.Kind == link.AdvisoryMaxDeliveries && a.Token == link.AdvisoryNotKept { o.Machine = consumerMachine(a.Stream, a.Consumer) o.Headline = clip(conditions.Capital(fmt.Sprintf("%s gave up on a message, not kept", @@ -582,6 +589,40 @@ func watchAdvisories(f *signalFacts) []conditions.Observation { return append(out, watchDeadLetters(f)...) } +// busUserClosed words an advisory about a bus user's connection the server closed (novox/hq ADR 0357): the +// user as the operator says it and the machine it is on, the dotted user name kept in the evidence only. +func busUserClosed(o *conditions.Observation, a link.Advisory, host string) { + who, machine := busUserWords(a.User) + if machine == "" && a.User == "controller" { + machine = host + } + o.Machine = machine + times, count := "once", "once" + if a.Count > 1 { + times = fmt.Sprintf("%d times since %s", a.Count, a.First.UTC().Format("15:04 MST")) + count = fmt.Sprintf("%d times", a.Count) + } + switch a.Kind { + case link.AdvisorySlowConsumer: + o.Summary = fmt.Sprintf("the bus closed the connection of %s as a slow consumer %s in the last hour, "+ + "newest at %s: it could not take what it was sent fast enough, and what was on its way to it was lost", + who, times, a.Last.UTC().Format("15:04 MST")) + o.Headline = clip(conditions.Capital(who)+" was too slow for the bus", 60) + o.Explanation = conditions.Capital(fmt.Sprintf("%s could not keep up with what the bus sent it, so the bus "+ + "closed its connection (%s in the last hour) and what was on its way was lost. It connects again "+ + "by itself; this clears an hour after the last time.", who, count)) + o.Resolved = "Resolved: " + who + " kept up for an hour" + default: + o.Summary = fmt.Sprintf("the bus closed the connection of %s for breaking its rules %s in the last hour, "+ + "newest at %s: %s", who, times, a.Last.UTC().Format("15:04 MST"), strings.ToLower(a.Reason)) + o.Headline = clip(conditions.Capital(who)+" was cut off by the bus", 60) + o.Explanation = conditions.Capital(fmt.Sprintf("The bus closed the connection of %s because it broke "+ + "the bus's rules (%s, %s in the last hour), so what it was sending was lost. This clears an hour "+ + "after the last time.", who, strings.ToLower(a.Reason), count)) + o.Resolved = "Resolved: " + who + " kept to the bus's rules for an hour" + } +} + // watchDeadLetters says each consumer that DEAD_LETTERS holds a message for (novox/hq issue 330): open // while it holds any, so it clears when they are delivered again or dropped, never because the server // stopped saying it. diff --git a/cmd/mesh-controller/watchdogs.go b/cmd/mesh-controller/watchdogs.go index 428744a2..a527df13 100644 --- a/cmd/mesh-controller/watchdogs.go +++ b/cmd/mesh-controller/watchdogs.go @@ -358,6 +358,14 @@ func (w *watchdogs) gather(ctx context.Context) *signalFacts { } f.advisories = link.Advisories.Since(now.Add(-advisoryQuiet)) f.lostConsumers, f.advisoriesErr = w.lostConsumers(ctx, f.advisories) + // And what the bus did to every other user's connection, as the bus module said it (novox/hq ADR 0357): + // read back for the same hour, and one advisory with the controller's own where both are its. + if f.advisoriesErr == nil && w.server != nil && w.js != nil { + var faults []link.Advisory + if faults, f.advisoriesErr = w.server.BusFaults(ctx, now.Add(-advisoryQuiet)); f.advisoriesErr == nil { + f.advisories = link.MergeAdvisories(f.advisories, faults) + } + } if f.advisoriesErr == nil && w.js != nil { f.deadLetters, f.advisoriesErr = link.HeldDeadLetters(w.js.Context()) } diff --git a/internal/broker/bus_module_emits_test.go b/internal/broker/bus_module_emits_test.go new file mode 100644 index 00000000..8e7a53a5 --- /dev/null +++ b/internal/broker/bus_module_emits_test.go @@ -0,0 +1,34 @@ +package broker + +import ( + "slices" + "testing" +) + +// **The bus module's watch speaks through the runtime that carries it** (novox/hq ADR 0357): the node's +// runtime is granted the events the bus module declares, under the module's name, while the bus's own +// account keeps the snapshot API and nothing else (ADR 0235). +func TestTheRuntimeCarryingTheBusModuleMayPublishItsEventsAndTheBussAccountMayNot(t *testing.T) { + events := []string{"bus.slow-consumer", "bus.connection-closed"} + runtime := Principal{Kind: KindNodeTools, Node: "anchor", Module: RuntimeModule, + Carries: []Declared{{Module: "nats", Emits: events}}} + p, err := PermissionsFor(runtime) + if err != nil { + t.Fatal(err) + } + for _, e := range events { + if !slices.Contains(p.Publish, "mesh.mod.nats.event."+e) { + t.Errorf("the runtime carrying the bus module may not publish %s: %v", e, p.Publish) + } + } + account := Principal{Kind: KindModule, Node: "anchor", Module: "nats", Emits: events, SnapshotsTheBus: true} + a, err := PermissionsFor(account) + if err != nil { + t.Fatal(err) + } + for _, s := range a.Publish { + if s == "mesh.mod.nats.event.bus.slow-consumer" || s == "mesh.mod.nats.event.bus.connection-closed" { + t.Errorf("the bus's own account is granted %s", s) + } + } +} diff --git a/internal/catalogue/bus_snapshot_test.go b/internal/catalogue/bus_snapshot_test.go index 945a23cc..e9904350 100644 --- a/internal/catalogue/bus_snapshot_test.go +++ b/internal/catalogue/bus_snapshot_test.go @@ -21,6 +21,16 @@ func TestTheBussAccountSaysNothingOnTheBus(t *testing.T) { t.Fatalf("a bus module declaring what it emits was not refused, or not for the reason: %v", err) } + // Its tools bundle says what it emits (novox/hq ADR 0357): the node's runtime carries the bundle and is + // granted the module's events, so a bus module that ships one may declare them. + bundled := []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"], + "build":{"artifacts":[{"name":"tools","kind":"bundle","language":"go","system":"arch","from":"cmd/t","binary":"t"}]}}`) + if _, err := ParseManifest(bundled); err != nil { + t.Fatalf("a bus module whose tools bundle declares what it emits was refused: %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"}, "tools":["bus_streams"]}`) diff --git a/internal/catalogue/manifest.go b/internal/catalogue/manifest.go index 4dae3034..21ec6363 100644 --- a/internal/catalogue/manifest.go +++ b/internal/catalogue/manifest.go @@ -1630,9 +1630,14 @@ func ParseManifest(raw []byte) (Manifest, error) { // **The bus's own account copies the bus and says nothing on it** (novox/hq ADR 0235). The module // holding mesh-broker is granted the snapshot API and nothing else, so anything it declared to say // or hear on the bus would be granted nothing — refused here rather than silently dropped. + // + // **But its tools bundle may say what it declares** (novox/hq ADR 0357): the node's runtime carries + // the bundle and is granted the events the module emits, under the module's name, as for any module it + // carries — the bus's account is not who publishes them. So emits are refused only for a bus module + // with no bundle to say them. if _, account := m.OwnSecrets["broker"]; account && m.ClaimsSeat(BrokerSeat) { var said []string - if len(m.EmitsAll()) > 0 { + if len(m.EmitsAll()) > 0 && !m.shipsABundle() { said = append(said, "emits") } for _, f := range []struct { @@ -2897,3 +2902,16 @@ 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. +func (m Manifest) shipsABundle() bool { + if m.Build == nil { + return false + } + for _, a := range m.Build.Artifacts { + if a.Kind == ArtifactBundle { + return true + } + } + return false +} diff --git a/internal/link/advisories.go b/internal/link/advisories.go index 013f2b5f..fc5d0e1d 100644 --- a/internal/link/advisories.go +++ b/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 diff --git a/internal/link/busfaults.go b/internal/link/busfaults.go new file mode 100644 index 00000000..e1d5eb28 --- /dev/null +++ b/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 +} diff --git a/internal/link/busfaults_test.go b/internal/link/busfaults_test.go new file mode 100644 index 00000000..29d48935 --- /dev/null +++ b/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) + } +} From 45870ab31cc57ae86600d40a3b685727006a774a Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 14 Oct 2026 09:02:47 +0200 Subject: [PATCH 2/2] Bound the read of bus-module events, keep S9 seeing when it fails, count the controller's own slow consumer once (hq issue 270, review of #326) A storm of closures must not spend a watchdog tick, so the read stops at 2000 events and says the count is a floor. A failed read is its own condition rather than blinding S9, so the controller's own signals are still said. The controller's fall behind, told to its client and closed by the server, is one fault, counted once. --- cmd/mesh-controller/bus_users_closed_test.go | 23 ++++++++++++++++ cmd/mesh-controller/plain_words.go | 6 +++++ cmd/mesh-controller/signals.go | 15 +++++++++++ cmd/mesh-controller/watchdogs.go | 7 +++-- internal/link/busfaults.go | 28 +++++++++++++++++--- internal/link/busfaults_test.go | 25 +++++++++++++++-- 6 files changed, 97 insertions(+), 7 deletions(-) diff --git a/cmd/mesh-controller/bus_users_closed_test.go b/cmd/mesh-controller/bus_users_closed_test.go index 10db7397..689bd9f8 100644 --- a/cmd/mesh-controller/bus_users_closed_test.go +++ b/cmd/mesh-controller/bus_users_closed_test.go @@ -1,6 +1,7 @@ package main import ( + "errors" "strings" "testing" "time" @@ -67,3 +68,25 @@ func TestAUserTheBusClosedAsASlowConsumerIsOneConditionNamingItsModuleAndMachine } } } + +// **A read of the bus module's events that fails does not blind S9** (review of #326): the controller's own +// signals are still said, and the failed read is a condition of its own. +func TestABusModuleReadThatFailsKeepsTheControllersOwnSignals(t *testing.T) { + at := time.Date(2026, 10, 14, 12, 0, 0, 0, time.UTC) + f := &signalFacts{now: at.Add(time.Minute), host: "anchor", + advisories: []link.Advisory{{Kind: link.AdvisorySlowConsumer, ID: "controller", Said: "the controller fell behind", + First: at, Last: at, Count: 1}}, + busFaultsErr: errors.New("nats: timeout")} + keys := map[string]bool{} + for _, o := range watchAdvisories(f) { + keys[o.Key()] = true + } + if !keys["bus.controller.slow-consumer"] || !keys["bus.bus-module.bus-faults-unread"] { + t.Fatalf("observed %v: the controller's own slow consumer and the unread events, both", keys) + } + for _, row := range signalsTable { + if row.Row == "S9" && row.needs(f) != nil { + t.Fatal("the S9 row went blind for a read of the bus module's events") + } + } +} diff --git a/cmd/mesh-controller/plain_words.go b/cmd/mesh-controller/plain_words.go index 0e16f0cc..da0511e5 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"} }), + 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 " + + "client the bus cut off would go unnoticed. It tries again at every check.", + Resolved: "Resolved: the bus's clients are checked again"} + }), "connection-closed": worded(func(o conditions.Observation) words { return words{Headline: "The bus cut a client off", Explanation: "The bus closed a client's connection because it broke the bus's rules, so what it was " + diff --git a/cmd/mesh-controller/signals.go b/cmd/mesh-controller/signals.go index 1d5f257c..5603533d 100644 --- a/cmd/mesh-controller/signals.go +++ b/cmd/mesh-controller/signals.go @@ -586,9 +586,24 @@ func watchAdvisories(f *signalFacts) []conditions.Observation { } out = append(out, o) } + if f.busFaultsErr != nil { + out = append(out, busFaultsUnread(f.busFaultsErr)) + } return append(out, watchDeadLetters(f)...) } +// 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" + +func busFaultsUnread(err error) conditions.Observation { + return conditions.Observation{Scope: conditions.ScopeBus, ID: "bus-module", Kind: kindBusFaultsUnread, + Severity: conditions.Warning, Confirm: true, + Summary: "what the bus module said of its users' connections could not be read back, so a slow consumer " + + "or a connection the bus closed is not said until it can", + Said: firstLine(err.Error())} +} + // busUserClosed words an advisory about a bus user's connection the server closed (novox/hq ADR 0357): the // user as the operator says it and the machine it is on, the dotted user name kept in the evidence only. func busUserClosed(o *conditions.Observation, a link.Advisory, host string) { diff --git a/cmd/mesh-controller/watchdogs.go b/cmd/mesh-controller/watchdogs.go index a527df13..a8d2f793 100644 --- a/cmd/mesh-controller/watchdogs.go +++ b/cmd/mesh-controller/watchdogs.go @@ -92,6 +92,8 @@ type signalFacts struct { // consumers: the condition counts the messages given up on, not the looks (novox/hq issue 440). deadLettersNewest map[string]time.Time advisoriesErr error + // busFaultsErr is why the bus module's events could not be read this tick (novox/hq ADR 0357). + busFaultsErr error selfCheck selfCheckFacts @@ -360,9 +362,10 @@ func (w *watchdogs) gather(ctx context.Context) *signalFacts { f.lostConsumers, f.advisoriesErr = w.lostConsumers(ctx, f.advisories) // And what the bus did to every other user's connection, as the bus module said it (novox/hq ADR 0357): // read back for the same hour, and one advisory with the controller's own where both are its. - if f.advisoriesErr == nil && w.server != nil && w.js != nil { + // A read that fails is its own condition, never S9 blind: the controller's own signals are still said. + if w.server != nil && w.js != nil { var faults []link.Advisory - if faults, f.advisoriesErr = w.server.BusFaults(ctx, now.Add(-advisoryQuiet)); f.advisoriesErr == nil { + if faults, f.busFaultsErr = w.server.BusFaults(ctx, now.Add(-advisoryQuiet)); f.busFaultsErr == nil { f.advisories = link.MergeAdvisories(f.advisories, faults) } } diff --git a/internal/link/busfaults.go b/internal/link/busfaults.go index e1d5eb28..ff4eb64a 100644 --- a/internal/link/busfaults.go +++ b/internal/link/busfaults.go @@ -154,7 +154,13 @@ func (s *Server) BusFaults(ctx context.Context, since time.Time) ([]Advisory, er defer func() { _ = sub.Unsubscribe() }() var closures []BusClosure var kinds []string + read, cut := 0, false for { + if read >= MaxBusEventsRead { + cut = true + break + } + read++ wait, cancel := context.WithTimeout(ctx, readQuiet) msg, err := sub.NextMsgWithContext(wait) cancel() @@ -174,11 +180,23 @@ func (s *Server) BusFaults(ctx context.Context, since time.Time) ([]Advisory, er break } } - return BusClosures(closures, kinds, since), nil + out := BusClosures(closures, kinds, since) + if cut { + // Said, never silent: a busy hour is read up to the bound, and each advisory says it is a floor. + for i := range out { + out[i].Said += fmt.Sprintf(" (at least: the read stopped at %d of the bus module's events this hour)", + MaxBusEventsRead) + } + } + return out, nil } +// MaxBusEventsRead bounds one tick's read of the bus module's events: a watchdog tick must not spend itself +// reading a storm of closures back. A read cut there says so in every advisory it makes. +var MaxBusEventsRead = 2000 + // 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 +// counted once (the larger count of the two), 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...) @@ -195,7 +213,11 @@ func MergeAdvisories(a, b []Advisory) []Advisory { continue } x := &out[i] - x.Count += y.Count + // The same fault seen twice, not two faults: the controller's own slow consumer is told to its client + // library and closed by the server, and is counted once — by whichever saw more of it. + if y.Count > x.Count { + x.Count = y.Count + } if y.First.Before(x.First) { x.First = y.First } diff --git a/internal/link/busfaults_test.go b/internal/link/busfaults_test.go index 29d48935..d1eda4ef 100644 --- a/internal/link/busfaults_test.go +++ b/internal/link/busfaults_test.go @@ -94,7 +94,28 @@ func TestTheControllersOwnSlowConsumerAndItsClosedConnectionAreOneAdvisory(t *te 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) + 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) } }