End healers H3 and H4 and the bus-health probe's consumer half with the controller

Nobody raises a lost or a far-behind consumer since the self-check's bus-consumer probe ended (issue 1126): the grants
part makes a missing account consumer itself and says one far behind as mesh-grants.consumer-behind, and H4's repair
was the controller's. The mesh defines no consumer of its own, so H-bus's consumer half judged nothing.

Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
This commit is contained in:
2026-11-02 12:29:56 +01:00
parent 9e9877e6ec
commit 3ed3266dde
8 changed files with 75 additions and 202 deletions
@@ -81,7 +81,7 @@ func Registry() []Probe {
Deferred: "the grants of restricted streams with what each machine composed for them: " + storeOnly},
{ID: "DW", Asserts: "the watchdogs answer: the watchman holds the seat mesh-watchdogs and answers signals",
From: "ADR 0227 rule 6: the watchers are watched", Kind: "watchdogs-silent", Run: probeWatchdogs},
{ID: "H-bus", Asserts: "every stream and durable consumer the mesh defines is on the bus, and a request crosses it " +
{ID: "H-bus", Asserts: "every stream the mesh defines is on the bus, and a request crosses it " +
"to the bus's machine and back", From: "ADR 0236, to-be 45 §8", Kind: "core-unhealthy", Run: probeBusHealth},
}
}
@@ -94,20 +94,10 @@ type streamsAnswer struct {
}
type streamRow struct {
Name string `json:"name"`
Subjects []string `json:"subjects"`
Retention string `json:"retention"`
MaxPerSubject int64 `json:"max_per_subject"`
Consumers []consumerRow `json:"consumer_detail"`
}
type consumerRow struct {
Name string `json:"name"`
Kind string `json:"kind"`
Filters []string `json:"filters"`
Pending int64 `json:"pending"`
Unacked int64 `json:"unacknowledged"`
MaxDeliver int64 `json:"max_deliver"`
Name string `json:"name"`
Subjects []string `json:"subjects"`
Retention string `json:"retention"`
MaxPerSubject int64 `json:"max_per_subject"`
}
// errNoStream is a stream the bus module says it does not hold.
@@ -133,20 +123,6 @@ func (c *Checker) streams(stream string) ([]streamRow, error) {
return a.Detail, nil
}
// consumerWords is a consumer as the mesh says it, with its name.
func consumerWords(c buscompose.Consumer) string {
return fmt.Sprintf("the consumer %s on %s", c.Name, c.Stream)
}
// expectedConsumers are the durable consumers the mesh defines, by stream.
func expectedConsumers() map[string][]buscompose.Consumer {
out := map[string][]buscompose.Consumer{}
for _, c := range buscompose.MeshConsumers() {
out[c.Stream] = append(out[c.Stream], c)
}
return out
}
// probeStreams is D7: every stream the mesh defines exists as defined.
func probeStreams(ctx context.Context, c *Checker) ([]Observation, error) {
rows, err := c.streams("")
@@ -308,8 +284,9 @@ func probeWatchdogs(_ context.Context, c *Checker) ([]Observation, error) {
return nil, nil
}
// probeBusHealth is H-bus: every stream and durable consumer the mesh defines is on the bus, and a request crosses it
// to the bus module's machine and back. The question to the bus module is that request: an answer is its round trip,
// probeBusHealth is H-bus: every stream the mesh defines is on the bus, and a request crosses it to the bus module's
// machine and back. The mesh defines no durable consumer of its own since the controller ended: the account consumers
// are the grants part's to make and judge (issue 1126). The question to the bus module is that request: an answer is its round trip,
// and no answer is no pass.
func probeBusHealth(ctx context.Context, c *Checker) ([]Observation, error) {
rows, err := c.streams("")
@@ -326,35 +303,6 @@ func probeBusHealth(ctx context.Context, c *Checker) ([]Observation, error) {
problems = append(problems, "the stream "+s.Name+" is missing")
}
}
byStream := expectedConsumers()
streams := make([]string, 0, len(byStream))
for s := range byStream {
streams = append(streams, s)
}
sort.Strings(streams)
for _, stream := range streams {
if !held[stream] {
continue // said above
}
if ctx.Err() != nil {
return nil, ctx.Err()
}
detail, err := c.streams(stream)
if err != nil && !errors.Is(err, errNoStream) {
return nil, err
}
have := map[string]bool{}
for _, r := range detail {
for _, cr := range r.Consumers {
have[cr.Name] = true
}
}
for _, want := range byStream[stream] {
if !have[want.Name] {
problems = append(problems, consumerWords(want)+" is missing")
}
}
}
if len(problems) == 0 {
return nil, nil
}
@@ -44,17 +44,6 @@ func (b bus) rows() map[string]*streamRow {
}
rows[s.Name] = &streamRow{Name: s.Name, Subjects: s.Subjects, Retention: retention, MaxPerSubject: per}
}
for _, c := range buscompose.MeshConsumers() {
r := rows[c.Stream]
if r == nil || b.drop[c.Stream+"."+c.Name] {
continue
}
kind := "pull"
if c.Push || c.Queue != "" {
kind = "push"
}
r.Consumers = append(r.Consumers, consumerRow{Name: c.Name, Kind: kind, Filters: c.Filters, MaxDeliver: int64(c.MaxDeliver)})
}
if b.change != nil {
b.change(rows)
}
@@ -71,9 +60,7 @@ func (b bus) ask(key string, body any) (json.RawMessage, error) {
var detail []streamRow
for name, r := range b.rows() {
if stream == "" {
row := *r
row.Consumers = nil
detail = append(detail, row)
detail = append(detail, *r)
} else if name == stream {
detail = append(detail, *r)
}
@@ -141,8 +128,8 @@ func TestTheBusProbes(t *testing.T) {
{"DW answers", probeWatchdogs, bus{}, nil, "", nil},
{"DW silent", probeWatchdogs, bus{signals: errors.New("no responders")}, []string{"core.watchdogs.silent"}, "", nil},
{"H-bus healthy", probeBusHealth, bus{}, nil, "", nil},
{"H-bus a stream and a consumer missing", probeBusHealth,
bus{drop: map[string]bool{"NODES": true, "EVENTS.controller-merges": true}}, []string{"bus.mesh.unhealthy"}, "", nil},
{"H-bus a stream missing", probeBusHealth,
bus{drop: map[string]bool{"NODES": true}}, []string{"bus.mesh.unhealthy"}, "", nil},
{"H-bus no round trip", probeBusHealth, bus{asksFail: errors.New("timeout")}, nil, "no request crossed", nil},
// D8 (issue 238): placement's addresses and endpoints against each heard holder's ban list.
{"D8 nothing of the mesh banned", probeBans, bus{banned: map[string]string{"ace": `{"banned":["198.51.100.7"]}`,