diff --git a/modules/mesh-conditions/cmd/mesh-conditions/compare.go b/modules/mesh-conditions/cmd/mesh-conditions/compare.go new file mode 100644 index 00000000..f2092381 --- /dev/null +++ b/modules/mesh-conditions/cmd/mesh-conditions/compare.go @@ -0,0 +1,218 @@ +package main + +// **Built dark, compared, then switched by the operator** (novox/hq ADR 0564 clause 9; issue 1952 C5, C8): `compare` +// lists the open keys of the keeper and of the controller, each difference with its reason; the kinds that end with the +// controller, the kinds only the keeper raises and the keys held while their module is delivered are listed apart and +// not counted. `take-over` raises an approve ask under the keeper's own name (ADR 0259), refused while compare shows a +// difference, and on the warrant writes the take-over key naming the keeper. `hand-back` writes it back to the +// controller until the lease write, and is refused after it. + +import ( + "encoding/json" + "errors" + "fmt" + "sort" + "strings" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/asks" + "git.novox.be/novox/mesh-sdk/go/health" +) + +// Difference is one key the two sides do not agree on, or one listed apart. +type Difference struct { + Key string `json:"key"` + Kind string `json:"kind"` + Reason string `json:"reason"` +} + +// Comparison is compare's answer. +type Comparison struct { + Differences []Difference `json:"differences"` + // Apart are the keys not counted: of kinds that end with the controller, raised only by the keeper, or held. + Apart []Difference `json:"apart"` + Same int `json:"same"` + At time.Time `json:"at"` +} + +// Empty says nothing differs. +func (c Comparison) Empty() bool { return len(c.Differences) == 0 } + +// Compare is the keeper's open conditions against the controller's. +func (k *Keeper) Compare() (Comparison, error) { + mine, err := k.Open() + if err != nil { + return Comparison{}, err + } + k.mu.Lock() + k.init() + theirs := map[string]health.Condition{} + for key, c := range k.ctrl { + theirs[key] = c + } + heldNow := map[string]held{} + for key, h := range k.held { + heldNow[key] = h + } + k.mu.Unlock() + out := Comparison{Differences: []Difference{}, Apart: []Difference{}, At: k.Now().UTC()} + open := map[string]health.Condition{} + for _, c := range mine { + open[c.Key] = c + } + keys := map[string]bool{} + for key := range open { + keys[key] = true + } + for key := range theirs { + keys[key] = true + } + for key := range heldNow { + keys[key] = true + } + sorted := make([]string, 0, len(keys)) + for key := range keys { + sorted = append(sorted, key) + } + sort.Strings(sorted) + for _, key := range sorted { + c, inMine := open[key] + t, inTheirs := theirs[key] + kind := c.Kind + if !inMine { + kind = t.Kind + } + home, homed := k.homeOf(kind) + if h, ok := heldNow[key]; ok { + out.Apart = append(out.Apart, Difference{Key: key, Kind: kind, + Reason: fmt.Sprintf("held while %s is delivered on %s (version %s)", h.Module, h.Node, short(h.Version))}) + continue + } + switch { + case homed && (home.Carried || home.Replaced): + why := "ends with the controller: carried from its bucket until the lease write" + if home.Replaced { + why = "ends with the controller: its fault is observed otherwise" + } + out.Apart = append(out.Apart, Difference{Key: key, Kind: kind, Reason: why}) + case inMine && !inTheirs && (!homed || !home.Controller): + out.Apart = append(out.Apart, Difference{Key: key, Kind: kind, Reason: "keeper only: the controller never raises this kind"}) + case inMine && !inTheirs: + out.Differences = append(out.Differences, Difference{Key: key, Kind: kind, + Reason: "open at the keeper, not at the controller"}) + case !inMine && !homed: + out.Differences = append(out.Differences, Difference{Key: key, Kind: kind, + Reason: fmt.Sprintf("open at the controller; no home is decided for the kind %s (ADR 0564's gap)", kind)}) + case !inMine: + out.Differences = append(out.Differences, Difference{Key: key, Kind: kind, + Reason: fmt.Sprintf("open at the controller; %s has not written its finding", home.Observer)}) + case c.Severity != t.Severity: + out.Differences = append(out.Differences, Difference{Key: key, Kind: kind, + Reason: fmt.Sprintf("%s at the keeper, %s at the controller", c.Severity, t.Severity)}) + default: + out.Same++ + } + } + return out, nil +} + +// Asks is the operator channel as the keeper asks it: an approve ask under its own name, and where one stands. +type Asks interface { + Ask(a asks.Ask) error +} + +// TakeOver raises the approve ask of a switch to the keeper, refused while compare differs or after the lease write. +func (k *Keeper) TakeOver(a Asks, by, why string) (map[string]any, error) { + if strings.TrimSpace(why) == "" { + return nil, errors.New("a take-over says why") + } + if k.LeaseAway() { + return nil, errors.New("the controller's lease is handed away: the keeper is the one source already") + } + cmp, err := k.Compare() + if err != nil { + return nil, err + } + if !cmp.Empty() { + return nil, fmt.Errorf("refused: compare shows %d difference(s), the first %s (%s); the take-over waits for none", + len(cmp.Differences), cmp.Differences[0].Key, cmp.Differences[0].Reason) + } + now := k.Now().UTC() + id := fmt.Sprintf("take-over-%d", now.UnixNano()) + act := asks.Act{"verb": Seat + ".take-over", "module": Module, "ask": id} + bind, err := asks.ActDigest(act) + if err != nil { + return nil, err + } + ask := asks.Ask{ID: id, Headline: "Switch the mesh's alarms to the new keeper?", Who: asks.Operator, + Explanation: fmt.Sprintf("The new condition keeper holds the same %d open alarms as the controller. Your yes makes "+ + "the alarm messages come from the keeper, with the same words. It can be handed back until the controller "+ + "gives up its lease.", cmp.Same), + Options: []asks.Option{{ID: "yes", Label: "Yes, switch", Does: "Alarms come from the keeper.", Level: asks.Approve, Binds: bind}, + {ID: "no", Label: "No, leave it", Does: "Alarms keep coming from the controller.", Level: asks.Acknowledge}}, + Expires: now.Add(asks.ApproveLasts - time.Minute), OnExpiry: "Alarms keep coming from the controller.", + About: Module + "-take-over", Details: "Asked by " + by + ": " + why} + if err := a.Ask(ask); err != nil { + return nil, err + } + return map[string]any{"asked": id, "same": cmp.Same, "apart": len(cmp.Apart)}, nil +} + +// Warranted is the act on the operator's answer: a yes, compare still empty, writes the take-over key naming the keeper. +func (k *Keeper) Warranted(o asks.Option, w asks.Warrant) error { + if o.ID != "yes" { + k.Logf("the take-over %s was answered %s: alarms stay with the controller", w.Ask, o.ID) + return nil + } + cmp, err := k.Compare() + if err != nil { + return err + } + if !cmp.Empty() { + return fmt.Errorf("the take-over %s was approved, and compare now shows %d difference(s): not switched", w.Ask, + len(cmp.Differences)) + } + return k.writeTakeover(health.Takeover{Follow: health.FollowKeeper, At: k.Now().UTC(), By: w.Says(), Ask: w.Ask}) +} + +// HandBack switches the readers back to the controller, until the lease write. +func (k *Keeper) HandBack(by, why string) (map[string]any, error) { + if strings.TrimSpace(why) == "" { + return nil, errors.New("a hand-back says why") + } + if k.LeaseAway() { + return nil, errors.New("refused: the controller's lease is handed away, so there is no controller to hand back to") + } + t := health.Takeover{Follow: health.FollowController, At: k.Now().UTC(), By: by, Why: why} + if err := k.writeTakeover(t); err != nil { + return nil, err + } + return map[string]any{"follow": t.Follow, "at": t.At}, nil +} + +// writeTakeover writes the take-over key, and keeps it in history as the record of the switch. +func (k *Keeper) writeTakeover(t health.Takeover) error { + raw, err := health.EncodeTakeover(t) + if err != nil { + return err + } + _, rev, found, err := k.Store.Get(TakeoverState, health.TakeoverKey) + if err != nil { + return err + } + if found { + _, err = k.Store.Update(TakeoverState, health.TakeoverKey, json.RawMessage(raw), rev) + } else { + _, err = k.Store.Create(TakeoverState, health.TakeoverKey, json.RawMessage(raw)) + } + if errors.Is(err, stdio.ErrChanged) { + return fmt.Errorf("the take-over key moved under this write: read it again and ask again") + } + if err != nil { + return err + } + k.Takeover(t) + k.Logf("the readers of conditions follow the %s from now (%s)", t.Follow, t.By) + return nil +} diff --git a/modules/mesh-conditions/cmd/mesh-conditions/hold.go b/modules/mesh-conditions/cmd/mesh-conditions/hold.go new file mode 100644 index 00000000..8cc8e7b8 --- /dev/null +++ b/modules/mesh-conditions/cmd/mesh-conditions/hold.go @@ -0,0 +1,308 @@ +package main + +// **A module being delivered raises no condition until its delivery ends** (novox/hq ADR 0564 clause 13 and Limits 12; +// issue 1952 C7, C8). The keeper reads the assigner's wants and the nodes' haves, and judges a module's delivery on a +// node with mesh-sdk's delivery.MotionOf, as the view does. While it is delivered there, a finding about it there, or +// about what it serves, and a condition carried from the controller's bucket on those subjects, is kept in the state +// `held` and raises nothing. When the delivery ends, healthy at the want or failed, what still stands is raised once, +// saying it happened during the delivery; what went is dropped with a log line. The hold lasts at most the delivery's +// bound, a key of the state `bounds` whose time to live is the bound: its expiry raises what is held, saying the delivery +// did not end. A finding about another module or about the machine itself is never held. A condition already open is +// not held: it is no news. + +import ( + "encoding/json" + "errors" + "fmt" + "sort" + "strings" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/health" +) + +// The modules whose delivery holds what they serve beyond their own findings (Limits 12). +const ( + controllerModule = "mesh-controller" + runtimeModule = "node-tools" +) + +// controllerServes are the sources of what the controller serves while it is delivered: its listening, its lease, its +// writes and its facts (ADR 0505 §2 (a)), matched to mesh-controller whatever its key names. +var controllerServes = map[string]bool{"S4": true, "S7": true, "S12": true, "S13": true, "S14": true, "S25": true} + +// about is a node and a module a finding or a condition is about, for the hold. +type about struct{ Node, Module string } + +// aboutOf are the modules on nodes a condition's subject is about (Limits 12): a module on a machine, a provider on its +// node, a seat of the same name as its module on a machine, the controller's own work wherever mesh-controller is +// delivered, and every module's tools on a machine whose tool runner is delivered. Never the machine itself. +func (k *Keeper) aboutOf(kind, source string, s health.Subject) []about { + var out []about + parts := strings.Split(s.ID, ".") + switch s.Scope { + case health.ScopeModule: + if len(parts) >= 2 { + out = append(out, about{Node: parts[len(parts)-1], Module: strings.Join(parts[:len(parts)-1], ".")}) + } + case health.ScopeProvider: + if len(parts) >= 2 { + out = append(out, about{Node: parts[1], Module: parts[0]}) + } + case health.ScopeSeat: + if len(parts) >= 2 { + out = append(out, about{Node: parts[len(parts)-1], Module: strings.Join(parts[:len(parts)-1], ".")}) + } + case health.ScopeMachine: + // The machine itself (presence, network, resolver, disk) is never held; its tools are, while its tool runner is + // delivered. + if source == "S11" || kind == "tools-silent" { + out = append(out, about{Node: s.Machine, Module: runtimeModule}) + if s.Machine == "" && len(parts) > 0 { + out[len(out)-1].Node = parts[0] + } + } + return out + } + if controllerServes[source] || (s.Scope == health.ScopeCore && strings.HasPrefix(s.ID, "controller")) { + for hk := range k.wants { + if node, module := splitNodeModule(hk); module == controllerModule { + out = append(out, about{Node: node, Module: module}) + } + } + } + if s.Machine != "" && s.Scope != health.ScopeMachine { + out = append(out, about{Node: s.Machine, Module: runtimeModule}) + } + return out +} + +func splitNodeModule(hk string) (string, string) { + node, module, _ := strings.Cut(hk, ".") + return node, module +} + +// motion is the delivery of module on node, from its want and have. +func (k *Keeper) motion(node, module string) (delivery.Motion, delivery.Want) { + w, wanted := k.wants[node+"."+module] + h, has := k.haves[node+"."+module] + return delivery.MotionOf(w, wanted, h, has, "", false), w +} + +// holdFor says whether key is held now, and as what. Called with mu held. +func (k *Keeper) holdFor(key string, f *health.Finding, carried *health.Condition, open bool) (held, bool) { + if open || (f == nil && carried == nil) { + return held{}, false + } + var kind, source string + var subject health.Subject + if f != nil { + kind, source, subject = f.Kind, f.Source, f.Subject + } else { + kind, source, subject = carried.Kind, carried.Source, carried.Subject + } + if _, raisedOut := k.outOfHold[key]; raisedOut { + return held{}, false + } + for _, a := range k.aboutOf(kind, source, subject) { + m, w := k.motion(a.Node, a.Module) + if !m.Delivering || m.Ended() || k.passed[a.Node+"."+a.Module] == w.FP { + continue + } + since := k.Now().UTC() + if was, ok := k.held[key]; ok { + since = was.Since + } + return held{Key: key, Node: a.Node, Module: a.Module, Since: since, Version: w.FP}, true + } + return held{}, false +} + +// keepHeld keeps a held key in the state `held`, and starts its delivery's bound once. +func (k *Keeper) keepHeld(h held) error { + k.mu.Lock() + _, was := k.held[h.Key] + k.held[h.Key] = h + bound := k.bound(h.Module) + k.mu.Unlock() + if was { + return nil + } + if _, err := k.Store.Create(HeldState, h.Key, h); err != nil && !errors.Is(err, stdio.ErrChanged) { + return fmt.Errorf("keeping %s held: %w", h.Key, err) + } + b := boundKey(h.Node, h.Module) + err := k.Store.CreateTTL(BoundsState, b, map[string]any{"node": h.Node, "module": h.Module, "version": h.Version, + "bound-seconds": int(bound / time.Second)}, bound) + if err != nil && !errors.Is(err, stdio.ErrChanged) { + return fmt.Errorf("starting the bound of %s's delivery on %s: %w", h.Module, h.Node, err) + } + k.Logf("held %s while %s is delivered on %s (version %s), at most %s", h.Key, h.Module, h.Node, h.Version, bound) + return nil +} + +// bound is a module's delivery bound: its manifest names none yet, so the default (clause 13 (d)). The state `bounds` +// keeps each key that long (its ttl-seconds), since the runtime writes no key with a time to live of its own; a bound a +// manifest names needs that first. +func (k *Keeper) bound(string) time.Duration { + if k.Bound > 0 { + return k.Bound + } + return delivery.DefaultDeliveryBound +} + +func boundKey(node, module string) string { return node + "." + module } + +// dropHeld lets a held key go: raised, or gone with why said. +func (k *Keeper) dropHeld(key, why string) { + k.mu.Lock() + h, ok := k.held[key] + delete(k.held, key) + k.mu.Unlock() + if !ok { + return + } + if err := k.Store.Delete(HeldState, key); err != nil { + k.Logf("the held %s could not be removed from held: %v", key, err) + } + if why != "" { + k.Logf("dropped %s, held while %s was delivered on %s: %s", key, h.Module, h.Node, why) + } +} + +// Want takes one change of the assigner's wants; Have one of the nodes' haves. Each may end a delivery, which raises or +// drops what it held. +func (k *Keeper) Want(key string, value []byte, deleted bool) error { + node, module, err := delivery.ParseWantKey(key) + if err != nil { + return nil // not a want of a module on a node (a held fingerprint's key) + } + k.mu.Lock() + k.init() + if deleted { + delete(k.wants, node+"."+module) + } else if w, err := delivery.DecodeWant(value); err == nil { + k.wants[node+"."+module] = w + } else { + k.Logf("the want %s is not readable: %v", key, err) + } + k.mu.Unlock() + return k.judge(node, module, "") +} + +// Have takes one change of a node's have. +func (k *Keeper) Have(key string, value []byte, deleted bool) error { + node, module, isNode, err := delivery.ParseHaveKey(key) + if err != nil || isNode { + return nil + } + k.mu.Lock() + k.init() + if deleted { + delete(k.haves, node+"."+module) + } else if h, err := delivery.DecodeHave(value); err == nil { + k.haves[node+"."+module] = h + } else { + k.Logf("the have %s is not readable: %v", key, err) + } + k.mu.Unlock() + return k.judge(node, module, "") +} + +// BoundPassed is a delivery's bound key expiring: what it holds is raised, saying the delivery did not end. +func (k *Keeper) BoundPassed(key string) error { + node, module := splitNodeModule(key) + k.mu.Lock() + k.init() + _, w := k.motion(node, module) + holding := false + for _, h := range k.held { + holding = holding || (h.Node == node && h.Module == module) + } + if !holding { + k.mu.Unlock() + return nil + } + k.passed[key] = w.FP + k.mu.Unlock() + return k.judge(node, module, fmt.Sprintf("Its delivery of version %s on %s did not end within %s.", short(w.FP), node, + k.bound(module))) +} + +// judge raises or drops what node and module hold when their delivery ended, or with line past its bound. +func (k *Keeper) judge(node, module, line string) error { + k.mu.Lock() + m, w := k.motion(node, module) + var keys []string + for key, h := range k.held { + if h.Node == node && h.Module == module { + keys = append(keys, key) + } + } + if line == "" && !m.Ended() && m.Delivering { + k.mu.Unlock() + return nil + } + if line == "" { + line = fmt.Sprintf("It happened during the delivery of version %s of %s on %s.", short(w.FP), module, node) + if m.Failed { + line = fmt.Sprintf("It happened as the delivery of version %s of %s on %s failed.", short(w.FP), module, node) + } + delete(k.passed, node+"."+module) + } + for _, key := range keys { + k.outOfHold[key] = line + } + k.mu.Unlock() + sort.Strings(keys) + var problems []string + for _, key := range keys { + k.dropHeld(key, "") + if err := k.Settle(key); err != nil { + problems = append(problems, err.Error()) + } + k.mu.Lock() + delete(k.outOfHold, key) + k.mu.Unlock() + } + if len(problems) > 0 { + return errors.New(strings.Join(problems, "; ")) + } + return nil +} + +func short(fp string) string { + if _, rest, ok := strings.Cut(fp, ":"); ok { + fp = rest + } + if len(fp) > 12 { + return fp[:12] + } + return fp +} + +// Held is every key held now, for the read tool and compare. +func (k *Keeper) Held() []held { + k.mu.Lock() + defer k.mu.Unlock() + out := make([]held, 0, len(k.held)) + for _, h := range k.held { + out = append(out, h) + } + sort.Slice(out, func(i, j int) bool { return out[i].Key < out[j].Key }) + return out +} + +// restoreHeld takes the state `held` as read at the start. +func (k *Keeper) restoreHeld(key string, value []byte) { + var h held + if json.Unmarshal(value, &h) != nil { + return + } + k.mu.Lock() + k.init() + k.held[key] = h + k.mu.Unlock() +} diff --git a/modules/mesh-conditions/cmd/mesh-conditions/hold_test.go b/modules/mesh-conditions/cmd/mesh-conditions/hold_test.go new file mode 100644 index 00000000..0ee296a2 --- /dev/null +++ b/modules/mesh-conditions/cmd/mesh-conditions/hold_test.go @@ -0,0 +1,109 @@ +package main + +import ( + "encoding/json" + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/health" +) + +// The fingerprints and composition of mesh-controller on novox, copied from its have (mesh-delivery-view.node_haves, +// 2026-10-29): the delivery replayed moves it from the running fingerprint to the next, through the node-engine's phases +// (delivery.go Phase*). No tool reads a have's or a want's history yet, so the sequence is the node-engine's documented +// order, not a capture; the values are captured. +const ( + ctrlRunning = "src2:fcaa7d7ccb35ea2beed5b9e4b87f4c24c99452244d49f4786d4022e5c543f39c" + ctrlComposed = "sha256:e9a4de75c9ca70a32cef3137cb0e6692b0f9558b699013caaee399d88dd1bac0" + ctrlNext = "src2:0d1e2f3a4b5c6d7e8f90a1b2c3d4e5f60718293a4b5c6d7e8f9001a2b3c4d5e6" + ctrlNextComp = "sha256:1f2e3d4c5b6a79881726354453627180918a7b6c5d4e3f2a1b0c9d8e7f605142" +) + +// ADR 0564 clause 13 and Limits 12 (issue 1952 C7, C8): a delivery of mesh-controller on novox replayed want by have. +// Its own findings and the carried controller-deaf raise nothing during it and none after when they went; one still +// standing at the end is raised once with its during-the-delivery line; one past the bound with its did-not-end line; a +// failed delivery raises what stands at once; another module's and the machine's own are raised at once; compare lists a +// held key as held. +func TestADeliveryHoldsItsOwnFindingsUntilItEnds(t *testing.T) { + want := func(fp, composed string) []byte { + raw, _ := json.Marshal(delivery.Want{V: delivery.Version, FP: fp, Composed: composed}) + return raw + } + have := func(fp, composed string, phase delivery.Phase, h delivery.Health) []byte { + raw, _ := json.Marshal(delivery.Have{V: delivery.Version, FP: fp, Composed: composed, Phase: phase, Health: h}) + return raw + } + deaf, _ := json.Marshal(health.Condition{Key: "core.controller.deaf", Kind: "controller-deaf", Source: "S4", + Subject: health.Subject{Scope: health.ScopeCore, ID: "controller"}, Severity: health.Urgent, Summary: "deaf", Resolver: "self"}) + for _, c := range []struct { + name string + // end is the have the delivery ends with, or nil for a delivery past its bound. + end []byte + standing bool + said string + line string + }{ + {name: "gone by its end", end: have(ctrlNext, ctrlNextComp, delivery.PhaseRunning, delivery.HealthOK)}, + {name: "still standing at its end", end: have(ctrlNext, ctrlNextComp, delivery.PhaseRunning, delivery.HealthOK), + standing: true, said: "raised module.mesh-controller.novox.unhealthy", line: "during the delivery of version 0d1e2f3a4b5c"}, + {name: "past its bound", standing: true, said: "raised module.mesh-controller.novox.unhealthy", line: "did not end within"}, + {name: "failed", end: have(ctrlNext, ctrlNextComp, delivery.PhaseFailed, delivery.HealthFailed), standing: true, + said: "raised module.mesh-controller.novox.unhealthy", line: "as the delivery of version 0d1e2f3a4b5c"}, + } { + r := newRig() + _ = r.k.Want("want.novox.mesh-controller", want(ctrlRunning, ctrlComposed), false) + _ = r.k.Have("novox.mesh-controller", have(ctrlRunning, ctrlComposed, delivery.PhaseRunning, delivery.HealthOK), false) + // The delivery starts: the assigner wants the next fingerprint; the node applies it. + _ = r.k.Want("want.novox.mesh-controller", want(ctrlNext, ctrlNextComp), false) + _ = r.k.Have("novox.mesh-controller", have(ctrlNext, ctrlNextComp, delivery.PhaseApplying, delivery.HealthUnknown), false) + own, ownRaw := finding(t, watchman, health.ScopeModule, "mesh-controller.novox", "unhealthy", "health", health.Warning, r.now) + other, otherRaw := finding(t, watchman, health.ScopeModule, "postgres.novox", "unhealthy", "health", health.Warning, r.now) + machine, machineRaw := finding(t, watchman, health.ScopeMachine, "novox", "silent", "S1", health.Urgent, r.now) + for _, f := range []struct { + key string + raw []byte + }{{own, ownRaw}, {other, otherRaw}, {machine, machineRaw}} { + if err := r.k.Finding(watchman, f.key, f.raw, false); err != nil { + t.Fatal(err) + } + } + if err := r.k.Controller("core.controller.deaf", deaf, false); err != nil { + t.Fatal(err) + } + if got := r.changes(); got != "raised "+other+", raised "+machine { + t.Errorf("%s: during the delivery said %q: another module's and the machine's at once, nothing of its own", c.name, got) + } + // C8: compare lists the held keys as held, never as a difference. + cmp, _ := r.k.Compare() + held := 0 + for _, d := range cmp.Apart { + if strings.HasPrefix(d.Reason, "held while mesh-controller is delivered on novox") { + held++ + } + } + if held != 2 { + t.Errorf("%s: compare lists %d held, %+v", c.name, held, cmp) + } + // The carried condition about the controller goes within the delivery; its own finding goes unless it stands. + _ = r.k.Controller("core.controller.deaf", nil, true) + if !c.standing { + _ = r.k.Finding(watchman, own, nil, true) + } + if c.end != nil { + _ = r.k.Have("novox.mesh-controller", c.end, false) + } else { + r.now = r.now.Add(delivery.DefaultDeliveryBound) + _ = r.k.BoundPassed("novox.mesh-controller") + } + if got := r.changes(); got != c.said { + t.Errorf("%s: at its end said %q, want %q", c.name, got, c.said) + } + if co, open := r.open(own); c.standing != open || (open && !strings.Contains(co.Explanation, c.line)) { + t.Errorf("%s: its own condition is %v: %q", c.name, open, co.Explanation) + } + if _, open := r.open("core.controller.deaf"); open || len(r.k.Held()) != 0 { + t.Errorf("%s: the carried deaf condition was raised, or something is still held: %+v", c.name, r.k.Held()) + } + } +} diff --git a/modules/mesh-conditions/cmd/mesh-conditions/homes.go b/modules/mesh-conditions/cmd/mesh-conditions/homes.go new file mode 100644 index 00000000..1f0d378c --- /dev/null +++ b/modules/mesh-conditions/cmd/mesh-conditions/homes.go @@ -0,0 +1,73 @@ +package main + +// **Where each condition kind lives after the take-over** (novox/hq ADR 0564 clauses 4, 6, 7 and 9, the Context's probe +// table; issue 1952 C3, C5): data, so a kind's home is a row and never code. `compare` reads it to tell a difference from +// a kind that ends with the controller or that the controller never raises, and the keeper reads it to carry, until the +// lease write, the controller's conditions of the kinds that end with it. +// +// The rows below are what ADR 0564 decides, by the controller's source of each kind as the live controller says it +// (mesh-controller.conditions, read 2026-10-29). **About twenty-five kinds the controller raises have no home in the +// record** (L4.md, "Missing decision"): they have no row here on purpose, and `compare` names each as "no home decided" +// until the lead's decision adds its row. + +// Home is where one condition kind lives. +type Home struct { + // Source is the controller's row or probe that raises it today, for whoever reads the table. + Source string + // Controller says the controller raises this kind today: a kind it never raises is the keeper's alone and is no + // difference in compare. + Controller bool + // Observer is the module whose findings raise it after the take-over; empty for a kind that ends. + Observer string + // Carried says it ends with the controller's lease and is carried from the controller's bucket until then + // (clause 9: S4, S7, S12, S13, S14, S25, D5, D9, H-controller). + Carried bool + // Replaced says the kind itself ends and another observes its fault (D1, D10, the stuck probe): listed apart in + // compare and not carried. + Replaced bool +} + +// The kinds' homes, by kind. +var Homes = map[string]Home{ + // The watchman's watchdogs (clause 4). + "silent": {Source: "S1", Controller: true, Observer: watchman}, + "max-deliveries": {Source: "S9", Controller: true, Observer: watchman}, + "slow-consumer": {Source: "S9", Controller: true, Observer: watchman}, + "tools-silent": {Source: "S11", Controller: true, Observer: watchman}, + "apply-failing": {Source: "S21", Controller: true, Observer: watchman}, + "module-unhealthy": {Source: "health", Controller: true, Observer: watchman}, + "stuck": {Source: "DM", Controller: true, Observer: watchman}, + // The self-check's probes (clause 6, the Context's table). + "holder-silent": {Source: "D3", Controller: true, Observer: selfCheck}, + "archives-missing": {Source: "D4", Controller: true, Observer: selfCheck}, + "consumer-wrong": {Source: "D6", Controller: true, Observer: selfCheck}, + "data-unmeasured": {Source: "D13", Controller: true, Observer: selfCheck}, + "agent-can-become-root": {Source: "DA", Controller: true, Observer: selfCheck}, + "root-not-free": {Source: "D-root", Controller: true, Observer: selfCheck}, + "probe-failed": {Source: "doctor", Controller: true, Observer: selfCheck}, + // The healers' own (clause 7, Limits 2). + "healer-wanted": {Source: "S15", Controller: true, Observer: healers}, + "healers-braked": {Source: "healers", Controller: true, Observer: healers}, + // The kinds that end with the controller and are carried until its lease write (clauses 4, 9). + "controller-deaf": {Source: "S4", Controller: true, Carried: true}, + "lease-lost": {Source: "S12", Controller: true, Carried: true}, + // The kinds that end, their faults observed otherwise (the Context's table). + "core-behind": {Source: "D10", Controller: true, Replaced: true}, +} + +// The observers, as the table names them. +const ( + watchman = "mesh-delivery-watchman" + selfCheck = "mesh-self-check" + healers = "mesh-healers" +) + +// homeOf is the home of a condition's kind, and whether the table has one. +func (k *Keeper) homeOf(kind string) (Home, bool) { + homes := k.Homes + if homes == nil { + homes = Homes + } + h, ok := homes[kind] + return h, ok +} diff --git a/modules/mesh-conditions/cmd/mesh-conditions/keeper.go b/modules/mesh-conditions/cmd/mesh-conditions/keeper.go new file mode 100644 index 00000000..70a82f3d --- /dev/null +++ b/modules/mesh-conditions/cmd/mesh-conditions/keeper.go @@ -0,0 +1,660 @@ +package main + +// The condition keeper (novox/hq ADR 0564 clauses 2, 3, 7, 8, 9 and 13; issue 1952 C2, C3, C7): the one writer of the +// mesh's open conditions. It watches every observer's `findings`, the healers' `attempts`, the controller's lease and, +// until the lease write, the controller's `conditions`, and the assigner's wants and the nodes' haves for the hold. From +// all of them it derives, per key, the condition that should be open, and writes the difference by compare-and-set, +// saying each transition as the controller says it (go/health) and keeping it in `history`. Nothing clears a condition +// but its finding going (ADR 0231 §14). + +import ( + "encoding/json" + "errors" + "fmt" + "sort" + "strings" + "sync" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/health" +) + +// The module, its seat, and the states it keeps. +const ( + Module = health.KeeperModule + Seat = "mesh-conditions" + + ConditionsState = health.ConditionsState + TakeoverState = health.TakeoverState + HistoryState = "history" + HeldState = "held" + BoundsState = "bounds" + // AskedState keeps the keeper's asks of the operator (go/asker). + AskedState = "asked" +) + +// HistoryKeptFor is how long a transition is kept (clause 3). +const HistoryKeptFor = 90 * 24 * time.Hour + +// Observers are the modules whose `findings` the keeper reads (clause 2; ADR 0565 §8 for the assigner's). +var Observers = []string{health.WatchmanModule, health.SelfCheckModule, health.HealersModule, delivery.AssignerModule} + +// The other states it reads, as `.`. +const ( + AttemptsRead = health.HealersModule + "." + health.AttemptsState + ControllerConditions = "mesh-controller.conditions" + ControllerLease = "mesh-controller.lease" + ControllerLeaseKey = "holder" + WantsRead = delivery.AssignerModule + ".wants" + HavesRead = "node-engine.have" + ConditionWrites = 8 +) + +// Store is the bus as the keeper uses it: its own states, written by compare-and-set, and a key with its own time to +// live for a hold's bound. +type Store interface { + Get(state, key string) (value []byte, revision uint64, found bool, err error) + Create(state, key string, value any) (uint64, error) + Update(state, key string, value any, revision uint64) (uint64, error) + Delete(state, key string) error + Keys(state string) ([]string, error) + CreateTTL(state, key string, value any, ttl time.Duration) error +} + +// held is a finding or a carried condition kept while its module is delivered (clause 13 (b)). +type held struct { + Key string `json:"key"` + Node string `json:"node"` + Module string `json:"module"` + Since time.Time `json:"since"` + Version string `json:"version"` + Wanted json.RawMessage `json:"-"` +} + +// clearedMemo is what a cleared condition leaves for ReopenWithin: the same fault again is a reopening. +type clearedMemo struct { + at time.Time + count int + began time.Time + gaps []health.Gap + silenced *health.Silence + tried []health.Tried +} + +// Keeper is the condition keeper's state and its writes. +type Keeper struct { + Store Store + // Say publishes one event on the seat; History keeps one transition. + Say func(e health.Event) error + Now func() time.Time + Logf func(format string, a ...any) + // Homes says, by kind, where each controller kind lives after the take-over (homes.go); Homes when nil. + Homes map[string]Home + // Bound is a delivery's hold bound; delivery.DefaultDeliveryBound when zero (hold.go). + Bound time.Duration + + mu sync.Mutex + findings map[string]map[string]health.Finding + blind map[string]string + attempts map[string]health.Attempt + ctrl map[string]health.Condition + leaseOut bool + takeover health.Takeover + wants map[string]delivery.Want + haves map[string]delivery.Have + stuck map[string]string + held map[string]held + // passed are the deliveries whose bound passed, by node and module, at the fingerprint wanted then; outOfHold the + // line a key raised out of a hold says (hold.go). + passed map[string]string + outOfHold map[string]string + cleared map[string]clearedMemo + seq uint64 +} + +func (k *Keeper) init() { + if k.findings == nil { + k.findings, k.blind, k.attempts, k.ctrl = map[string]map[string]health.Finding{}, map[string]string{}, + map[string]health.Attempt{}, map[string]health.Condition{} + k.wants, k.haves, k.stuck, k.held, k.cleared = map[string]delivery.Want{}, map[string]delivery.Have{}, + map[string]string{}, map[string]held{}, map[string]clearedMemo{} + k.takeover = health.Takeover{V: health.TakeoverVersion, Follow: health.FollowController} + k.passed, k.outOfHold = map[string]string{}, map[string]string{} + } + if k.Now == nil { + k.Now = time.Now + } + if k.Logf == nil { + k.Logf = func(string, ...any) {} + } +} + +// Finding takes one change of an observer's finding: a key put raises or changes its condition, a key deleted clears +// it (clause 2). A value that is not a finding is the observer's fault, said and left as it was. +func (k *Keeper) Finding(observer, key string, value []byte, deleted bool) error { + k.mu.Lock() + k.init() + if k.findings[observer] == nil { + k.findings[observer] = map[string]health.Finding{} + } + if deleted { + delete(k.findings[observer], key) + } else { + f, err := health.DecodeFinding(value) + if err != nil { + k.mu.Unlock() + k.Logf("%s's finding %s is not one the keeper can raise, so it is left as it stood: %v", observer, key, err) + return nil + } + if f.Key != key { + k.mu.Unlock() + k.Logf("%s's finding under %s names the key %s, so it is left as it stood", observer, key, f.Key) + return nil + } + k.findings[observer][key] = f + } + k.mu.Unlock() + return k.Settle(key) +} + +// Blind says an observer's findings cannot be read: its keys stay as they were, and nothing is cleared for not seeing +// (ADR 0227 §4). Empty why says it is read again. +func (k *Keeper) Blind(observer, why string) { + k.mu.Lock() + defer k.mu.Unlock() + k.init() + if why == "" { + delete(k.blind, observer) + } else { + k.blind[observer] = why + } +} + +// Attempt takes one change of the healers' `attempts`: the condition it names gets its tried, resolver and escalation +// (clause 7). +func (k *Keeper) Attempt(key string, value []byte, deleted bool) error { + k.mu.Lock() + k.init() + var condition string + if old, ok := k.attempts[key]; ok { + condition = old.Condition + } + if deleted { + delete(k.attempts, key) + } else { + var a health.Attempt + if err := json.Unmarshal(value, &a); err != nil || a.Condition == "" { + k.mu.Unlock() + k.Logf("the healers' attempt %s is not readable, so it is left out: %v", key, err) + return nil + } + k.attempts[key], condition = a, a.Condition + } + k.mu.Unlock() + if condition == "" { + return nil + } + return k.Settle(condition) +} + +// Controller takes one change of the controller's `conditions` bucket (clauses 8, 9): read until the lease write. +func (k *Keeper) Controller(key string, value []byte, deleted bool) error { + k.mu.Lock() + k.init() + if deleted { + delete(k.ctrl, key) + } else { + var c health.Condition + if err := json.Unmarshal(value, &c); err != nil { + k.mu.Unlock() + k.Logf("the controller's condition %s is not readable, so nothing is carried from it: %v", key, err) + return nil + } + k.ctrl[key] = c + } + k.mu.Unlock() + return k.Settle(key) +} + +// Lease takes the controller's lease key as read (clause 8, Limits 8): once it is handed to a holder that is no +// controller, nothing more is carried from the controller's bucket, and the kinds that ended with it clear. +func (k *Keeper) Lease(away bool) error { + k.mu.Lock() + k.init() + was := k.leaseOut + k.leaseOut = away + keys := make([]string, 0, len(k.ctrl)) + for key := range k.ctrl { + keys = append(keys, key) + } + k.mu.Unlock() + if was == away { + return nil + } + return k.settleAll(keys) +} + +// LeaseAway says whether the controller's lease is handed away. +func (k *Keeper) LeaseAway() bool { + k.mu.Lock() + defer k.mu.Unlock() + return k.leaseOut +} + +// Takeover takes the keeper's own take-over key as read. +func (k *Keeper) Takeover(t health.Takeover) { + k.mu.Lock() + defer k.mu.Unlock() + k.init() + k.takeover = t +} + +// Settle derives the condition key should be and writes the difference (clause 2), retried while another write moves +// it under this one. +func (k *Keeper) Settle(key string) error { + var err error + for i := 0; i < ConditionWrites; i++ { + if err = k.settleOnce(key); !errors.Is(err, stdio.ErrChanged) { + return err + } + } + return fmt.Errorf("the condition %s kept moving under this write, %d tries: %w", key, ConditionWrites, err) +} + +func (k *Keeper) settleAll(keys []string) error { + sort.Strings(keys) + var problems []string + for _, key := range keys { + if err := k.Settle(key); err != nil { + problems = append(problems, err.Error()) + } + } + if len(problems) > 0 { + return errors.New(strings.Join(problems, "; ")) + } + return nil +} + +// wanted is what the inputs say key should be: a finding of an observer, or a condition carried from the controller's +// bucket of a kind that ends with the controller, until the lease write (clause 9). Nil when nothing says it. +func (k *Keeper) wanted(key string) (*health.Finding, *health.Condition) { + observers := make([]string, 0, len(k.findings)) + for o := range k.findings { + observers = append(observers, o) + } + sort.Strings(observers) + for _, o := range observers { + if f, ok := k.findings[o][key]; ok { + return &f, nil + } + } + if c, ok := k.ctrl[key]; ok && !k.leaseOut && k.endsWithController(c) { + return nil, &c + } + return nil, nil +} + +func (k *Keeper) endsWithController(c health.Condition) bool { + h, ok := k.homeOf(c.Kind) + return ok && h.Carried +} + +// settleOnce is one try of Settle. +func (k *Keeper) settleOnce(key string) error { + now := k.Now().UTC() + raw, rev, open, err := k.Store.Get(ConditionsState, key) + if err != nil { + return fmt.Errorf("reading the condition %s: %w", key, err) + } + var was health.Condition + if open { + if err := json.Unmarshal(raw, &was); err != nil { + return fmt.Errorf("the condition %s on the bus cannot be read: %w", key, err) + } + } + k.mu.Lock() + k.init() + f, carried := k.wanted(key) + hold, holding := k.holdFor(key, f, carried, open) + k.mu.Unlock() + + if f == nil && carried == nil { + k.dropHeld(key, "its finding went while its module was delivered") + if !open { + return nil + } + return k.clear(key, was, rev, now, "its finding went") + } + if holding { + return k.keepHeld(hold) + } + k.dropHeld(key, "") + next := k.derive(key, was, open, f, carried, now) + events := transitions(was, next, open) + if len(events) == 0 && open && sameBody(was, next) { + return nil + } + if open { + _, err = k.Store.Update(ConditionsState, key, next, rev) + } else { + _, err = k.Store.Create(ConditionsState, key, next) + } + if err != nil { + return err + } + k.mu.Lock() + delete(k.cleared, key) + k.mu.Unlock() + for _, e := range events { + e.Condition, e.At, e.Event = next, now, health.EventFor(e.Change) + e.Show = k.show(next, carried != nil) + k.tell(e) + } + return nil +} + +func (k *Keeper) show(c health.Condition, carried bool) string { + if carried { + return health.ShowController(c.Key) + } + return health.ShowKeeper(c.Key) +} + +// derive is the condition key should be now, from what was open and what the inputs say. +func (k *Keeper) derive(key string, was health.Condition, open bool, f *health.Finding, carried *health.Condition, now time.Time) health.Condition { + var c health.Condition + switch { + case carried != nil: + // Raised and cleared as the controller's bucket says, with its show and actions (Limits 9). + c = *carried + c.Epoch = 0 + c.LooksUnread = "" + case open: + c = was + k.fromFinding(&c, *f, now) + default: + c = health.Condition{Key: key, Raised: now, Count: 1, Resolver: health.ResolverSelf} + k.fromFinding(&c, *f, now) + k.mu.Lock() + if before, ok := k.cleared[key]; ok && now.Sub(before.at) <= health.ReopenWithin { + c.Count = before.count + 1 + c.First = before.began + c.Gaps = append(append([]health.Gap(nil), before.gaps...), health.Gap{Cleared: before.at, Reopened: now}) + if over := len(c.Gaps) - health.KeptGaps; over > 0 { + c.First, c.Gaps = c.Gaps[over-1].Reopened, c.Gaps[over:] + } + if before.silenced != nil && now.Before(before.silenced.Until) { + c.Silenced = before.silenced + } + c.Tried = before.tried + } + k.mu.Unlock() + } + if carried == nil { + k.overlay(&c, now) + } + if !open { + k.mu.Lock() + if line := k.outOfHold[key]; line != "" { + c.Explanation = strings.TrimSpace(c.Explanation + " " + line) + } + k.mu.Unlock() + } + if c.Silenced != nil && !now.Before(c.Silenced.Until) { + c.Silenced = nil + } + return c +} + +// fromFinding is what the condition says of its finding: its words, subject, severity and source, and its evidence +// newest first (clause 2). A healer's escalation is not undone by the finding saying a warning again. +func (k *Keeper) fromFinding(c *health.Condition, f health.Finding, now time.Time) { + escalated := c.Escalated() + c.Kind, c.Subject, c.Summary, c.Source = f.Kind, f.Subject, f.Summary, f.Source + c.Headline, c.Explanation, c.Resolved, c.Needs, c.Actions = f.Headline, f.Explanation, f.Resolved, f.Needs, f.Actions + if !(escalated && f.Severity != health.Urgent) { + c.Severity = f.Severity + } + if !escalated { + c.Resolver = f.Resolver + if c.Resolver == "" { + c.Resolver = health.ResolverSelf + } + } + if len(f.Evidence) > 0 { + c.Evidence = append([]health.Evidence(nil), f.Evidence...) + c.LastObserved = f.Evidence[0].At + } else if c.Evidence == nil { + c.Evidence = []health.Evidence{{At: f.At, Said: f.Summary}} + c.LastObserved = f.At + } + c.Observations = len(c.Evidence) + if c.Actions == nil { + c.Actions = []health.Action{} + } + _ = now +} + +// overlay writes the healers' attempts into the condition (clause 7) and, until the lease write, what the controller's +// bucket holds for the same key (clause 8): its tried, its resolver and escalation, and its silence where the keeper has +// none of its own. +func (k *Keeper) overlay(c *health.Condition, now time.Time) { + k.mu.Lock() + var mine []health.Attempt + for _, a := range k.attempts { + if a.Condition == c.Key { + mine = append(mine, a) + } + } + ctrl, carry := k.ctrl[c.Key] + carry = carry && !k.leaseOut + k.mu.Unlock() + sort.Slice(mine, func(i, j int) bool { return mine[i].Begun.Before(mine[j].Begun) }) + if carry && len(ctrl.Tried) > 0 { + c.Tried = append([]health.Tried(nil), ctrl.Tried...) + if ctrl.Escalated() { + c.Resolver, c.Severity = health.ResolverOperator, health.Urgent + } else if ctrl.Resolver != "" && ctrl.Resolver != health.ResolverSelf { + c.Resolver = ctrl.Resolver + } + } + for _, a := range mine { + t := a.Tried() + if !containsTried(c.Tried, t) { + c.Tried = append(c.Tried, t) + } + switch a.Outcome { + case health.OutcomeEscalated: + c.Resolver, c.Severity = health.ResolverOperator, health.Urgent + default: + if c.Resolver != health.ResolverOperator { + c.Resolver = health.ResolverHealer(a.Healer) + } + } + } + if len(c.Tried) > health.KeptTried { + c.Tried = c.Tried[len(c.Tried)-health.KeptTried:] + } + if carry && c.Silenced == nil && ctrl.Silenced != nil && now.Before(ctrl.Silenced.Until) { + c.Silenced = ctrl.Silenced + } +} + +func containsTried(list []health.Tried, t health.Tried) bool { + for _, x := range list { + if x.At.Equal(t.At) && x.By == t.By && x.What == t.What { + return true + } + } + return false +} + +// transitions are the events between what was open and what is now, in the controller's order: raised or reopened; +// else severity, resolver, silence and words changes. An observation that changes none of these says nothing. +func transitions(was, now health.Condition, open bool) []health.Event { + if !open { + change := health.ChangeRaised + if now.Count > 1 { + change = health.ChangeReopened + } + return []health.Event{{Change: change}} + } + var out []health.Event + if was.Severity != now.Severity { + out = append(out, health.Event{Change: health.ChangeSeverity, Was: string(was.Severity)}) + } + if was.Resolver != now.Resolver { + out = append(out, health.Event{Change: health.ChangeResolver, Was: was.Resolver}) + } + switch { + case was.Silenced == nil && now.Silenced != nil: + out = append(out, health.Event{Change: health.ChangeSilenced, Why: now.Silenced.Why}) + case was.Silenced != nil && now.Silenced == nil: + out = append(out, health.Event{Change: health.ChangeUnsilenced}) + } + if was.Needs != now.Needs || !sameActions(was.Actions, now.Actions) { + out = append(out, health.Event{Change: health.ChangeWords, Was: was.Needs}) + } + return out +} + +func sameActions(a, b []health.Action) bool { + x, _ := json.Marshal(a) + y, _ := json.Marshal(b) + return string(x) == string(y) || (len(a) == 0 && len(b) == 0) +} + +func sameBody(a, b health.Condition) bool { + x, _ := json.Marshal(a) + y, _ := json.Marshal(b) + return string(x) == string(y) +} + +// clear removes an open condition whose finding went, and says it (clause 2). +func (k *Keeper) clear(key string, was health.Condition, rev uint64, now time.Time, why string) error { + // A delete at the revision read: a write that moved it meanwhile is read again. + if _, err := k.Store.Update(ConditionsState, key, json.RawMessage("null"), rev); err != nil { + return err + } + if err := k.Store.Delete(ConditionsState, key); err != nil { + return err + } + k.mu.Lock() + k.cleared[key] = clearedMemo{at: now, count: was.Count, began: was.Began(), gaps: was.Gaps, silenced: was.Silenced, tried: was.Tried} + k.mu.Unlock() + cleared := now + k.tell(health.Event{Condition: was, Event: health.EventCleared, At: now, Change: health.ChangeCleared, Why: why, + Cleared: &cleared, Show: k.show(was, false)}) + return nil +} + +// tell keeps the transition in history and says it on the seat. A transition not said is said in the log: the condition +// and its history still hold it. +func (k *Keeper) tell(e health.Event) { + k.mu.Lock() + k.seq++ + hk := fmt.Sprintf("%020d-%06d", e.At.UnixNano(), k.seq) + k.mu.Unlock() + if _, err := k.Store.Create(HistoryState, hk, e); err != nil { + k.Logf("the %s of %s could not be kept in history: %v", e.Change, e.Key, err) + } + if k.Say == nil { + return + } + if err := k.Say(e); err != nil { + k.Logf("the %s of %s could NOT be said on the seat: %v", e.Change, e.Key, err) + } +} + +// Open is every open condition, ordered as `conditions` says them. +func (k *Keeper) Open() ([]health.Condition, error) { + keys, err := k.Store.Keys(ConditionsState) + if err != nil { + return nil, err + } + var out []health.Condition + for _, key := range keys { + raw, _, found, err := k.Store.Get(ConditionsState, key) + if err != nil { + return nil, err + } + if !found || string(raw) == "null" { + continue + } + var c health.Condition + if err := json.Unmarshal(raw, &c); err != nil { + return nil, fmt.Errorf("the condition %s on the bus cannot be read: %w", key, err) + } + out = append(out, c) + } + health.Order(out) + return out, nil +} + +// Resync settles every key any input names or any open condition holds (clause 12): what differs is raised or cleared. +func (k *Keeper) Resync() (int, error) { + open, err := k.Open() + if err != nil { + return 0, err + } + seen := map[string]bool{} + for _, c := range open { + seen[c.Key] = true + } + k.mu.Lock() + k.init() + for o, fs := range k.findings { + if k.blind[o] != "" { + continue + } + for key := range fs { + seen[key] = true + } + } + for key := range k.ctrl { + seen[key] = true + } + k.mu.Unlock() + keys := make([]string, 0, len(seen)) + for key := range seen { + keys = append(keys, key) + } + return len(keys), k.settleAll(keys) +} + +// Silence stops a condition's messages for a while, a hand act with why (clause 3, Limits 1): kept as its transition in +// history, naming who and why. +func (k *Keeper) Silence(key string, d time.Duration, by, why string) (health.Condition, error) { + if strings.TrimSpace(why) == "" { + return health.Condition{}, errors.New("a silence says why") + } + if d <= 0 || d > health.MaxSilence { + return health.Condition{}, fmt.Errorf("a silence lasts more than nothing and at most %s, not %s", health.MaxSilence, d) + } + for i := 0; i < ConditionWrites; i++ { + raw, rev, found, err := k.Store.Get(ConditionsState, key) + if err != nil { + return health.Condition{}, err + } + if !found || string(raw) == "null" { + return health.Condition{}, fmt.Errorf("no condition %s is open", key) + } + var c health.Condition + if err := json.Unmarshal(raw, &c); err != nil { + return health.Condition{}, err + } + now := k.Now().UTC() + c.Silenced = &health.Silence{Until: now.Add(d), By: by, Why: why, Since: now} + if _, err := k.Store.Update(ConditionsState, key, c, rev); errors.Is(err, stdio.ErrChanged) { + continue + } else if err != nil { + return health.Condition{}, err + } + k.tell(health.Event{Condition: c, Event: health.EventChanged, At: now, Change: health.ChangeSilenced, + Why: why + " (" + by + ")", Show: health.ShowKeeper(key)}) + return c, nil + } + return health.Condition{}, fmt.Errorf("the condition %s kept moving under the silence", key) +} diff --git a/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go b/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go new file mode 100644 index 00000000..6f8dd1d8 --- /dev/null +++ b/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go @@ -0,0 +1,306 @@ +package main + +import ( + "encoding/json" + "fmt" + "sort" + "strings" + "sync" + "testing" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/health" +) + +// fakeStore is the bus's states in memory, with revisions, as the runtime answers them. +type fakeStore struct { + mu sync.Mutex + vals map[string][]byte + revs map[string]uint64 + next uint64 +} + +func newFakeStore() *fakeStore { + return &fakeStore{vals: map[string][]byte{}, revs: map[string]uint64{}} +} + +func (f *fakeStore) Get(state, key string) ([]byte, uint64, bool, error) { + f.mu.Lock() + defer f.mu.Unlock() + v, ok := f.vals[state+"/"+key] + return v, f.revs[state+"/"+key], ok, nil +} + +func (f *fakeStore) put(state, key string, value any) uint64 { + raw, _ := json.Marshal(value) + f.next++ + f.vals[state+"/"+key], f.revs[state+"/"+key] = raw, f.next + return f.next +} + +func (f *fakeStore) Create(state, key string, value any) (uint64, error) { + f.mu.Lock() + defer f.mu.Unlock() + if _, ok := f.vals[state+"/"+key]; ok { + return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged) + } + return f.put(state, key, value), nil +} + +func (f *fakeStore) Update(state, key string, value any, rev uint64) (uint64, error) { + f.mu.Lock() + defer f.mu.Unlock() + if f.revs[state+"/"+key] != rev { + return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged) + } + return f.put(state, key, value), nil +} + +func (f *fakeStore) Delete(state, key string) error { + f.mu.Lock() + defer f.mu.Unlock() + delete(f.vals, state+"/"+key) + delete(f.revs, state+"/"+key) + return nil +} + +func (f *fakeStore) Keys(state string) ([]string, error) { + f.mu.Lock() + defer f.mu.Unlock() + var out []string + for k := range f.vals { + if s, key, _ := strings.Cut(k, "/"); s == state { + out = append(out, key) + } + } + sort.Strings(out) + return out, nil +} + +func (f *fakeStore) CreateTTL(state, key string, value any, _ time.Duration) error { + _, err := f.Create(state, key, value) + return err +} + +// rig is a keeper over a fake store, its clock and what it said. +type rig struct { + k *Keeper + st *fakeStore + now time.Time + said []health.Event +} + +func newRig() *rig { + r := &rig{st: newFakeStore(), now: time.Date(2026, 10, 29, 12, 0, 0, 0, time.UTC)} + r.k = &Keeper{Store: r.st, Now: func() time.Time { return r.now }, Say: func(e health.Event) error { + r.said = append(r.said, e) + return nil + }} + return r +} + +func (r *rig) changes() string { + var out []string + for _, e := range r.said { + out = append(out, e.Change+" "+e.Key) + } + r.said = nil + return strings.Join(out, ", ") +} + +func (r *rig) open(key string) (health.Condition, bool) { + raw, _, ok, _ := r.st.Get(ConditionsState, key) + var c health.Condition + if ok { + _ = json.Unmarshal(raw, &c) + } + return c, ok +} + +// finding is an observer's finding as mesh-sdk's layout writes it (health.EncodeFinding). +func finding(t *testing.T, observer, scope, id, kind, source string, sev health.Severity, at time.Time) (string, []byte) { + t.Helper() + f := health.Finding{Key: health.Key(scope, id, kind), Kind: kind, Subject: health.Subject{Scope: scope, ID: id}, + Severity: sev, Summary: kind + " of " + id, Headline: kind, Explanation: "It is " + kind + ".", Source: source, + Observer: observer, At: at, Evidence: []health.Evidence{{At: at, Said: kind}}} + raw, err := health.EncodeFinding(f) + if err != nil { + t.Fatal(err) + } + return f.Key, raw +} + +// ADR 0564 clauses 2, 3, 7, 8 (issue 1952 C2, C3): a finding raises, changes and clears its condition; a blind observer +// clears nothing; the healers' attempts and the controller's escalation reach the condition; a reopening within ten +// minutes is the same fault; a silence is a transition in history. +func TestTheKeeperRaisesChangesAndClearsFromFindings(t *testing.T) { + r := newRig() + key, raw := finding(t, watchman, health.ScopeMachine, "g14", "silent", "S1", health.Warning, r.now) + steps := []struct { + name string + do func() error + said string + check func(c health.Condition, open bool) string + }{ + {name: "a finding written raises", do: func() error { return r.k.Finding(watchman, key, raw, false) }, + said: "raised machine.g14.silent", check: func(c health.Condition, open bool) string { + if !open || c.Severity != health.Warning || c.Resolver != health.ResolverSelf || c.Epoch != 0 || c.Count != 1 { + return fmt.Sprintf("%+v", c) + } + return "" + }}, + {name: "the same finding again says nothing", do: func() error { return r.k.Finding(watchman, key, raw, false) }}, + {name: "an observer gone blind clears nothing", do: func() error { r.k.Blind(watchman, "refused"); _, err := r.k.Resync(); return err }, + check: func(c health.Condition, open bool) string { + if !open { + return "cleared by blindness" + } + return "" + }}, + {name: "a healer's attempt reaches tried and the resolver", do: func() error { + a, _ := json.Marshal(health.Attempt{Healer: "H3", Condition: key, Act: "asserted", Outcome: health.OutcomeActed, Begun: r.now}) + return r.k.Attempt("h3-1", a, false) + }, said: "resolver machine.g14.silent", check: func(c health.Condition, _ bool) string { + if c.Resolver != "healer:H3" || len(c.Tried) != 1 || c.Tried[0].By != "healer H3" { + return fmt.Sprintf("%+v", c) + } + return "" + }}, + {name: "an escalation makes it urgent, the operator's", do: func() error { + a, _ := json.Marshal(health.Attempt{Healer: "H3", Condition: key, Act: "handed over", Outcome: health.OutcomeEscalated, + Begun: r.now.Add(time.Minute)}) + return r.k.Attempt("h3-2", a, false) + }, said: "severity machine.g14.silent, resolver machine.g14.silent", check: func(c health.Condition, _ bool) string { + if c.Severity != health.Urgent || c.Resolver != health.ResolverOperator || len(c.Tried) != 2 { + return fmt.Sprintf("%+v", c) + } + return "" + }}, + {name: "a silence is a transition with who and why", do: func() error { + _, err := r.k.Silence(key, time.Hour, "jochen", "known, a cable") + return err + }, said: "silenced machine.g14.silent"}, + {name: "the finding deleted clears", do: func() error { return r.k.Finding(watchman, key, nil, true) }, + said: "cleared machine.g14.silent", check: func(_ health.Condition, open bool) string { + if open { + return "still open" + } + return "" + }}, + {name: "raised again within ten minutes is a reopening", do: func() error { + r.now = r.now.Add(5 * time.Minute) + return r.k.Finding(watchman, key, raw, false) + }, said: "reopened machine.g14.silent", check: func(c health.Condition, _ bool) string { + if c.Count != 2 || len(c.Gaps) != 1 { + return fmt.Sprintf("%+v", c) + } + return "" + }}, + // Until the lease write the controller's escalation for the same key reaches the condition (clause 8). + {name: "the controller's escalation is carried", do: func() error { + ctrl, _ := json.Marshal(health.Condition{Key: key, Kind: "silent", Severity: health.Urgent, Resolver: health.ResolverOperator, + Tried: []health.Tried{{At: r.now, What: "x", Outcome: "escalated", By: "healer H1"}}}) + _ = r.k.Attempt("h3-1", nil, true) + _ = r.k.Attempt("h3-2", nil, true) + return r.k.Controller(key, ctrl, false) + }, check: func(c health.Condition, _ bool) string { + if c.Resolver != health.ResolverOperator || c.Severity != health.Urgent { + return fmt.Sprintf("%+v", c) + } + return "" + }}, + } + for _, s := range steps { + if err := s.do(); err != nil { + t.Fatalf("%s: %v", s.name, err) + } + if got := r.changes(); s.said != "" && got != s.said || s.said == "" && got != "" && s.name != "the controller's escalation is carried" { + t.Errorf("%s: said %q, want %q", s.name, got, s.said) + } + if s.check != nil { + c, open := r.open(key) + if why := s.check(c, open); why != "" { + t.Errorf("%s: %s", s.name, why) + } + } + } + // Every transition is kept in history, and each event carries the keeper's show. + keys, _ := r.st.Keys(HistoryState) + if len(keys) < 7 { + t.Errorf("history keeps %d transitions", len(keys)) + } + raw0, _, _, _ := r.st.Get(HistoryState, keys[0]) + if !strings.Contains(string(raw0), `"show":"mesh-conditions.conditions key=machine.g14.silent"`) { + t.Errorf("the first transition is %s", raw0) + } + // A carried kind (it ends with the controller) is raised and cleared as the controller's bucket says, until the + // lease is handed away, with the controller's show (clause 9, Limits 9). + deaf, _ := json.Marshal(health.Condition{Key: "core.controller.deaf", Kind: "controller-deaf", Source: "S4", + Subject: health.Subject{Scope: health.ScopeCore, ID: "controller"}, Severity: health.Urgent, Summary: "deaf", Resolver: "self", Epoch: 9}) + if err := r.k.Controller("core.controller.deaf", deaf, false); err != nil { + t.Fatal(err) + } + if c, open := r.open("core.controller.deaf"); !open || c.Epoch != 0 || r.said[0].Show != "mesh-controller.conditions key=core.controller.deaf" { + t.Errorf("a carried kind is %+v, said %+v", c, r.said) + } + r.said = nil + if err := r.k.Lease(true); err != nil { + t.Fatal(err) + } + if _, open := r.open("core.controller.deaf"); open || r.changes() != "cleared core.controller.deaf" { + t.Error("a carried kind outlived the lease write") + } +} + +// ADR 0564 clause 9 (issue 1952 C5, C8; the lead 2026-10-29): compare counts only what both sides can raise. A key held +// while its module is delivered, a kind that ends with the controller and a kind the controller never raises are listed +// apart; a kind with no home decided is a difference naming the gap; take-over is refused while compare differs, and +// hand-back after the lease write. +func TestCompareCountsOnlyWhatBothSidesRaise(t *testing.T) { + r := newRig() + put := func(key, kind string) []byte { + raw, _ := json.Marshal(health.Condition{Key: key, Kind: kind, Severity: health.Warning, Source: "x"}) + return raw + } + key, raw := finding(t, watchman, health.ScopeMachine, "g14", "silent", "S1", health.Warning, r.now) + _ = r.k.Finding(watchman, key, raw, false) + _ = r.k.Controller(key, put(key, "silent"), false) + only, onlyRaw := finding(t, delivery.AssignerModule, health.ScopeSeat, "mesh-placement.ace", "states-unbound", "say", health.Warning, r.now) + _ = r.k.Finding(delivery.AssignerModule, only, onlyRaw, false) + _ = r.k.Controller("core.controller.deaf", put("core.controller.deaf", "controller-deaf"), false) + _ = r.k.Controller("delivery.x.stalled", put("delivery.x.stalled", "stalled"), false) + _ = r.k.Controller("seat.vault.ace.holder-silent", put("seat.vault.ace.holder-silent", "holder-silent"), false) + cmp, err := r.k.Compare() + if err != nil { + t.Fatal(err) + } + reasons := map[string]string{} + for _, d := range append(append([]Difference(nil), cmp.Differences...), cmp.Apart...) { + reasons[d.Key] = d.Reason + } + for key, says := range map[string]string{ + only: "keeper only", + "core.controller.deaf": "carried", + "delivery.x.stalled": "no home is decided for the kind stalled", + "seat.vault.ace.holder-silent": "mesh-self-check has not written its finding", + } { + if !strings.Contains(reasons[key], says) { + t.Errorf("%s: %q, want it to say %q", key, reasons[key], says) + } + } + if cmp.Same != 1 || len(cmp.Differences) != 2 { + t.Errorf("compare: %+v", cmp) + } + if _, err := r.k.TakeOver(nil, "jochen", "now"); err == nil || !strings.Contains(err.Error(), "2 difference") { + t.Errorf("take-over while compare differs: %v", err) + } + if _, err := r.k.HandBack("jochen", "back"); err != nil { + t.Errorf("hand-back before the lease write: %v", err) + } + _ = r.k.Lease(true) + if _, err := r.k.HandBack("jochen", "back"); err == nil { + t.Error("hand-back after the lease write was taken") + } +} diff --git a/modules/mesh-conditions/cmd/mesh-conditions/main.go b/modules/mesh-conditions/cmd/mesh-conditions/main.go new file mode 100644 index 00000000..2a1c654e --- /dev/null +++ b/modules/mesh-conditions/cmd/mesh-conditions/main.go @@ -0,0 +1,161 @@ +// mesh-conditions: the condition keeper, holder of the seat mesh-conditions (novox/hq ADR 0564 clauses 1-3, 9, 13). A +// Go bundle served by a runtime of its own, on its own bus account (`runs-as`), so that its account alone writes its +// states: `conditions` (one key per open condition), `history` (every transition, ninety days), `takeover` (the key the +// readers follow), `held` and `bounds` (what a delivery holds, and for how long). It watches every observer's +// `findings`, the healers' `attempts`, the assigner's wants, the nodes' haves and the controller's lease and, until the +// lease write, its `conditions`; it says every transition on its seat with the controller's body. stdout is the MCP +// channel; what this module says, it says on stderr. +package main + +import ( + "fmt" + "os" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/asker" + "git.novox.be/novox/mesh-sdk/go/asks" + "git.novox.be/novox/mesh-sdk/go/buscompose" + "git.novox.be/novox/mesh-sdk/go/health" +) + +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, uint64, bool, error) { + e, err := stdio.State(state).Get(key) + if err != nil || e == nil { + return nil, 0, false, err + } + return e.Value, e.Revision, true, nil +} + +func (runtimeStore) Create(state, key string, value any) (uint64, error) { + return stdio.State(state).Create(key, value) +} + +func (runtimeStore) Update(state, key string, value any, revision uint64) (uint64, error) { + return stdio.State(state).Update(key, value, revision) +} + +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() } + +// CreateTTL creates the key in a state whose own age is the time to live (`bounds`, ttl-seconds): the runtime writes no +// key with a time to live of its own. +func (runtimeStore) CreateTTL(state, key string, value any, _ time.Duration) error { + _, err := stdio.State(state).Create(key, value) + return err +} + +// sayOnSeat says a transition on the seat, as the controller says it on its own (ADR 0564 clause 3). +func sayOnSeat(e health.Event) error { + return stdio.SeatPublish("mesh.seat."+Seat+".event."+e.Event, e, "") +} + +// NewKeeper is the keeper over the runtime. +func NewKeeper() *Keeper { + return &Keeper{Store: runtimeStore{}, Say: sayOnSeat, Now: time.Now, Logf: logf} +} + +// watch keeps a watch of one state, said blind while it cannot be read: a state not on the bus yet (an observer not +// delivered) is read again when it is. +func watch(k *Keeper, state, observer string, handle func(stdio.StateChange) error) { + go func() { + for wait := 5 * time.Second; ; wait = min(wait*2, 5*time.Minute) { + err := stdio.State(state).Watch("", handle) + if err == nil { + if observer != "" { + k.Blind(observer, "") + } + return + } + if observer != "" { + k.Blind(observer, err.Error()) + } + logf("%s cannot be watched (%v); watched again in %s", state, err, wait) + time.Sleep(wait) + } + }() +} + +// Keep starts every watch the keeper reads by (clauses 2, 8, 9, 13). +func Keep(k *Keeper) { + for _, s := range []string{HeldState} { + keys, _ := stdio.State(s).Keys() + for _, key := range keys { + if e, err := stdio.State(s).Get(key); err == nil && e != nil { + k.restoreHeld(key, e.Value) + } + } + } + if e, err := stdio.State(TakeoverState).Get(health.TakeoverKey); err == nil { + var raw []byte + if e != nil { + raw = e.Value + } + if t, err := health.DecodeTakeover(raw, e != nil); err == nil { + k.Takeover(t) + } + } + said := func(err error) error { + if err != nil { + logf("%v", err) + } + return nil + } + watch(k, ControllerLease, "", func(c stdio.StateChange) error { + if c.Key != ControllerLeaseKey { + return nil + } + away, why := buscompose.LeaseHandedAway(c.Value, c.Op == "put", nil) + logf("the controller's lease: %s", why) + return said(k.Lease(away)) + }) + watch(k, WantsRead, "", func(c stdio.StateChange) error { return said(k.Want(c.Key, c.Value, c.Op != "put")) }) + watch(k, HavesRead, "", func(c stdio.StateChange) error { return said(k.Have(c.Key, c.Value, c.Op != "put")) }) + watch(k, AttemptsRead, health.HealersModule+" attempts", func(c stdio.StateChange) error { + return said(k.Attempt(c.Key, c.Value, c.Op != "put")) + }) + watch(k, ControllerConditions, "", func(c stdio.StateChange) error { + if k.LeaseAway() { + return nil + } + return said(k.Controller(c.Key, c.Value, c.Op != "put")) + }) + for _, o := range Observers { + watch(k, o+"."+health.FindingsState, o, func(c stdio.StateChange) error { + return said(k.Finding(o, c.Key, c.Value, c.Op != "put")) + }) + } + watch(k, BoundsState, "", func(c stdio.StateChange) error { + if c.Op != "put" && !c.Current { + return said(k.BoundPassed(c.Key)) + } + return nil + }) +} + +func main() { + k := NewKeeper() + client := asker.New(AskedState) + client.Module = Module + go func() { + <-stdio.Served() + Keep(k) + if err := client.Listen(k.Warranted, func(h asker.Handled, w asks.Warrant, err error) { + if err != nil { + logf("the answer to %s (%s): %v", w.Ask, h, err) + } + }); err != nil { + logf("the operator's answers cannot be heard: %v", err) + } + }() + if err := stdio.Serve("", Tools(k, client)); err != nil { + logf("%v", err) + os.Exit(1) + } +} diff --git a/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go b/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go new file mode 100644 index 00000000..66174fa2 --- /dev/null +++ b/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go @@ -0,0 +1,67 @@ +package main + +import ( + "os" + "path/filepath" + "slices" + "testing" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/health" +) + +// ADR 0564 clauses 1, 2, 3 and 12 (issue 1952 C1): the manifest says what the code does. It claims the seat +// mesh-conditions once for the mesh with the verbs mesh-sdk defines for it; reads every observer's findings, the healers' +// attempts, the controller's conditions and lease, the wants and the haves, and never the healers' would; keeps its +// states, history ninety days and each delivery's bound as long as the bound; and every state has a tool that reads it, +// every act a tool that starts it. +func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) { + raw, err := os.ReadFile(filepath.Join("..", "..", "module.json")) + if err != nil { + t.Fatal(err) + } + m, err := catalogue.ParseManifest(raw) + if err != nil { + t.Fatalf("the manifest is refused: %v", err) + } + seat, _ := catalogue.SeatNamed(Seat) + if m.Module != Module || m.RunsAs != Module || len(m.Claims) != 1 || m.Claims[0].Name != Seat || + !slices.Equal(Verbs, catalogue.VerbNames(seat.Serves)) { + t.Errorf("module %s runs as %s and claims %+v; the seat serves %v", m.Module, m.RunsAs, m.Claims, catalogue.VerbNames(seat.Serves)) + } + reads := []string{ControllerConditions, ControllerLease, WantsRead, HavesRead, AttemptsRead} + for _, o := range Observers { + reads = append(reads, o+"."+health.FindingsState) + } + slices.Sort(reads) + got := slices.Clone(m.Reads) + slices.Sort(got) + if !slices.Equal(got, reads) || slices.Contains(m.Reads, health.HealersModule+"."+health.WouldState) { + t.Errorf("reads %v, want %v and never the healers' would", got, reads) + } + ttl := map[string]int{} + for _, s := range m.State { + ttl[s.Name] = s.TTLSeconds + } + if ttl[HistoryState] != int(HistoryKeptFor.Seconds()) || ttl[BoundsState] != 900 { + t.Errorf("history is kept %d s and a bound %d s", ttl[HistoryState], ttl[BoundsState]) + } + // Clause 12: each state with the tool that reads it, each act with the tool that starts it. + readBy := map[string]string{ConditionsState: Seat + ".conditions", HistoryState: HistoryTool, TakeoverState: TakeoverTool, + HeldState: HeldTool, BoundsState: HeldTool, AskedState: Seat + ".take-over", "a resync": Seat + ".resync", + "what it reads": InputsTool} + tools := map[string]bool{} + for _, tool := range Tools(&Keeper{}, nil) { + tools[tool.Name] = true + } + for name := range ttl { + if readBy[name] == "" { + t.Errorf("the state %s has no tool that reads it", name) + } + } + for what, tool := range readBy { + if !tools[tool] { + t.Errorf("%s is read or started by %s, which is not served", what, tool) + } + } +} diff --git a/modules/mesh-conditions/cmd/mesh-conditions/reads-outside b/modules/mesh-conditions/cmd/mesh-conditions/reads-outside new file mode 100644 index 00000000..c20a67a5 --- /dev/null +++ b/modules/mesh-conditions/cmd/mesh-conditions/reads-outside @@ -0,0 +1,2 @@ +# What these tests read outside their package, as paths from the repository's root (issue 1936). +modules/mesh-conditions/module.json diff --git a/modules/mesh-conditions/cmd/mesh-conditions/tools.go b/modules/mesh-conditions/cmd/mesh-conditions/tools.go new file mode 100644 index 00000000..330e153e --- /dev/null +++ b/modules/mesh-conditions/cmd/mesh-conditions/tools.go @@ -0,0 +1,252 @@ +package main + +// The seat's verbs and the keeper's own tools (novox/hq ADR 0564 clauses 3, 9 and 12; issue 1952 C4): `conditions` +// answers as the controller's verb answers, so a caller changes only the seat it asks; and a read of every state the +// keeper keeps and of what it reads, so a person sees all it holds and a recovery is a call. + +import ( + "encoding/json" + "fmt" + "slices" + "strconv" + "strings" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/health" +) + +// Verbs are the seat's verbs, as the manifest claims them. +var Verbs = []string{"conditions", "silence", "compare", "take-over", "hand-back", "resync"} + +// The keeper's own read tools. +const ( + HistoryTool = "conditions_history" + HeldTool = "conditions_held" + TakeoverTool = "conditions_takeover" + InputsTool = "conditions_inputs" +) + +// briefEvidence is how many observations a condition carries in a list: its newest (the controller's, issue 314). +const briefEvidence = 1 + +// inBrief is a condition in a list: its newest evidence, with how many it keeps and how to read it whole. +type inBrief struct { + health.Condition + EvidenceKept int `json:"evidence-kept,omitempty"` + More string `json:"more,omitempty"` +} + +func obj(props map[string]any, required ...string) map[string]any { + out := map[string]any{"type": "object", "properties": props} + if len(required) > 0 { + out["required"] = required + } + return out +} + +func str(description string) map[string]any { + return map[string]any{"type": "string", "description": description} +} + +// Tools are what the runtime serves. +func Tools(k *Keeper, a Asks) []stdio.Tool { + return []stdio.Tool{ + {Name: Seat + ".conditions", Description: "What is wrong with the mesh now: every open condition, urgent first, then " + + "oldest, raised from the observers' findings and cleared when the finding goes, never by hand. Given a key, that " + + "one whole; with history, every transition lately.", + Input: obj(map[string]any{"key": str("a condition's key"), "scope": str("only this scope"), + "severity": str("urgent or warning"), "machine": str("only those about this machine"), + "history": str("\"true\": every transition lately"), "days": str("with history: days back (7, at most 90)")}), + Run: func(args map[string]any) (any, error) { return k.ConditionsVerb(args) }}, + {Name: Seat + ".silence", Description: "Stop one condition's messages for a while, a hand act with why, kept as its " + + "transition in history; it stays open.", + Input: obj(map[string]any{"key": str("the condition's key"), "for": str("30m, 4h, 2d; at most 7d"), + "why": str("why"), "cause": str("the cause in a word")}, "key", "for", "why"), + Run: func(args map[string]any) (any, error) { + d, err := duration(strArg(args, "for")) + if err != nil { + return nil, err + } + return k.Silence(strArg(args, "key"), d, stdio.CallerOf(args), strArg(args, "why")) + }}, + {Name: Seat + ".compare", Description: "The keeper's open conditions against the controller's, each difference with " + + "its reason; what ends with the controller, what only the keeper raises and what is held listed apart.", + Run: func(map[string]any) (any, error) { return k.Compare() }}, + {Name: Seat + ".take-over", Description: "Ask the operator's Approve to switch the readers of conditions to the " + + "keeper; refused while compare differs.", + Input: obj(map[string]any{"why": str("why now")}, "why"), + Run: func(args map[string]any) (any, error) { + return k.TakeOver(a, stdio.CallerOf(args), strArg(args, "why")) + }}, + {Name: Seat + ".hand-back", Description: "Switch the readers of conditions back to the controller, until its lease " + + "is handed away.", + Input: obj(map[string]any{"why": str("why")}, "why"), + Run: func(args map[string]any) (any, error) { return k.HandBack(stdio.CallerOf(args), strArg(args, "why")) }}, + {Name: Seat + ".resync", Description: "Read every finding again and raise and clear what differs.", + Run: func(map[string]any) (any, error) { + n, err := k.Resync() + return map[string]any{"settled": n}, err + }}, + {Name: HistoryTool, Description: "The keeper's history: every transition it said, newest first, at most limit.", + Input: obj(map[string]any{"limit": str("how many (default 50)"), "key": str("only this condition")}), + Run: func(args map[string]any) (any, error) { + return k.History(strArg(args, "key"), intArg(args, "limit", 50)) + }}, + {Name: HeldTool, Description: "What the keeper holds while its module is delivered (ADR 0564 clause 13): each key " + + "with its module, node, version and since when, and the deliveries' bounds.", + Run: func(map[string]any) (any, error) { + bounds, err := k.Store.Keys(BoundsState) + return map[string]any{"held": k.Held(), "bounds": bounds}, err + }}, + {Name: TakeoverTool, Description: "The take-over key: which source the router, the grants part and mesh-watcher hear.", + Run: func(map[string]any) (any, error) { + raw, _, found, err := k.Store.Get(TakeoverState, health.TakeoverKey) + if err != nil { + return nil, err + } + return health.DecodeTakeover(raw, found) + }}, + {Name: InputsTool, Description: "What the keeper reads, as it stands: each observer's findings counted, an observer it " + + "cannot read and why, the healers' attempts, the controller's conditions and whether its lease is handed away.", + Run: func(map[string]any) (any, error) { return k.Inputs(), nil }}, + } +} + +// ConditionsVerb is `conditions`: the list, one whole, or the history. +func (k *Keeper) ConditionsVerb(args map[string]any) (any, error) { + key := strArg(args, "key") + if strArg(args, "history") == "true" { + days := intArg(args, "days", 7) + if days < 1 || days > 90 { + return nil, fmt.Errorf("days is from 1 to 90, not %d", days) + } + return k.HistorySince(key, k.Now().Add(-time.Duration(days)*24*time.Hour), days) + } + if key != "" { + raw, _, found, err := k.Store.Get(ConditionsState, key) + if err != nil { + return nil, err + } + if !found || string(raw) == "null" { + return nil, fmt.Errorf("no condition %s is open", key) + } + var c health.Condition + return c, json.Unmarshal(raw, &c) + } + scope, severity, machine := strArg(args, "scope"), strArg(args, "severity"), strArg(args, "machine") + if severity != "" && severity != string(health.Urgent) && severity != string(health.Warning) { + return nil, fmt.Errorf("a severity is urgent or warning, not %q", severity) + } + open, err := k.Open() + if err != nil { + return nil, err + } + list := []inBrief{} + counted := map[string]int{} + for _, c := range open { + if (scope != "" && c.Subject.Scope != scope) || (severity != "" && string(c.Severity) != severity) || + (machine != "" && !concerns(c, machine)) { + continue + } + b := inBrief{Condition: c} + if len(c.Evidence) > briefEvidence { + b.EvidenceKept, b.Evidence, b.More = len(c.Evidence), c.Evidence[:briefEvidence:briefEvidence], health.ShowKeeper(c.Key) + } + list = append(list, b) + counted[c.Source+" "+c.Kind]++ + } + return map[string]any{"conditions": list, "open": len(open), "counted": counted, + "note": "urgent first, then oldest first; a condition clears when observation says so, never by hand; each with " + + "its newest evidence — `conditions key=` gives one whole"}, nil +} + +// concerns says whether a condition is about a machine: it names it, or its key does. +func concerns(c health.Condition, machine string) bool { + return c.Subject.Machine == machine || slices.Contains(c.Subject.Also, machine) || + slices.Contains(strings.Split(c.Key, "."), machine) +} + +// History is the newest transitions, at most limit, of one key or all. +func (k *Keeper) History(key string, limit int) ([]health.Event, error) { + all, err := k.HistorySince(key, time.Time{}, 0) + if err != nil { + return nil, err + } + events := all["history"].([]health.Event) + slices.Reverse(events) + if len(events) > limit { + events = events[:limit] + } + return events, nil +} + +// HistorySince is every transition since a moment, oldest first, as the controller's verb answers it. +func (k *Keeper) HistorySince(key string, since time.Time, days int) (map[string]any, error) { + keys, err := k.Store.Keys(HistoryState) + if err != nil { + return nil, err + } + events := []health.Event{} + for _, hk := range keys { + raw, _, found, err := k.Store.Get(HistoryState, hk) + if err != nil || !found { + continue + } + var e health.Event + if json.Unmarshal(raw, &e) != nil || e.At.Before(since) || (key != "" && e.Key != key) { + continue + } + events = append(events, e) + } + return map[string]any{"history": events, "days": days}, nil +} + +// Inputs is what the keeper reads, as it stands. +func (k *Keeper) Inputs() map[string]any { + k.mu.Lock() + defer k.mu.Unlock() + k.init() + findings := map[string]int{} + for o, fs := range k.findings { + findings[o] = len(fs) + } + return map[string]any{"findings": findings, "blind": k.blind, "attempts": len(k.attempts), + "controller-conditions": len(k.ctrl), "lease-handed-away": k.leaseOut, "follow": k.takeover.Follow, + "wants": len(k.wants), "haves": len(k.haves)} +} + +func strArg(a map[string]any, name string) string { + switch v := a[name].(type) { + case string: + return strings.TrimSpace(v) + case bool: + return strconv.FormatBool(v) + case float64: + return strconv.FormatFloat(v, 'f', -1, 64) + } + return "" +} + +func intArg(a map[string]any, name string, def int) int { + if n, err := strconv.Atoi(strArg(a, name)); err == nil { + return n + } + return def +} + +// duration reads 30m, 4h or 2d. +func duration(s string) (time.Duration, error) { + if n, ok := strings.CutSuffix(s, "d"); ok { + days, err := strconv.Atoi(n) + if err != nil { + return 0, fmt.Errorf("%q is not a duration: 30m, 4h or 2d", s) + } + return time.Duration(days) * 24 * time.Hour, nil + } + d, err := time.ParseDuration(s) + if err != nil { + return 0, fmt.Errorf("%q is not a duration: 30m, 4h or 2d", s) + } + return d, nil +} diff --git a/modules/mesh-conditions/cmd/mesh-conditions/whole_integration_test.go b/modules/mesh-conditions/cmd/mesh-conditions/whole_integration_test.go new file mode 100644 index 00000000..877a8ce1 --- /dev/null +++ b/modules/mesh-conditions/cmd/mesh-conditions/whole_integration_test.go @@ -0,0 +1,335 @@ +//go:build integration + +package main + +// The condition keeper whole against a real bus (novox/hq ADR 0564 clauses 2 and 3, issue 1952 C6): the built binary, +// played over its real stdin and stdout as the node's tool runtime plays it (tools/call; mesh/state.* answered from a +// real nats-server's key-value buckets, watches handed over as the runtime hands them, mesh/seat.publish onto its +// subjects). An observer's finding written to its bucket raises a condition the keeper keeps in its bucket and says as +// condition-raised with the controller's body; the seat's verb lists it; the finding deleted clears it, said as +// condition-cleared. + +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 keeper 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 +} + +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 a state's bucket: the keeper's own by its local name, another's by `.`. +func (p *player) kv(state string) (jetstream.KeyValue, error) { + if owner, name, read := strings.Cut(state, "."); read { + return p.js.KeyValue(p.ctx, owner+"_"+name) + } + return p.js.KeyValue(p.ctx, Module+"_"+state) +} + +func (p *player) handle(m wire) { + var a struct { + State string `json:"state"` + Key string `json:"key"` + Value json.RawMessage `json:"value"` + Revision uint64 `json:"revision"` + Subject string `json:"subject"` + Body json.RawMessage `json:"body"` + } + _ = json.Unmarshal(m.Params, &a) + kv, kvErr := p.kv(a.State) + switch m.Method { + case "mesh/state.get": + if kvErr != nil { + p.answer(m.ID, nil, kvErr) + 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.create", "mesh/state.update", "mesh/state.put": + if kvErr != nil { + p.answer(m.ID, nil, kvErr) + return + } + var rev uint64 + var err error + switch m.Method { + case "mesh/state.create": + rev, err = kv.Create(p.ctx, a.Key, a.Value) + case "mesh/state.update": + rev, err = kv.Update(p.ctx, a.Key, a.Value, a.Revision) + default: + rev, err = kv.Put(p.ctx, a.Key, a.Value) + } + if errors.Is(err, jetstream.ErrKeyExists) || errors.Is(err, jetstream.ErrKeyRevisionMismatch) { + p.send(map[string]any{"id": m.ID, "error": map[string]any{"code": -32010, "message": "changed: " + err.Error()}}) + return + } + p.answer(m.ID, map[string]any{"revision": rev}, err) + case "mesh/state.keys": + if kvErr != nil { + p.answer(m.ID, nil, kvErr) + 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": + err := kvErr + if err == nil { + err = kv.Delete(p.ctx, a.Key) + } + p.answer(m.ID, map[string]any{}, err) + case "mesh/state.watch": + if kvErr != nil { + p.answer(m.ID, nil, kvErr) + return + } + w, err := kv.WatchAll(p.ctx) + if err != nil { + p.answer(m.ID, nil, err) + return + } + var watchID string + _ = json.Unmarshal(m.ID, &watchID) + 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" + } + 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": watchID, + "change": map[string]any{"state": a.State, "key": e.Key(), "op": op, "value": json.RawMessage(nullIfEmpty(e.Value())), + "revision": e.Revision(), "current": current}}}) + } + case "mesh/seat.publish": + p.answer(m.ID, map[string]any{"sequence": 1}, p.nc.Publish(a.Subject, a.Body)) + case "mesh/subscribe": + p.answer(m.ID, map[string]any{}, nil) + default: + p.answer(m.ID, nil, errors.New("nats: no responders available for request")) + } +} + +func nullIfEmpty(b []byte) []byte { + if len(b) == 0 { + return []byte("null") + } + return b +} + +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) + } +} + +// 1952/C6: an observer's finding raises a condition on the bus, said with the controller's body, and its deletion clears +// it. +func TestTheKeeperWholeRaisesAndClearsOnTheBus(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + bin := filepath.Join(t.TempDir(), Module) + build := exec.Command("go", "build", "-o", bin, ".") + build.Stderr = os.Stderr + if err := build.Run(); err != nil { + t.Fatalf("the keeper 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) + } + buckets := []string{"mesh-controller_conditions", "mesh-controller_lease", "mesh-delivery-assigner_wants", "node-engine_have", + "mesh-healers_attempts"} + for _, o := range Observers { + buckets = append(buckets, o+"_findings") + } + for _, s := range []string{ConditionsState, HistoryState, TakeoverState, HeldState, BoundsState, AskedState} { + buckets = append(buckets, Module+"_"+s) + } + for _, b := range buckets { + if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: b}); err != nil { + t.Fatal(err) + } + } + events, err := nc.SubscribeSync("mesh.seat." + Seat + ".event.>") + 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) + + // The first call serves; the watches start after it. + if _, err := p.call(InputsTool, map[string]any{}); err != nil { + t.Fatal(err) + } + time.Sleep(time.Second) + at := time.Now().UTC() + f := health.Finding{Key: "machine.g14.silent", Kind: "silent", Subject: health.Subject{Scope: health.ScopeMachine, ID: "g14", + Machine: "g14"}, Severity: health.Warning, Summary: "g14 is silent", Headline: "g14 is offline", + Explanation: "g14 said nothing for three minutes.", Resolved: "g14 is back", Source: "S1", + Observer: health.WatchmanModule, At: at, Evidence: []health.Evidence{{At: at, Said: "no presence for 3m"}}} + raw, err := health.EncodeFinding(f) + if err != nil { + t.Fatal(err) + } + findings, _ := js.KeyValue(ctx, health.WatchmanModule+"_findings") + if _, err := findings.Put(ctx, f.Key, raw); err != nil { + t.Fatal(err) + } + msg, err := events.NextMsg(10 * time.Second) + if err != nil { + t.Fatalf("no condition-raised was said: %v", err) + } + var e health.Event + if err := json.Unmarshal(msg.Data, &e); err != nil || msg.Subject != "mesh.seat.mesh-conditions.event.condition-raised" || + e.Key != f.Key || e.Change != health.ChangeRaised || e.Show != health.ShowKeeper(f.Key) || e.Headline != f.Headline { + t.Fatalf("said %s on %s", msg.Data, msg.Subject) + } + listed, err := p.call(Seat+".conditions", map[string]any{}) + if err != nil || !strings.Contains(string(listed), `"key":"machine.g14.silent"`) { + t.Fatalf("conditions answered %s, %v", listed, err) + } + if err := findings.Delete(ctx, f.Key); err != nil { + t.Fatal(err) + } + msg, err = events.NextMsg(10 * time.Second) + if err != nil || msg.Subject != "mesh.seat.mesh-conditions.event.condition-cleared" { + t.Fatalf("no condition-cleared was said: %v", err) + } + conditions, _ := js.KeyValue(ctx, Module+"_"+ConditionsState) + if _, err := conditions.Get(ctx, f.Key); !errors.Is(err, jetstream.ErrKeyNotFound) { + t.Errorf("the cleared condition is still kept: %v", err) + } +} diff --git a/modules/mesh-conditions/go.mod b/modules/mesh-conditions/go.mod new file mode 100644 index 00000000..fcfe14e1 --- /dev/null +++ b/modules/mesh-conditions/go.mod @@ -0,0 +1,22 @@ +module mesh-conditions + +go 1.26.0 + +require ( + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029210043-32f5ce371268 + 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-conditions/go.sum b/modules/mesh-conditions/go.sum new file mode 100644 index 00000000..8a5c3b7a --- /dev/null +++ b/modules/mesh-conditions/go.sum @@ -0,0 +1,27 @@ +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029210043-32f5ce371268 h1:+jH77IJmxUiFDdbJljhynQRY0Cc5/5jtwmMV7UUqRI4= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029210043-32f5ce371268/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-conditions/module.json b/modules/mesh-conditions/module.json new file mode 100644 index 00000000..20805f2b --- /dev/null +++ b/modules/mesh-conditions/module.json @@ -0,0 +1,84 @@ +{ + "module": "mesh-conditions", + "version": "1", + "runs-as": "mesh-conditions", + "claims": [ + { + "name": "mesh-conditions", + "scope": "mesh", + "serves": [ + "conditions", + "silence", + "compare", + "take-over", + "hand-back", + "resync" + ] + } + ], + "uses": [ + "operator-channel" + ], + "reads": [ + "mesh-delivery-watchman.findings", + "mesh-self-check.findings", + "mesh-healers.findings", + "mesh-delivery-assigner.findings", + "mesh-healers.attempts", + "mesh-controller.conditions", + "mesh-controller.lease", + "mesh-delivery-assigner.wants", + "node-engine.have" + ], + "state": [ + "conditions", + { + "name": "history", + "ttl-seconds": 7776000 + }, + "takeover", + "held", + { + "name": "bounds", + "ttl-seconds": 900, + "key-ttl": true + }, + "asked" + ], + "own-secrets": { + "broker": "${dir:state}/broker" + }, + "secrets-owner": "mesh-conditions", + "resources": [ + { + "id": "account", + "type": "user", + "name": "mesh-conditions", + "shell": "/usr/bin/nologin", + "home": "/var/lib/mesh-conditions" + }, + { + "id": "state", + "type": "directory", + "mode": "0700", + "owner": "mesh-conditions", + "place": ".", + "unknown": "program" + } + ], + "build": { + "artifacts": [ + { + "name": "tools", + "kind": "bundle", + "language": "go", + "system": "arch", + "from": "cmd/mesh-conditions", + "binary": "mesh-conditions", + "loads": [ + "mesh-conditions" + ] + } + ] + } +}