mesh-self-check: D-fill judges every module's bucket and log against its cap from the bus module's nats_streams, with the controller's kind, key and words (issue 1952, ADR 0297 §6)

Its registry row and test rows are already in f7f02abb; this is the probe they call. Each KV_<module>_<name> and
LOG_<module>_<name> stream is said filling at three quarters of its max_bytes, and kept while a finding is written
until under seven tenths. One read of every stream, so the bus module not answering is the probe failing to run, which
clears nothing; bucket-or-log-unread is not raised.

modules/mesh-self-check/cmd/mesh-self-check/probes_fill.go

@@ -0,0 +1,112 @@
+package main
+
+// D-fill (novox/hq ADR 0297 §6, issue 501): a module's bucket or log filling toward its cap is said before it is full.
+// Every module's bucket (`KV_<module>_<name>`) and log (`LOG_<module>_<name>`) is read from the bus module's
+// nats_streams, which answers each stream's size and its cap (max_bytes, mesh-catalog cab031bf). Said at three
+// quarters of its cap, cleared below seven tenths: between the two a finding already written is kept, so one at its
+// threshold is not raised and cleared on every run. The kind, key and words are the controller's
+// (mesh-controller bucket_log_fill.go).
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"sort"
+	"strings"
+
+	"git.novox.be/novox/mesh-sdk/go/buscompose"
+)
+
+// The fill thresholds, in percent of a bucket's or log's cap.
+const (
+	fillRaiseAt    = 75
+	fillClearBelow = 70
+)
+
+// fillRow is what nats_streams answers of a stream's size and cap.
+type fillRow struct {
+	Name     string `json:"name"`
+	Bytes    uint64 `json:"bytes"`
+	MaxBytes uint64 `json:"max_bytes"`
+}
+
+// bucketOrLog is a module's bucket or log by its stream: the module, `bucket` or `log`, and its local name. False for
+// a stream that is no module's bucket or log (the mesh's own streams).
+func bucketOrLog(stream string) (module, kind, name string, ok bool) {
+	var rest string
+	switch {
+	case strings.HasPrefix(stream, "KV_"):
+		kind, rest = "bucket", strings.TrimPrefix(stream, "KV_")
+	case strings.HasPrefix(stream, buscompose.LogStreamPrefix):
+		kind, rest = "log", strings.TrimPrefix(stream, buscompose.LogStreamPrefix)
+	default:
+		return "", "", "", false
+	}
+	module, name, ok = strings.Cut(rest, "_")
+	if !ok || module == "" || name == "" {
+		return "", "", "", false
+	}
+	return module, kind, name, true
+}
+
+// mibWords is a size as a person reads it, in MiB with one decimal.
+func mibWords(b uint64) string { return fmt.Sprintf("%.1f MiB", float64(b)/(1024*1024)) }
+
+// fillKey is where a bucket's or log's fill finding is kept.
+func fillKey(stream string) string {
+	return Observation{Subject: Subject{Scope: ScopeBus, ID: stream}, Token: "filling"}.Key()
+}
+
+// fillObservation is what one bucket's or log's fill says: a finding at fillRaiseAt or above, and, while its finding
+// is written, at fillClearBelow or above too; nothing otherwise, which is what clears it.
+func fillObservation(stream string, r fillRow, open bool) (Observation, bool) {
+	module, kind, name, ok := bucketOrLog(stream)
+	if !ok || r.MaxBytes == 0 {
+		return Observation{}, false // no module's, or with no cap: it cannot fill
+	}
+	// Compared in bytes, not rounded percent: 74.9% is not 75%.
+	atRaise := r.Bytes*100 >= r.MaxBytes*fillRaiseAt
+	aboveClear := r.Bytes*100 >= r.MaxBytes*fillClearBelow
+	if !atRaise && !(open && aboveClear) {
+		return Observation{}, false
+	}
+	pct := r.Bytes * 100 / r.MaxBytes
+	return Observation{Kind: "bucket-or-log-filling", Token: "filling", Severity: Warning,
+		Subject: Subject{Scope: ScopeBus, ID: stream},
+		Summary: fmt.Sprintf("%s's %s %s holds %s of %s, %d%%: at its cap the bus refuses the module's next write",
+			module, kind, name, mibWords(r.Bytes), mibWords(r.MaxBytes), pct),
+		Said: fmt.Sprintf("module %s, %s %s (stream %s): %d of %d bytes, %d%%", module, kind, name, stream, r.Bytes,
+			r.MaxBytes, pct),
+		Headline: "A module's bucket or log on the bus is filling up",
+		Needs:    "decide whether to raise its cap or have the module keep less.",
+		Explanation: "A bucket or log a module keeps on the bus holds three quarters of its cap or more. When it is " +
+			"full, the bus refuses what the module writes there.",
+		Resolved: "It has room again"}, true
+}
+
+// probeFill is D-fill: every module's bucket and log on the bus judged against its cap. The bus module not answering
+// is the probe failing to run, which clears nothing; a finding already written counts as open.
+func probeFill(_ context.Context, c *Checker) ([]Observation, error) {
+	raw, err := c.Ask(StreamsTool, map[string]any{})
+	if err != nil {
+		return nil, fmt.Errorf("the bus module did not answer %s: %w", StreamsTool, err)
+	}
+	var a struct {
+		Detail []fillRow `json:"detail"`
+	}
+	if err := json.Unmarshal(raw, &a); err != nil {
+		return nil, fmt.Errorf("the bus module answered %s with something unreadable: %w", StreamsTool, err)
+	}
+	written, err := c.written()
+	if err != nil {
+		return nil, fmt.Errorf("its own findings cannot be read, so a fill between the thresholds cannot be judged: %w", err)
+	}
+	sort.Slice(a.Detail, func(i, j int) bool { return a.Detail[i].Name < a.Detail[j].Name })
+	var out []Observation
+	for _, r := range a.Detail {
+		if o, said := fillObservation(r.Name, r, written.has(fillKey(r.Name))); said {
+			out = append(out, o)
+		}
+	}
+	return sorted(out), nil
+}