mesh-controller/cmd/mesh-controller/bucket_log_fill.go

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. Only ever made from a condition
// read back as open — its own summary and severity, never words made up for it.
func keptFilling(s bucketOrLog, open conditions.Condition, why error) conditions.Observation {
	return conditions.Observation{
		Scope: conditions.ScopeBus, ID: s.Stream, Token: "filling", Kind: kindBucketOrLogFilling,
		Severity: open.Severity, Summary: open.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, or why it could not be read.
type openFill func(ctx context.Context, key string) (conditions.Condition, bool, error)

// judgeFills judges every bucket and log on its own: one that cannot be read is said as unread, with
// its fill condition kept only when that condition was read back and is open — when it cannot be read
// either, nothing is said of its fill, which the unread finding already covers. Every other is judged
// by its fill; for one of those, a condition that cannot be read is taken as open, so an unknown never
// clears a fill measured between fillClearBelow and fillRaiseAt.
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, cerr := open(ctx, fillKey(s)); cerr == nil && 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, cerr := open(ctx, fillKey(s))
		isOpen = isOpen || cerr != nil
		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, error) {
		if d.keeper == nil {
			return conditions.Condition{}, false, nil
		}
		return d.keeper.Get(ctx, key)
	}
	return judgeFills(ctx, all, read, open), nil
}