S9 raises every bus user's slow consumer and closed connection, from the bus module's events (hq issue 270, ADR 0357) #326

Merged
mesh-admin merged 2 commits from fix/270-the-bus-module-says-every-users-slow-consumer into main 2026-10-14 07:14:33 +00:00
6 changed files with 97 additions and 7 deletions
Showing only changes of commit 45870ab31c - Show all commits
@@ -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")
}
}
}
+6
View File
@@ -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 " +
+15
View File
@@ -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) {
+5 -2
View File
@@ -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)
}
}
+25 -3
View File
@@ -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
}
+23 -2
View File
@@ -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)
}
}