From 4c11c2c0c0fb8e18539d4faef62b1783bf9b2828 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 29 Oct 2026 21:48:09 +0100 Subject: [PATCH] Move the healers out of the controller into mesh-healers, acting on the keeper's events and only once the lease names no controller (issue 1952) Until the lease write each act is written to would, kept a week by its bucket's age, so a would-act counts for no budget or brake. Acts begin by a create in attempts; H3 asks the grants part's assert; the brake and S15 are its own findings. --- modules/mesh-healers/.gitignore | 2 + modules/mesh-healers/README.md | 37 ++ .../mesh-healers/cmd/mesh-healers/healers.go | 602 ++++++++++++++++++ .../cmd/mesh-healers/healers_test.go | 377 +++++++++++ modules/mesh-healers/cmd/mesh-healers/main.go | 92 +++ .../cmd/mesh-healers/manifest_test.go | 91 +++ .../cmd/mesh-healers/observation.go | 77 +++ .../cmd/mesh-healers/reads-outside | 2 + .../mesh-healers/cmd/mesh-healers/tools.go | 186 ++++++ .../mesh-healers/cmd/mesh-healers/wanted.go | 152 +++++ .../mesh-healers/whole_integration_test.go | 416 ++++++++++++ modules/mesh-healers/go.mod | 22 + modules/mesh-healers/go.sum | 27 + modules/mesh-healers/module.json | 78 +++ 14 files changed, 2161 insertions(+) create mode 100644 modules/mesh-healers/.gitignore create mode 100644 modules/mesh-healers/README.md create mode 100644 modules/mesh-healers/cmd/mesh-healers/healers.go create mode 100644 modules/mesh-healers/cmd/mesh-healers/healers_test.go create mode 100644 modules/mesh-healers/cmd/mesh-healers/main.go create mode 100644 modules/mesh-healers/cmd/mesh-healers/manifest_test.go create mode 100644 modules/mesh-healers/cmd/mesh-healers/observation.go create mode 100644 modules/mesh-healers/cmd/mesh-healers/reads-outside create mode 100644 modules/mesh-healers/cmd/mesh-healers/tools.go create mode 100644 modules/mesh-healers/cmd/mesh-healers/wanted.go create mode 100644 modules/mesh-healers/cmd/mesh-healers/whole_integration_test.go create mode 100644 modules/mesh-healers/go.mod create mode 100644 modules/mesh-healers/go.sum create mode 100644 modules/mesh-healers/module.json diff --git a/modules/mesh-healers/.gitignore b/modules/mesh-healers/.gitignore new file mode 100644 index 00000000..0bafa572 --- /dev/null +++ b/modules/mesh-healers/.gitignore @@ -0,0 +1,2 @@ +/mesh-healers +/cmd/mesh-healers/mesh-healers diff --git a/modules/mesh-healers/README.md b/modules/mesh-healers/README.md new file mode 100644 index 00000000..0431657c --- /dev/null +++ b/modules/mesh-healers/README.md @@ -0,0 +1,37 @@ +# mesh-healers + +Holds the seat `mesh-healers` (novox/hq ADR 0564 §7, §8): the mesh's healers, moved out of the controller. Go, one +binary (`cmd/mesh-healers`), served by a runtime of its own on its own bus account (`runs-as`). + +## What it does + +It hears the condition keeper's `condition-raised` and `condition-changed` (seat `mesh-conditions`) on a durable +consumer. A condition whose kind a healer answers is acted on within the healer's budget, its settle and the mesh-wide +brake (twelve acts an hour). A repair is a call of the seat or tool that owns the act: H3 asks the grants part's +`grants_assert` for the module whose consumer is gone. H4 (the controller's own consumer) and H5 (the provider's own +repair) are registered and run elsewhere. + +Each act is begun by a create of its key in `attempts` before it is made and finished after, so a holder that dies +mid-act has spent it and a second instance cannot begin it again. The condition keeper writes the condition's +`tried`, resolver and escalation from `attempts`; a healer writes no condition. + +**Only once the controller's lease names no controller.** Until the write that hands the lease away, each act it would +take is written to `would`, never to `attempts`: it counts for no budget, settle or brake and reaches no condition. + +S15: it watches the controller's hand-act log and writes `healer-wanted` for a cause repaired by hand twice within a +fortnight; a drill and a person's decision by design never count. + +## What it keeps + +- `mesh-healers_attempts`: every act and escalation, kept a week (the bucket's age). +- `mesh-healers_would`: the newest act it would take per healer and condition while the controller holds the lease, + kept a week (the bucket's age); nothing else deletes it, and nothing is written there once the lease is handed away. +- `mesh-healers_findings`: the brake, its own blindness, and S15. + +## Its verbs and tools + +- `mesh-healers.healers`: the registry, the acts of the last days, the would-acts, the brake, whether it acts and why; + `recount=true` counts the hand-act log again. +- `mesh-healers.heal`: one act now on an open condition, by key and why. +- `mesh-healers.drill`: refused until the hand-act log's home is delivered (issue 1953), naming `mesh-controller.drill`. +- `healers_findings`, `healers_would`: the reads of its buckets. diff --git a/modules/mesh-healers/cmd/mesh-healers/healers.go b/modules/mesh-healers/cmd/mesh-healers/healers.go new file mode 100644 index 00000000..170f0442 --- /dev/null +++ b/modules/mesh-healers/cmd/mesh-healers/healers.go @@ -0,0 +1,602 @@ +package main + +// The healers (novox/hq ADR 0564 §7, §8; ADR 0231): a known failure heals itself, under a brake, and every repair is +// said. The registry is compiled in; mesh-healers hears `condition-raised` and `condition-changed` of the seat +// mesh-conditions on a durable consumer, instead of looking every half minute. A repair is a call of the seat or tool +// that owns the act (H3: the grants part's grants_assert). Each act is begun by a create of its key in the bucket +// `attempts` before it is made and finished after, so a holder that dies mid-act has spent it and a second instance +// cannot begin the same act; budgets, settles and the mesh-wide brake count from there. Success is never the +// healer's to say: the condition clears when its observer no longer sees it, and the condition keeper writes `tried`, +// the resolver and the escalation into the condition from `attempts`. +// +// **Only once the controller's lease names no controller** (ADR 0564 §8): until then each act it would take is +// written to the bucket `would`, never to `attempts`, so a would-act counts for no budget, settle or brake, reaches +// no `tried` and escalates nothing. `would` keeps each key a week (its bucket's age): nothing else deletes it, and once +// the lease is handed away nothing is written there again. + +import ( + "encoding/json" + "errors" + "fmt" + "slices" + "sort" + "strconv" + "strings" + "sync" + "time" + + "git.novox.be/novox/mesh-sdk/go/buscompose" + "git.novox.be/novox/mesh-sdk/go/health" +) + +// Module and Seat are this module and the seat it holds once for the mesh. +const ( + Module = "mesh-healers" + Seat = "mesh-healers" +) + +// The states it keeps and reads, and the event it says. +const ( + AttemptsState = health.AttemptsState + WouldState = health.WouldState + FindingsState = health.FindingsState + LeaseRead = buscompose.ControllerLease + LeaseKey = buscompose.ControllerLeaseKey + ConditionsRead = "mesh-conditions.conditions" + HandActsRead = "mesh-controller.hand-acts" + ActedEvent = health.EventHealerActed + // The events it hears: the keeper's, on the seat mesh-conditions. + RaisedEvent = "mesh-conditions.condition-raised" + ChangedEvent = "mesh-conditions.condition-changed" + // AssertTool is the grants part's tool that asserts a module's buckets, logs and streams again. + AssertTool = "mesh-grants.grants_assert" +) + +// The mesh-wide brake (to-be 45 §7): how many acts all healers together may take in an hour. +const ( + BrakeLimit = 12 + BrakeWindow = time.Hour +) + +// The outcomes an attempt has. +const ( + OutcomeActing = health.OutcomeActing + OutcomeActed = health.OutcomeActed + OutcomeFailed = health.OutcomeFailed + OutcomeEscalated = health.OutcomeEscalated +) + +// Store is the bus's buckets as this module reaches them, through its runtime. Create writes a key only when it has +// none, and says ErrExists otherwise. +type Store interface { + Get(state, key string) ([]byte, bool, error) + Put(state, key string, value any) error + Create(state, key string, value any) error + Delete(state, key string) error + Keys(state string) ([]string, error) +} + +// ErrExists is a create of a key another writer made first. +var ErrExists = errors.New("the key was made by another writer first") + +// Condition is what a healer reads of a condition the keeper said: the controller's condition body, its looks aside. +type Condition struct { + Key string `json:"key"` + Kind string `json:"kind"` + Subject Subject `json:"subject"` + Severity string `json:"severity"` + Resolver string `json:"resolver"` + Tried []json.RawMessage `json:"tried,omitempty"` +} + +// Escalated says a healer's budget was spent: the operator resolves it now, and no healer touches it again until +// observation clears it. +func (c Condition) Escalated() bool { return c.Resolver == "operator" && len(c.Tried) > 0 } + +// Attempt, Would and Acted are mesh-sdk's health layouts: an act begun in `attempts` (its key +// `..` is the healers' own), an act written to `would` while the lease names a +// controller (one key per healer and condition, `.`, its newest), and the body of healer-acted, +// epoch left out (ADR 0564 Limits 10). +type ( + Attempt = health.Attempt + Would = health.Would + Acted = health.HealerActed +) + +// Healer is one row of the registry. +type Healer struct { + ID string + Kinds []string + Condition string + Repair string + Then string + From string + Budget int + Window time.Duration + Settle time.Duration + // ActsIn is where the repair runs: here, or a module that repairs its own (H5), or the controller (H4). + ActsIn string + // applies says whether this condition is one it may act on, and what its budget counts against; why when not. + applies func(c Condition) (budget string, ok bool, why string) + // repair is the act: what it did, what came of it, or why it could not. + repair func(h *Healers, c Condition) (act, said string, err error) +} + +// actsHere marks a healer whose repair mesh-healers makes. +const actsHere = "mesh-healers" + +// Registry is to-be 45 §7's table, compiled in: every healer the mesh has, the ones mesh-healers runs with their act. +func Registry() []Healer { + return []Healer{ + {ID: "H3", Kinds: []string{"holder-silent", "consumer-lost"}, + Condition: "a durable consumer of a module the bus does not hold (D6), or a seat's holder that does not answer (D3)", + Repair: "assert the module's buckets, logs and streams again through the grants part (grants_assert)", + Then: "resolver operator, urgent", From: "a controller restarted to make a missing object again (208, 248)", + Budget: 1, Window: time.Hour, Settle: 6 * time.Minute, ActsIn: actsHere, + applies: appliesToAModulesObject, repair: assertAgain}, + {ID: "H4", Kinds: []string{"consumer-behind"}, + Condition: "a durable consumer far behind its stream's head (D6) that the stream table marks resettable", + Repair: "re-make the consumer to deliver from now; the one marked resettable is the controller's own", + Then: "resolver operator, urgent", From: "a consumer re-made from now by hand (248)", + Budget: 1, Window: 24 * time.Hour, Settle: 6 * time.Minute, + ActsIn: "the controller, until its lease: the one consumer marked resettable is its own (ADR 0564 §7)"}, + {ID: "H5", Kinds: []string{"provider-failing"}, + Condition: "the identity provider's administrator refusing the mesh's secret (provider-failing)", + Repair: "the provider repairs it itself through the server's own bootstrap command (ADR 0224 §5)", + Then: "announced failing by the provider, braked from ten minutes doubling to six hours", + From: "the identity provider's admin reset through its bootstrap command (179)", + Budget: 1, Window: 10 * time.Minute, Settle: 10 * time.Minute, + ActsIn: "the provider module (keycloak), ADR 0224 §5"}, + } +} + +// Healers is mesh-healers: the registry, what its acts reach, and what it holds. +type Healers struct { + Store Store + Ask func(key string, body any) (json.RawMessage, error) + Emit func(event string, body any) error + Now func() time.Time + Logf func(format string, args ...any) + Registry []Healer + // After wakes the healers when a settle ends, to judge the condition again; nil wakes nothing. + After func(d time.Duration, f func()) + + mu sync.Mutex + // lease is the controller's lease as last read: acting is on only while it is handed away. + acting bool + leaseWhy string + // declined is why each condition was last passed over, said once. + declined map[string]string + // acts are the hand-act log as watched, for S15. + acts map[string]HandAct + // woken is the settle ends already waited for, by budget key. + woken map[string]time.Time + s15At time.Time +} + +// LeaseHeard records what the lease says: acting is on only while it names a holder that is no controller (ADR 0564 +// §8, Limits 8, as ADR 0561 §3's reader reads it). +func (h *Healers) LeaseHeard(raw []byte, found bool, readErr error) { + on, why := buscompose.LeaseHandedAway(raw, found, readErr) + h.mu.Lock() + flipped := on != h.acting || h.leaseWhy == "" + h.acting, h.leaseWhy = on, leaseWords(on, why) + h.mu.Unlock() + if flipped { + h.Logf("[%s] the healers %s: %s", Module, map[bool]string{true: "ACT", false: "act in shadow"}[on], h.leaseWhy) + } +} + +// leaseWords says the lease's state as the healers read it; the SDK's words speak of minting. +func leaseWords(on bool, why string) string { + if on { + return "the controller's lease is handed away: the healers act" + } + if strings.Contains(why, "cannot be read") || strings.Contains(why, "not readable") { + return "the controller's lease cannot be read, so it is taken to name a controller: every act is written as would" + } + return "the controller's lease names a controller (or none is written, so one may take it): every act is written as would" +} + +// Acting says whether the healers act, and why. +func (h *Healers) Acting() (bool, string) { + h.mu.Lock() + defer h.mu.Unlock() + if h.leaseWhy == "" { + return false, "the controller's lease has not been read yet: every act is written as would" + } + return h.acting, h.leaseWhy +} + +// healerFor is the healer that acts here on a kind. +func (h *Healers) healerFor(kind string) (Healer, bool) { + for _, r := range h.Registry { + if r.repair != nil && slices.Contains(r.Kinds, kind) { + return r, true + } + } + return Healer{}, false +} + +// Heard is a condition the keeper raised or changed: judged, and acted on, written as would, or passed over. +func (h *Healers) Heard(body []byte) error { + var c Condition + if err := json.Unmarshal(body, &c); err != nil { + h.Logf("[%s] a condition event is not readable, and is passed over: %v", Module, err) + return nil + } + _, err := h.Judge(c, "") + return err +} + +// Judge is one condition judged by the healer registered for its kind. by names a person's `heal`, empty for an +// event. It answers what was done, in words. +func (h *Healers) Judge(c Condition, by string) (string, error) { + h.mu.Lock() + defer h.mu.Unlock() + row, ok := h.healerFor(c.Kind) + if !ok { + return "no healer here answers the kind " + c.Kind, nil + } + if c.Escalated() { + return "it is handed to the operator already: no healer touches it until observation clears it", nil + } + budget, applies, why := row.applies(c) + if !applies { + h.decline(c.Key, row.ID, why) + return fmt.Sprintf("healer %s leaves it alone: %s", row.ID, why), nil + } + delete(h.declined, c.Key) + now := h.Now() + if !h.acting { + w := Would{Healer: row.ID, Condition: c.Key, Kind: c.Kind, Act: row.Repair, At: now.UTC(), Why: h.whyNot()} + if err := h.Store.Put(WouldState, row.ID+"."+c.Key, w); err != nil { + return "", fmt.Errorf("the would-act of %s on %s is not written: %w", row.ID, c.Key, err) + } + return fmt.Sprintf("healer %s would act on %s: %s", row.ID, c.Key, w.Why), nil + } + all, err := h.attempts() + if err != nil { + h.blind(err) + return "", fmt.Errorf("the healers cannot read what they did, so they count no budget and act on nothing: %w", err) + } + h.see("blind", nil) + acts := actsWithin(all, now.Add(-BrakeWindow)) + if braked, said := h.brakeHolds(acts, now); braked { + h.see("brake", []Observation{brakeFinding(acts, said)}) + return "every healer has stopped: " + said, nil + } + h.see("brake", nil) + spent := spentOn(all, row, budget, now) + if n := len(spent); n > 0 && now.Sub(spent[n-1].Begun) < row.Settle { + h.wake(budget, spent[n-1].Begun.Add(row.Settle), c.Key) + return fmt.Sprintf("healer %s waits: its last act on it is still the observation's to judge until %s", row.ID, + spent[n-1].Begun.Add(row.Settle).UTC().Format(time.RFC3339)), nil + } + if len(spent) >= row.Budget { + return h.escalate(row, c, budget, spent) + } + if len(acts) >= BrakeLimit { + return "the brake holds from the next act on", nil + } + return h.act(row, c, budget, len(spent), by) +} + +func (h *Healers) whyNot() string { + if h.leaseWhy == "" { + return "the controller's lease has not been read yet" + } + return h.leaseWhy +} + +// act is one healer's act: begun by a create in attempts, made, finished, and said. +func (h *Healers) act(row Healer, c Condition, budget string, spent int, by string) (string, error) { + begun := h.Now().UTC() + key := attemptKey(row.ID, c.Key, begun) + if by == "" { + by = "healer " + row.ID + } + window := fmt.Sprintf("act %d of %d within %s", spent+1, row.Budget, row.Window) + a := Attempt{Healer: row.ID, Condition: c.Key, Kind: c.Kind, BudgetKey: budget, Act: row.Repair, Outcome: OutcomeActing, + Begun: begun} + if by != "healer "+row.ID { + a.Said = "asked by " + by + } + if err := h.Store.Create(AttemptsState, key, a); err != nil { + if errors.Is(err, ErrExists) { + return "another instance began this act first", nil + } + return "", fmt.Errorf("healer %s did not act on %s: its attempt could not be begun: %w", row.ID, c.Key, err) + } + act, said, err := row.repair(h, c) + a.Outcome = OutcomeActed + if err != nil { + a.Outcome, said = OutcomeFailed, err.Error() + } + if act != "" { + a.Act = act + } + finished := h.Now().UTC() + if a.Said != "" { + said = a.Said + "; " + said + } + a.Said, a.Finished = said+" ("+window+")", &finished + if err := h.Store.Put(AttemptsState, key, a); err != nil { + h.Logf("[%s] healer %s %s %s and could not record what came of it: %v", Module, row.ID, a.Outcome, c.Key, err) + } + h.tell(Acted{Healer: row.ID, Condition: c.Key, Kind: c.Kind, Act: a.Act, Outcome: a.Outcome, Said: said, Budget: window}) + h.wake(budget, begun.Add(row.Settle), c.Key) + return fmt.Sprintf("healer %s %s %s: %s — %s (%s)", row.ID, a.Outcome, c.Key, a.Act, said, window), nil +} + +// escalate is a spent budget: the condition is the operator's now. Written once as an attempt the keeper turns into +// the condition's resolver and urgency (ADR 0564 §7); an escalated condition is passed over until it clears. +func (h *Healers) escalate(row Healer, c Condition, budget string, spent []Attempt) (string, error) { + var tried []string + for _, s := range spent { + tried = append(tried, fmt.Sprintf("%s at %s (%s)", s.Outcome, s.Begun.UTC().Format("15:04 MST"), firstLine(s.Said))) + } + said := fmt.Sprintf("its budget of %d act(s) within %s is spent and %s is still open: %s", row.Budget, row.Window, + c.Key, strings.Join(tried, "; ")) + now := h.Now().UTC() + window := fmt.Sprintf("%d of %d within %s", len(spent), row.Budget, row.Window) + a := Attempt{Healer: row.ID, Condition: c.Key, Kind: c.Kind, BudgetKey: budget, Act: "handed the condition to the operator", + Outcome: OutcomeEscalated, Said: said, Begun: now, Finished: &now} + if err := h.Store.Create(AttemptsState, attemptKey(row.ID, c.Key, now), a); err != nil && !errors.Is(err, ErrExists) { + return "", fmt.Errorf("healer %s could not record handing %s to the operator, and does not: %w", row.ID, c.Key, err) + } + h.tell(Acted{Healer: row.ID, Condition: c.Key, Kind: c.Kind, Act: "handed to the operator", Outcome: OutcomeEscalated, + Said: said, Budget: window}) + return fmt.Sprintf("healer %s handed %s to the operator: %s", row.ID, c.Key, said), nil +} + +// tell says a heal as healer-acted. Not said is said in the log: the attempt holds it. +func (h *Healers) tell(e Acted) { + e.Event, e.At, e.By = ActedEvent, h.Now().UTC(), "healer "+e.Healer + e.Show = "mesh-conditions.conditions key=" + e.Condition + if h.Emit == nil { + return + } + if err := h.Emit(ActedEvent, e); err != nil { + h.Logf("[%s] healer %s %s %s, and that could NOT be said on the bus: %v", Module, e.Healer, e.Outcome, e.Condition, err) + } +} + +// wake judges a condition again when its settle ends: the one wait of the healers, set by an act, never a look on a +// schedule. The settle is read from the attempt's time, so a wake lost to a restart is the next event's to make. +func (h *Healers) wake(budget string, at time.Time, key string) { + if h.After == nil || !h.woken[budget].Before(at) { + return + } + if h.woken == nil { + h.woken = map[string]time.Time{} + } + h.woken[budget] = at + h.After(at.Sub(h.Now())+time.Second, func() { h.again(key) }) +} + +// again judges a condition again from the keeper's bucket, if it is still open. +func (h *Healers) again(key string) { + raw, found, err := h.Store.Get(ConditionsRead, key) + if err != nil || !found { + return + } + if _, err := h.Judge(conditionOf(raw), ""); err != nil { + h.Logf("[%s] %s judged again at its settle's end: %v", Module, key, err) + } +} + +func conditionOf(raw []byte) Condition { + var c Condition + _ = json.Unmarshal(raw, &c) + return c +} + +// decline remembers why a healer passed a condition over, said once. +func (h *Healers) decline(key, healer, why string) { + if h.declined == nil { + h.declined = map[string]string{} + } + if h.declined[key] != why { + h.Logf("[%s] healer %s leaves %s alone: %s", Module, healer, key, why) + } + h.declined[key] = why +} + +func attemptKey(healer, condition string, begun time.Time) string { + return healer + "." + condition + "." + strconv.FormatInt(begun.UnixNano(), 10) +} + +// attempts are every attempt kept, oldest first. +func (h *Healers) attempts() ([]Attempt, error) { + keys, err := h.Store.Keys(AttemptsState) + if err != nil { + return nil, err + } + var out []Attempt + for _, k := range keys { + raw, found, err := h.Store.Get(AttemptsState, k) + if err != nil { + return nil, err + } + if !found { + continue + } + var a Attempt + if err := json.Unmarshal(raw, &a); err != nil { + return nil, fmt.Errorf("the attempt %s is not readable: %w", k, err) + } + out = append(out, a) + } + sort.SliceStable(out, func(i, j int) bool { return out[i].Begun.Before(out[j].Begun) }) + return out, nil +} + +// actsWithin are the attempts since a moment that were acts: an escalation is a saying, and the brake does not count it. +func actsWithin(all []Attempt, since time.Time) []Attempt { + var out []Attempt + for _, a := range all { + if !a.Begun.Before(since) && a.Outcome != OutcomeEscalated { + out = append(out, a) + } + } + return out +} + +// spentOn is one healer's acts against one budget key within its window, oldest first. +func spentOn(all []Attempt, row Healer, budget string, now time.Time) []Attempt { + var out []Attempt + for _, a := range actsWithin(all, now.Add(-row.Window)) { + if a.Healer == row.ID && a.BudgetKey == budget { + out = append(out, a) + } + } + return out +} + +// brakeHolds says whether the mesh-wide brake holds: the limit reached within the hour, or the brake already written +// and an act within the hour still — it lets go an hour after the last act, not the first. +func (h *Healers) brakeHolds(acts []Attempt, now time.Time) (bool, string) { + said := fmt.Sprintf("%d act(s) by the healers in the last %s, the limit %d", len(acts), BrakeWindow, BrakeLimit) + if len(acts) >= BrakeLimit { + return true, said + } + _, held, _ := h.Store.Get(FindingsState, brakeKey) + if held && len(acts) > 0 { + last := acts[len(acts)-1].Begun + return true, fmt.Sprintf("%s; held until an hour after the last, at %s", said, + last.Add(BrakeWindow).UTC().Format("15:04 MST")) + } + return false, said +} + +var brakeKey = health.Key(ScopeMesh, "healers", "braked") + +// brakeFinding is the brake, as the urgent finding that says it. +func brakeFinding(acts []Attempt, said string) Observation { + counts := map[string]int{} + for _, a := range acts { + counts[a.Healer+" on "+a.Condition]++ + } + var most []string + for what, n := range counts { + most = append(most, fmt.Sprintf("%s ×%d", what, n)) + } + sort.Strings(most) + return Observation{Kind: "healers-braked", Token: "braked", Severity: Urgent, Subject: Subject{Scope: ScopeMesh, ID: "healers"}, + Summary: "every healer has stopped: the healers acted more often in an hour than the mesh allows, which is a " + + "healer looping, not repairing — " + said, + Said: strings.Join(most, ", "), + Headline: "All automatic repairs stopped", + Needs: "read the details and decide whether the repairs may run again.", + Explanation: "The healers (the mesh's automatic repairs) acted more often in an hour than allowed, which looks " + + "like a loop, so all of them stopped.", + Resolved: "Automatic repairs run again"} +} + +// blind writes the healers' own blindness: they cannot read what they did. +func (h *Healers) blind(err error) { + h.see("blind", []Observation{{Kind: "probe-failed", Token: "failed", Severity: Warning, + Subject: Subject{Scope: ScopeProbe, ID: "healers"}, + Summary: "the healers cannot read what they did, so they count no budget and act on nothing", + Said: firstLine(err.Error()), Headline: "A check of the mesh could not run", + Explanation: "The automatic repairs cannot read what they did, so they act on nothing until they can.", + Resolved: "The automatic repairs read what they did again"}}) +} + +// see writes the healers' own findings of one source, each only when it changes, and deletes those of the source it +// no longer sees (ADR 0564 §2). +func (h *Healers) see(source string, found []Observation) { + now := h.Now() + want := map[string]health.Finding{} + for _, o := range found { + o.Source = source + want[o.Key()] = o.Finding("", now) + } + keys, err := h.Store.Keys(FindingsState) + if err != nil { + h.Logf("[%s] its own findings cannot be read, so none is written or cleared: %v", Module, err) + return + } + for _, k := range keys { + raw, found, err := h.Store.Get(FindingsState, k) + if err != nil || !found { + continue + } + was, err := health.DecodeFinding(raw) + if err != nil || was.Source != source { + continue + } + if f, still := want[k]; still { + if says(was) == says(f) { + delete(want, k) + } + continue + } + if err := h.Store.Delete(FindingsState, k); err != nil { + h.Logf("[%s] the finding %s no longer seen is not deleted: %v", Module, k, err) + } + } + for k, f := range want { + raw, err := health.EncodeFinding(f) + if err == nil { + err = h.Store.Put(FindingsState, k, json.RawMessage(raw)) + } + if err != nil { + h.Logf("[%s] the finding %s is not written: %v", Module, k, err) + } + } +} + +// --- H3 -------------------------------------------------------------------------------------------------------- + +// moduleOf is the module whose objects a condition names: a consumer of a module's own (`.`). The +// controller's own consumers are not a module's: the controller asserts them itself until its lease. +func moduleOf(c Condition) (string, string) { + stream, consumer, ok := strings.Cut(c.Subject.ID, ".") + if c.Kind != "consumer-lost" || !ok || stream == "" || consumer == "" { + return "", "it names no consumer of a module whose objects the grants part asserts" + } + if consumer == buscompose.ControllerName || strings.HasPrefix(consumer, buscompose.ControllerName+"-") { + return "", "the consumer is the controller's own, which the controller asserts itself while it holds its lease" + } + return consumer, "" +} + +// appliesToAModulesObject is H3's: a module's consumer the bus does not hold, its budget per condition. +func appliesToAModulesObject(c Condition) (string, bool, string) { + if _, why := moduleOf(c); why != "" { + return "", false, why + } + return c.Key, true, "" +} + +// assertAgain is H3: the grants part asserts the module's buckets, logs and streams again. +func assertAgain(h *Healers, c Condition) (string, string, error) { + module, why := moduleOf(c) + if module == "" { + return "", "", errors.New(why) + } + raw, err := h.Ask(AssertTool, map[string]any{"module": module}) + if err != nil { + return "", "", fmt.Errorf("the grants part did not assert %s: %w", module, err) + } + var a struct { + Says string `json:"says"` + } + _ = json.Unmarshal(raw, &a) + return "asserted " + module + "'s buckets, logs and streams again through the grants part", + orSaid(a.Says, "the grants part answered") + "; the probe that raised it says on its next run whether it is there", nil +} + +func orSaid(s, otherwise string) string { + if strings.TrimSpace(s) == "" { + return otherwise + } + return s +} + +func firstLine(s string) string { + line, _, _ := strings.Cut(s, "\n") + return line +} diff --git a/modules/mesh-healers/cmd/mesh-healers/healers_test.go b/modules/mesh-healers/cmd/mesh-healers/healers_test.go new file mode 100644 index 00000000..b3c4fb6a --- /dev/null +++ b/modules/mesh-healers/cmd/mesh-healers/healers_test.go @@ -0,0 +1,377 @@ +package main + +import ( + "encoding/json" + "errors" + "slices" + "sort" + "strings" + "sync" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/health" +) + +type memStore struct { + mu sync.Mutex + kv map[string]map[string][]byte + keysErr map[string]error +} + +func newStore() *memStore { + return &memStore{kv: map[string]map[string][]byte{}, keysErr: map[string]error{}} +} + +func (s *memStore) Get(state, key string) ([]byte, bool, error) { + s.mu.Lock() + defer s.mu.Unlock() + v, ok := s.kv[state][key] + return v, ok, nil +} + +func (s *memStore) put(state, key string, value any, create bool) error { + raw, err := json.Marshal(value) + if err != nil { + return err + } + s.mu.Lock() + defer s.mu.Unlock() + if s.kv[state] == nil { + s.kv[state] = map[string][]byte{} + } + if _, ok := s.kv[state][key]; ok && create { + return ErrExists + } + s.kv[state][key] = raw + return nil +} + +func (s *memStore) Put(state, key string, value any) error { return s.put(state, key, value, false) } +func (s *memStore) Create(state, key string, value any) error { return s.put(state, key, value, true) } + +func (s *memStore) Delete(state, key string) error { + s.mu.Lock() + defer s.mu.Unlock() + delete(s.kv[state], key) + return nil +} + +func (s *memStore) Keys(state string) ([]string, error) { + s.mu.Lock() + defer s.mu.Unlock() + if err := s.keysErr[state]; err != nil { + return nil, err + } + var out []string + for k := range s.kv[state] { + out = append(out, k) + } + sort.Strings(out) + return out, nil +} + +func (s *memStore) keys(state string) []string { + k, _ := s.Keys(state) + return k +} + +func (s *memStore) attempt(t *testing.T, key string) Attempt { + t.Helper() + raw, _, _ := s.Get(AttemptsState, key) + var a Attempt + if err := json.Unmarshal(raw, &a); err != nil { + t.Fatalf("%s: %s", key, raw) + } + return a +} + +// world is mesh-healers over memory, with its clock, its asks, what it said and the wakes it set. +type world struct { + h *Healers + store *memStore + now time.Time + asked []string + said []Acted + wakes []time.Duration +} + +func newWorld() *world { + w := &world{store: newStore(), now: time.Date(2026, 10, 29, 21, 0, 0, 0, time.UTC)} + w.h = &Healers{Store: w.store, Registry: Registry(), Logf: func(string, ...any) {}, + Now: func() time.Time { return w.now }, + Ask: func(key string, body any) (json.RawMessage, error) { + raw, _ := json.Marshal(body) + w.asked = append(w.asked, key+" "+string(raw)) + return json.RawMessage(`{"says":"2 the bus lacks: created at this ask"}`), nil + }, + Emit: func(event string, body any) error { + if a, ok := body.(Acted); ok && event == ActedEvent { + w.said = append(w.said, a) + } + return nil + }, + After: func(d time.Duration, _ func()) { w.wakes = append(w.wakes, d) }} + return w +} + +func (w *world) handAway() { + w.h.LeaseHeard([]byte(`{"handed-to":"the operator's hand-over"}`), true, nil) +} + +// lost is a module's consumer the bus does not hold, as D6 raises it. +func lost(module string) Condition { + return Condition{Key: "bus.EVENTS." + module + ".missing", Kind: "consumer-lost", Severity: "warning", Resolver: "self", + Subject: Subject{Scope: "bus", ID: "EVENTS." + module}} +} + +func body(c Condition) []byte { + raw, _ := json.Marshal(map[string]any{"key": c.Key, "kind": c.Kind, "subject": c.Subject, "severity": c.Severity, + "resolver": c.Resolver, "tried": c.Tried, "event": "condition-raised", "show": "mesh-conditions.conditions key=" + c.Key}) + return raw +} + +// No act is made while the lease names a controller: each is written to would, never to attempts, so it counts for no +// budget or brake and nothing is asked or said; one is made from the write that hands the lease away (ADR 0564 §8). +func TestTheHealersActOnlyOnceTheLeaseNamesNoController(t *testing.T) { + cases := []struct { + name string + lease func(h *Healers) + acts bool + }{ + {"never read", func(*Healers) {}, false}, + {"absent", func(h *Healers) { h.LeaseHeard(nil, false, nil) }, false}, + {"unreadable", func(h *Healers) { h.LeaseHeard(nil, false, errors.New("timeout")) }, false}, + {"a controller holds it", func(h *Healers) { + h.LeaseHeard([]byte(`{"instance":"novox/1/2026-10-27","epoch":57}`), true, nil) + }, false}, + {"handed away", func(h *Healers) { h.LeaseHeard([]byte(`{"handed-to":"the operator"}`), true, nil) }, true}, + } + for _, tc := range cases { + w := newWorld() + tc.lease(w.h) + if err := w.h.Heard(body(lost("records"))); err != nil { + t.Fatalf("%s: %v", tc.name, err) + } + attempts, would := w.store.keys(AttemptsState), w.store.keys(WouldState) + if tc.acts { + if len(attempts) != 1 || len(would) != 0 || len(w.asked) != 1 || !strings.Contains(w.asked[0], `"module":"records"`) || + len(w.said) != 1 || w.said[0].Outcome != OutcomeActed { + t.Errorf("%s: attempts %v, would %v, asked %v, said %+v", tc.name, attempts, would, w.asked, w.said) + continue + } + a := w.store.attempt(t, attempts[0]) + if a.Outcome != OutcomeActed || a.Finished == nil || a.Healer != "H3" || a.Condition != "bus.EVENTS.records.missing" { + t.Errorf("%s: the attempt is %+v", tc.name, a) + } + continue + } + if len(attempts) != 0 || len(w.asked) != 0 || len(w.said) != 0 || !slices.Equal(would, []string{"H3.bus.EVENTS.records.missing"}) { + t.Errorf("%s: attempts %v, would %v, asked %v, said %v", tc.name, attempts, would, w.asked, w.said) + } + if _, braked, _ := w.store.Get(FindingsState, brakeKey); braked { + t.Errorf("%s: a would-act moved the brake", tc.name) + } + } +} + +// An act waits out its settle, then its spent budget hands the condition to the operator once; an escalated condition +// is passed over; a second instance cannot begin an act already begun; the brake stops every healer past twelve acts +// in an hour and lets go an hour after the last (ADR 0231 §21, ADR 0564 §7). +func TestBudgetSettleEscalationAndBrake(t *testing.T) { + w := newWorld() + w.handAway() + c := lost("records") + steps := []struct { + name string + after time.Duration + cond Condition + attempts int + outcome string // the newest said outcome, empty for none new + }{ + {"acts", 0, c, 1, OutcomeActed}, + {"within its settle: waits", 2 * time.Minute, c, 1, ""}, + {"after its settle, budget spent: escalates", 5 * time.Minute, c, 2, OutcomeEscalated}, + {"escalated: passed over", time.Minute, Condition{Key: c.Key, Kind: c.Kind, Subject: c.Subject, Resolver: "operator", + Tried: []json.RawMessage{json.RawMessage(`{}`)}}, 2, ""}, + } + for _, s := range steps { + w.now = w.now.Add(s.after) + before := len(w.said) + if _, err := w.h.Judge(s.cond, ""); err != nil { + t.Fatalf("%s: %v", s.name, err) + } + if n := len(w.store.keys(AttemptsState)); n != s.attempts { + t.Errorf("%s: %d attempts, want %d", s.name, n, s.attempts) + } + switch { + case s.outcome == "" && len(w.said) != before: + t.Errorf("%s: said %+v", s.name, w.said[before:]) + case s.outcome != "" && (len(w.said) != before+1 || w.said[before].Outcome != s.outcome): + t.Errorf("%s: said %+v, want %s", s.name, w.said[before:], s.outcome) + } + } + if len(w.wakes) == 0 || w.wakes[0] != 6*time.Minute+time.Second { + t.Errorf("the settle's end wakes %v, want once after 6m1s", w.wakes) + } + // A second instance cannot begin the act an attempt's key already holds. + x := newWorld() + x.handAway() + _ = x.store.Create(AttemptsState, attemptKey("H3", lost("mail").Key, x.now.UTC()), Attempt{}) + if said, _ := x.h.Judge(lost("mail"), ""); !strings.Contains(said, "another instance") || len(x.asked) != 0 { + t.Errorf("a second instance acted: %q, asked %v", said, x.asked) + } + // The brake: twelve acts in the hour stop every healer, said as an urgent finding. + b := newWorld() + b.handAway() + for i := 0; i < BrakeLimit; i++ { + b.now = b.now.Add(time.Minute) + _ = b.store.Create(AttemptsState, attemptKey("H3", "x", b.now), Attempt{Healer: "H3", Condition: "x", BudgetKey: "x", + Outcome: OutcomeActed, Begun: b.now}) + } + if said, _ := b.h.Judge(lost("mail"), ""); !strings.Contains(said, "stopped") || len(b.asked) != 0 { + t.Errorf("past the brake: %q, asked %v", said, b.asked) + } + raw, braked, _ := b.store.Get(FindingsState, brakeKey) + f, _ := health.DecodeFinding(raw) + if !braked || f.Severity != Urgent || f.Kind != "healers-braked" || f.Observer != Module { + t.Errorf("the brake is written as %s", raw) + } + b.now = b.now.Add(55 * time.Minute) // fewer than twelve within the hour, but the brake is said: held + if said, _ := b.h.Judge(lost("mail"), ""); !strings.Contains(said, "held until") { + t.Errorf("the brake let go before an hour after the last act: %q", said) + } + b.now = b.now.Add(6 * time.Minute) + if _, err := b.h.Judge(lost("mail"), ""); err != nil || len(b.asked) != 1 { + t.Errorf("an hour after the last act the healers do not act again: %v, asked %v", err, b.asked) + } + if _, braked, _ := b.store.Get(FindingsState, brakeKey); braked { + t.Error("the brake's finding outlived the brake") + } + // Blind to what they did, they act on nothing and say so. + z := newWorld() + z.handAway() + z.store.keysErr[AttemptsState] = errors.New("the bus is away") + if _, err := z.h.Judge(lost("mail"), ""); err == nil || len(z.asked) != 0 { + t.Errorf("blind, the healers acted: %v", err) + } + if keys := z.store.keys(FindingsState); !slices.Equal(keys, []string{"probe.healers.failed"}) { + t.Errorf("blind, the findings are %v", keys) + } +} + +// H3 answers a module's consumer the bus does not hold, never the controller's own or a condition naming no consumer; +// the registry holds every healer the mesh has, H3 run here, H4 and H5 run elsewhere (ADR 0231 §2, ADR 0564 §7). +func TestTheRegistryAndWhatH3Answers(t *testing.T) { + cases := []struct { + c Condition + holds bool + }{ + {lost("records"), true}, + {lost("controller"), false}, + {lost("controller-merges"), false}, + {Condition{Key: "seat.x.silent", Kind: "holder-silent", Subject: Subject{Scope: "seat", ID: "issue-tracker"}}, false}, + {Condition{Key: "bus.EVENTS.records.behind", Kind: "consumer-behind", Subject: Subject{Scope: "bus", ID: "EVENTS.records"}}, false}, + } + for _, tc := range cases { + w := newWorld() + w.handAway() + said, _ := w.h.Judge(tc.c, "") + if acted := len(w.asked) == 1; acted != tc.holds { + t.Errorf("%s: acted %v, want %v (%s)", tc.c.Key, acted, tc.holds, said) + } + } + runs := map[string]bool{} + for _, r := range Registry() { + if len(r.Kinds) == 0 || r.Budget < 1 || r.Window <= 0 || r.Settle <= 0 || r.Repair == "" || r.ActsIn == "" { + t.Errorf("%s is not a whole row: %+v", r.ID, r) + } + runs[r.ID] = r.repair != nil + } + if !runs["H3"] || runs["H4"] || runs["H5"] || len(runs) != 3 { + t.Errorf("the healers run here are %v, want H3 alone of H3, H4, H5", runs) + } +} + +// S15: a cause repaired by hand twice within a fortnight is a finding; a decision by design never counts; it clears +// when fewer than two remain, and the healers wake when the oldest leaves the fortnight (ADR 0564 Limits 2). +func TestACauseRepairedByHandTwiceIsAHealerWanted(t *testing.T) { + w := newWorld() + act := func(key, verb, cause string, ago time.Duration) { + raw, _ := json.Marshal(HandAct{ID: key, At: w.now.Add(-ago), By: "person.jochen", Verb: verb, Why: "x", Cause: cause}) + w.h.HandActHeard(key, "put", raw) + } + act("a1", "broker consumer-reset", "consumer-behind", 10*24*time.Hour) + if keys := w.store.keys(FindingsState); len(keys) != 0 { + t.Fatalf("one act is a finding: %v", keys) + } + act("a2", "broker consumer-reset", "consumer-behind", 24*time.Hour) + act("a3", "hand-act drill", "drill", time.Hour) + act("a4", "hand-act drill", "drill", 2*time.Hour) + act("a5", "conditions silence", "operator-answer", time.Hour) + act("a6", "conditions silence", "operator-answer", time.Hour) + act("a7", "hand-act record", "relogin", 20*24*time.Hour) + act("a8", "hand-act record", "relogin", time.Hour) + want := []string{"mesh.hand-acts.consumer-behind.healer-wanted"} + if keys := w.store.keys(FindingsState); !slices.Equal(keys, want) { + t.Fatalf("findings %v, want %v", keys, want) + } + raw, _, _ := w.store.Get(FindingsState, want[0]) + if !strings.Contains(string(raw), "healer H4 answers this cause") { + t.Errorf("the finding does not say its healer is not enough: %s", raw) + } + if n := len(w.wakes); n == 0 || w.wakes[n-1] != 4*24*time.Hour+time.Second { + t.Errorf("wakes %v, want the oldest act's leaving the fortnight in 4 days", w.wakes) + } + w.now = w.now.Add(4*24*time.Hour + time.Second) + w.h.Wanted() + if keys := w.store.keys(FindingsState); len(keys) != 0 { + t.Errorf("past the fortnight the finding stays: %v", keys) + } + w.h.HandActHeard("a2", "delete", nil) + if len(w.h.acts) != 7 { + t.Errorf("a deleted entry is kept: %d", len(w.h.acts)) + } +} + +// The verbs: heal acts on an open condition by key and says why it was asked; drill waits for the log's home; the reads +// answer every bucket. +func TestTheVerbsAndReads(t *testing.T) { + w := newWorld() + w.handAway() + c := lost("records") + _ = w.store.Put(ConditionsRead, c.Key, c) + if _, err := w.h.Heal(c.Key, "", "person.jochen"); err == nil { + t.Error("heal took no why") + } + if _, err := w.h.Heal("bus.none", "x", "person.jochen"); err == nil { + t.Error("heal acted on no open condition") + } + if _, err := w.h.Heal(c.Key, "it is gone again", "person.jochen"); err != nil { + t.Fatal(err) + } + keys := w.store.keys(AttemptsState) + if len(keys) != 1 || !strings.Contains(w.store.attempt(t, keys[0]).Said, "asked by person.jochen (it is gone again)") { + t.Errorf("heal's attempt is %v", keys) + } + answer, err := w.h.Answer(7, true) + if err != nil { + t.Fatal(err) + } + a := answer.(map[string]any) + if a["acting"] != true || len(a["acts"].([]Attempt)) != 1 || len(a["healers"].([]map[string]any)) != 3 { + t.Errorf("healers answered %v", a) + } + var drill error + for _, tool := range Tools(w.h) { + if tool.Name == DrillTool { + _, drill = tool.Run(map[string]any{}) + } + } + if drill == nil || !strings.Contains(drill.Error(), "mesh-controller.drill") { + t.Errorf("drill answered %v", drill) + } + if _, err := w.h.read(WouldState, ""); err != nil { + t.Error(err) + } +} diff --git a/modules/mesh-healers/cmd/mesh-healers/main.go b/modules/mesh-healers/cmd/mesh-healers/main.go new file mode 100644 index 00000000..86ec7539 --- /dev/null +++ b/modules/mesh-healers/cmd/mesh-healers/main.go @@ -0,0 +1,92 @@ +// mesh-healers: the holder of the seat mesh-healers (novox/hq ADR 0564 §7, §8). A Go bundle served by a runtime of its +// own, on its own bus account (`runs-as`), so that its account alone writes its buckets: `attempts` (every act begun, +// made and finished, and every escalation, kept a week), `would` (what it would have done while the controller holds +// its lease, kept a week) and `findings` (the brake, its blindness, and S15). It hears the condition keeper's +// condition-raised and condition-changed on a durable consumer, watches the controller's lease, and watches the +// controller's hand-act log for S15. stdout is the MCP channel; what this module says, it says on stderr. +package main + +import ( + "encoding/json" + "errors" + "fmt" + "os" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +func logf(format string, a ...any) { fmt.Fprintf(os.Stderr, format+"\n", a...) } + +// runtimeStore is the bus, through the runtime that launched this bundle. +type runtimeStore struct{} + +func (runtimeStore) Get(state, key string) ([]byte, bool, error) { + e, err := stdio.State(state).Get(key) + if err != nil || e == nil { + return nil, false, err + } + return e.Value, true, nil +} + +func (runtimeStore) Put(state, key string, value any) error { + _, err := stdio.State(state).Put(key, value) + return err +} + +func (runtimeStore) Create(state, key string, value any) error { + _, err := stdio.State(state).Create(key, value) + if errors.Is(err, stdio.ErrChanged) { + return ErrExists + } + return err +} + +func (runtimeStore) Delete(state, key string) error { return stdio.State(state).Delete(key) } + +func (runtimeStore) Keys(state string) ([]string, error) { return stdio.State(state).Keys() } + +// watched asks a watch of the runtime again until it is taken: the runtime answers only once Serve runs. +func watched(state, key string, handle func(stdio.StateChange) error) { + for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { + time.Sleep(wait) + if err := stdio.State(state).Watch(key, handle); err != nil { + logf("[%s] %s is not watched yet (%v); asking again in %s", Module, state, err, wait) + continue + } + return + } +} + +func main() { + h := &Healers{Store: runtimeStore{}, Registry: Registry(), Now: time.Now, Logf: logf, + Ask: func(key string, body any) (json.RawMessage, error) { return stdio.Ask(key, body) }, + Emit: func(event string, body any) error { return stdio.Emit(event, body) }, + After: func(d time.Duration, f func()) { time.AfterFunc(d, f) }} + // The lease first: until it is read the healers write would-acts only. + go watched(LeaseRead, LeaseKey, func(c stdio.StateChange) error { + h.LeaseHeard(c.Value, c.Op == "put", nil) + return nil + }) + go watched(HandActsRead, "", func(c stdio.StateChange) error { + h.HandActHeard(c.Key, c.Op, c.Value) + return nil + }) + for _, event := range []string{RaisedEvent, ChangedEvent} { + go func(event string) { + for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { + time.Sleep(wait) + err := stdio.Subscribe(event, func(e stdio.Envelope) error { return h.Heard(e.Body) }) + if err == nil { + logf("[%s] hears %s", Module, event) + return + } + logf("[%s] does not hear %s yet (%v); asking again in %s", Module, event, err, wait) + } + }(event) + } + if err := stdio.Serve("", Tools(h)); err != nil { + logf("%v", err) + os.Exit(1) + } +} diff --git a/modules/mesh-healers/cmd/mesh-healers/manifest_test.go b/modules/mesh-healers/cmd/mesh-healers/manifest_test.go new file mode 100644 index 00000000..86588a98 --- /dev/null +++ b/modules/mesh-healers/cmd/mesh-healers/manifest_test.go @@ -0,0 +1,91 @@ +package main + +import ( + "encoding/json" + "os" + "path/filepath" + "slices" + "strings" + "testing" +) + +type manifest struct { + Module string `json:"module"` + RunsAs string `json:"runs-as"` + Claims []struct { + Name string `json:"name"` + Scope string `json:"scope"` + Serves []string `json:"serves"` + } `json:"claims"` + Emits []string `json:"emits"` + Consumes []string `json:"consumes"` + Invokes []string `json:"invokes"` + Reads []string `json:"reads"` + State []json.RawMessage `json:"state"` +} + +// The manifest says what the code does: it claims the seat mesh-healers once for the mesh with its verbs (ADR 0564 §1, +// §7), runs on its own account, says healer-acted, hears the keeper's raised and changed, asks only the grants part's +// assert, reads the lease, the keeper's conditions and the hand-act log, and keeps attempts and would a week each. +// Every state it keeps has a tool that reads it, and every act it takes by itself a tool that starts it (ADR 0564 §12). +func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) { + raw, err := os.ReadFile(filepath.Join("..", "..", "module.json")) + if err != nil { + t.Fatal(err) + } + var m manifest + if err := json.Unmarshal(raw, &m); err != nil { + t.Fatal(err) + } + if m.Module != Module || m.RunsAs != Module { + t.Errorf("module %q runs as %q", m.Module, m.RunsAs) + } + if len(m.Claims) != 1 || m.Claims[0].Name != Seat || m.Claims[0].Scope != "mesh" || !slices.Equal(m.Claims[0].Serves, Verbs) { + t.Errorf("claims %+v, want %s at mesh scope serving %v", m.Claims, Seat, Verbs) + } + for _, c := range []struct { + name string + got, want []string + }{ + {"emits", m.Emits, []string{ActedEvent}}, + {"consumes", m.Consumes, []string{RaisedEvent, ChangedEvent}}, + {"invokes", m.Invokes, []string{AssertTool}}, + {"reads", m.Reads, []string{LeaseRead, ConditionsRead, HandActsRead}}, + } { + if !slices.Equal(c.got, c.want) { + t.Errorf("%s %v, want %v", c.name, c.got, c.want) + } + } + states := map[string]int{} + for _, s := range m.State { + var d struct { + Name string `json:"name"` + TTL int `json:"ttl-seconds"` + } + if json.Unmarshal(s, &d.Name) != nil { + _ = json.Unmarshal(s, &d) + } + states[d.Name] = d.TTL + } + week := 7 * 24 * 60 * 60 + if states[AttemptsState] != week || states[WouldState] != week { + t.Errorf("attempts kept %d s and would %d s, want a week each: would has no other deleter", states[AttemptsState], + states[WouldState]) + } + readBy := map[string]string{AttemptsState: HealersTool, WouldState: WouldTool, FindingsState: FindingsTool, + "an act on an event": HealTool, "S15's count": HealersTool + " recount=true", "the lease as read": HealersTool} + tools := map[string]bool{} + for _, tool := range Tools(&Healers{}) { + tools[tool.Name] = true + } + for state := range states { + if _, ok := readBy[state]; !ok { + t.Errorf("the state %s has no tool that reads it", state) + } + } + for what, tool := range readBy { + if name, _, _ := strings.Cut(tool, " "); !tools[name] { + t.Errorf("%s is read or started by %s, which is not served", what, tool) + } + } +} diff --git a/modules/mesh-healers/cmd/mesh-healers/observation.go b/modules/mesh-healers/cmd/mesh-healers/observation.go new file mode 100644 index 00000000..853550b5 --- /dev/null +++ b/modules/mesh-healers/cmd/mesh-healers/observation.go @@ -0,0 +1,77 @@ +package main + +// What the healers observe of themselves, and the finding it is written as (novox/hq ADR 0564 §2): the brake, their +// blindness and S15, each written into the bucket `findings` as mesh-sdk's health.Finding, whose key is the +// controller's key for the same condition. + +import ( + "encoding/json" + "time" + + "git.novox.be/novox/mesh-sdk/go/health" +) + +// The severities and scopes, as mesh-sdk's health names them. +const ( + Urgent = health.Urgent + Warning = health.Warning + ScopeProbe = health.ScopeProbe + ScopeMesh = health.ScopeMesh +) + +// Subject is what a finding is about. +type Subject = health.Subject + +// Observation is one thing a probe sees wrong now. +type Observation struct { + Kind string + Token string + Subject Subject + Severity health.Severity + Summary string + Said string + Headline string + Explanation string + Resolved string + Needs string + // Source is what saw it: the brake, the healers' blindness, or S15. + Source string +} + +// Key is where it is kept: `..`, the controller's key for the same condition. +func (o Observation) Key() string { + token := o.Token + if token == "" { + token = o.Kind + } + return health.Key(o.Subject.Scope, o.Subject.ID, token) +} + +// Finding is the observation as written by run at, its evidence the one line it said. +func (o Observation) Finding(run string, at time.Time) health.Finding { + f := health.Finding{Key: o.Key(), Kind: o.Kind, Subject: o.Subject, Severity: o.Severity, Summary: o.Summary, + Headline: o.Headline, Explanation: o.Explanation, Resolved: o.Resolved, Needs: o.Needs, Source: o.Source, + Observer: Module, Run: run, At: at.UTC(), Evidence: []health.Evidence{}} + if o.Token != "" && o.Token != o.Kind { + f.Token = o.Token + } + said := o.Said + if said == "" { + said = o.Summary + } + f.Evidence = append(f.Evidence, health.Evidence{At: at.UTC(), Said: said}) + return f +} + +// says is what a finding says, its times and the run that wrote it aside: two findings that say the same are one write +// (ADR 0289 §1). +func says(f health.Finding) string { + f.At, f.Run, f.V = time.Time{}, "", 0 + ev := make([]health.Evidence, len(f.Evidence)) + for i, e := range f.Evidence { + ev[i] = health.Evidence{Said: e.Said} + } + f.Evidence = ev + raw, _ := json.Marshal(f) + return string(raw) +} diff --git a/modules/mesh-healers/cmd/mesh-healers/reads-outside b/modules/mesh-healers/cmd/mesh-healers/reads-outside new file mode 100644 index 00000000..cdc54f27 --- /dev/null +++ b/modules/mesh-healers/cmd/mesh-healers/reads-outside @@ -0,0 +1,2 @@ +# What these tests read outside their package, as paths from the repository's root (issue 1936). +modules/mesh-healers/module.json diff --git a/modules/mesh-healers/cmd/mesh-healers/tools.go b/modules/mesh-healers/cmd/mesh-healers/tools.go new file mode 100644 index 00000000..949f4e2a --- /dev/null +++ b/modules/mesh-healers/cmd/mesh-healers/tools.go @@ -0,0 +1,186 @@ +package main + +// The seat's verbs and the module's own reads (novox/hq ADR 0564 §7, §12): `healers` is the registry, what the healers +// did and would do, the brake and whether they act; `heal` takes one act now on a condition, within budget and brake; +// `drill` waits for the hand-act log's home; and a read of each bucket it keeps, so a recovery is a call. + +import ( + "encoding/json" + "errors" + "fmt" + "sort" + "strings" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// Verbs are the seat's verbs, as the manifest claims them. +var Verbs = []string{"healers", "heal", "drill"} + +// Tool names. +const ( + HealersTool = Seat + ".healers" + HealTool = Seat + ".heal" + DrillTool = Seat + ".drill" + FindingsTool = "healers_findings" + WouldTool = "healers_would" +) + +func obj(props map[string]any) map[string]any { + return map[string]any{"type": "object", "properties": props} +} + +// Tools are what the runtime serves. +func Tools(h *Healers) []stdio.Tool { + return []stdio.Tool{ + {Name: HealersTool, + Description: "The mesh's healers (novox/hq ADR 0564 §7, §8): the registry, every act and escalation of the last " + + "days (the bucket attempts), whether the mesh-wide brake holds, whether they act or only write what they " + + "would do while the controller holds its lease, and the would-acts. recount=true counts the hand-act log " + + "again for S15 (a cause repaired by hand twice in a fortnight).", + Input: obj(map[string]any{ + "days": map[string]any{"type": "number", "description": "how many days of acts back (default 7)"}, + "recount": map[string]any{"type": "boolean", "description": "count the hand-act log again now"}}), + Run: func(a map[string]any) (any, error) { return h.Answer(intArg(a, "days", 7), boolArg(a, "recount")) }}, + {Name: HealTool, + Description: "One healer's act now on an open condition, by its key, within its budget and the brake, as an " + + "event would have it — or, while the controller holds its lease, the act it would take written as would. " + + "Says what was done.", + Input: obj(map[string]any{ + "key": map[string]any{"type": "string", "description": "the condition's key"}, + "why": map[string]any{"type": "string", "description": "why, in a sentence"}}), + Run: func(a map[string]any) (any, error) { + return h.Heal(strArg(a, "key"), strArg(a, "why"), stdio.CallerOf(a)) + }}, + {Name: DrillTool, + Description: "A drill: something broken on purpose to see the mesh raise and clear it, recorded in the hand-act " + + "log. Refused until the log's home is delivered (issue 1953): the controller's drill records it meanwhile.", + Input: obj(map[string]any{"why": map[string]any{"type": "string"}}), + Run: func(map[string]any) (any, error) { + return nil, errors.New("a drill is recorded in the hand-act log, whose one writer is the controller until its " + + "home is delivered (issue 1953): use mesh-controller.drill, which records it and S15 never counts") + }}, + {Name: FindingsTool, + Description: "What the healers find of themselves (their bucket findings): the brake, their blindness, and " + + "each cause repaired by hand twice in a fortnight (S15); or one whole by key.", + Input: obj(map[string]any{"key": map[string]any{"type": "string"}}), + Run: func(a map[string]any) (any, error) { return h.read(FindingsState, strArg(a, "key")) }}, + {Name: WouldTool, + Description: "What the healers would have done while the controller holds its lease (their bucket would, each " + + "key kept a week): one per healer and condition, its newest; or one whole by key.", + Input: obj(map[string]any{"key": map[string]any{"type": "string"}}), + Run: func(a map[string]any) (any, error) { return h.read(WouldState, strArg(a, "key")) }}, + } +} + +// Heal is the verb heal: one condition judged now by its healer. +func (h *Healers) Heal(key, why, caller string) (any, error) { + if key == "" || why == "" { + return nil, errors.New("heal names the condition's key and says why") + } + raw, found, err := h.Store.Get(ConditionsRead, key) + if err != nil { + return nil, fmt.Errorf("the open conditions cannot be read: %w", err) + } + if !found { + return nil, fmt.Errorf("no condition is open under %s", key) + } + said, err := h.Judge(conditionOf(raw), caller+" ("+why+")") + if err != nil { + return nil, err + } + return map[string]any{"said": said}, nil +} + +// Answer is the verb healers. +func (h *Healers) Answer(days int, recount bool) (any, error) { + if recount { + h.Wanted() + } + now := h.Now() + all, err := h.attempts() + if err != nil { + return nil, fmt.Errorf("what the healers did cannot be read: %w", err) + } + var recent []Attempt + for _, a := range all { + if now.Sub(a.Begun) <= time.Duration(days)*24*time.Hour { + recent = append(recent, a) + } + } + sort.SliceStable(recent, func(i, j int) bool { return recent[i].Begun.After(recent[j].Begun) }) + braked, said := h.brakeHolds(actsWithin(all, now.Add(-BrakeWindow)), now) + var rows []map[string]any + for _, r := range h.Registry { + rows = append(rows, map[string]any{"id": r.ID, "kinds": r.Kinds, "condition": r.Condition, "repair": r.Repair, + "budget": fmt.Sprintf("%d act(s) within %s, %s apart at least", r.Budget, r.Window, r.Settle), + "then": r.Then, "acts-in": r.ActsIn, "from": r.From}) + } + acting, why := h.Acting() + would, err := h.read(WouldState, "") + if err != nil { + would = map[string]any{"unread": err.Error()} + } + h.mu.Lock() + declined := map[string]string{} + for k, v := range h.declined { + declined[k] = v + } + h.mu.Unlock() + if recent == nil { + recent = []Attempt{} + } + return map[string]any{"healers": rows, "acts": recent, "days": days, "acting": acting, "why": why, "would": would, + "brake": map[string]any{"holds": braked, "said": said, "limit": BrakeLimit, "window": BrakeWindow.String()}, + "declined": declined, + "note": "a heal is never a hand act; whether it repaired anything is the condition's clearing to say"}, nil +} + +// read is every key of a bucket with its value, or one whole. +func (h *Healers) read(state, key string) (any, error) { + if key != "" { + raw, found, err := h.Store.Get(state, key) + if err != nil { + return nil, err + } + if !found { + return nil, fmt.Errorf("nothing is kept under %s in %s", key, state) + } + return json.RawMessage(raw), nil + } + keys, err := h.Store.Keys(state) + if err != nil { + return nil, err + } + rows := map[string]json.RawMessage{} + for _, k := range keys { + if raw, found, err := h.Store.Get(state, k); err == nil && found { + rows[k] = raw + } + } + return map[string]any{state: rows, "count": len(rows)}, nil +} + +func strArg(a map[string]any, k string) string { + s, _ := a[k].(string) + return strings.TrimSpace(s) +} + +// boolArg reads a flag as a boolean or as "true". +func boolArg(a map[string]any, k string) bool { + switch v := a[k].(type) { + case bool: + return v + case string: + return strings.EqualFold(strings.TrimSpace(v), "true") + } + return false +} + +func intArg(a map[string]any, k string, otherwise int) int { + if n, ok := a[k].(float64); ok && n > 0 { + return int(n) + } + return otherwise +} diff --git a/modules/mesh-healers/cmd/mesh-healers/wanted.go b/modules/mesh-healers/cmd/mesh-healers/wanted.go new file mode 100644 index 00000000..d07ec541 --- /dev/null +++ b/modules/mesh-healers/cmd/mesh-healers/wanted.go @@ -0,0 +1,152 @@ +package main + +// S15: a cause repaired by hand twice within a fortnight is a healer wanted (ADR 0231 §27, §28; ADR 0564 Limits 2). +// The hand-act log is the controller's bucket mesh-controller_hand-acts, a key-value bucket with no events: mesh-healers +// watches it, the one of the four holders granted its read, counts each cause again on every put, and writes the +// finding `healer-wanted` per cause. **A heal is never a hand act**, and an act that is a person's decision by design +// (a drill above all) never counts. The finding clears when fewer than two acts of its cause remain within the +// fortnight: the healers wake once, when the oldest counted act leaves it. From the log's home (issue 1953) they +// watch that instead. + +import ( + "encoding/json" + "fmt" + "slices" + "sort" + "strings" + "time" +) + +// Within is how far back a repeated cause counts. +const Within = 14 * 24 * time.Hour + +// HandAct is what S15 reads of one entry of the controller's hand-act log (mesh-controller internal/link HandAct). +type HandAct struct { + ID string `json:"id"` + At time.Time `json:"at"` + By string `json:"by"` + Verb string `json:"verb"` + Args []string `json:"arguments,omitempty"` + Why string `json:"why"` + Cause string `json:"cause"` + Kind string `json:"kind,omitempty"` +} + +// decided are the verbs whose entries are a person's decision by design, with the causes they hold for (every cause +// when none): the controller's handActVerbs, read as S15 reads them. A verb not listed counts as a repair. +var decided = map[string][]string{ + "warrant": nil, + "seat": {"seat-passed"}, + "conditions silence": {"operator-answer"}, + "hand-act record": {"drill"}, + "secret issued": nil, + "hand-act drill": nil, + "retire approve": nil, + "retire reject": nil, + "cleanup delete": nil, + "dead-letters deliver": nil, + "dead-letters drop": nil, + "collect": nil, + "mirrors": nil, + "move-placement": nil, + "move-config": nil, + "bus upgrade": nil, + "upgrade release-backlog": nil, + "secret accept": nil, + "secret rotate": {"leaked-in-logs"}, + "rotate": {"leaked-in-logs"}, +} + +// personsDecision is whether an act is a person's decision by design, by the verb that recorded it. +func personsDecision(a HandAct) bool { + causes, ok := decided[a.Verb] + if !ok { + return false + } + return len(causes) == 0 || slices.Contains(causes, a.Cause) +} + +// HandActHeard is one change of the hand-act log: kept or forgotten, and every cause counted again. +func (h *Healers) HandActHeard(key, op string, value []byte) { + h.mu.Lock() + if h.acts == nil { + h.acts = map[string]HandAct{} + } + if op != "put" { + delete(h.acts, key) + } else { + var a HandAct + if json.Unmarshal(value, &a) == nil { + h.acts[key] = a + } + } + h.mu.Unlock() + h.Wanted() +} + +// Wanted counts every cause repaired by hand within the fortnight and writes a finding for each counted twice or more, +// deleting the rest; it wakes again when the oldest counted act of a wanted cause leaves the fortnight. +func (h *Healers) Wanted() []Observation { + h.mu.Lock() + defer h.mu.Unlock() + now := h.Now() + byCause := map[string][]HandAct{} + for _, a := range h.acts { + if personsDecision(a) || now.Sub(a.At) > Within { + continue + } + byCause[a.Cause] = append(byCause[a.Cause], a) + } + var causes []string + for c, acts := range byCause { + if len(acts) >= 2 { + causes = append(causes, c) + } + } + sort.Strings(causes) + var found []Observation + var next time.Time + for _, cause := range causes { + acts := byCause[cause] + sort.Slice(acts, func(i, j int) bool { return acts[i].At.Before(acts[j].At) }) + var lines []string + for _, a := range acts { + lines = append(lines, fmt.Sprintf("%s %s by %s: %s", a.At.UTC().Format("2006-01-02 15:04"), + strings.TrimSpace(a.Verb+" "+strings.Join(a.Args, " ")), a.By, a.Why)) + } + newest := acts[len(acts)-1] + wanted := "a healer is wanted for it" + for _, r := range h.Registry { + if slices.Contains(r.Kinds, cause) { + wanted = fmt.Sprintf("healer %s answers this cause and a person still repaired it: its reach or its budget "+ + "is not enough", r.ID) + } + } + words := strings.NewReplacer(".", " ", "_", " ").Replace(cause) + found = append(found, Observation{Kind: "healer-wanted", Token: "healer-wanted", Severity: Warning, + Subject: Subject{Scope: ScopeMesh, ID: "hand-acts." + cause}, + Summary: fmt.Sprintf("%q was repaired by hand %d times in %d days, the last by %s: %s", cause, len(acts), + int(Within.Hours()/24), newest.By, wanted), + Said: strings.Join(lines, "; "), + Headline: capital(words) + " keeps being fixed by hand", + Explanation: fmt.Sprintf("A person repaired %s by hand %d times in %d days, so an automatic repair is wanted "+ + "for it. Nothing is broken now.", words, len(acts), int(Within.Hours()/24)), + Resolved: "Resolved: no more hand repairs of " + words}) + if leaves := acts[0].At.Add(Within); next.IsZero() || leaves.Before(next) { + next = leaves + } + } + h.see("S15", found) + if h.After != nil && !next.IsZero() && !next.Equal(h.s15At) { + h.s15At = next + h.After(next.Sub(now)+time.Second, func() { h.Wanted() }) + } + return found +} + +func capital(s string) string { + if s == "" { + return s + } + return strings.ToUpper(s[:1]) + s[1:] +} diff --git a/modules/mesh-healers/cmd/mesh-healers/whole_integration_test.go b/modules/mesh-healers/cmd/mesh-healers/whole_integration_test.go new file mode 100644 index 00000000..5c7ec972 --- /dev/null +++ b/modules/mesh-healers/cmd/mesh-healers/whole_integration_test.go @@ -0,0 +1,416 @@ +//go:build integration + +package main + +// mesh-healers whole against a real bus (novox/hq ADR 0564 §7, §8; issue 1952 H3): the built binary, played over its +// real stdin and stdout as the node's tool runtime plays it (mesh/state.* and watches answered from a real nats-server's +// key-value buckets, the keeper's events handed over as the runtime hands them, the grants part's assert answered). An +// induced failure, a module's consumer gone, is only written as would while the controller holds its lease; from the +// write that hands the lease away a healer acts, its attempt on the bus and healer-acted said; past twelve acts in the +// hour the brake stops every healer, said as an urgent finding on the bus. + +import ( + "bufio" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "os" + "os/exec" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/health" + "git.novox.be/novox/mesh-sdk/go/testbus" +) + +type wire struct { + ID json.RawMessage `json:"id,omitempty"` + Method string `json:"method,omitempty"` + Params json.RawMessage `json:"params,omitempty"` + Result json.RawMessage `json:"result,omitempty"` + Error *struct { + Message string `json:"message"` + } `json:"error,omitempty"` +} + +// player is the node's tool runtime, as far as the self-check needs it, over a real bus. +type player struct { + ctx context.Context + nc *nats.Conn + js jetstream.JetStream + mu sync.Mutex + in io.Writer + next int + pending map[string]chan wire + asked []string +} + +func (p *player) send(m map[string]any) { + raw, _ := json.Marshal(m) + p.mu.Lock() + defer p.mu.Unlock() + _, _ = p.in.Write(append(raw, '\n')) +} + +func (p *player) answer(id json.RawMessage, result any, err error) { + if err != nil { + p.send(map[string]any{"id": id, "error": map[string]any{"code": -32000, "message": err.Error()}}) + return + } + p.send(map[string]any{"id": id, "result": result}) +} + +func (p *player) serve(out io.Reader) { + scanner := bufio.NewScanner(out) + scanner.Buffer(make([]byte, 1<<20), 1<<24) + for scanner.Scan() { + var m wire + if json.Unmarshal(scanner.Bytes(), &m) != nil { + continue + } + if m.Method == "" { + var key string + _ = json.Unmarshal(m.ID, &key) + p.mu.Lock() + ch := p.pending[key] + delete(p.pending, key) + p.mu.Unlock() + if ch != nil { + ch <- m + } + continue + } + go p.handle(m) + } +} + +// kv is the bucket a state names: this module's own by its local name, another's as `.`. +func (p *player) kv(state string) (jetstream.KeyValue, error) { + if owner, name, ok := strings.Cut(state, "."); ok { + return p.js.KeyValue(p.ctx, owner+"_"+name) + } + return p.js.KeyValue(p.ctx, Module+"_"+state) +} + +// deliver hands the bundle one event as the runtime does from its consumer. +func (p *player) deliver(key string, body []byte) { + p.mu.Lock() + p.next++ + id := fmt.Sprintf("e-%d", p.next) + p.mu.Unlock() + p.send(map[string]any{"id": id, "method": "mesh/event", "params": map[string]any{"envelope": map[string]any{ + "key": key, "node": "ace", "body": json.RawMessage(body)}}}) +} + +func (p *player) handle(m wire) { + var a struct { + State string `json:"state"` + Key string `json:"key"` + Value json.RawMessage `json:"value"` + Body json.RawMessage `json:"body"` + } + _ = json.Unmarshal(m.Params, &a) + switch m.Method { + case "mesh/state.get": + kv, err := p.kv(a.State) + if err != nil { + p.answer(m.ID, nil, err) + return + } + e, err := kv.Get(p.ctx, a.Key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + p.answer(m.ID, nil, nil) + return + } + if err != nil { + p.answer(m.ID, nil, err) + return + } + p.answer(m.ID, map[string]any{"key": e.Key(), "value": json.RawMessage(e.Value()), "revision": e.Revision()}, nil) + case "mesh/state.put": + kv, err := p.kv(a.State) + if err == nil { + var rev uint64 + if rev, err = kv.Put(p.ctx, a.Key, a.Value); err == nil { + p.answer(m.ID, map[string]any{"revision": rev}, nil) + return + } + } + p.answer(m.ID, nil, err) + case "mesh/state.create": + kv, err := p.kv(a.State) + if err == nil { + var rev uint64 + if rev, err = kv.Create(p.ctx, a.Key, a.Value); err == nil { + p.answer(m.ID, map[string]any{"revision": rev}, nil) + return + } + if errors.Is(err, jetstream.ErrKeyExists) { + p.send(map[string]any{"id": m.ID, "error": map[string]any{"code": -32001, "message": "the key was written by another writer first"}}) + return + } + } + p.answer(m.ID, nil, err) + case "mesh/state.watch": + kv, err := p.kv(a.State) + if err != nil { + p.answer(m.ID, nil, err) + return + } + key := a.Key + if key == "" { + key = ">" + } + w, err := kv.Watch(p.ctx, key) + if err != nil { + p.answer(m.ID, nil, err) + return + } + var watch string + _ = json.Unmarshal(m.ID, &watch) + current := true + for e := range w.Updates() { + if e == nil { + current = false + p.answer(m.ID, map[string]any{}, nil) + continue + } + op := "put" + if e.Operation() != jetstream.KeyValuePut { + op = "delete" + } + change := map[string]any{"state": a.State, "key": e.Key(), "op": op, "revision": e.Revision(), "current": current} + if op == "put" { + change["value"] = json.RawMessage(e.Value()) + } + p.mu.Lock() + p.next++ + id := fmt.Sprintf("w-%d", p.next) + p.mu.Unlock() + p.send(map[string]any{"id": id, "method": "mesh/state", "params": map[string]any{"watch": watch, "change": change}}) + } + case "mesh/subscribe": + p.answer(m.ID, map[string]any{}, nil) + case "mesh/state.keys": + kv, err := p.kv(a.State) + if err != nil { + p.answer(m.ID, nil, err) + return + } + keys, err := kv.Keys(p.ctx) + if errors.Is(err, jetstream.ErrNoKeysFound) { + keys, err = []string{}, nil + } + p.answer(m.ID, keys, err) + case "mesh/state.delete": + kv, err := p.kv(a.State) + if err == nil { + err = kv.Delete(p.ctx, a.Key) + } + p.answer(m.ID, map[string]any{}, err) + case "mesh/publish": + var e struct { + Key string `json:"key"` + Body json.RawMessage `json:"body"` + } + _ = json.Unmarshal(m.Params, &e) + p.answer(m.ID, map[string]any{}, p.nc.Publish("mesh.mod."+Module+".event."+e.Key, e.Body)) + case "mesh/ask": + var ask struct { + Key string `json:"key"` + Body json.RawMessage `json:"body"` + } + _ = json.Unmarshal(m.Params, &ask) + p.mu.Lock() + p.asked = append(p.asked, ask.Key+" "+string(ask.Body)) + p.mu.Unlock() + if ask.Key != AssertTool { + p.answer(m.ID, nil, errors.New("nats: no responders available for request")) + return + } + p.answer(m.ID, map[string]any{"says": "1 the bus lacks: created at this ask"}, nil) + default: + p.answer(m.ID, nil, errors.New("nats: no responders available for request")) + } +} + +func (p *player) call(name string, args map[string]any) (json.RawMessage, error) { + p.mu.Lock() + p.next++ + id := fmt.Sprintf("c-%d", p.next) + ch := make(chan wire, 1) + p.pending[id] = ch + p.mu.Unlock() + p.send(map[string]any{"id": id, "method": "tools/call", "params": map[string]any{"name": name, "arguments": args, + "_meta": map[string]any{"mesh/caller": "person.jochen"}}}) + select { + case m := <-ch: + if m.Error != nil { + return nil, errors.New(m.Error.Message) + } + var r struct { + Content []struct { + Text string `json:"text"` + } `json:"content"` + IsError bool `json:"isError"` + } + if err := json.Unmarshal(m.Result, &r); err != nil || len(r.Content) != 1 { + return nil, fmt.Errorf("%s answered %s", name, m.Result) + } + if r.IsError { + return nil, errors.New(r.Content[0].Text) + } + return json.RawMessage(r.Content[0].Text), nil + case <-time.After(60 * time.Second): + return nil, fmt.Errorf("%s did not answer in 60s", name) + } +} + +func eventually(t *testing.T, what string, ok func() bool) { + t.Helper() + for end := time.Now().Add(30 * time.Second); time.Now().Before(end); time.Sleep(50 * time.Millisecond) { + if ok() { + return + } + } + t.Fatalf("never: %s", what) +} + +func raised(key, module string) []byte { + raw, _ := json.Marshal(map[string]any{"key": key, "kind": "consumer-lost", "severity": "warning", "resolver": "self", + "subject": map[string]any{"scope": "bus", "id": "EVENTS." + module}, "event": "condition-raised", + "show": health.ShowKeeper(key)}) + return raw +} + +// 1952/H3: an induced failure; the healer acts only from the lease write, and brakes. +func TestTheHealersWholeActOnTheBusAndBrake(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + dir := t.TempDir() + bin := filepath.Join(dir, Module) + build := exec.Command("go", "build", "-o", bin, ".") + build.Stderr = os.Stderr + if err := build.Run(); err != nil { + t.Fatalf("mesh-healers does not build: %v", err) + } + nc, err := nats.Connect(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(nc.Close) + js, err := jetstream.New(nc) + if err != nil { + t.Fatal(err) + } + for _, b := range []string{Module + "_" + AttemptsState, Module + "_" + WouldState, Module + "_" + FindingsState, + "mesh-controller_lease", "mesh-conditions_conditions", "mesh-controller_hand-acts"} { + if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: b}); err != nil { + t.Fatal(err) + } + } + bucket := func(name string) jetstream.KeyValue { + kv, err := js.KeyValue(ctx, name) + if err != nil { + t.Fatal(err) + } + return kv + } + keys := func(name string) []string { + k, _ := bucket(name).Keys(ctx) + return k + } + lease := bucket("mesh-controller_lease") + if _, err := lease.Put(ctx, LeaseKey, []byte(`{"instance":"novox/1/2026-10-27","epoch":57}`)); err != nil { + t.Fatal(err) + } + acted, err := nc.SubscribeSync("mesh.mod." + Module + ".event." + ActedEvent) + if err != nil { + t.Fatal(err) + } + cmd := exec.CommandContext(ctx, bin) + cmd.Env = append(os.Environ(), "MESH_NODE=ace", "MESH_MODULE="+Module) + stdin, _ := cmd.StdinPipe() + stdout, _ := cmd.StdoutPipe() + cmd.Stderr = os.Stderr + if err := cmd.Start(); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = stdin.Close(); cancel(); _ = cmd.Wait() }) + p := &player{ctx: ctx, nc: nc, js: js, in: stdin, pending: map[string]chan wire{}} + go p.serve(stdout) + healers := func() map[string]any { + raw, err := p.call(HealersTool, map[string]any{}) + if err != nil { + return nil + } + var a map[string]any + _ = json.Unmarshal(raw, &a) + return a + } + eventually(t, "the lease read as a controller's", func() bool { + a := healers() + return a != nil && strings.Contains(fmt.Sprint(a["why"]), "names a controller") + }) + + // The controller holds the lease: the induced failure is written as would, never acted on. + p.deliver(RaisedEvent, raised("bus.EVENTS.records.missing", "records")) + eventually(t, "the would-act written", func() bool { return len(keys(Module+"_"+WouldState)) == 1 }) + if k := keys(Module + "_" + AttemptsState); len(k) != 0 { + t.Fatalf("acted under the controller's lease: %v", k) + } + // The write that hands the lease away: the same failure is acted on. + if _, err := lease.Put(ctx, LeaseKey, []byte(`{"handed-to":"the operator's hand-over"}`)); err != nil { + t.Fatal(err) + } + eventually(t, "the healers acting", func() bool { return healers()["acting"] == true }) + p.deliver(RaisedEvent, raised("bus.EVENTS.records.missing", "records")) + eventually(t, "the act begun and finished on the bus", func() bool { + k := keys(Module + "_" + AttemptsState) + if len(k) != 1 { + return false + } + e, err := bucket(Module+"_"+AttemptsState).Get(ctx, k[0]) + return err == nil && strings.Contains(string(e.Value()), `"outcome":"acted"`) + }) + msg, err := acted.NextMsg(10 * time.Second) + if err != nil || !strings.Contains(string(msg.Data), `"healer":"H3"`) { + t.Fatalf("healer-acted was not said: %v", err) + } + p.mu.Lock() + asked := strings.Join(p.asked, "\n") + p.mu.Unlock() + if !strings.Contains(asked, AssertTool+` {"module":"records"}`) { + t.Errorf("the grants part was asked %s", asked) + } + // Eleven more acts within the hour: the twelfth was the limit, and the next condition meets the brake. + attempts := bucket(Module + "_" + AttemptsState) + for i := 0; i < BrakeLimit-1; i++ { + begun := time.Now().Add(-time.Duration(i+1) * time.Minute).UTC() + raw, _ := json.Marshal(Attempt{Healer: "H3", Condition: fmt.Sprintf("bus.EVENTS.m%d.missing", i), BudgetKey: "x", + Outcome: OutcomeActed, Begun: begun}) + if _, err := attempts.Put(ctx, attemptKey("H3", fmt.Sprintf("bus.EVENTS.m%d.missing", i), begun), raw); err != nil { + t.Fatal(err) + } + } + p.deliver(RaisedEvent, raised("bus.EVENTS.mail.missing", "mail")) + eventually(t, "the brake written as a finding on the bus", func() bool { + e, err := bucket(Module+"_"+FindingsState).Get(ctx, brakeKey) + if err != nil { + return false + } + f, err := health.DecodeFinding(e.Value()) + return err == nil && f.Severity == health.Urgent && f.Kind == "healers-braked" + }) + if n := len(keys(Module + "_" + AttemptsState)); n != BrakeLimit { + t.Errorf("past the brake the attempts are %d, want %d", n, BrakeLimit) + } +} diff --git a/modules/mesh-healers/go.mod b/modules/mesh-healers/go.mod new file mode 100644 index 00000000..587a34e4 --- /dev/null +++ b/modules/mesh-healers/go.mod @@ -0,0 +1,22 @@ +module mesh-healers + +go 1.26.0 + +require ( + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204453-69fcb6c1c1ba + github.com/nats-io/nats.go v1.54.0 +) + +require ( + github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect + github.com/google/go-tpm v0.9.8 // indirect + github.com/klauspost/compress v1.20.0 // indirect + github.com/minio/highwayhash v1.0.4 // indirect + github.com/nats-io/jwt/v2 v2.8.1 // indirect + github.com/nats-io/nats-server/v2 v2.11.17 // indirect + github.com/nats-io/nkeys v0.4.16 // indirect + github.com/nats-io/nuid v1.0.1 // indirect + golang.org/x/crypto v0.57.0 // indirect + golang.org/x/sys v0.48.0 // indirect + golang.org/x/time v0.15.0 // indirect +) diff --git a/modules/mesh-healers/go.sum b/modules/mesh-healers/go.sum new file mode 100644 index 00000000..ae67790f --- /dev/null +++ b/modules/mesh-healers/go.sum @@ -0,0 +1,27 @@ +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204453-69fcb6c1c1ba h1:uXwkcQWgbu/UthizGHFaKPiA4NBVJgqL9q2jJnnGR1o= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204453-69fcb6c1c1ba/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= +github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= +github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= +github.com/google/go-tpm v0.9.8/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY= +github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA= +github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= +github.com/minio/highwayhash v1.0.4 h1:asJizugGgchQod2ja9NJlGOWq4s7KsAWr5XUc9Clgl4= +github.com/minio/highwayhash v1.0.4/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ= +github.com/nats-io/jwt/v2 v2.8.1 h1:V0xpGuD/N8Mi+fQNDynXohVvp7ZztevW5io8CUWlPmU= +github.com/nats-io/jwt/v2 v2.8.1/go.mod h1:nWnOEEiVMiKHQpnAy4eXlizVEtSfzacZ1Q43LIRavZg= +github.com/nats-io/nats-server/v2 v2.11.17 h1:GKEghcFK6A+aFx11Yf1LjgLC3txAwvyhnYzhBIQZA8I= +github.com/nats-io/nats-server/v2 v2.11.17/go.mod h1:B1sFVz4StNosQ903ak4N1G01Fl/9f8e06mXpFIE2K24= +github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA= +github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k= +github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg= +github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs= +github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= +github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= +golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= +golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA= +golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= +golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= +golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= +golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= diff --git a/modules/mesh-healers/module.json b/modules/mesh-healers/module.json new file mode 100644 index 00000000..186965ec --- /dev/null +++ b/modules/mesh-healers/module.json @@ -0,0 +1,78 @@ +{ + "module": "mesh-healers", + "version": "1", + "runs-as": "mesh-healers", + "claims": [ + { + "name": "mesh-healers", + "scope": "mesh", + "serves": [ + "healers", + "heal", + "drill" + ] + } + ], + "emits": [ + "healer-acted" + ], + "consumes": [ + "mesh-conditions.condition-raised", + "mesh-conditions.condition-changed" + ], + "invokes": [ + "mesh-grants.grants_assert" + ], + "reads": [ + "mesh-controller.lease", + "mesh-conditions.conditions", + "mesh-controller.hand-acts" + ], + "state": [ + { + "name": "attempts", + "ttl-seconds": 604800 + }, + { + "name": "would", + "ttl-seconds": 604800 + }, + "findings" + ], + "own-secrets": { + "broker": "${dir:state}/broker" + }, + "secrets-owner": "mesh-healers", + "resources": [ + { + "id": "account", + "type": "user", + "name": "mesh-healers", + "shell": "/usr/bin/nologin", + "home": "/var/lib/mesh-healers" + }, + { + "id": "state", + "type": "directory", + "mode": "0700", + "owner": "mesh-healers", + "place": ".", + "unknown": "program" + } + ], + "build": { + "artifacts": [ + { + "name": "tools", + "kind": "bundle", + "language": "go", + "system": "arch", + "from": "cmd/mesh-healers", + "binary": "mesh-healers", + "loads": [ + "mesh-healers" + ] + } + ] + } +}