S9 raises every bus user's slow consumer and closed connection, from the bus module's events (hq issue 270, ADR 0357) #326
@@ -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")
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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. " +
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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())
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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"]}`)
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user