jochen
45870ab31c
mesh/merge-gate head 45870ab31cc5 merged into trunk 3479a35462b2 — pass: builds build-agent, mesh-controller → ace, g14, novox, shanks; no bus step; every …
mesh/repo-check head 45870ab31cc5 merged into trunk 3479a35462b2 — pass: its merge-check.sh passed
mesh/delivery delivered
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.
93 lines
4.2 KiB
Go
93 lines
4.2 KiB
Go
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")
|
|
}
|
|
}
|
|
}
|