From 4b2d5aeaab019b720959cd26a67417854cb93c2f Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 12:01:04 +0100 Subject: [PATCH] CAT-N6: the keeper's health tool nats_keeper_health, which the process keeper's tool check asks It reads the keeper's status from its own state asked, what the keeper holds, and how many accepted verdicts wait on its consumer _nats_keeper on EVENTS from the monitoring endpoint, and answers HealthOf's judgement: unhealthy, naming why, past the three-minute bound; the entries refused, waiting and past the raise mark named beside. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig --- modules/nats/cmd/nats-tools/keep_test.go | 30 +++++++ modules/nats/cmd/nats-tools/keeper_health.go | 89 ++++++++++++++++++++ modules/nats/cmd/nats-tools/main.go | 1 + modules/nats/module.json | 3 +- 4 files changed, 122 insertions(+), 1 deletion(-) create mode 100644 modules/nats/cmd/nats-tools/keeper_health.go diff --git a/modules/nats/cmd/nats-tools/keep_test.go b/modules/nats/cmd/nats-tools/keep_test.go index 61f525d1..40a78a30 100644 --- a/modules/nats/cmd/nats-tools/keep_test.go +++ b/modules/nats/cmd/nats-tools/keep_test.go @@ -4,8 +4,11 @@ import ( "fmt" "strings" "testing" + "time" "git.novox.be/novox/mesh-sdk/go/catalogue" + + "nats-tools/internal/keeper" ) // The keeper's inputs (design 25 § 12.2; CAT-N2): a manifest-accepted read with the indexer's revision when it carries @@ -63,3 +66,30 @@ func line(module, digest string, revision uint64, entries []catalogue.BusStateEn } return fmt.Sprintf("%s %s %d [%s]", module, digest, revision, strings.Join(es, " ")) } + +// The keeper's health tool (CAT-N6): healthy naming its pass; unhealthy, naming why, when its pass is old or verdicts +// wait on its consumer `_nats_keeper` past the bound. +func TestTheKeepersHealthToolReadsItsStatusAndItsConsumer(t *testing.T) { + at := time.Date(2026, 11, 4, 12, 0, 0, 0, time.UTC) + jsz := `{"account_details":[{"stream_detail":[{"name":"EVENTS","consumer_detail":[{"name":"anchor_nats_keeper","num_pending":3},{"name":"anchor_nats","num_pending":9}]}]}]}` + status := keeper.Status{Started: at.Add(-time.Hour), Pass: at.Add(-time.Minute), Heard: at.Add(-time.Minute), + FirstView: keeper.FirstViewStatus{Owner: keeper.FirstViewOwner, Verb: keeper.FirstViewVerb, Answered: at.Add(-time.Hour)}} + none := func() (map[string]keeper.Asked, error) { return nil, nil } + for _, row := range []struct { + name string + now time.Time + state string + why string + }{ + {"its pass is fresh", at, "healthy", "last pass 2026-11-04T11:59:00Z"}, + {"its pass and its verdicts are old", at.Add(5 * time.Minute), "unhealthy", "3 accepted verdict(s) wait on its consumer"}, + } { + t.Run(row.name, func(t *testing.T) { + a := keeperHealth(func() (keeper.Status, error) { return status, nil }, none, + func() ([]byte, error) { return []byte(jsz), nil }, func() time.Time { return row.now }) + if a.State != row.state || !strings.Contains(a.Why, row.why) { + t.Errorf("answered %+v: %s, naming %q", a, row.state, row.why) + } + }) + } +} diff --git a/modules/nats/cmd/nats-tools/keeper_health.go b/modules/nats/cmd/nats-tools/keeper_health.go new file mode 100644 index 00000000..66f8ef75 --- /dev/null +++ b/modules/nats/cmd/nats-tools/keeper_health.go @@ -0,0 +1,89 @@ +package main + +import ( + "encoding/json" + "strings" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/health" + + "nats-tools/internal/keeper" +) + +// The keeper's health, as the node-engine asks it of the process keeper (design 25 § 12.2-12.3; CAT-N6): its status +// from its own state asked, what it holds, and how many accepted verdicts wait on its consumer, read from the bus's +// monitoring endpoint. + +// StatusInState reads the keeper's status through the runtime. +func StatusInState() (keeper.Status, error) { + e, err := stdio.State(keeper.StateAsked).Get(keeper.KeyStatus) + if err != nil || e == nil { + return keeper.Status{}, err + } + var s keeper.Status + return s, json.Unmarshal(e.Value, &s) +} + +// keeperPending is how many messages wait on the keeper's consumer on EVENTS (`_nats_keeper`), or -1 unread. +func keeperPending(jsz []byte) int { + var doc struct { + Accounts []struct { + Streams []struct { + Name string `json:"name"` + Consumers []struct { + Name string `json:"name"` + Pending int `json:"num_pending"` + } `json:"consumer_detail"` + } `json:"stream_detail"` + } `json:"account_details"` + } + if json.Unmarshal(jsz, &doc) != nil { + return -1 + } + for _, a := range doc.Accounts { + for _, s := range a.Streams { + for _, c := range s.Consumers { + if s.Name == "EVENTS" && strings.HasSuffix(c.Name, "_"+keeper.Module+"_keeper") { + return c.Pending + } + } + } + } + return -1 +} + +// keeperHealth is the tool's answer. +func keeperHealth(status func() (keeper.Status, error), read heldEntries, jsz func() ([]byte, error), now func() time.Time) health.Answer { + s, err := status() + if err != nil { + return health.Unhealthy("The keeper's status cannot be read from its own state: " + err.Error()) + } + held, _ := read() + pending := -1 + if doc, err := jsz(); err == nil { + pending = keeperPending(doc) + } + h := keeper.HealthOf(s, held, pending, now()) + why := h.Why + "; last pass " + h.Pass + for _, l := range []struct { + name string + lines []string + }{{"refused", h.Refused}, {"waiting", h.Waiting}, {"past the raise mark", h.Full}} { + if len(l.lines) > 0 { + why += "; " + l.name + ": " + strings.Join(l.lines, ", ") + } + } + if !h.Healthy { + return health.Unhealthy("The keeper is unhealthy: " + why + ".") + } + return health.Healthy("The keeper is healthy: " + why + ".") +} + +func keeperHealthTool(status func() (keeper.Status, error), read heldEntries, jsz func() ([]byte, error)) stdio.Tool { + return stdio.Tool{Name: "nats_keeper_health", + Description: "The bus module's keeper's health, as the node-engine asks it: unhealthy, naming why, when its fill pass is more than three minutes old, its first view (the indexer's indexer_bus_state) has not answered, or accepted verdicts wait on its consumer unanswered past three minutes; it names beside the entries it could not create, the ones waiting, and those past the raise mark (another module's full entry raises that module's condition and leaves the keeper healthy).", + Run: func(map[string]any) (any, error) { + return keeperHealth(status, read, jsz, func() time.Time { return time.Now().UTC() }), nil + }} +} diff --git a/modules/nats/cmd/nats-tools/main.go b/modules/nats/cmd/nats-tools/main.go index 8f79d0ff..561e3536 100644 --- a/modules/nats/cmd/nats-tools/main.go +++ b/modules/nats/cmd/nats-tools/main.go @@ -182,5 +182,6 @@ func allTools() []stdio.Tool { out := append(tools(monitor, ReadUsers, UsersWritten), removalTools(LiveBus(), SaidByTheRuntime, HeldInState)...) out = append(out, expiryTools(LiveBus(), SaidByTheRuntime, RecordedBySeat)...) out = append(out, brokerTools(HeldInState, jsz, StreamHistory(LiveBus()))...) + out = append(out, keeperHealthTool(StatusInState, HeldInState, jsz)) return append(out, requestTools(RequestsInState())...) } diff --git a/modules/nats/module.json b/modules/nats/module.json index b59e56ac..cb4e5f19 100644 --- a/modules/nats/module.json +++ b/modules/nats/module.json @@ -119,7 +119,8 @@ "nats_create", "nats_apply", "nats_recreate", - "nats_drop" + "nats_drop", + "nats_keeper_health" ], "guards": [ 8222