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..84fe351a --- /dev/null +++ b/modules/nats/cmd/nats-tools/watch.go @@ -0,0 +1,433 @@ +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" + // 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 +// (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 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 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, started: time.Now(), said: map[string]time.Time{}, + saidN: map[string]int{}, failed: map[string]failedClosure{}} +} + +// 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() + 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() + closures, err := w.readSince(read, since) + + w.mu.Lock() + w.lastLook = now + for id, closed := range w.said { + if closed.Before(hour) { + delete(w.said, id) + } + } + 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[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 + } + 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. + 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 + } + 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() + } + w.beat() +} + +// 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) { + // 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) + } +} + +// 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{ + "healthy": healthy, + "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 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() + } + if w.lastErr != "" { + out["last_error"] = w.lastErr + } + if len(w.failed) > 0 { + 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 new file mode 100644 index 00000000..06ddbbd3 --- /dev/null +++ b/modules/nats/cmd/nats-tools/watch_test.go @@ -0,0 +1,336 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "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 + beats []Beat + 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 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)) + } + 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-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) + } +} + +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) + } +} + +// **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 { + 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+","+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 { + 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..510ae258 100644 --- a/modules/nats/module.json +++ b/modules/nats/module.json @@ -24,7 +24,11 @@ "capabilities": [ "container-runtime" ], - "emits": [], + "emits": [ + "bus.slow-consumer", + "bus.connection-closed", + "bus.watch" + ], "consumes": [], "listens": [ { @@ -39,6 +43,7 @@ "nats_server", "nats_connections", "nats_closed_connections", + "nats_bus_watch", "nats_subscriptions", "nats_streams", "nats_backlog",