Name the fill condition for buckets and logs, and judge each one on its own

Review of #231: "store" is the glossary's word for the postgres server, so
the condition is bucket-or-log-filling. And one stream that cannot be read
no longer fails the whole check: it is said as unread with its reason, an
open fill condition for it is kept, and every other bucket and log is
still judged (ADR 0297 §6).

cmd/mesh-controller/bucket_log_fill.go

@@ -0,0 +1,212 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sort"
+
+	"github.com/nats-io/nats.go"
+
+	"github.com/novox/mesh-controller/internal/conditions"
+)
+
+// **A filling bucket or log is said before it is full** (novox/hq ADR 0297 §6, issue 501).
+//
+// A module's bucket and its log each have a cap on the bus, and when either is full the bus refuses
+// the next write: the module stops doing what it writes for, and nothing said so before. On
+// 2026-10-11 the issue tracker's bucket held a sixth of its cap after two days, growing toward a stop
+// nobody would have heard coming. So the self-check reads every module's bucket and log, and one at
+// fillRaiseAt of its cap or more raises a condition naming the module, the bucket or log, and how full
+// it is.
+//
+// **Cleared by observation below fillClearBelow, not below fillRaiseAt**: a bucket or log hovering at
+// its threshold would otherwise be raised and cleared on every run. Between the two, an open condition
+// is kept and none is raised.
+//
+// **One that cannot be read is said, and the rest are still judged.** An unanswered question about one
+// stream is no reason to know nothing of the others: it is a finding of its own, naming the bucket or
+// log and why, and a fill condition already open for it is kept, because not knowing is not a pass.
+
+// The fill thresholds, in percent of a bucket's or log's cap.
+const (
+	fillRaiseAt    = 75
+	fillClearBelow = 70
+)
+
+// The conditions the fill probe raises: one filling, and one that could not be read.
+const (
+	kindBucketOrLogFilling = "bucket-or-log-filling"
+	kindBucketOrLogUnread  = "bucket-or-log-unread"
+)
+
+// probeFillID is the probe's id in the registry.
+const probeFillID = "D-fill"
+
+// bucketOrLog is one module's bucket or log as the fill probe reads it.
+type bucketOrLog struct {
+	Module string
+	// Kind is `bucket` or `log`; Name its local name; Stream the stream it is on the bus.
+	Kind, Name, Stream string
+}
+
+// filled is how full one bucket or log is, read from the bus.
+type filled struct {
+	Bytes, Max uint64
+}
+
+// percent is how full, in whole percent, rounded down.
+func (f filled) percent() uint64 {
+	if f.Max == 0 {
+		return 0
+	}
+	return f.Bytes * 100 / f.Max
+}
+
+// fillKey is where a bucket's or log's fill condition is kept.
+func fillKey(s bucketOrLog) string { return conditions.Key(conditions.ScopeBus, s.Stream, "filling") }
+
+// fillObservation is what one bucket's or log's fill says: a finding at fillRaiseAt or above, and,
+// while its condition is open, at fillClearBelow or above too; nothing otherwise, which is what clears
+// it. The operator's words are the kind's (plain_words.go); the summary and the evidence name the
+// module, the bucket or log, and the fill.
+func fillObservation(s bucketOrLog, f filled, open bool) (conditions.Observation, bool) {
+	if f.Max == 0 {
+		return conditions.Observation{}, false // one with no cap cannot fill
+	}
+	// Compared in bytes, not rounded percent: 74.9% is not 75%.
+	atRaise := f.Bytes*100 >= f.Max*fillRaiseAt
+	aboveClear := f.Bytes*100 >= f.Max*fillClearBelow
+	if !atRaise && !(open && aboveClear) {
+		return conditions.Observation{}, false
+	}
+	pct := f.percent()
+	return conditions.Observation{
+		Scope: conditions.ScopeBus, ID: s.Stream, Token: "filling", Kind: kindBucketOrLogFilling,
+		Severity: conditions.Warning,
+		Summary: fmt.Sprintf("%s's %s %s holds %s of %s, %d%%: at its cap the bus refuses the module's next write",
+			s.Module, s.Kind, s.Name, mibWords(f.Bytes), mibWords(f.Max), pct),
+		Said: fmt.Sprintf("module %s, %s %s (stream %s): %d of %d bytes, %d%%", s.Module, s.Kind, s.Name,
+			s.Stream, f.Bytes, f.Max, pct),
+	}, true
+}
+
+// unreadObservation says one bucket or log could not be read, and why. Asked over the network, so one
+// look can be wrong: raised on the second look in a row (confirm.go).
+func unreadObservation(s bucketOrLog, why error) conditions.Observation {
+	return conditions.Observation{
+		Scope: conditions.ScopeBus, ID: s.Stream, Token: "unread", Kind: kindBucketOrLogUnread,
+		Severity: conditions.Warning, Confirm: true,
+		Summary: fmt.Sprintf("%s's %s %s could not be read, so how full it is is not known", s.Module, s.Kind, s.Name),
+		Said:    fmt.Sprintf("module %s, %s %s (stream %s): %v", s.Module, s.Kind, s.Name, s.Stream, why),
+	}
+}
+
+// keptFilling is an open fill condition said again for a bucket or log that could not be read this
+// time: not knowing how full it is now is no reason to say it has room.
+func keptFilling(s bucketOrLog, open conditions.Condition, why error) conditions.Observation {
+	severity := open.Severity
+	if severity == "" {
+		severity = conditions.Warning
+	}
+	summary := open.Summary
+	if summary == "" {
+		summary = fmt.Sprintf("%s's %s %s was filling when it was last read", s.Module, s.Kind, s.Name)
+	}
+	return conditions.Observation{
+		Scope: conditions.ScopeBus, ID: s.Stream, Token: "filling", Kind: kindBucketOrLogFilling,
+		Severity: severity, Summary: summary,
+		Said: fmt.Sprintf("module %s, %s %s (stream %s): not read this time (%v); kept open as it was",
+			s.Module, s.Kind, s.Name, s.Stream, why),
+	}
+}
+
+// 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))
+}
+
+// readFill is how full one bucket or log is on the bus; found false when it is not on the bus.
+type readFill func(ctx context.Context, s bucketOrLog) (f filled, found bool, err error)
+
+// openFill is the fill condition open under a key, if one is.
+type openFill func(ctx context.Context, key string) (conditions.Condition, bool)
+
+// judgeFills judges every bucket and log on its own: one that cannot be read is said as unread, with
+// its open fill condition kept, and every other is judged by its fill.
+func judgeFills(ctx context.Context, all []bucketOrLog, read readFill, open openFill) []conditions.Observation {
+	var out []conditions.Observation
+	for _, s := range all {
+		f, found, err := read(ctx, s)
+		if err != nil {
+			out = append(out, unreadObservation(s, err))
+			if c, isOpen := open(ctx, fillKey(s)); isOpen {
+				out = append(out, keptFilling(s, c, err))
+			}
+			continue
+		}
+		if !found {
+			continue // not on the bus: created on the controller's next raise, and nothing there can fill
+		}
+		_, isOpen := open(ctx, fillKey(s))
+		if o, said := fillObservation(s, f, isOpen); said {
+			out = append(out, o)
+		}
+	}
+	sort.SliceStable(out, func(i, j int) bool { return out[i].Key() < out[j].Key() })
+	return out
+}
+
+// declaredBucketsAndLogs is every module's bucket and log the catalogue declares, by its stream.
+func declaredBucketsAndLogs(ctx context.Context, d *doctor) ([]bucketOrLog, error) {
+	buckets, err := d.open.inventory.DeclaredBuckets(ctx)
+	if err != nil {
+		return nil, err
+	}
+	logs, err := d.open.inventory.DeclaredLogs(ctx)
+	if err != nil {
+		return nil, err
+	}
+	var out []bucketOrLog
+	for _, b := range buckets {
+		out = append(out, bucketOrLog{Module: b.Module, Kind: "bucket", Name: b.Name, Stream: "KV_" + b.Bucket()})
+	}
+	for _, l := range logs {
+		out = append(out, bucketOrLog{Module: l.Module, Kind: "log", Name: l.Name, Stream: l.Stream()})
+	}
+	return out, nil
+}
+
+// probeFill reads every module's bucket and log on the bus and says each one filling toward its cap,
+// and each one that could not be read.
+func probeFill(ctx context.Context, d *doctor) ([]conditions.Observation, error) {
+	all, err := declaredBucketsAndLogs(ctx, d)
+	if err != nil {
+		return nil, err
+	}
+	js := d.js.Context()
+	read := func(ctx context.Context, s bucketOrLog) (filled, bool, error) {
+		if err := ctx.Err(); err != nil {
+			return filled{}, false, err
+		}
+		info, err := js.StreamInfo(s.Stream, nats.Context(ctx))
+		switch {
+		case errors.Is(err, nats.ErrStreamNotFound):
+			return filled{}, false, nil
+		case err != nil:
+			return filled{}, false, err
+		case info.Config.MaxBytes <= 0:
+			return filled{}, true, nil
+		}
+		return filled{Bytes: info.State.Bytes, Max: uint64(info.Config.MaxBytes)}, true, nil
+	}
+	open := func(ctx context.Context, key string) (conditions.Condition, bool) {
+		if d.keeper == nil {
+			return conditions.Condition{}, false
+		}
+		// One that cannot be read is taken as open, as confirm.go takes it: unknown is no reason to clear.
+		c, isOpen, err := d.keeper.Get(ctx, key)
+		return c, isOpen || err != nil
+	}
+	return judgeFills(ctx, all, read, open), nil
+}