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": [