From 9e50c1eb36f032c93cd197eabd84a6604288cb27 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 14 Oct 2026 08:41:47 +0200 Subject: [PATCH] Say every bus user's slow consumer and closed connection on the bus (hq issue 270, ADR 0357) The server records the user of every connection it closes as a slow consumer or for breaking its rules, and says it nowhere the controller hears. The nats tools bundle now reads that record every 30 seconds and publishes bus.slow-consumer and bus.connection-closed, naming the user, never a credential. --- modules/nats/cmd/nats-tools/main.go | 18 +- modules/nats/cmd/nats-tools/watch.go | 262 ++++++++++++++++++++++ modules/nats/cmd/nats-tools/watch_test.go | 206 +++++++++++++++++ modules/nats/module.json | 6 +- 4 files changed, 488 insertions(+), 4 deletions(-) create mode 100644 modules/nats/cmd/nats-tools/watch.go create mode 100644 modules/nats/cmd/nats-tools/watch_test.go diff --git a/modules/nats/cmd/nats-tools/main.go b/modules/nats/cmd/nats-tools/main.go index f2e0f6d2..ac7ccce7 100644 --- a/modules/nats/cmd/nats-tools/main.go +++ b/modules/nats/cmd/nats-tools/main.go @@ -1,6 +1,8 @@ // nats-tools: the mesh bus's own tools (novox/hq design 25), a Go bundle the node's runtime launches on the // machine that runs the bus. Everything here reads — the server's monitoring API and the user list the -// controller composed — and nothing changes the bus. stdout is the MCP channel; this says nothing else. +// controller composed — and nothing changes the bus. Beside the tools runs the watch (watch.go, novox/hq +// ADR 0357), which says on the bus every user's slow consumer the server closed. stdout is the MCP +// channel; this says nothing else. package main import ( @@ -29,7 +31,7 @@ func limitOf(a map[string]any, fallback int) int { return fallback } -func tools(monitor Monitor, users func(context.Context) ([]byte, error), written func(context.Context) (time.Time, error)) []stdio.Tool { +func tools(monitor Monitor, users func(context.Context) ([]byte, error), written func(context.Context) (time.Time, error), watch *Watch) []stdio.Tool { get := func(path string) ([]byte, error) { ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) defer cancel() @@ -112,6 +114,9 @@ func tools(monitor Monitor, users func(context.Context) ([]byte, error), written } return Buckets(doc) }}, + {Name: "nats_bus_watch", + Description: "The watch that says every bus user's slow consumer and every connection the bus closed for breaking its rules, as the events bus.slow-consumer and bus.connection-closed the controller raises as S9 (novox/hq ADR 0357): when it last read the server's record of closed connections, what it said, the newest closures said, the reasons it leaves to D15 and any it could not publish.", + Run: func(map[string]any) (any, error) { return watch.Report(), nil }}, {Name: "nats_users", Description: "The bus's users as the controller composed them — never a password hash. Without a user, each with how many grants it holds; with part of a user's name, its publish and subscribe grants in full.", Input: map[string]any{"user": str("part of a user name, e.g. g14.node-tools or controller")}, @@ -144,7 +149,14 @@ func tools(monitor Monitor, users func(context.Context) ([]byte, error), written } func main() { - if err := stdio.Serve("", tools(LiveMonitor(), ReadUsers, UsersWritten)); err != nil { + monitor := LiveMonitor() + // The watch publishes through the runtime that launched this bundle, under the module's name: the + // events the manifest declares, and nothing else (novox/hq ADR 0357). + watch := NewWatch(monitor, func(event string, body any) error { return stdio.Emit(event, body) }) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go watch.Run(ctx) + if err := stdio.Serve("", tools(monitor, ReadUsers, UsersWritten, watch)); err != nil { fmt.Fprintln(os.Stderr, err) os.Exit(1) } diff --git a/modules/nats/cmd/nats-tools/watch.go b/modules/nats/cmd/nats-tools/watch.go new file mode 100644 index 00000000..7eb40b08 --- /dev/null +++ b/modules/nats/cmd/nats-tools/watch.go @@ -0,0 +1,262 @@ +package main + +// The watch: every user's slow consumer and every connection the bus closed for breaking its rules, +// said on the bus (novox/hq ADR 0357, issue 270). +// +// **The server says these of other users only in its record of closed connections.** It closes every +// client it judges a slow consumer, and keeps the last 10,000 closed connections with the user and the +// reason (`/connz?state=closed&auth=1`) — about two days of the mesh's traffic. It publishes neither in +// the mesh's account; a system account would hear them, and the bus has none. So the bundle that already +// reads the record for `nats_closed_connections` looks at it every WatchEvery and publishes each closure +// that matters as an event the controller raises as S9. +// +// **Said at least once, counted once.** A closure is its user, its connection id and when it opened — +// the id alone is reused after the server restarts. Each one said is remembered while this process runs; +// one whose publish failed is tried again at the next look. On start the last hour (S9's window) is said +// again, so a restart repeats closures rather than losing them; the controller counts each by its +// identity, so a repeat changes no count. + +import ( + "context" + "encoding/json" + "fmt" + "os" + "sort" + "strings" + "sync" + "time" +) + +const ( + // WatchEvery is how often the watch reads the record of closed connections. With the controller's + // watchdogs ticking every 30 s too, a closure is a condition about a minute after the server closed it. + WatchEvery = 30 * time.Second + // WatchSaysAgain is how far back the first look says closures again: S9's window, an hour. + WatchSaysAgain = time.Hour + // EventSlowConsumer and EventConnectionClosed are the events the watch publishes (module.json `emits`). + EventSlowConsumer = "bus.slow-consumer" + EventConnectionClosed = "bus.connection-closed" +) + +// notSaid are the reasons a closure is routine: the client's own choice, the network or a machine asleep +// (D15 says a user dropped in a loop), the server's own shutdown, and a refused login, which names no user. +// Every other reason is said — one the module does not know included — because a new way for the bus to +// throw a client out is exactly what must not pass in silence. +var notSaid = map[string]bool{ + ClientClosed: true, + "Stale Connection": true, + "Read Error": true, + "Write Error": true, + "TLS Handshake Failure": true, + "Authentication Timeout": true, + "Authentication Failure": true, + "Server Shutdown": true, +} + +// isSlowConsumer says a closure's reason is the server judging the client a slow consumer: "Slow Consumer +// (Pending Bytes)" or "Slow Consumer (Write Deadline)". +func isSlowConsumer(reason string) bool { return strings.HasPrefix(reason, "Slow Consumer") } + +// Closure is one closed connection as the watch says it. The user and the connection's own name, never a +// credential: the record the server keeps names who logged in, and nothing they logged in with. +type Closure 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"` + // Lived is how long the connection was open, in seconds. + Lived float64 `json:"lived_seconds"` + // Pending is what the server still had to write to it when it closed, in bytes, where it says. + Pending int64 `json:"pending_bytes,omitempty"` +} + +// identity is what tells one closure from another across a server restart. +func (c Closure) identity() string { + return fmt.Sprintf("%s/%d/%s", c.User, c.CID, c.Opened.UTC().Format(time.RFC3339Nano)) +} + +// event is the event a closure is said as. +func (c Closure) event() string { + if isSlowConsumer(c.Reason) { + return EventSlowConsumer + } + return EventConnectionClosed +} + +// Closures are the closed connections in the record that are to be said, closed at or after since, oldest +// first. +func Closures(connz []byte, since time.Time) ([]Closure, error) { + var doc struct { + Connections []struct { + CID uint64 `json:"cid"` + Name string `json:"name"` + AuthorizedUser string `json:"authorized_user"` + Start time.Time `json:"start"` + Stop *time.Time `json:"stop"` + Reason string `json:"reason"` + Pending int64 `json:"pending_bytes"` + } `json:"connections"` + } + if err := json.Unmarshal(connz, &doc); err != nil { + return nil, fmt.Errorf("the bus's record of closed connections cannot be read: %w", err) + } + var out []Closure + for _, c := range doc.Connections { + if c.Stop == nil || c.Stop.Before(since) || c.AuthorizedUser == "" || notSaid[c.Reason] { + continue + } + reason := c.Reason + if reason == "" { + reason = "unsaid" + } + out = append(out, Closure{User: c.AuthorizedUser, Name: c.Name, CID: c.CID, Reason: reason, + Opened: c.Start.UTC(), Closed: c.Stop.UTC(), Lived: c.Stop.Sub(c.Start).Seconds(), Pending: c.Pending}) + } + sort.SliceStable(out, func(i, j int) bool { return out[i].Closed.Before(out[j].Closed) }) + return out, nil +} + +// Watch is the watch's state, shared with the tool that reports it. +type Watch struct { + monitor Monitor + emit func(string, any) error + now func() time.Time + + mu sync.Mutex + // said is every closure said, by identity, with when it closed: forgotten once older than the look's + // window, since the record will not be read that far back again. + said map[string]time.Time + lastLook time.Time + lastErr string + saidN map[string]int + failed map[string]string + newest []Closure + stopped string +} + +// NewWatch is a watch that has not looked yet. +func NewWatch(monitor Monitor, emit func(string, any) error) *Watch { + return &Watch{monitor: monitor, emit: emit, now: time.Now, said: map[string]time.Time{}, + saidN: map[string]int{}, failed: map[string]string{}} +} + +// Look reads the record once and says every closure not said yet. +func (w *Watch) Look(ctx context.Context) { + now := w.now() + // The first look says the last hour again; every later one reads that far back too, so a closure whose + // publish failed is still in reach, and what is said is told apart by the said set, not by time. + since := now.Add(-WatchSaysAgain) + read, cancel := context.WithTimeout(ctx, 20*time.Second) + defer cancel() + connz, err := w.monitor(read, fmt.Sprintf("/connz?state=closed&auth=1&limit=%d", maxClosedKept)) + var closures []Closure + if err == nil { + closures, err = Closures(connz, since) + } + w.mu.Lock() + w.lastLook = now + if err != nil { + w.lastErr = err.Error() + w.mu.Unlock() + fmt.Fprintf(os.Stderr, "the watch could not read the bus's closed connections: %v\n", err) + return + } + w.lastErr = "" + for id, closed := range w.said { + if closed.Before(since) { + delete(w.said, id) + } + } + var todo []Closure + for _, c := range closures { + if _, done := w.said[c.identity()]; !done { + todo = append(todo, c) + } + } + w.mu.Unlock() + + for _, c := range todo { + if w.emit == nil { + continue + } + err := w.emit(c.event(), c) + w.mu.Lock() + if err != nil { + // Tried again at the next look: it stays out of the said set. + w.failed[c.identity()] = err.Error() + w.mu.Unlock() + fmt.Fprintf(os.Stderr, "%s for %s not published, tried again at the next look: %v\n", c.event(), c.User, err) + continue + } + delete(w.failed, c.identity()) + w.said[c.identity()] = c.Closed + w.saidN[c.event()]++ + w.newest = append(w.newest, c) + if len(w.newest) > 10 { + w.newest = w.newest[len(w.newest)-10:] + } + w.mu.Unlock() + } +} + +// Run is the watch's life: a look every WatchEvery until the context ends. +func (w *Watch) Run(ctx context.Context) { + defer func() { + if r := recover(); r != nil { + w.mu.Lock() + w.stopped = fmt.Sprintf("the watch stopped on a fault: %v", r) + w.mu.Unlock() + fmt.Fprintln(os.Stderr, w.stopped) + } + }() + // The first look one interval in, not at once: the runtime that publishes for this bundle answers only + // once the bundle serves, and the first look reads the last hour back anyway. + t := time.NewTicker(WatchEvery) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + } + w.Look(ctx) + } +} + +// Report is the watch as the tool `nats_bus_watch` shows it. +func (w *Watch) Report() map[string]any { + w.mu.Lock() + defer w.mu.Unlock() + out := map[string]any{ + "running": w.stopped == "", + "every": WatchEvery.String(), + "reads_back": WatchSaysAgain.String(), + "events": []string{EventSlowConsumer, EventConnectionClosed}, + "said": w.saidN, + "remembered": len(w.said), + "newest_said": w.newest, + "not_said_reasons": func() []string { + var r []string + for k := range notSaid { + r = append(r, k) + } + sort.Strings(r) + return r + }(), + } + if w.stopped != "" { + out["stopped"] = w.stopped + } + if !w.lastLook.IsZero() { + out["last_look"] = w.lastLook.UTC() + } + if w.lastErr != "" { + out["last_error"] = w.lastErr + } + if len(w.failed) > 0 { + out["not_published"] = w.failed + } + return out +} diff --git a/modules/nats/cmd/nats-tools/watch_test.go b/modules/nats/cmd/nats-tools/watch_test.go new file mode 100644 index 00000000..24ebada5 --- /dev/null +++ b/modules/nats/cmd/nats-tools/watch_test.go @@ -0,0 +1,206 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "os" + "sort" + "strings" + "sync" + "testing" + "time" +) + +// The watch (novox/hq ADR 0357, issue 270): every user's slow consumer and every connection the bus closed +// for breaking its rules, said once as an event the controller raises as S9. + +var watchNow = time.Date(2026, 10, 14, 12, 0, 0, 0, time.UTC) + +// closedRecord is a record of closed connections as the server serves it, with a password-shaped field the +// server never serves, to prove nothing but the user is carried. +const closedRecord = `{"connections":[ + {"cid":10,"name":"g14.claude-code","authorized_user":"g14.claude-code","start":"2026-10-14T11:40:00Z","stop":"2026-10-14T11:58:00Z","reason":"Slow Consumer (Write Deadline)","pending_bytes":334000}, + {"cid":11,"name":"mesh-controller verb node","authorized_user":"controller","start":"2026-10-14T11:58:00Z","stop":"2026-10-14T11:58:01Z","reason":"Client Closed"}, + {"cid":12,"name":"mesh-host/g14","authorized_user":"node.g14","start":"2026-10-14T10:00:00Z","stop":"2026-10-14T11:59:00Z","reason":"Stale Connection"}, + {"cid":13,"name":"g14.node-tools","authorized_user":"g14.node-tools","start":"2026-10-14T10:00:00Z","stop":"2026-10-14T11:59:10Z","reason":"Read Error"}, + {"cid":14,"name":"shanks.desk","authorized_user":"shanks.desk","start":"2026-10-14T11:00:00Z","stop":"2026-10-14T11:59:20Z","reason":"Maximum Message Payload Exceeded"}, + {"cid":15,"name":"","authorized_user":"","start":"2026-10-14T11:59:00Z","stop":"2026-10-14T11:59:30Z","reason":"Authentication Failure"}, + {"cid":16,"name":"ace.media","authorized_user":"ace.media","start":"2026-10-14T09:00:00Z","stop":"2026-10-14T10:30:00Z","reason":"Slow Consumer (Pending Bytes)"}, + {"cid":17,"name":"ace.future","authorized_user":"ace.future","start":"2026-10-14T11:00:00Z","stop":"2026-10-14T11:59:40Z","reason":"A Reason Not Yet Invented","password":"never-carried"}, + {"cid":18,"name":"ace.route-proxy","authorized_user":"ace.route-proxy","start":"2026-10-14T11:00:00Z","stop":"2026-10-14T11:59:50Z","reason":"Server Shutdown"} +]}` + +type published struct { + event string + body Closure +} + +type recorder struct { + mu sync.Mutex + got []published + fail error +} + +func (r *recorder) emit(event string, body any) error { + r.mu.Lock() + defer r.mu.Unlock() + if r.fail != nil { + return r.fail + } + raw, _ := json.Marshal(body) + if strings.Contains(string(raw), "never-carried") || strings.Contains(strings.ToLower(string(raw)), "password") { + panic("a credential was published: " + string(raw)) + } + var c Closure + _ = json.Unmarshal(raw, &c) + r.got = append(r.got, published{event, c}) + return nil +} + +func (r *recorder) said() []string { + r.mu.Lock() + defer r.mu.Unlock() + var out []string + for _, p := range r.got { + out = append(out, p.event+" "+p.body.User) + } + sort.Strings(out) + return out +} + +func recordOf(doc *string) Monitor { + return func(_ context.Context, path string) ([]byte, error) { + if !strings.HasPrefix(path, "/connz?state=closed&auth=1") { + return nil, errors.New("the watch asked " + path) + } + return []byte(*doc), nil + } +} + +func watchAt(doc *string, r *recorder) *Watch { + w := NewWatch(recordOf(doc), r.emit) + w.now = func() time.Time { return watchNow } + return w +} + +func TestTheWatchSaysEachClosureTheServerDroppedNamingItsUserAndNothingRoutine(t *testing.T) { + doc := closedRecord + r := &recorder{} + watchAt(&doc, r).Look(context.Background()) + want := []string{ + "bus.connection-closed ace.future", // a reason the module does not know is said + "bus.connection-closed shanks.desk", // a limit broken + "bus.slow-consumer g14.claude-code", // the slow consumer research 036 could not name + } + if got := r.said(); strings.Join(got, "|") != strings.Join(want, "|") { + t.Fatalf("said %v, want %v (none for the client's own close, the network, a refused login, the "+ + "server's shutdown, or a closure older than the hour)", got, want) + } + for _, p := range r.got { + if p.body.User == "g14.claude-code" { + if p.body.CID != 10 || p.body.Reason != "Slow Consumer (Write Deadline)" || p.body.Pending != 334000 || + !p.body.Closed.Equal(time.Date(2026, 10, 14, 11, 58, 0, 0, time.UTC)) || p.body.Lived != 18*60 { + t.Fatalf("the slow consumer said as %+v", p.body) + } + } + } +} + +func TestTheWatchSaysAClosureOnceWhileItRuns(t *testing.T) { + doc := closedRecord + r := &recorder{} + w := watchAt(&doc, r) + w.Look(context.Background()) + w.Look(context.Background()) + if got := len(r.said()); got != 3 { + t.Fatalf("said %d events over two looks at the same record, want 3: %v", got, r.said()) + } +} + +func TestAClosureWhosePublishFailedIsTriedAgainAtTheNextLook(t *testing.T) { + doc := closedRecord + r := &recorder{fail: errors.New("the runtime did not answer")} + w := watchAt(&doc, r) + w.Look(context.Background()) + if len(r.said()) != 0 { + t.Fatal("a failed publish was counted as said") + } + if report := w.Report(); len(report["not_published"].(map[string]string)) != 3 { + t.Fatalf("the watch does not say what it could not publish: %v", report) + } + r.fail = nil + w.Look(context.Background()) + if got := len(r.said()); got != 3 { + t.Fatalf("said %d after the runtime came back, want the 3 that failed", got) + } + if _, left := w.Report()["not_published"]; left { + t.Fatal("a closure published at last is still said not published") + } +} + +func TestAWatchStartedAgainSaysTheLastHourAgain(t *testing.T) { + doc := closedRecord + first := &recorder{} + watchAt(&doc, first).Look(context.Background()) + again := &recorder{} + watchAt(&doc, again).Look(context.Background()) + if strings.Join(first.said(), "|") != strings.Join(again.said(), "|") || len(again.said()) != 3 { + t.Fatalf("a restarted watch said %v, the first %v: the last hour must be said again, for the controller "+ + "to count once", again.said(), first.said()) + } +} + +func TestAReusedConnectionIDWithANewOpeningIsANewClosure(t *testing.T) { + doc := `{"connections":[{"cid":10,"authorized_user":"g14.claude-code","start":"2026-10-14T11:40:00Z","stop":"2026-10-14T11:50:00Z","reason":"Slow Consumer (Pending Bytes)"}]}` + r := &recorder{} + w := watchAt(&doc, r) + w.Look(context.Background()) + // The server restarted and numbered connections from one again. + doc = `{"connections":[ + {"cid":10,"authorized_user":"g14.claude-code","start":"2026-10-14T11:40:00Z","stop":"2026-10-14T11:50:00Z","reason":"Slow Consumer (Pending Bytes)"}, + {"cid":10,"authorized_user":"g14.claude-code","start":"2026-10-14T11:55:00Z","stop":"2026-10-14T11:59:00Z","reason":"Slow Consumer (Pending Bytes)"}]}` + w.Look(context.Background()) + if got := len(r.said()); got != 2 { + t.Fatalf("said %d, want 2: the same connection id opened again is another closure", got) + } +} + +func TestTheWatchSaysItCannotReadTheRecord(t *testing.T) { + r := &recorder{} + w := NewWatch(func(context.Context, string) ([]byte, error) { return nil, errors.New("connection refused") }, r.emit) + w.Look(context.Background()) + if report := w.Report(); !strings.Contains(report["last_error"].(string), "connection refused") { + t.Fatalf("a record that could not be read is not said: %v", report) + } +} + +func TestTheManifestDeclaresTheToolsServedTheWatchesEventsAndItsInterval(t *testing.T) { + raw, err := os.ReadFile("../../module.json") + if err != nil { + t.Fatal(err) + } + var m struct { + Tools []string `json:"tools"` + Emits []string `json:"emits"` + } + if err := json.Unmarshal(raw, &m); err != nil { + t.Fatal(err) + } + var served []string + for _, tool := range tools(nil, nil, nil, NewWatch(nil, nil)) { + served = append(served, tool.Name) + } + sort.Strings(served) + sort.Strings(m.Tools) + if strings.Join(served, ",") != strings.Join(m.Tools, ",") { + t.Fatalf("served %v, the manifest lists %v", served, m.Tools) + } + sort.Strings(m.Emits) + if strings.Join(m.Emits, ",") != EventConnectionClosed+","+EventSlowConsumer { + t.Fatalf("the manifest emits %v: the runtime may publish only what is declared", m.Emits) + } + if WatchEvery != 30*time.Second || WatchSaysAgain != time.Hour { + t.Fatalf("the watch looks every %s and reads back %s; ADR 0357 says 30s and an hour", WatchEvery, WatchSaysAgain) + } +} diff --git a/modules/nats/module.json b/modules/nats/module.json index d8647b3e..1cc6000e 100644 --- a/modules/nats/module.json +++ b/modules/nats/module.json @@ -24,7 +24,10 @@ "capabilities": [ "container-runtime" ], - "emits": [], + "emits": [ + "bus.slow-consumer", + "bus.connection-closed" + ], "consumes": [], "listens": [ { @@ -39,6 +42,7 @@ "nats_server", "nats_connections", "nats_closed_connections", + "nats_bus_watch", "nats_subscriptions", "nats_streams", "nats_backlog",