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..689bd9f8 --- /dev/null +++ b/cmd/mesh-controller/bus_users_closed_test.go @@ -0,0 +1,92 @@ +package main + +import ( + "errors" + "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) + } + } +} + +// **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 7d60b16a..da0511e5 100644 --- a/cmd/mesh-controller/plain_words.go +++ b/cmd/mesh-controller/plain_words.go @@ -620,6 +620,18 @@ 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 " + + "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..5603533d 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", @@ -579,9 +586,58 @@ 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) { + 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..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 @@ -358,6 +360,15 @@ 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. + // 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.busFaultsErr = w.server.BusFaults(ctx, now.Add(-advisoryQuiet)); f.busFaultsErr == 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..ff4eb64a --- /dev/null +++ b/internal/link/busfaults.go @@ -0,0 +1,235 @@ +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 + read, cut := 0, false + for { + if read >= MaxBusEventsRead { + cut = true + break + } + read++ + 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 + } + } + 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, +// 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...) + 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] + // 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 + } + 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..d1eda4ef --- /dev/null +++ b/internal/link/busfaults_test.go @@ -0,0 +1,121 @@ +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 != 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) + } +}