From 9e50c1eb36f032c93cd197eabd84a6604288cb27 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 14 Oct 2026 08:41:47 +0200 Subject: [PATCH 1/2] 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", -- 2.54.0 From 950e37a46e8dad69dec53afd68d78f46ce4d34df Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 14 Oct 2026 09:19:56 +0200 Subject: [PATCH 2/2] Keep the nats watch alive and heard: recover a faulted look, beat after every look, read only what closed since (hq issue 270, review 274) A watch that panicked stopped for good, and one that could not read said so only to stderr, so a dead watch read as no slow consumers. A faulted look is now recovered and the loop goes on; every look ends with a bus.watch heartbeat saying whether the watch is healthy, which the controller raises when it stops or says it cannot see. The record is read newest first back to the last look, and closures that could not be published are forgotten after the hour. --- modules/nats/cmd/nats-tools/watch.go | 261 ++++++++++++++++++---- modules/nats/cmd/nats-tools/watch_test.go | 140 +++++++++++- modules/nats/module.json | 3 +- 3 files changed, 353 insertions(+), 51 deletions(-) diff --git a/modules/nats/cmd/nats-tools/watch.go b/modules/nats/cmd/nats-tools/watch.go index 7eb40b08..84fe351a 100644 --- a/modules/nats/cmd/nats-tools/watch.go +++ b/modules/nats/cmd/nats-tools/watch.go @@ -36,6 +36,10 @@ const ( // EventSlowConsumer and EventConnectionClosed are the events the watch publishes (module.json `emits`). EventSlowConsumer = "bus.slow-consumer" EventConnectionClosed = "bus.connection-closed" + // EventWatch is the watch's heartbeat, after every look: whether it is healthy and why not. The controller + // raises a watch that stops saying it, or says it is not healthy (novox/hq ADR 0357, review 274): a dead + // watch must never read as no slow consumers. + EventWatch = "bus.watch" ) // notSaid are the reasons a closure is routine: the client's own choice, the network or a machine asleep @@ -118,66 +122,155 @@ func Closures(connz []byte, since time.Time) ([]Closure, error) { return out, nil } -// Watch is the watch's state, shared with the tool that reports it. +// Watch is the watch's state, shared with the tools that report it. type Watch struct { monitor Monitor emit func(string, any) error now func() time.Time + started 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 + // said is every closure said, by identity, with when it closed: forgotten once older than the hour, since + // the record is never read that far back again. + said map[string]time.Time + // failed are the closures whose publish failed, by identity, tried again at every look until said or + // older than the hour. + failed map[string]failedClosure + // lastLook is when the watch last looked; lastRead when a look last read the record whole. + lastLook, lastRead time.Time + // failingSince is when the reads began to fail, zero while they succeed. + failingSince time.Time + lastErr string + saidN map[string]int + newest []Closure + // faults are the looks that panicked and were recovered: the loop goes on, and each is said. + faults int + lastFault string +} + +type failedClosure struct { + closure Closure + err string + since time.Time } // 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{}} + return &Watch{monitor: monitor, emit: emit, now: time.Now, started: time.Now(), said: map[string]time.Time{}, + saidN: map[string]int{}, failed: map[string]failedClosure{}} } -// Look reads the record once and says every closure not said yet. +// readPage is how many closed connections one request asks for, newest first. +const readPage = 256 + +// readSince reads the record newest first (`sort=stop`), a page at a time, until a page reaches back past since +// or the record ends: a look reads only what closed since the last one, not all 10,000 kept. +func (w *Watch) readSince(ctx context.Context, since time.Time) ([]Closure, error) { + var out []Closure + for offset := 0; offset < maxClosedKept; offset += readPage { + page, err := w.monitor(ctx, fmt.Sprintf("/connz?state=closed&auth=1&sort=stop&limit=%d&offset=%d", readPage, offset)) + if err != nil { + return nil, err + } + closures, oldest, n, err := closuresOfPage(page, since) + if err != nil { + return nil, err + } + out = append(out, closures...) + if n < readPage || (!oldest.IsZero() && oldest.Before(since)) { + break + } + } + sort.SliceStable(out, func(i, j int) bool { return out[i].Closed.Before(out[j].Closed) }) + return out, nil +} + +// closuresOfPage is one page's closures to say, its oldest stop and how many connections it held. +func closuresOfPage(page []byte, since time.Time) ([]Closure, time.Time, int, error) { + var doc struct { + Connections []struct { + Stop *time.Time `json:"stop"` + } `json:"connections"` + } + if err := json.Unmarshal(page, &doc); err != nil { + return nil, time.Time{}, 0, fmt.Errorf("the bus's record of closed connections cannot be read: %w", err) + } + var oldest time.Time + for _, c := range doc.Connections { + if c.Stop != nil && (oldest.IsZero() || c.Stop.Before(oldest)) { + oldest = *c.Stop + } + } + closures, err := Closures(page, since) + return closures, oldest, len(doc.Connections), err +} + +// Look reads what closed since the last look and says every closure not said yet, and tries again every +// closure whose publish failed. A fault in a look is recovered and said: the watch goes on. func (w *Watch) Look(ctx context.Context) { + defer func() { + if r := recover(); r != nil { + w.mu.Lock() + w.faults++ + w.lastFault = fmt.Sprintf("a look stopped on a fault at %s: %v", w.now().UTC().Format(time.RFC3339), r) + w.mu.Unlock() + fmt.Fprintln(os.Stderr, w.lastFault) + } + }() 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) + hour := now.Add(-WatchSaysAgain) + w.mu.Lock() + // The first look says the last hour again; a later one reads from one interval before the last whole + // read, so a closure the server recorded late is not missed — the said set tells what was said. + since := hour + if !w.lastRead.IsZero() && w.lastRead.Add(-WatchEvery).After(hour) { + since = w.lastRead.Add(-WatchEvery) + } + w.mu.Unlock() 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) - } + closures, err := w.readSince(read, 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) { + if closed.Before(hour) { delete(w.said, id) } } - var todo []Closure + for id, f := range w.failed { + if f.closure.Closed.Before(hour) { + delete(w.failed, id) // S9 would no longer say it + } + } + if err != nil { + w.lastErr = err.Error() + if w.failingSince.IsZero() { + w.failingSince = now + } + fmt.Fprintf(os.Stderr, "the watch could not read the bus's closed connections: %v\n", err) + } else { + w.lastErr, w.failingSince, w.lastRead = "", time.Time{}, now + } + todo := map[string]Closure{} for _, c := range closures { if _, done := w.said[c.identity()]; !done { - todo = append(todo, c) + todo[c.identity()] = c + } + } + for id, f := range w.failed { + if _, done := w.said[id]; !done { + todo[id] = f.closure } } w.mu.Unlock() + order := make([]Closure, 0, len(todo)) for _, c := range todo { + order = append(order, c) + } + sort.SliceStable(order, func(i, j int) bool { return order[i].Closed.Before(order[j].Closed) }) + for _, c := range order { if w.emit == nil { continue } @@ -185,7 +278,12 @@ func (w *Watch) Look(ctx context.Context) { 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() + f := w.failed[c.identity()] + if f.since.IsZero() { + f.since = now + } + f.closure, f.err = c, err.Error() + w.failed[c.identity()] = f 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 @@ -199,18 +297,40 @@ func (w *Watch) Look(ctx context.Context) { } w.mu.Unlock() } + w.beat() } -// Run is the watch's life: a look every WatchEvery until the context ends. +// Beat is the watch's heartbeat as it publishes it. +type Beat struct { + At time.Time `json:"at"` + Healthy bool `json:"healthy"` + Why string `json:"why,omitempty"` + LastRead *time.Time `json:"last_read,omitempty"` + Every string `json:"every"` +} + +// beat says the watch is there, and whether it is healthy. One that cannot be published is the controller's to +// notice by its absence. +func (w *Watch) beat() { + if w.emit == nil { + return + } + w.mu.Lock() + healthy, why := w.healthLocked() + b := Beat{At: w.now().UTC(), Healthy: healthy, Why: why, Every: WatchEvery.String()} + if !w.lastRead.IsZero() { + at := w.lastRead.UTC() + b.LastRead = &at + } + w.mu.Unlock() + if err := w.emit(EventWatch, b); err != nil { + fmt.Fprintf(os.Stderr, "the watch's heartbeat not published: %v\n", err) + } +} + +// Run is the watch's life: a look every WatchEvery until the context ends. A look that faults is recovered in +// Look, so the loop never ends but with the context. 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) @@ -225,12 +345,53 @@ func (w *Watch) Run(ctx context.Context) { } } +// WatchStale is how long the watch may go without reading the record whole, or keep a closure unpublished, +// before it is not healthy: three looks. +const WatchStale = 3 * WatchEvery + +// Health is the watch's own verdict: healthy while it has read the record whole within WatchStale (after the +// same grace from its start) and holds no closure unpublished for longer. Said in every heartbeat, so the +// controller raises a watch that cannot read (novox/hq ADR 0357, review 274). +func (w *Watch) Health() (bool, string) { + w.mu.Lock() + defer w.mu.Unlock() + return w.healthLocked() +} + +func (w *Watch) healthLocked() (bool, string) { + now := w.now() + switch { + case w.lastRead.IsZero() && now.Sub(w.started) > WatchStale: + why := fmt.Sprintf("the watch of slow consumers has not read the bus's closed connections since it started %s ago", + now.Sub(w.started).Round(time.Second)) + if w.lastErr != "" { + why += ": " + w.lastErr + } + return false, why + case !w.lastRead.IsZero() && now.Sub(w.lastRead) > WatchStale: + why := fmt.Sprintf("the watch of slow consumers last read the bus's closed connections %s ago", + now.Sub(w.lastRead).Round(time.Second)) + if w.lastErr != "" { + why += ": " + w.lastErr + } + return false, why + } + for _, f := range w.failed { + if now.Sub(f.since) > WatchStale { + return false, fmt.Sprintf("the watch could not publish %s for %s for %s: %s", f.closure.event(), + f.closure.User, now.Sub(f.since).Round(time.Second), f.err) + } + } + return true, "" +} + // 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() + healthy, why := w.healthLocked() out := map[string]any{ - "running": w.stopped == "", + "healthy": healthy, "every": WatchEvery.String(), "reads_back": WatchSaysAgain.String(), "events": []string{EventSlowConsumer, EventConnectionClosed}, @@ -246,8 +407,14 @@ func (w *Watch) Report() map[string]any { return r }(), } - if w.stopped != "" { - out["stopped"] = w.stopped + if why != "" { + out["why"] = why + } + if w.faults > 0 { + out["faults_recovered"], out["last_fault"] = w.faults, w.lastFault + } + if !w.lastRead.IsZero() { + out["last_read"] = w.lastRead.UTC() } if !w.lastLook.IsZero() { out["last_look"] = w.lastLook.UTC() @@ -256,7 +423,11 @@ func (w *Watch) Report() map[string]any { out["last_error"] = w.lastErr } if len(w.failed) > 0 { - out["not_published"] = w.failed + failed := map[string]string{} + for id, f := range w.failed { + failed[id] = f.err + } + out["not_published"] = failed } return out } diff --git a/modules/nats/cmd/nats-tools/watch_test.go b/modules/nats/cmd/nats-tools/watch_test.go index 24ebada5..06ddbbd3 100644 --- a/modules/nats/cmd/nats-tools/watch_test.go +++ b/modules/nats/cmd/nats-tools/watch_test.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "os" "sort" "strings" @@ -37,9 +38,10 @@ type published struct { } type recorder struct { - mu sync.Mutex - got []published - fail error + mu sync.Mutex + got []published + beats []Beat + fail error } func (r *recorder) emit(event string, body any) error { @@ -49,6 +51,12 @@ func (r *recorder) emit(event string, body any) error { return r.fail } raw, _ := json.Marshal(body) + if event == EventWatch { + var b Beat + _ = json.Unmarshal(raw, &b) + r.beats = append(r.beats, b) + return nil + } if strings.Contains(string(raw), "never-carried") || strings.Contains(strings.ToLower(string(raw)), "password") { panic("a credential was published: " + string(raw)) } @@ -159,7 +167,8 @@ func TestAReusedConnectionIDWithANewOpeningIsANewClosure(t *testing.T) { // 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)"}]}` + {"cid":10,"authorized_user":"g14.claude-code","start":"2026-10-14T12:00:05Z","stop":"2026-10-14T12:00:20Z","reason":"Slow Consumer (Pending Bytes)"}]}` + w.now = func() time.Time { return watchNow.Add(WatchEvery) } 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) @@ -175,6 +184,127 @@ func TestTheWatchSaysItCannotReadTheRecord(t *testing.T) { } } +// **A dead watch is heard** (review 274): every look ends with a heartbeat saying whether the watch is +// healthy; a record it cannot read for longer than three looks is said not healthy, with why, rather than read +// as no slow consumers; and it recovers with a read. +func TestAWatchThatCannotReadForThreeLooksSaysSoInItsHeartbeat(t *testing.T) { + broken := true + doc := closedRecord + clock := watchNow + r := &recorder{} + w := NewWatch(func(ctx context.Context, path string) ([]byte, error) { + if broken { + return nil, errors.New("connection refused") + } + return recordOf(&doc)(ctx, path) + }, r.emit) + w.now, w.started = func() time.Time { return clock }, clock + w.Look(context.Background()) + if len(r.beats) != 1 || !r.beats[0].Healthy { + t.Fatalf("a watch that just started beat %+v", r.beats) + } + clock = clock.Add(WatchStale + time.Second) + w.Look(context.Background()) + last := r.beats[len(r.beats)-1] + if last.Healthy || !strings.Contains(last.Why, "connection refused") || !last.At.Equal(clock) { + t.Fatalf("a watch that has not read for %s beat %+v", WatchStale, last) + } + broken = false + w.Look(context.Background()) + if last := r.beats[len(r.beats)-1]; !last.Healthy || last.LastRead == nil { + t.Fatalf("a watch that reads again beat %+v", last) + } +} + +// A closure that cannot be published for longer than three looks is not healthy either. +func TestAClosureUnpublishedForThreeLooksIsNotHealthy(t *testing.T) { + doc := closedRecord + r := &recorder{fail: errors.New("the runtime did not answer")} + clock := watchNow + w := watchAt(&doc, r) + w.now, w.started = func() time.Time { return clock }, clock + w.Look(context.Background()) + clock = clock.Add(WatchStale + time.Second) + w.Look(context.Background()) + if ok, why := w.Health(); ok || !strings.Contains(why, "could not publish") { + t.Fatalf("closures unpublished for %s read healthy %v: %q", WatchStale, ok, why) + } +} + +// **A look that panics does not end the watch** (review 274): it is recovered and said, and the next look runs. +func TestALookThatPanicsIsRecoveredAndTheWatchGoesOn(t *testing.T) { + calls := 0 + doc := closedRecord + r := &recorder{} + w := NewWatch(func(ctx context.Context, path string) ([]byte, error) { + calls++ + if calls == 1 { + panic("a nil map somewhere") + } + return recordOf(&doc)(ctx, path) + }, r.emit) + w.now = func() time.Time { return watchNow } + w.Look(context.Background()) + w.Look(context.Background()) + if len(r.said()) != 3 { + t.Fatalf("after a look that panicked the next said %v", r.said()) + } + if report := w.Report(); report["faults_recovered"] != 1 { + t.Fatalf("the recovered fault is not said: %v", report) + } +} + +// **The record is read newest first, and only back to the last look** (review 274), never all 10,000 kept. +func TestALaterLookReadsOnlyWhatClosedSinceTheLastOne(t *testing.T) { + var asked []string + page := func(n int, newest time.Time) []byte { + var conns []string + for i := range n { + stop := newest.Add(-time.Duration(i) * time.Second) + conns = append(conns, fmt.Sprintf(`{"cid":%d,"authorized_user":"u","start":"%s","stop":"%s","reason":"Client Closed"}`, + 1000-i, stop.Add(-time.Minute).Format(time.RFC3339), stop.Format(time.RFC3339))) + } + return []byte(`{"connections":[` + strings.Join(conns, ",") + `]}`) + } + clock := watchNow + w := NewWatch(func(_ context.Context, path string) ([]byte, error) { + asked = append(asked, path) + if !strings.Contains(path, "sort=stop") { + return nil, errors.New("not read newest first: " + path) + } + if strings.Contains(path, "offset=0") { + return page(readPage, clock), nil // a full page of the last few minutes + } + return page(readPage, clock.Add(-2*time.Hour)), nil // older than any look reaches + }, (&recorder{}).emit) + w.now = func() time.Time { return clock } + w.Look(context.Background()) + first := len(asked) + asked = nil + clock = clock.Add(WatchEvery) + w.Look(context.Background()) + if len(asked) != 1 || first > 2 { + t.Fatalf("the first look asked %d pages and a later one %d (%v): a later look reads back only to the last", first, + len(asked), asked) + } +} + +// The failed-publish record forgets a closure older than the hour: S9 would no longer say it. +func TestAClosureThatCouldNotBePublishedIsForgottenAfterTheHour(t *testing.T) { + doc := closedRecord + r := &recorder{fail: errors.New("the runtime did not answer")} + clock := watchNow + w := watchAt(&doc, r) + w.now = func() time.Time { return clock } + w.Look(context.Background()) + doc = `{"connections":[]}` + clock = clock.Add(2 * time.Hour) + w.Look(context.Background()) + if _, left := w.Report()["not_published"]; left { + t.Fatal("closures older than the hour are still kept as not published") + } +} + func TestTheManifestDeclaresTheToolsServedTheWatchesEventsAndItsInterval(t *testing.T) { raw, err := os.ReadFile("../../module.json") if err != nil { @@ -197,7 +327,7 @@ func TestTheManifestDeclaresTheToolsServedTheWatchesEventsAndItsInterval(t *test t.Fatalf("served %v, the manifest lists %v", served, m.Tools) } sort.Strings(m.Emits) - if strings.Join(m.Emits, ",") != EventConnectionClosed+","+EventSlowConsumer { + if strings.Join(m.Emits, ",") != EventConnectionClosed+","+EventSlowConsumer+","+EventWatch { t.Fatalf("the manifest emits %v: the runtime may publish only what is declared", m.Emits) } if WatchEvery != 30*time.Second || WatchSaysAgain != time.Hour { diff --git a/modules/nats/module.json b/modules/nats/module.json index 1cc6000e..510ae258 100644 --- a/modules/nats/module.json +++ b/modules/nats/module.json @@ -26,7 +26,8 @@ ], "emits": [ "bus.slow-consumer", - "bus.connection-closed" + "bus.connection-closed", + "bus.watch" ], "consumes": [], "listens": [ -- 2.54.0