diff --git a/modules/nats/internal/keeper/fill.go b/modules/nats/internal/keeper/fill.go new file mode 100644 index 000000000..b28806b89 --- /dev/null +++ b/modules/nats/internal/keeper/fill.go @@ -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() }) +} diff --git a/modules/nats/internal/keeper/health.go b/modules/nats/internal/keeper/health.go new file mode 100644 index 000000000..fb623b1cd --- /dev/null +++ b/modules/nats/internal/keeper/health.go @@ -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 +} diff --git a/modules/nats/internal/keeper/keeper_test.go b/modules/nats/internal/keeper/keeper_test.go index 0515c06cd..95934c3d9 100644 --- a/modules/nats/internal/keeper/keeper_test.go +++ b/modules/nats/internal/keeper/keeper_test.go @@ -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})