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.
This commit is contained in:
@@ -0,0 +1,2 @@
|
||||
/mesh-healers
|
||||
/cmd/mesh-healers/mesh-healers
|
||||
@@ -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.
|
||||
@@ -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
|
||||
// `<healer>.<condition key>.<begun ns>` is the healers' own), an act written to `would` while the lease names a
|
||||
// controller (one key per healer and condition, `<healer>.<condition key>`, 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 (`<stream>.<module>`). 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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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: `<scope>.<id>.<token>`, 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)
|
||||
}
|
||||
@@ -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
|
||||
@@ -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
|
||||
}
|
||||
@@ -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:]
|
||||
}
|
||||
@@ -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 `<owner>.<name>`.
|
||||
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)
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
)
|
||||
@@ -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=
|
||||
@@ -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"
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user