CAT-N6: the fill pass raises full at the raise mark and clears it below the clear mark; the keeper's health

The keeper's one periodic read (ADR 0595 §6): every entry that refuses when
full, by its fullest object; full raised at the raise mark (75 by default),
cleared only below the clear mark (70), so an entry at the mark does not flap;
an entry dropping its oldest is never read. HealthOf is the keeper's health
for its tool: unhealthy when its pass is more than three minutes old, its
first view has not answered (naming the indexer's verb), or verdicts wait on
its consumer unanswered past the bound; it names the entries refused, waiting
and full beside.

Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
2026-11-04 12:34:59 +01:00
parent 93bfdd9e2f
commit cc1a942235
3 changed files with 186 additions and 0 deletions
+72
View File
@@ -0,0 +1,72 @@
package keeper
import (
"context"
"fmt"
"time"
"git.novox.be/novox/mesh-sdk/go/asks"
)
// FillPass reads the fill of every entry that refuses when full (ADR 0595 §6), raising `full` at the raise mark and
// clearing it only below the clear mark, so an entry at the mark does not flap. It is the keeper's one periodic read.
func (k *Keeper) FillPass() {
raise, clear := k.marks()
keys, err := k.Store.Keys(StateAsked)
if err != nil {
k.logf("the fill pass cannot read its own %s: %v", StateAsked, err)
return
}
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
for _, module := range keys {
held, _, found := k.asked(module)
if !found {
continue
}
changed := false
for _, local := range sortedKeys(held.Made) {
m := held.Made[local]
if !m.Created {
continue
}
var most int64 = -1
for _, o := range m.Plan.Objects {
if !o.Settings.Refuses() || o.Settings.MaxBytes <= 0 {
continue
}
_, f, there, err := k.Bus.Held(ctx, o)
if err != nil || !there {
continue
}
if p := int64(f.Bytes) * 100 / o.Settings.MaxBytes; p > most {
most = p
}
}
if most < 0 {
continue
}
was := m
m.Fill = most
switch {
case !m.Full && most >= raise:
m.Full = true
k.emit(EventFull, m.Plan, held.Revision, fmt.Sprintf("%d%% of its cap, past the raise mark %d%%", most, raise), nil, nil)
k.raise(conditionFull, module, local, asks.SeverityWarning, "Kept data is filling up",
"a larger cap in the module's manifest, or data removed",
fmt.Sprintf("%s of %s holds %d%% of its cap and refuses new data when full; it is cleared below %d%%", local, module, most, clear))
case m.Full && most < clear:
m.Full = false
k.emit(EventFullCleared, m.Plan, held.Revision, fmt.Sprintf("%d%% of its cap, below the clear mark %d%%", most, clear), nil, nil)
k.clear(conditionFull, module, local)
}
if m.Fill != was.Fill || m.Full != was.Full {
held.Made[local], changed = m, true
}
}
if changed {
k.write(held)
}
}
k.setStatus(func(s *Status) { s.Pass = k.now() })
}
+71
View File
@@ -0,0 +1,71 @@
package keeper
import (
"fmt"
"strings"
"time"
)
// Bound is how long the keeper may go without its fill pass, or leave an accepted verdict unanswered, before it is
// unhealthy (design 25 § 12.2): three minutes, so the only guard against a store filling never stops without a sound.
const Bound = 3 * time.Minute
// Health is the keeper's health as its tool answers it: healthy or not, why, and what it names beside (the entries it
// could not create, the ones waiting, the ones past the raise mark). Another module's entry past the raise mark raises
// that module's condition and leaves the keeper healthy.
type Health struct {
Healthy bool `json:"healthy"`
Why string `json:"why"`
Pass string `json:"pass"`
Refused []string `json:"refused,omitempty"`
Waiting []string `json:"waiting,omitempty"`
Full []string `json:"full,omitempty"`
}
// HealthOf judges the keeper from its status, what it holds, and pending: how many accepted verdicts wait on its
// durable consumer unanswered (-1 when it could not be read). Pure.
func HealthOf(s Status, held map[string]Asked, pending int, now time.Time) Health {
h := Health{Healthy: true}
var why []string
switch {
case s.Pass.IsZero() && (s.Started.IsZero() || now.Sub(s.Started) > Bound):
why = append(why, "the keeper has made no fill pass since its start")
h.Pass = "none"
case !s.Pass.IsZero():
h.Pass = s.Pass.Format(time.RFC3339)
if now.Sub(s.Pass) > Bound {
why = append(why, fmt.Sprintf("its last fill pass is %s old, more than %s", now.Sub(s.Pass).Round(time.Second), Bound))
}
}
if s.FirstView.Answered.IsZero() {
w := fmt.Sprintf("its first view, %s.%s, has not answered", s.FirstView.Owner, s.FirstView.Verb)
if s.FirstView.Error != "" {
w += ": " + s.FirstView.Error
}
why = append(why, w)
}
if pending > 0 && (s.Heard.IsZero() || now.Sub(s.Heard) > Bound) {
why = append(why, fmt.Sprintf("%d accepted verdict(s) wait on its consumer unanswered for more than %s", pending, Bound))
}
for _, module := range sortedKeys(held) {
for _, local := range sortedKeys(held[module].Made) {
m := held[module].Made[local]
name := BusName(module, local)
switch {
case m.Refused != "":
h.Refused = append(h.Refused, name+": "+m.Refused)
case m.Waiting != "":
h.Waiting = append(h.Waiting, name+": "+m.Waiting)
}
if m.Full {
h.Full = append(h.Full, fmt.Sprintf("%s: %d%%", name, m.Fill))
}
}
}
if len(why) > 0 {
h.Healthy, h.Why = false, strings.Join(why, "; ")
} else {
h.Why = "its fill pass runs and its first view is held"
}
return h
}
@@ -401,6 +401,49 @@ func TestTheKeeperMakesWhatAVerdictAsksForAndSaysIt(t *testing.T) {
t.Errorf("n's key is still held: nothing of it was made, so its empty verdict ends it")
}
}},
// CAT-N6: fill (ADR 0595 §6): an entry that refuses when full raises `full` at the raise mark and clears only
// below the clear mark; one that drops its oldest is never read; the health names the pass.
{"full at the raise mark, cleared only below the clear mark", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
events := entry("events", catalogue.StorageStream, catalogue.TemplateEvents, map[string]string{"max-mib": "1"})
k.Heard(Accepted{Module: "m", Revision: 1, BusState: []catalogue.BusStateEntry{drafts, events}})
out.reset()
bus.fill["m_events"] = 1 * uint64(MiB)
var steps []string
for _, mib := range []uint64{40, 48, 46, 45, 44} {
bus.fill["KV_m_drafts"] = mib * uint64(MiB)
k.FillPass()
steps = append(steps, fmt.Sprintf("%d:%d", mib, len(out.events)))
}
if fmt.Sprint(steps) != "[40:0 48:1 46:1 45:1 44:2]" || fmt.Sprint(out.events) != "[full m drafts full-cleared m drafts]" ||
fmt.Sprint(out.conds) != "[raised nats.full.m_drafts cleared nats.full.m_drafts]" {
t.Errorf("said %v %v: full at 48 of 64 MiB (75%%), kept at 46 and 45 (71%%, 70%%), cleared at 44 (68%%)", out.events, out.conds)
}
h := HealthOf(k.Status(), k.Held(), 0, k.now())
if !h.Healthy || h.Pass == "none" {
t.Errorf("health %+v: healthy, naming its pass", h)
}
}},
{"the health is unhealthy past its bound, naming why", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
at := k.now()
k.FillPass()
later := at.Add(4 * time.Minute)
if h := HealthOf(k.Status(), nil, 0, later); h.Healthy || !strings.Contains(h.Why, "last fill pass is 4m0s old") {
t.Errorf("health %+v: unhealthy, its pass too old", h)
}
if h := HealthOf(k.Status(), nil, 2, later); !strings.Contains(h.Why, "2 accepted verdict(s) wait") {
t.Errorf("health %+v: names the verdicts waiting", h)
}
k.View = func() (FirstView, error) { return FirstView{}, errors.New("no responders") }
k.mu.Lock()
k.status.FirstView.Answered = time.Time{}
k.mu.Unlock()
if next, err := k.AskFirstView(10 * time.Second); err == nil || next != 20*time.Second {
t.Fatalf("asked again in %s (%v): on the backoff", next, err)
}
if h := HealthOf(k.Status(), nil, 0, at); h.Healthy || !strings.Contains(h.Why, "mesh-delivery-indexer.indexer_bus_state") {
t.Errorf("health %+v: unhealthy, naming the verb", h)
}
}},
} {
t.Run(row.name, func(t *testing.T) {
k, bus, _, out := newKeeper(FirstView{Revision: 1})