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 <node>_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
This commit is contained in:
2026-11-04 12:34:59 +01:00
parent 68dd69f4ef
commit 4b2d5aeaab
4 changed files with 122 additions and 1 deletions
+30
View File
@@ -4,8 +4,11 @@ import (
"fmt" "fmt"
"strings" "strings"
"testing" "testing"
"time"
"git.novox.be/novox/mesh-sdk/go/catalogue" "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 // 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, " ")) 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 `<node>_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)
}
})
}
}
@@ -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 (`<node>_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
}}
}
+1
View File
@@ -182,5 +182,6 @@ func allTools() []stdio.Tool {
out := append(tools(monitor, ReadUsers, UsersWritten), removalTools(LiveBus(), SaidByTheRuntime, HeldInState)...) out := append(tools(monitor, ReadUsers, UsersWritten), removalTools(LiveBus(), SaidByTheRuntime, HeldInState)...)
out = append(out, expiryTools(LiveBus(), SaidByTheRuntime, RecordedBySeat)...) out = append(out, expiryTools(LiveBus(), SaidByTheRuntime, RecordedBySeat)...)
out = append(out, brokerTools(HeldInState, jsz, StreamHistory(LiveBus()))...) out = append(out, brokerTools(HeldInState, jsz, StreamHistory(LiveBus()))...)
out = append(out, keeperHealthTool(StatusInState, HeldInState, jsz))
return append(out, requestTools(RequestsInState())...) return append(out, requestTools(RequestsInState())...)
} }
+2 -1
View File
@@ -119,7 +119,8 @@
"nats_create", "nats_create",
"nats_apply", "nats_apply",
"nats_recreate", "nats_recreate",
"nats_drop" "nats_drop",
"nats_keeper_health"
], ],
"guards": [ "guards": [
8222 8222