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) } }