From 45870ab31cc57ae86600d40a3b685727006a774a Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 14 Oct 2026 09:02:47 +0200 Subject: [PATCH] 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) } }