Merge branch 'feat/47-G-cond6' into feat/47-G-join8

This commit is contained in:
2026-11-12 00:14:16 +01:00
17 changed files with 809 additions and 105 deletions
@@ -347,6 +347,7 @@ type keptView struct {
Pending map[string]time.Time `json:"pending,omitempty"`
Asks map[string]time.Time `json:"asks,omitempty"`
Findings []seat.Finding `json:"findings,omitempty"`
Sets map[string]time.Time `json:"sets,omitempty"`
}
// saveView keeps what the keeper heard of the observer on the node, by compare-and-set; a source is kept once it is heard.
@@ -360,7 +361,7 @@ func (k *Keeper) saveView(observer, node string) {
}
v.kept = true
value := keptView{Observer: observer, Node: node, At: v.At, Pending: maps.Clone(v.Pending), Asks: maps.Clone(v.Asks),
Findings: slices.Clone(v.Findings)}
Findings: slices.Clone(v.Findings), Sets: maps.Clone(v.Sets)}
k.mu.Unlock()
key := viewKey(observer, node)
for range ConditionWrites {
@@ -403,6 +404,9 @@ func (k *Keeper) RestoreViews() {
v.At, v.Findings, v.kept = kv.At, kv.Findings, true
maps.Copy(v.Pending, kv.Pending)
maps.Copy(v.Asks, kv.Asks)
if len(kv.Sets) > 0 {
v.Sets = maps.Clone(kv.Sets)
}
k.mu.Unlock()
}
}
@@ -0,0 +1,201 @@
package main
// What a node runs and what the healers attempt, heard as events (design 47 G §541): the keeper's `reads` of
// `node-engine.have` and `mesh-healers.attempts` are gone. The node-engine seat's `composed`, `running` (with `healthy`
// and `stopped`), `wait-held` and `wait-ended` make the have the hold compares with the wants (ADR 0564 §13); the seat
// `mesh-healers`' `attempt-changed` makes an attempt of a condition, dropped at its `until` on the keeper's own clock or
// when the healer says `removed`. Each is read as §556 has it: a message older by `at` than what the keeper holds changes nothing.
import (
"encoding/json"
"errors"
"strings"
"time"
engine "git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
meshhealers "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshhealers"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/motion"
)
// haveParts are the newest of each event a node-engine said of one module on one node.
type haveParts struct {
composed *engine.Composed
running *engine.Running
wait *engine.Wait
waitAt time.Time
}
// The prefix of the keys of `bounds` that are an attempt's end.
const attemptPrefix = "attempt="
// engineKey reads the event and the node and module a key names: `node-engine.<event>[.<node>[.<module>]]`.
func engineKey(key string) (event, node, module string, ok bool) {
rest, found := strings.CutPrefix(key, engine.Seat+".")
if !found {
return "", "", "", false
}
parts := strings.SplitN(rest, ".", 3)
event = parts[0]
if len(parts) > 1 {
node = parts[1]
}
if len(parts) > 2 {
module = parts[2]
}
return event, node, module, true
}
// EngineEvent takes one `composed`, `running`, `wait-held` or `wait-ended` of a node-engine.
func (k *Keeper) EngineEvent(e stdio.Envelope) error {
event, node, module, ok := engineKey(e.Key)
if !ok {
return nil
}
switch event {
case engine.EventRunning:
var r engine.Running
if err := json.Unmarshal(e.Body, &r); err != nil || r.Module == "" || r.Node == "" {
k.Logf("the node-engine's running on %s is not readable: %v", node, err)
return nil
}
err := k.haveChanged(r.Node, r.Module, func(p *haveParts) {
if p.running == nil || !r.At.Before(p.running.At) {
p.running = &r
}
})
if err != nil {
return err
}
return k.EngineRunning(node, e.Body)
case engine.EventComposed:
var c engine.Composed
if err := json.Unmarshal(e.Body, &c); err != nil || c.Module == "" || c.Node == "" {
k.Logf("the node-engine's composed on %s is not readable: %v", node, err)
return nil
}
return k.haveChanged(c.Node, c.Module, func(p *haveParts) {
if p.composed == nil || !c.At.Before(p.composed.At) {
p.composed = &c
}
})
case engine.EventWaitHeld, engine.EventWaitEnded:
var w engine.Wait
if err := json.Unmarshal(e.Body, &w); err != nil || node == "" || module == "" {
k.Logf("the node-engine's %s on %s is not readable, or names no node and module: %v", event, node, err)
return nil
}
at := w.Since
return k.haveChanged(node, module, func(p *haveParts) {
if !p.waitAt.IsZero() && at.Before(p.waitAt) {
return
}
p.waitAt = at
if p.wait = &w; event == engine.EventWaitEnded {
p.wait = nil
}
})
}
return nil
}
// haveChanged applies one change to what a node-engine said of a module on a node, makes the have the hold reads, and
// judges the delivery.
func (k *Keeper) haveChanged(node, module string, apply func(*haveParts)) error {
k.mu.Lock()
k.init()
if k.parts == nil {
k.parts = map[string]*haveParts{}
}
p := k.parts[node+"."+module]
if p == nil {
p = &haveParts{}
k.parts[node+"."+module] = p
}
apply(p)
h, has := p.have()
if has {
k.haves[node+"."+module] = h
} else {
delete(k.haves, node+"."+module)
}
k.mu.Unlock()
return k.judge(node, module, "")
}
// have is the have those events make: a hold is the phase `held` with its reason; a running build with its health, on the
// composition of the composed with its fingerprint; a composed alone is applying; a stopped module has none.
func (p *haveParts) have() (motion.Have, bool) {
switch {
case p.wait != nil && p.wait.Step == engine.WaitStepCompose:
h := motion.Have{Phase: motion.PhaseHeld, Detail: p.wait.Reason}
if p.composed != nil {
h.FP, h.Composed = p.composed.Fingerprint, p.composed.Digest
}
return h, true
case p.running != nil && !p.running.Stopped && (p.composed == nil || p.composed.At.Before(p.running.At) || p.composed.Fingerprint == p.running.Fingerprint):
h := motion.Have{FP: p.running.Fingerprint, Phase: motion.PhaseRunning, Health: motion.HealthOK}
if p.running.Healthy != nil && !*p.running.Healthy {
h.Health = motion.HealthFailed
}
if p.composed != nil && p.composed.Fingerprint == h.FP {
h.Composed = p.composed.Digest
}
return h, true
case p.composed != nil && (p.running == nil || p.composed.At.After(p.running.At)):
return motion.Have{FP: p.composed.Fingerprint, Composed: p.composed.Digest, Phase: motion.PhaseApplying}, true
}
return motion.Have{}, false
}
// AttemptChanged takes one `attempt-changed` of the healers, kept by its subject's `<condition>.<attempt>`: the
// attempt reaches its condition's tried, resolver and escalation (ADR 0564 §7); it is dropped when the healer says
// `removed` and at its `until`, on the keeper's own clock.
func (k *Keeper) AttemptChanged(key string, body []byte) error {
rest, ok := strings.CutPrefix(key, meshhealers.Seat+".attempt-changed.")
if !ok || rest == "" {
return nil
}
var e meshhealers.AttemptChangedEvent
if err := json.Unmarshal(body, &e); err != nil || e.Condition == "" {
k.Logf("the healers' attempt-changed %s is not readable, so it is left out: %v", key, err)
return nil
}
now := k.Now().UTC()
k.mu.Lock()
k.init()
if e.Removed || e.Until != nil && !now.Before(*e.Until) {
delete(k.attempts, rest)
k.mu.Unlock()
_ = k.Store.Delete(BoundsState, attemptPrefix+rest)
return k.Settle(e.Condition)
}
a := meshhealers.AttemptsByAttemptValue{Act: e.Act, BudgetKey: e.BudgetKey, Condition: e.Condition, Finished: e.Finished,
Healer: e.Healer, Kind: e.Kind, Outcome: e.Outcome, Said: e.Said}
if e.Begun != nil {
a.Begun = *e.Begun
}
k.attempts[rest] = a
k.mu.Unlock()
if e.Until != nil {
if err := k.Store.CreateTTL(BoundsState, attemptPrefix+rest, map[string]any{"key": rest, "condition": e.Condition, "until": *e.Until},
e.Until.Sub(now)); err != nil && !errors.Is(err, stdio.ErrChanged) {
k.Logf("the end of the attempt %s could not be timed: %v", rest, err)
}
}
return k.Settle(e.Condition)
}
// AttemptPassed is an attempt's `until` passing: it is dropped from its condition.
func (k *Keeper) AttemptPassed(bound string) error {
rest := strings.TrimPrefix(bound, attemptPrefix)
k.mu.Lock()
k.init()
a, ok := k.attempts[rest]
delete(k.attempts, rest)
k.mu.Unlock()
if !ok {
return nil
}
return k.Settle(a.Condition)
}
@@ -18,6 +18,7 @@ import (
engine "git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
cc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
seat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
meshhealers "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshhealers"
watchdogs "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshwatchdogs"
stdio "git.novox.be/novox/mesh-sdk/go"
)
@@ -35,13 +36,35 @@ func emitterOf(key, event string) (string, bool) {
return m, ok && m != "" && !strings.Contains(m, ".")
}
// subjectOf reads the observer, the event and the node out of an event's key in either form of a part's finding stream
// (design 47 G §529): the old form `mesh.stream.<module>.event.<event>.<node>`, which the runtime delivers as
// `<module>.<event>.<node>` (or as `<module>.<event>` with the node in the envelope), and the template form
// `mesh.bus.<module>.finding-events.<event>.<node>`. The node of the key is the subject's alone; the envelope's is the
// fallback of a key without one.
func subjectOf(key, envelopeNode string) (module, event, node string, ok bool) {
parts := strings.Split(key, ".")
switch {
case len(parts) == 6 && parts[0] == "mesh" && parts[1] == "bus" && parts[3] == FindingEventsTemplate:
return parts[2], parts[4], parts[5], true
case len(parts) == 3:
return parts[0], parts[1], parts[2], true
case len(parts) == 2:
return parts[0], parts[1], envelopeNode, true
}
return "", "", "", false
}
// FindingEvent is one `finding-written` or `finding-cleared` heard from an observer on a node.
func (k *Keeper) FindingEvent(e stdio.Envelope) error {
if observer, ok := emitterOf(e.Key, EventFindingWritten); ok {
return k.FindingWritten(observer, e.Node, e.Body)
}
if observer, ok := emitterOf(e.Key, EventFindingCleared); ok {
return k.FindingCleared(observer, e.Node, e.Body)
if module, event, node, ok := subjectOf(e.Key, e.Node); ok {
switch event {
case EventFindingWritten:
return k.FindingWritten(module, node, e.Body)
case EventFindingCleared:
return k.FindingCleared(module, node, e.Body)
case EventFindingsStated:
return k.FindingsStated(module, node, e.Body)
}
}
if _, ok := emitterOf(e.Key, watchdogs.EventMachineAway); ok && strings.HasPrefix(e.Key, watchdogs.Seat+".") {
return k.MachinePresence(e.Body, true)
@@ -49,15 +72,15 @@ func (k *Keeper) FindingEvent(e stdio.Envelope) error {
if _, ok := emitterOf(e.Key, watchdogs.EventMachineBack); ok && strings.HasPrefix(e.Key, watchdogs.Seat+".") {
return k.MachinePresence(e.Body, false)
}
if e.Key == engine.Seat+"."+engine.EventRunning {
return k.EngineRunning(e.Node, e.Body)
if _, _, _, ok := engineKey(e.Key); ok {
return k.EngineEvent(e)
}
if strings.HasPrefix(e.Key, meshhealers.Seat+".attempt-changed.") {
return k.AttemptChanged(e.Key, e.Body)
}
if e.Key == broker.Seat+"."+broker.EventCreated {
return k.BrokerCreated(e.Body)
}
if observer, ok := emitterOf(e.Key, EventFindingsStated); ok {
return k.FindingsStated(observer, e.Node, e.Body)
}
return nil
}
@@ -211,7 +234,9 @@ func (k *Keeper) MachinePresence(body []byte, away bool) error {
func hear(k *Keeper) {
patterns := []string{"*." + EventFindingWritten, "*." + EventFindingCleared, "*." + EventFindingsStated,
watchdogs.Seat + "." + watchdogs.EventMachineAway, watchdogs.Seat + "." + watchdogs.EventMachineBack,
engine.Seat + "." + engine.EventRunning, broker.Seat + "." + broker.EventCreated}
engine.Seat + "." + engine.EventRunning, engine.Seat + "." + engine.EventComposed, engine.Seat + "." + engine.EventWaitHeld,
engine.Seat + "." + engine.EventWaitEnded, meshhealers.Seat + "." + meshhealers.EventAttemptChanged,
broker.Seat + "." + broker.EventCreated}
for _, pattern := range patterns {
go func() {
<-stdio.Served()
@@ -244,6 +269,9 @@ type sourceView struct {
At time.Time
Pending map[string]time.Time
Findings []seat.Finding
// Sets are the last findings-stated's at of each set of kinds a statement covers (design 47 G §580), keyed by the kinds
// sorted and joined by a comma; the empty key is a statement with no kinds, which covers every kind.
Sets map[string]time.Time
// Asks are the asks it has not answered, with the end of their bound (asks.go).
Asks map[string]time.Time
// kept says the source is in the state `findings-views` (catalog.go).
@@ -313,13 +341,29 @@ func (k *Keeper) FindingsStated(observer, node string, body []byte) error {
k.mu.Lock()
k.init()
v := k.view(observer, node)
if !v.At.IsZero() && !s.At.After(v.At) {
last := v.Sets[setOf(s.Kinds)]
if len(s.Kinds) == 0 && last.IsZero() {
last = v.At
}
if !last.IsZero() && !s.At.After(last) {
k.mu.Unlock()
k.Logf("%s's findings-stated on %s at %s is not newer than the one at %s: left out", observer, node, s.At, v.At)
k.Logf("%s's findings-stated on %s at %s is not newer than the one at %s: left out", observer, node, s.At, last)
return nil
}
v.At = s.At
v.Findings = s.Findings
covered := func(kind string) bool { return len(s.Kinds) == 0 || slices.Contains(s.Kinds, kind) }
v.At = later(v.At, s.At)
if v.Sets == nil {
v.Sets = map[string]time.Time{}
}
v.Sets[setOf(s.Kinds)] = s.At
// Only the kinds the statement covers are replaced: another bundle's findings and pending keys stand (§580).
kept := v.Findings[:0:0]
for _, f := range v.Findings {
if !covered(f.Kind) {
kept = append(kept, f)
}
}
v.Findings = append(kept, s.Findings...)
answered := v.Asks
v.Asks = map[string]time.Time{}
named := map[string]bool{}
@@ -328,6 +372,11 @@ func (k *Keeper) FindingsStated(observer, node string, body []byte) error {
}
oldPending := v.Pending
v.Pending = map[string]time.Time{}
for key, until := range oldPending {
if !covered(keyKind(key)) {
v.Pending[key] = until
}
}
type pend struct {
key string
deadline time.Time
@@ -343,8 +392,8 @@ func (k *Keeper) FindingsStated(observer, node string, body []byte) error {
pending = append(pending, pend{p.Key, s.At.Add(bound)})
}
var drop []string
for key := range k.findings[observer] {
if !named[key] && (once || k.keyNode[observer][key] == node) {
for key, f := range k.findings[observer] {
if !named[key] && covered(f.Kind) && (once || k.keyNode[observer][key] == node) {
drop = append(drop, key)
}
}
@@ -402,6 +451,28 @@ func (k *Keeper) FindingsStated(observer, node string, body []byte) error {
return nil
}
// setOf is the key of a set of kinds: the kinds sorted and joined by a comma; none is the empty key, every kind.
func setOf(kinds []string) string {
sorted := slices.Clone(kinds)
sort.Strings(sorted)
return strings.Join(sorted, ",")
}
// keyKind is the last token of a finding's key, its kind (or its ref, §14 of ADR 0564).
func keyKind(key string) string {
if i := strings.LastIndex(key, "."); i >= 0 {
return key[i+1:]
}
return key
}
func later(a, b time.Time) time.Time {
if b.After(a) {
return b
}
return a
}
// PendingPassed is a pending key's bound passing: unless a finding or a later set named it since, its condition is cleared.
func (k *Keeper) PendingPassed(bound string) error {
rest := strings.TrimPrefix(bound, pendingPrefix)
@@ -18,7 +18,6 @@ import (
"strings"
"time"
engineclient "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
meshwants "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/motion"
@@ -161,8 +160,8 @@ func (k *Keeper) dropHeld(key, why string) {
}
}
// 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.
// Want takes one change of the assigner's wants, which may end a delivery, which raises or drops what it held (the nodes' haves
// are heard as the node-engine's events, engine_events.go).
func (k *Keeper) Want(key string, value []byte, deleted bool) error {
node, module, ok := meshwants.ParseWantsByWantNodeModuleKey(key)
if !ok {
@@ -181,25 +180,6 @@ func (k *Keeper) Want(key string, value []byte, deleted bool) error {
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 := engineclient.ParseHaveKey(key)
if err != nil || isNode {
return nil
}
k.mu.Lock()
k.init()
if deleted {
delete(k.haves, node+"."+module)
} else if h, err := engineclient.DecodeHave(value); err == nil {
k.haves[node+"."+module] = motion.Have{FP: h.Fp, Composed: h.Composed, Phase: h.Phase, Health: h.Health, Detail: h.Detail}
} else {
k.Logf("the have %s is not readable: %v", key, err)
}
k.mu.Unlock()
return k.judge(node, module, "")
}
// Assignments takes one change of the assignments: which module holds which seat on a node. The keys are a node key and
// its module keys (assignerclient.Keys; research 072 row 37).
func (k *Keeper) Assignments(key string, value []byte, deleted bool) {
@@ -235,6 +215,12 @@ func (k *Keeper) BoundPassed(key string) error {
if strings.HasPrefix(key, askedPrefix) {
return k.AskPassed(key)
}
if strings.HasPrefix(key, ownerPrefix) {
return k.OwnerPassed(key)
}
if strings.HasPrefix(key, attemptPrefix) {
return k.AttemptPassed(key)
}
node, module := splitNodeModule(key)
k.mu.Lock()
k.init()
@@ -3,6 +3,9 @@ package main
import (
"encoding/json"
"strings"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
"testing"
assignerclient "git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient"
@@ -35,31 +38,28 @@ func TestADeliveryHoldsItsOwnFindingsUntilItEnds(t *testing.T) {
raw, _ := json.Marshal(meshwants.WantsByWantNodeModuleValue{V: 1, Fp: fp, Composed: composed})
return raw
}
have := func(fp, composed string, phase, h string) []byte {
raw, _ := json.Marshal(nodeengine.HaveByNodeModuleValue{V: 1, Fp: fp, Composed: composed, Phase: phase, Health: h})
return raw
}
have := func(fp, composed string, phase, h string) haveSpec { return haveSpec{fp, composed, phase, h} }
for _, c := range []struct {
name string
// end is the have the delivery ends with, or nil for a delivery past its bound.
end []byte
end *haveSpec
standing bool
said string
line string
}{
{name: "gone by its end", end: have(ctrlNext, ctrlNextComp, motion.PhaseRunning, engineclient.HealthOK)},
{name: "still standing at its end", end: have(ctrlNext, ctrlNextComp, motion.PhaseRunning, engineclient.HealthOK),
{name: "gone by its end", end: ptr(have(ctrlNext, ctrlNextComp, motion.PhaseRunning, engineclient.HealthOK))},
{name: "still standing at its end", end: ptr(have(ctrlNext, ctrlNextComp, motion.PhaseRunning, engineclient.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, motion.PhaseFailed, engineclient.HealthFailed), standing: true,
{name: "failed", end: ptr(have(ctrlNext, ctrlNextComp, motion.PhaseFailed, engineclient.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, motion.PhaseRunning, engineclient.HealthOK), false)
r.engineHave("novox.mesh-controller", have(ctrlRunning, ctrlComposed, motion.PhaseRunning, engineclient.HealthOK))
// 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, motion.PhaseApplying, engineclient.HealthUnknown), false)
r.engineHave("novox.mesh-controller", have(ctrlNext, ctrlNextComp, motion.PhaseApplying, engineclient.HealthUnknown))
own, ownRaw := finding(t, watchman, cc.ScopeModule, "mesh-controller.novox", "unhealthy", "health", cc.Warning, r.now)
other, otherRaw := finding(t, watchman, cc.ScopeModule, "postgres.novox", "unhealthy", "health", cc.Warning, r.now)
machine, machineRaw := finding(t, watchman, cc.ScopeMachine, "novox", "silent", "S1", cc.Urgent, r.now)
@@ -83,7 +83,7 @@ func TestADeliveryHoldsItsOwnFindingsUntilItEnds(t *testing.T) {
_ = r.k.Finding(watchman, own, nil, true)
}
if c.end != nil {
_ = r.k.Have("novox.mesh-controller", c.end, false)
r.engineHave("novox.mesh-controller", *c.end)
} else {
r.now = r.now.Add(motion.DefaultDeliveryBound)
_ = r.k.BoundPassed("novox.mesh-controller")
@@ -110,10 +110,34 @@ func TestADeliveryHoldsItsOwnFindingsUntilItEnds(t *testing.T) {
r.k.Assignments("ace.mesh-delivery-assigner", placed["ace.mesh-delivery-assigner"], false)
r.k.Assignments("ace.node", placed["ace.node"], false)
_ = r.k.Want("want.ace.mesh-delivery-assigner", want(ctrlNext, ctrlNextComp), false)
_ = r.k.Have("ace.mesh-delivery-assigner", have(ctrlRunning, ctrlComposed, motion.PhaseRunning, engineclient.HealthOK), false)
r.engineHave("ace.mesh-delivery-assigner", have(ctrlRunning, ctrlComposed, motion.PhaseRunning, engineclient.HealthOK))
seatKey, seatRaw := finding(t, selfCheck, cc.ScopeSeat, "mesh-module-assigner.ace", "holder-silent", "D3", cc.Warning, r.now)
_ = r.k.Finding(selfCheck, seatKey, seatRaw, false)
if got := r.changes(); got != "" || len(r.k.Held()) != 1 || r.k.Held()[0].Module != "mesh-delivery-assigner" {
t.Errorf("a seat's finding while its holder is delivered: said %q, held %+v", got, r.k.Held())
}
}
// haveSpec is a have as the old tests spoke it; engineHave says it as the node-engine's events do (design 47 G §541).
type haveSpec struct{ fp, composed, phase, health string }
func ptr[T any](v T) *T { return &v }
func (r *rig) engineHave(nodeModule string, h haveSpec) {
node, module, _ := strings.Cut(nodeModule, ".")
say := func(event string, body any) {
raw, _ := json.Marshal(body)
if err := r.k.EngineEvent(stdio.Envelope{Key: nodeengine.Seat + "." + event + "." + node + "." + module, Node: node, Body: raw}); err != nil {
panic(err)
}
}
r.hevent++
at := r.now.Add(time.Duration(r.hevent) * time.Millisecond)
if h.phase != motion.PhaseFailed {
say(nodeengine.EventComposed, nodeengine.Composed{At: at, Digest: h.composed, Fingerprint: h.fp, Module: module, Node: node, V: 1})
}
if h.phase == motion.PhaseRunning || h.phase == motion.PhaseFailed {
healthy := h.health == engineclient.HealthOK
say(nodeengine.EventRunning, nodeengine.Running{At: at.Add(time.Microsecond), Fingerprint: h.fp, Healthy: &healthy, Module: module, Node: node, V: 1})
}
}
@@ -12,6 +12,7 @@ import (
"encoding/json"
"errors"
"fmt"
"regexp"
"sort"
"strings"
"sync"
@@ -51,17 +52,25 @@ const (
BoundsState = "bounds"
)
// The hand-over of a failed head (design 47 G §540): the condition `pipeline-failed` is its commit's author's, and OwnerHandover
// after the finding's `since` it is the operator's, the seat the operator is asked through (a seat's name, §451).
const (
HandoverKind = "pipeline-failed"
HandoverOwner = "operator-channel"
OwnerHandover = 2 * time.Hour
ownerPrefix = "owner="
handedOverWhy = "handed to the operator after %s"
)
// HistoryKeptFor is how long a transition is kept (clause 3).
const HistoryKeptFor = 90 * 24 * time.Hour
// The other states it reads, as `<module>.<name>`.
const (
AttemptsRead = healersModule + "." + meshhealers.StateAttempts
WantsRead = assignerModule + ".wants"
WantsRead = assignerModule + ".wants"
// AssignmentsRead is the mesh-module-assigner seat's assignments, read for which module holds a seat on a node (hold.go; the
// lead, 2026-10-29, for ADR 0564 clause 2's inputs).
AssignmentsRead = assignerModule + "." + mesh.StateAssignments
HavesRead = "node-engine.have"
ConditionWrites = 8
)
@@ -109,6 +118,8 @@ type Keeper struct {
Bound time.Duration
// AskBound is how long an observer has to state its findings after an ask (the setting ask-bound-seconds, §557); AskedWithin when zero.
AskBound time.Duration
// Handover is how long a failed head stays its author's (the setting owner-handover-seconds, §557); OwnerHandover when zero.
Handover time.Duration
mu sync.Mutex
findings map[string]map[string]seat.Finding
@@ -116,6 +127,8 @@ type Keeper struct {
attempts map[string]meshhealers.AttemptsByAttemptValue
wants map[string]motion.Want
haves map[string]motion.Have
// parts are the newest of each event a node-engine said of a module on a node, which make its have (engine_events.go).
parts map[string]*haveParts
// holders are, by `<node>.<seat>`, the module the assignments says holds the seat there.
holders map[string]string
// assignments is the assignments' keys as heard, assembled per node.
@@ -215,33 +228,6 @@ func (k *Keeper) Blind(observer, why string) {
}
}
// 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 meshhealers.AttemptsByAttemptValue
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)
}
// endedWhy is why a condition of a kind that ended with the controller is cleared.
const endedWhy = "its kind ended with the controller, which was unassigned (issue 1126): no finding raises it again"
@@ -366,6 +352,7 @@ func (k *Keeper) settleOnce(key string) error {
}
k.dropHeld(key, "")
next := k.derive(key, was, open, f, now)
k.armHandover(next, f, now)
events := transitions(was, next, open)
if len(events) == 0 && open && sameBody(was, next) {
return nil
@@ -382,6 +369,9 @@ func (k *Keeper) settleOnce(key string) error {
delete(k.cleared, key)
k.mu.Unlock()
for _, e := range events {
if e.Change == cc.ChangeOwner && next.Owner == HandoverOwner {
e.Why = fmt.Sprintf(handedOverWhy, k.handover())
}
e.Condition, e.At, e.Event = next, now, cc.EventFor(e.Change)
e.Show = cc.ShowKeeper(next.Key)
k.tell(e)
@@ -414,6 +404,10 @@ func (k *Keeper) derive(key string, was Condition, open bool, f *seat.Finding, n
}
k.mu.Unlock()
}
if !open || c.Owner == "" && c.WaitingOn == "" {
c.Owner, c.WaitingOn = f.Owner, f.WaitingOn
}
k.handOver(&c, *f, now)
k.overlay(&c, now)
if !open {
k.mu.Lock()
@@ -428,6 +422,54 @@ func (k *Keeper) derive(key string, was Condition, open bool, f *seat.Finding, n
return c
}
// handOver makes a failed head the operator's once its Handover has passed since the finding's `since`: only while it is still
// the author's, so a hand's choice of another owner stands (design 47 G §540).
func (k *Keeper) handOver(c *Condition, f seat.Finding, now time.Time) {
if c.Kind != HandoverKind || c.Owner == "" || c.Owner != f.Owner || c.Owner == HandoverOwner {
return
}
since := f.Since
if since.IsZero() {
since = f.At
}
if !now.Before(since.Add(k.handover())) {
c.Owner = HandoverOwner
}
}
func (k *Keeper) handover() time.Duration {
if k.Handover > 0 {
return k.Handover
}
return OwnerHandover
}
// armHandover keeps the end of a failed head's author's time as a bound with its own time to live, so that the keeper hears it
// end on the bus and settles the condition again; after a restart the same key stands, or the time has passed and derive
// hands it over at once (design 47 G §540: timed from `since`).
func (k *Keeper) armHandover(c Condition, f *seat.Finding, now time.Time) {
if f == nil || c.Kind != HandoverKind || c.Owner != f.Owner || c.Owner == HandoverOwner {
return
}
since := f.Since
if since.IsZero() {
since = f.At
}
ttl := since.Add(k.handover()).Sub(now)
if ttl < time.Second {
return
}
if err := k.Store.CreateTTL(BoundsState, ownerPrefix+c.Key, map[string]any{"key": c.Key, "until": since.Add(k.handover())},
ttl); err != nil && !errors.Is(err, stdio.ErrChanged) {
k.Logf("the hand-over of %s to the operator could not be timed: %v", c.Key, err)
}
}
// OwnerPassed is a failed head's author's time ending: the condition is settled again, which hands it over.
func (k *Keeper) OwnerPassed(bound string) error {
return k.Settle(strings.TrimPrefix(bound, ownerPrefix))
}
// 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 *Condition, f seat.Finding, now time.Time) {
@@ -519,6 +561,9 @@ func transitions(was, now Condition, open bool) []Event {
case was.Silenced != nil && now.Silenced == nil:
out = append(out, Event{Change: cc.ChangeUnsilenced})
}
if was.Owner != now.Owner {
out = append(out, Event{Change: cc.ChangeOwner, Was: was.Owner})
}
if was.Needs != now.Needs || !sameActions(was.Actions, now.Actions) {
out = append(out, Event{Change: cc.ChangeWords, Was: was.Needs})
}
@@ -626,6 +671,45 @@ func (k *Keeper) Resync() (int, error) {
return len(keys), k.settleAll(keys)
}
// SetOwner is the hand act `conditions_owner`: it sets an open condition's owner now, in a form §678 gives, and says the
// change as any change of a condition (design 47 G §540, §685).
func (k *Keeper) SetOwner(key, owner, by string) (Condition, error) {
if !ownerForm.MatchString(owner) {
return Condition{}, fmt.Errorf("the owner %q is none of a session who@node, a seat's name or forge:<account> (design 47 G §678)", owner)
}
for i := 0; i < ConditionWrites; i++ {
raw, rev, found, err := k.Store.Get(ConditionsState, key)
if err != nil {
return Condition{}, err
}
if !found || string(raw) == "null" {
return Condition{}, fmt.Errorf("no condition %s is open", key)
}
c, err := decodeCondition(raw)
if err != nil {
return Condition{}, err
}
was := c.Owner
if was == owner {
return c, nil
}
now := k.Now().UTC()
c.Owner = owner
if _, err := k.Store.Update(ConditionsState, key, c.Stored(), rev); errors.Is(err, stdio.ErrChanged) {
continue
} else if err != nil {
return Condition{}, err
}
k.tell(Event{Condition: c, Event: seat.EventConditionChanged, At: now, Change: cc.ChangeOwner, Was: was,
Why: "set by " + by, Show: cc.ShowKeeper(key)})
return c, nil
}
return Condition{}, fmt.Errorf("the condition %s kept moving under the owner change", key)
}
// ownerForm is §678's three forms of an owner: who@node, a seat's name, forge:<account>.
var ownerForm = regexp.MustCompile(`^([a-z0-9][a-z0-9._-]*@[a-z0-9][a-z0-9-]*|[a-z][a-z0-9-]*|forge:[A-Za-z0-9._-]+)$`)
// 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) (Condition, error) {
@@ -111,6 +111,8 @@ type rig struct {
st *fakeStore
now time.Time
said []Event
// hevent orders the node-engine events a test says.
hevent int
}
func newRig() *rig {
@@ -231,8 +233,8 @@ func TestTheKeeperRaisesChangesAndClearsFromFindings(t *testing.T) {
return ""
}},
{name: "a healer's attempt reaches tried and the resolver", do: func() error {
a, _ := json.Marshal(meshhealers.AttemptsByAttemptValue{Healer: "H3", Condition: key, Act: "asserted", Outcome: meshhealers.AttemptsByAttemptValueOutcomeActed, Begun: r.now})
return r.k.Attempt("h3-1", a, false)
a, _ := json.Marshal(meshhealers.AttemptChangedEvent{Healer: "H3", Condition: key, Act: "asserted", Outcome: meshhealers.AttemptChangedEventOutcomeActed, Begun: &r.now})
return r.k.AttemptChanged(meshhealers.Seat+".attempt-changed."+key+".h3-1", a)
}, said: "resolver machine.g14.silent", check: func(c Condition, _ bool) string {
if c.Resolver != "healer:H3" || len(c.Tried) != 1 || c.Tried[0].By != "healer H3" {
return fmt.Sprintf("%+v", c)
@@ -240,9 +242,10 @@ func TestTheKeeperRaisesChangesAndClearsFromFindings(t *testing.T) {
return ""
}},
{name: "an escalation makes it urgent, the operator's", do: func() error {
a, _ := json.Marshal(meshhealers.AttemptsByAttemptValue{Healer: "H3", Condition: key, Act: "handed over", Outcome: meshhealers.AttemptsByAttemptValueOutcomeEscalated,
Begun: r.now.Add(time.Minute)})
return r.k.Attempt("h3-2", a, false)
begun := r.now.Add(time.Minute)
a, _ := json.Marshal(meshhealers.AttemptChangedEvent{Healer: "H3", Condition: key, Act: "handed over", Outcome: meshhealers.AttemptChangedEventOutcomeEscalated,
Begun: &begun})
return r.k.AttemptChanged(meshhealers.Seat+".attempt-changed."+key+".h3-2", a)
}, said: "severity machine.g14.silent, resolver machine.g14.silent", check: func(c Condition, _ bool) string {
if c.Severity != cc.Urgent || c.Resolver != cc.ResolverOperator || len(c.Tried) != 2 {
return fmt.Sprintf("%+v", c)
@@ -126,10 +126,6 @@ func Keep(k *Keeper, v *views.Reader) {
logf("cleared %d condition(s) of kinds that ended with the controller: %v", n, err)
}
watch(k, v, WantsRead, "", func(c stdio.StateChange) error { return said(k.Want(c.Key, c.Value, c.Op != "put")) })
watch(k, v, HavesRead, "", func(c stdio.StateChange) error { return said(k.Have(c.Key, c.Value, c.Op != "put")) })
watch(k, v, AttemptsRead, healersModule+" attempts", func(c stdio.StateChange) error {
return said(k.Attempt(c.Key, c.Value, c.Op != "put"))
})
hear(k)
go func() { <-stdio.Served(); ListBound(k, stdio.Ask) }()
watch(k, v, BoundsState, "", func(c stdio.StateChange) error {
@@ -35,11 +35,11 @@ func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) {
}
// The assigner's wants and assignments are asked once of their read tools and followed by their events (ADR 0615 §1);
// the observers' findings are heard as events and no `findings` is read (design 47 G §383).
reads := []string{HavesRead, AttemptsRead}
var reads []string
if !slices.Equal(m.Invokes, []string{WantsTool, AssignmentsTool, "seat:mesh-hand-acts.hand-act", "seat:mesh-wants.revision",
"seat:mesh-catalog.registrations", "seat:mesh-catalog.revision", "seat:mesh-broker.list"}) ||
!slices.Equal(m.Consumes, []string{"mesh-delivery-assigner.views.wants", "mesh-delivery-assigner.views.assignments",
"*.finding-written", "*.finding-cleared", "*.findings-stated", "mesh-watchdogs.machine-away", "mesh-watchdogs.machine-back", "mesh-wants.assignment-revision", "mesh-catalog.registration-changed", "mesh-catalog.catalog-revision", "mesh-broker.created", "node-engine.running"}) {
"*.finding-written", "*.finding-cleared", "*.findings-stated", "mesh-watchdogs.machine-away", "mesh-watchdogs.machine-back", "mesh-wants.assignment-revision", "mesh-catalog.registration-changed", "mesh-catalog.catalog-revision", "mesh-broker.created", "node-engine.running", "node-engine.composed", "node-engine.wait-held", "node-engine.wait-ended", "mesh-healers.attempt-changed"}) {
t.Errorf("invokes %v, consumes %v", m.Invokes, m.Consumes)
}
for _, r := range m.Reads {
@@ -64,7 +64,7 @@ func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) {
}
// 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,
HeldState: HeldTool, BoundsState: HeldTool, ViewsState: FindingsViewsTool, "a resync": Seat + ".resync", "what it holds of observers": FindingsViewsTool, "what it reads": InputsTool}
HeldState: HeldTool, BoundsState: HeldTool, ViewsState: FindingsViewsTool, "a resync": Seat + ".resync", "an owner": OwnerTool, "what it holds of observers": FindingsViewsTool, "what it reads": InputsTool}
tools := map[string]bool{}
for _, tool := range Tools(&Keeper{}) {
tools[tool.Name] = true
@@ -60,6 +60,9 @@ type Condition struct {
Count int
Tried []Tried
Resolver string
// Owner and WaitingOn are who the condition is about and what it waits on (design 47 G §685): from the finding when
// the condition is raised, and changed only as §540 says.
Owner, WaitingOn string
// Silenced is nil when no silence is in force.
Silenced *Silence
Epoch int64
@@ -110,7 +113,7 @@ func (c Condition) Stored() seat.ConditionsByKeyValue {
Headline: c.Headline, Explanation: c.Explanation, Resolved: c.Resolved, Needs: c.Needs, Actions: c.Actions,
Evidence: c.Evidence, Source: c.Source, Raised: c.Raised, LastObserved: c.LastObserved, First: timeP(c.First),
Observations: int64(c.Observations), Count: int64(c.Count), Resolver: c.Resolver, Epoch: c.Epoch,
LooksUnread: c.LooksUnread}
LooksUnread: c.LooksUnread, Owner: c.Owner, WaitingOn: c.WaitingOn}
for _, g := range c.Gaps {
v.Gaps = append(v.Gaps, seat.ConditionsByKeyValueGapsItem{Cleared: timeP(g.Cleared), Reopened: timeP(g.Reopened)})
}
@@ -130,7 +133,7 @@ func conditionOf(v seat.ConditionsByKeyValue) Condition {
Headline: v.Headline, Explanation: v.Explanation, Resolved: v.Resolved, Needs: v.Needs, Actions: v.Actions,
Evidence: v.Evidence, Source: v.Source, Raised: v.Raised, LastObserved: v.LastObserved, First: timeOf(v.First),
Observations: int(v.Observations), Count: int(v.Count), Resolver: v.Resolver, Epoch: v.Epoch,
LooksUnread: v.LooksUnread}
LooksUnread: v.LooksUnread, Owner: v.Owner, WaitingOn: v.WaitingOn}
for _, g := range v.Gaps {
c.Gaps = append(c.Gaps, Gap{Cleared: timeOf(g.Cleared), Reopened: timeOf(g.Reopened)})
}
@@ -161,7 +164,7 @@ func (e Event) Said() seat.ConditionEvent {
Headline: s.Headline, Explanation: s.Explanation, Resolved: s.Resolved, Needs: s.Needs, Actions: s.Actions,
Evidence: s.Evidence, Source: s.Source, Raised: timeP(s.Raised), LastObserved: timeP(s.LastObserved), First: s.First,
Observations: s.Observations, Count: s.Count, Resolver: s.Resolver, Epoch: s.Epoch, LooksUnread: s.LooksUnread,
Event: e.Event, At: e.At, Change: e.Change, Was: e.Was, Why: e.Why, Cleared: e.Cleared, Show: e.Show}
Owner: s.Owner, WaitingOn: s.WaitingOn, Event: e.Event, At: e.At, Change: e.Change, Was: e.Was, Why: e.Why, Cleared: e.Cleared, Show: e.Show}
for _, g := range s.Gaps {
v.Gaps = append(v.Gaps, seat.ConditionEventGapsItem{Cleared: g.Cleared, Reopened: g.Reopened})
}
@@ -181,7 +184,7 @@ func eventOf(v seat.ConditionEvent) Event {
Headline: v.Headline, Explanation: v.Explanation, Resolved: v.Resolved, Needs: v.Needs, Actions: v.Actions,
Evidence: v.Evidence, Source: v.Source, Raised: timeOf(v.Raised), LastObserved: timeOf(v.LastObserved),
First: v.First, Observations: v.Observations, Count: v.Count, Resolver: v.Resolver, Epoch: v.Epoch,
LooksUnread: v.LooksUnread}
LooksUnread: v.LooksUnread, Owner: v.Owner, WaitingOn: v.WaitingOn}
if v.Subject != nil {
s.Subject = *v.Subject
}
@@ -0,0 +1,209 @@
package main
import (
"testing"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
cc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
seat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
)
// Design 47 G §540, §685, A-58: a condition carries the finding's owner and waiting-on; a failed head is its author's for two
// hours after the finding's `since` and then the operator's, said as the change `owner`; a hand sets an owner in §678's forms;
// a newer finding does not undo it.
func TestTheKeeperHandsAFailedHeadToTheOperatorAfterTwoHours(t *testing.T) {
r := newRig()
failed := func(owner string) (string, []byte) {
f := seat.Finding{Scope: "pipeline", ID: "abc123", Kind: HandoverKind, Since: r.now, Subject: seat.Subject{Scope: "pipeline", ID: "abc123"},
Severity: cc.Warning, Summary: "a head failed", Headline: "failed", Explanation: "It failed.", Source: "runner",
Observer: watchman, At: r.now, Owner: owner, WaitingOn: "node-engine", Evidence: []seat.Evidence{{At: r.now, Said: "failed"}}}
raw, err := encodeFinding(f)
if err != nil {
t.Fatal(err)
}
return cc.Key("pipeline", "abc123", HandoverKind), raw
}
key, raw := failed("forge:jochen")
if err := r.k.Finding(watchman, key, raw, false); err != nil {
t.Fatal(err)
}
if c, ok := r.open(key); !ok || c.Owner != "forge:jochen" || c.WaitingOn != "node-engine" {
t.Fatalf("raised with the finding's owner and waiting-on: %+v", c)
}
if _, ok := r.st.ttls["bounds/"+ownerPrefix+key]; !ok || r.st.ttls["bounds/"+ownerPrefix+key] != 2*time.Hour {
t.Fatalf("the two hours are timed as a bound of their own: %v", r.st.ttls)
}
r.changes()
// An hour in, the finding said again changes nothing; at two hours the bound's end settles it into the operator's.
r.now = r.now.Add(time.Hour)
if err := r.k.Finding(watchman, key, raw, false); err != nil || r.changes() != "" {
t.Fatalf("an hour in nothing changes: %v", err)
}
r.now = r.now.Add(time.Hour)
if err := r.k.BoundPassed(ownerPrefix + key); err != nil {
t.Fatal(err)
}
if got := r.changes(); got != "owner "+key {
t.Fatalf("said %q", got)
}
if c, _ := r.open(key); c.Owner != HandoverOwner || c.WaitingOn != "node-engine" {
t.Fatalf("handed to the operator: %+v", c)
}
// The finding said again does not give it back to the author, and a restarted keeper that finds the two hours past hands it over alone.
if err := r.k.Finding(watchman, key, raw, false); err != nil || r.changes() != "" {
t.Fatalf("the hand-over stands: %v", err)
}
// A hand sets another owner in a form of §678, and refuses a form it does not give.
if _, err := r.k.SetOwner(key, "Not An Owner", "jochen"); err == nil {
t.Error("a form §678 does not give is refused")
}
if c, err := r.k.SetOwner(key, "jochen@shanks", "jochen"); err != nil || c.Owner != "jochen@shanks" || r.changes() != "owner "+key {
t.Fatalf("a hand's owner: %+v %v", c, err)
}
if err := r.k.Finding(watchman, key, raw, false); err != nil || r.changes() != "" {
t.Fatalf("a hand's owner stands: %v", err)
}
// A newer failing head is a finding of its own with two hours of its own.
r2 := newRig()
r2.now = r.now
r2.k.Handover = time.Hour
other, rawOther := failed("forge:marie")
if err := r2.k.Finding(watchman, other, rawOther, false); err != nil {
t.Fatal(err)
}
r2.now = r2.now.Add(2 * time.Hour)
if err := r2.k.Settle(other); err != nil {
t.Fatal(err)
}
if c, _ := r2.open(other); c.Owner != HandoverOwner {
t.Fatalf("a restarted keeper, the time past, hands it over from `since`: %+v", c)
}
}
// Design 47 G §580, §529: a statement carrying `kinds` clears only the conditions of those kinds the statement does not name,
// so two bundles of one observer on one node keep each other's conditions open; a statement with no kinds covers every
// kind; and the finding events are taken from either form of the part's finding stream.
func TestAStatementCoversOnlyItsKindsAndEventsComeInEitherForm(t *testing.T) {
r := newRig()
t0 := r.now
mk := func(kind string) seat.Finding {
f := eventFinding(watchman, t0)
f.Kind, f.ID, f.Subject.ID = kind, "g14", "g14"
return f
}
stated := func(at time.Time, kinds []string, fs ...seat.Finding) {
t.Helper()
if err := r.k.FindingsStated(watchman, "ace", mustJSON(t, seat.FindingsStated{At: at, Kinds: kinds, Findings: fs})); err != nil {
t.Fatal(err)
}
}
keyOf := func(kind string) string { return cc.Key(cc.ScopeMachine, "g14", kind) }
// Both bundles state their kinds; each raises its own.
stated(t0, []string{"alpha"}, mk("alpha"))
stated(t0.Add(time.Second), []string{"beta"}, mk("beta"))
for _, kind := range []string{"alpha", "beta"} {
if _, ok := r.open(keyOf(kind)); !ok {
t.Fatalf("%s is raised", kind)
}
}
// Alpha's bundle states nothing now: only alpha's condition is cleared, beta's stays open.
stated(t0.Add(2*time.Second), []string{"alpha"})
if _, ok := r.open(keyOf("alpha")); ok {
t.Error("alpha's condition is cleared by alpha's own statement")
}
if _, ok := r.open(keyOf("beta")); !ok {
t.Error("beta's condition stays open: another bundle's statement does not clear it")
}
// A statement with no kinds covers every kind.
stated(t0.Add(3*time.Second), nil)
if _, ok := r.open(keyOf("beta")); ok {
t.Error("a statement with no kinds clears every kind it does not name")
}
// Either form of the stream's subject reaches the same handler, the node taken from the subject.
body := mustJSON(t, eventFinding(watchman, t0.Add(time.Minute)))
for _, key := range []string{watchman + ".finding-written.ace", "mesh.bus." + watchman + ".finding-events.finding-written.ace"} {
r.k.Settle(keyOf("silent"))
if err := r.k.FindingEvent(stdioEnvelope(key, "", body)); err != nil {
t.Fatal(err)
}
if _, ok := r.open(keyOf("silent")); !ok {
t.Errorf("%s raised nothing", key)
}
if err := r.k.FindingEvent(stdioEnvelope(watchman+".finding-cleared.ace", "", mustJSON(t, seat.FindingCleared{At: t0.Add(2 * time.Minute),
ID: "g14", Kind: "silent", Scope: cc.ScopeMachine, Why: "back"}))); err != nil {
t.Fatal(err)
}
}
}
func stdioEnvelope(key, node string, body []byte) stdio.Envelope {
return stdio.Envelope{Key: key, Node: node, Body: body}
}
// Design 47 G §541: the healers' attempts and a node's holds are heard as events: an attempt shows on its condition and
// is dropped at its `until` or when the healer says it removed; a wait-held makes the module's have held with its
// reason, and a wait-ended ends it; a message older by `at` than the one held changes nothing.
func TestTheKeeperHearsAttemptsAndHoldsAsEvents(t *testing.T) {
r := newRig()
key, raw := finding(t, watchman, cc.ScopeMachine, "g14", "silent", "S1", cc.Warning, r.now)
if err := r.k.Finding(watchman, key, raw, false); err != nil {
t.Fatal(err)
}
subject := meshhealersSeat + ".attempt-changed." + key + ".a1"
until := r.now.Add(time.Hour)
a := mustJSON(t, map[string]any{"healer": "H3", "condition": key, "act": "asserted", "outcome": "acted", "begun": r.now, "until": until})
if err := r.k.AttemptChanged(subject, a); err != nil {
t.Fatal(err)
}
if c, _ := r.open(key); len(c.Tried) != 1 || c.Resolver != "healer:H3" {
t.Fatalf("an attempt said is shown on its condition: %+v", c)
}
if r.st.ttls["bounds/"+attemptPrefix+key+".a1"] != time.Hour {
t.Errorf("its end is a bound of its own: %v", r.st.ttls)
}
r.now = until
if err := r.k.BoundPassed(attemptPrefix + key + ".a1"); err != nil {
t.Fatal(err)
}
r.k.mu.Lock()
left := len(r.k.attempts)
r.k.mu.Unlock()
if left != 0 {
t.Errorf("an attempt past its until is dropped: %d held", left)
}
// The healer ending one says removed.
a = mustJSON(t, map[string]any{"healer": "H3", "condition": key, "act": "asserted", "outcome": "acted", "begun": r.now})
_ = r.k.AttemptChanged(meshhealersSeat+".attempt-changed."+key+".a2", a)
_ = r.k.AttemptChanged(meshhealersSeat+".attempt-changed."+key+".a2", mustJSON(t, map[string]any{"condition": key, "removed": true}))
r.k.mu.Lock()
left = len(r.k.attempts)
r.k.mu.Unlock()
if left != 0 {
t.Errorf("a removed attempt is dropped: %d held", left)
}
// A hold of a node-engine is the module's have held, until its wait-ended; an older message changes nothing.
say := func(event string, w map[string]any) {
if err := r.k.EngineEvent(stdioEnvelope(nodeEngineSeat+"."+event+".ace.mod", "ace", mustJSON(t, w))); err != nil {
t.Fatal(err)
}
}
say("wait-held", map[string]any{"step": "compose", "reason": "restore point missing", "since": r.now})
if h := r.k.haves["ace.mod"]; h.Phase != "held" || h.Detail != "restore point missing" {
t.Fatalf("a hold is heard: %+v", h)
}
say("wait-held", map[string]any{"step": "compose", "reason": "older", "since": r.now.Add(-time.Hour)})
if h := r.k.haves["ace.mod"]; h.Detail != "restore point missing" {
t.Errorf("an older message changes nothing: %+v", h)
}
say("wait-ended", map[string]any{"step": "compose", "since": r.now.Add(time.Minute)})
if h := r.k.haves["ace.mod"]; h.Phase == "held" {
t.Errorf("wait-ended ends the hold: %+v", h)
}
}
const (
meshhealersSeat = "mesh-healers"
nodeEngineSeat = "node-engine"
)
@@ -12,6 +12,7 @@ import (
const (
SettingAskBound = "ask-bound-seconds"
SettingViewWaits = "view-waits-seconds"
SettingHandover = "owner-handover-seconds"
defaultViewWaits = 5 * time.Minute
)
@@ -32,5 +33,6 @@ func seconds(read func(key string) (json.RawMessage, error), key string, def tim
// ReadSettings sets the keeper's bounds and the first views' wait from the settings.
func ReadSettings(k *Keeper, read func(key string) (json.RawMessage, error)) (viewWaits time.Duration) {
k.AskBound = seconds(read, SettingAskBound, AskedWithin)
k.Handover = seconds(read, SettingHandover, OwnerHandover)
return seconds(read, SettingViewWaits, defaultViewWaits)
}
@@ -6,6 +6,7 @@ package main
import (
"encoding/json"
"errors"
"fmt"
"slices"
"strconv"
@@ -21,8 +22,9 @@ import (
// Verbs are the seat's verbs, as the manifest claims them.
var Verbs = []string{"conditions", "silence", "resync"}
// The keeper's own read tools.
// The keeper's own tools: the owner act, and the read tools.
const (
OwnerTool = "conditions_owner"
HistoryTool = "conditions_history"
HeldTool = "conditions_held"
InputsTool = "conditions_inputs"
@@ -78,6 +80,13 @@ func Tools(k *Keeper) []stdio.Tool {
Run: func(args map[string]any) (any, error) {
return k.ResyncVerb(stdio.CallerOf(args), strArg(args, "module"))
}},
{Name: OwnerTool, Description: "Set an open condition's owner now: a session who@node, a seat's name or forge:<account>; " +
"said as the condition's change and written in the hand-act log (design 47 G §540).",
Input: obj(map[string]any{"key": str("the condition's key"), "owner": str("the new owner"), "why": str("why")},
"key", "owner", "why"),
Run: func(args map[string]any) (any, error) {
return k.OwnerVerb(stdio.CallerOf(args), strArg(args, "key"), strArg(args, "owner"), strArg(args, "why"))
}},
{Name: FindingsViewsTool, Description: "What the keeper holds of each observer and node: its last stated set with its at, " +
"its pending keys and its open asks (design 47 G §390).",
Run: func(map[string]any) (any, error) { return k.FindingsViews(), nil }},
@@ -98,6 +107,22 @@ func Tools(k *Keeper) []stdio.Tool {
}
}
// OwnerVerb is `conditions_owner`: it writes the call in the hand-act log, then sets the owner.
func (k *Keeper) OwnerVerb(by, key, owner, why string) (any, error) {
if strings.TrimSpace(why) == "" {
return nil, errors.New("an owner change says why")
}
record := k.Record
if record == nil {
record = recordHandAct
}
if err := record("set the owner of "+key+" to "+owner, why+" (by "+by+")", OwnerTool); err != nil {
return nil, fmt.Errorf("the hand act was not recorded, so nothing was changed: %w", err)
}
c, err := k.SetOwner(key, owner, by)
return c.Stored(), err
}
// ConditionsVerb is `conditions`: the list, one whole, or the history.
func (k *Keeper) ConditionsVerb(args map[string]any) (any, error) {
key := strArg(args, "key")
+11 -5
View File
@@ -13,10 +13,6 @@
]
}
],
"reads": [
"mesh-healers.attempts",
"node-engine.have"
],
"invokes": [
"mesh-delivery-assigner.assigner_read_wants",
"mesh-delivery-assigner.assigner_read_assignments",
@@ -38,7 +34,11 @@
"mesh-catalog.registration-changed",
"mesh-catalog.catalog-revision",
"mesh-broker.created",
"node-engine.running"
"node-engine.running",
"node-engine.composed",
"node-engine.wait-held",
"node-engine.wait-ended",
"mesh-healers.attempt-changed"
],
"state": [
"conditions",
@@ -60,6 +60,12 @@
"minimum": 1,
"default": 300
},
"owner-handover-seconds": {
"type": "number",
"description": "How long a failed head stays its commit's author's, in seconds, before the keeper hands its condition to the operator: 7200 is two hours.",
"minimum": 1,
"default": 7200
},
"view-waits-seconds": {
"type": "number",
"description": "How long a first view may stay unanswered, in seconds, before the keeper raises view-waits for it: 300 is five minutes.",
@@ -30,6 +30,7 @@ import (
handacts "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshhandacts"
meshhealers "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshhealers"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/keys"
"git.novox.be/novox/mesh-sdk/go/views"
)
@@ -287,6 +288,7 @@ func (h *Healers) act(row Healer, c Condition, budget string, spent int, by stri
}
return "", fmt.Errorf("healer %s did not act on %s: its attempt could not be begun: %w", row.ID, c.Key, err)
}
h.sayAttempt(key, a, begun)
act, said, err := row.repair(h, c)
a.Outcome = OutcomeActed
if err != nil {
@@ -302,6 +304,8 @@ func (h *Healers) act(row Healer, c Condition, budget string, spent int, by stri
a.Said, a.Finished = said+" ("+window+")", &finished
if err := h.Store.PutTTL(AttemptsState, key, a, KeptFor); err != nil {
h.Logf("[%s] healer %s %s %s and could not record what came of it: %v", Module, row.ID, a.Outcome, c.Key, err)
} else {
h.sayAttempt(key, a, finished)
}
h.tell(Acted{Healer: row.ID, Condition: c.Key, Kind: c.Kind, Act: a.Act, Outcome: a.Outcome, Said: said, Budget: window})
h.wake(budget, begun.Add(row.Settle), c.Key)
@@ -321,14 +325,55 @@ func (h *Healers) escalate(row Healer, c Condition, budget string, spent []Attem
window := fmt.Sprintf("%d of %d within %s", len(spent), row.Budget, row.Window)
a := Attempt{Healer: row.ID, Condition: c.Key, Kind: c.Kind, BudgetKey: budget, Act: "handed the condition to the operator",
Outcome: OutcomeEscalated, Said: said, Begun: now, Finished: &now}
if err := h.Store.CreateTTL(AttemptsState, attemptKey(row.ID, c.Key, now), a, KeptFor); err != nil && !errors.Is(err, ErrExists) {
ekey := attemptKey(row.ID, c.Key, now)
if err := h.Store.CreateTTL(AttemptsState, ekey, a, KeptFor); err != nil && !errors.Is(err, ErrExists) {
return "", fmt.Errorf("healer %s could not record handing %s to the operator, and does not: %w", row.ID, c.Key, err)
} else if err == nil {
h.sayAttempt(ekey, a, now)
}
h.tell(Acted{Healer: row.ID, Condition: c.Key, Kind: c.Kind, Act: "handed to the operator", Outcome: OutcomeEscalated,
Said: said, Budget: window})
return fmt.Sprintf("healer %s handed %s to the operator: %s", row.ID, c.Key, said), nil
}
// sayAttempt says an attempt as the seat's attempt-changed (design 47 G §541, §665): the attempt whole as its key in `attempts`
// holds it, with `until`, the end of that key's lifetime (KeptFor from its write at wrote), on the subject keyed by the
// condition and the attempt, each escaped as a key's name is (§73). Not said is said in the journal: the attempt holds it.
func (h *Healers) sayAttempt(key string, a Attempt, wrote time.Time) {
if h.Publish == nil {
return
}
until := wrote.Add(KeptFor)
e := meshhealers.AttemptChangedEvent{Act: a.Act, Begun: &a.Begun, BudgetKey: a.BudgetKey, Condition: a.Condition, Finished: a.Finished,
Healer: a.Healer, Kind: a.Kind, Outcome: a.Outcome, Said: a.Said, Until: &until}
subject := meshhealers.SubjectAttemptChanged(keys.EscapeName(a.Condition), keys.EscapeName(key))
if err := h.Publish(subject, e, key+"."+strconv.FormatInt(wrote.UnixNano(), 10)); err != nil {
h.Logf("[%s] the attempt %s of %s could NOT be said on the bus: %v", Module, key, a.Condition, err)
}
}
// sayOpen says every attempt still kept, at the healers' start, so a keeper that missed one hears it (design 47 G §541). The
// end of an attempt's lifetime is a week from its last write: its finish, or its beginning while it is not finished.
func (h *Healers) sayOpen() {
names, err := h.Store.Keys(AttemptsState)
if err != nil {
h.Logf("[%s] the attempts kept cannot be read, so none is said again at the start: %v", Module, err)
return
}
for _, key := range names {
raw, found, err := h.Store.Get(AttemptsState, key)
var a Attempt
if err != nil || !found || json.Unmarshal(raw, &a) != nil {
continue
}
wrote := a.Begun
if a.Finished != nil {
wrote = *a.Finished
}
h.sayAttempt(key, a, wrote)
}
}
// tell says a heal as healer-acted. Not said is said in the log: the attempt holds it.
func (h *Healers) tell(e Acted) {
e.Event, e.At, e.By = ActedEvent, h.Now().UTC(), "healer "+e.Healer
@@ -345,6 +390,7 @@ func (h *Healers) tell(e Acted) {
// no expiry left to hear, so each newest act whose settle ended within its window is judged at once. A settle still
// running keeps its key on the bus and needs nothing.
func (h *Healers) Rearm() {
h.sayOpen()
all, err := h.attempts()
if err != nil {
h.Logf("[%s] what the healers did cannot be read, so no ended settle is judged: %v", Module, err)
@@ -16,6 +16,7 @@ import (
meshconditions "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
handacts "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshhandacts"
meshhealers "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshhealers"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/views"
)
@@ -136,6 +137,8 @@ type world struct {
said []Acted
// events are the subjects the finding writer said its events on.
events []string
// attempted are the attempt-changed events said: subject, outcome and until.
attempted []string
}
func newWorld() *world {
@@ -157,6 +160,9 @@ func (w *world) fresh() {
if a, ok := body.(Acted); ok && subject == "mesh.seat.mesh-healers.event.healer-acted" {
w.said = append(w.said, a)
}
if a, ok := body.(meshhealers.AttemptChangedEvent); ok {
w.attempted = append(w.attempted, subject+" "+a.Outcome+" "+a.Until.Format(time.RFC3339))
}
return nil
},
}
@@ -232,6 +238,25 @@ func TestTheHealersActWithNoLease(t *testing.T) {
}
}
// Design 47 G §541, §665: each attempt written is said as attempt-changed on the subject keyed by the condition and the attempt,
// escaped as a key's name is, with its until a week from its write; and said again for every attempt kept at the healers' start.
func TestAnAttemptIsSaidAsAttemptChangedAndAgainAtTheStart(t *testing.T) {
w := newWorld()
if err := w.h.Heard(body(lost("records"))); err != nil {
t.Fatal(err)
}
if len(w.attempted) != 2 || !strings.Contains(w.attempted[0], "mesh.seat.mesh-healers.event.attempt-changed.bus") ||
!strings.HasSuffix(w.attempted[0], " acting "+w.now.Add(KeptFor).Format(time.RFC3339)) || !strings.Contains(w.attempted[1], " acted ") {
t.Fatalf("said %v: the begun attempt, then the finished one", w.attempted)
}
w.attempted = nil
w.fresh()
w.h.sayOpen()
if len(w.attempted) != 1 || !strings.Contains(w.attempted[0], " acted ") {
t.Errorf("at the start every attempt kept is said again: %v", w.attempted)
}
}
// An act waits out its settle, then its spent budget hands the condition to the operator once; an escalated condition
// is passed over; a second instance cannot begin an act already begun; the brake stops every healer past twelve acts
// in an hour and lets go an hour after the last (ADR 0231 §21, ADR 0564 §7).
@@ -22,6 +22,12 @@ type manifest struct {
Invokes []string `json:"invokes"`
Reads []string `json:"reads"`
State []json.RawMessage `json:"state"`
Streams []json.RawMessage `json:"streams"`
Requires []struct {
Name string `json:"name"`
Template string `json:"template"`
Options map[string]any `json:"options"`
} `json:"requires"`
}
// The manifest says what the code does: it claims the seat mesh-healers once for the mesh with its verbs (ADR 0564 §1,
@@ -72,6 +78,19 @@ func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) {
if !states[AttemptsState] || !states[WouldState] || !states[WakesState] || states[FindingsState] {
t.Errorf("the states take a key's own time to live as %v, want attempts, would and wakes alone", states)
}
// Design 47 G §529: the healers are no foundation part, so the state `findings` and the finding stream are asked in
// `requires` (the templates `state` and `events`, the stream named `finding-events`), and the old stream
// `mesh-healers-findings` is declared no more.
got := map[string]string{}
for _, r := range m.Requires {
got[r.Name] = r.Template
if r.Name == "finding-events" && (len(r.Options["events"].([]any)) != 3 || r.Options["max-age-days"] != 7.0 || r.Options["max-mib"] != 16.0) {
t.Errorf("the finding stream's options are %v, want the three finding events, 7 days and 16 MiB", r.Options)
}
}
if got[FindingsState] != "state" || got["finding-events"] != "events" || len(m.Streams) != 0 {
t.Errorf("requires %v and %d old streams, want findings by state, finding-events by events and no old stream", got, len(m.Streams))
}
readBy := map[string]string{AttemptsState: HealersTool, WouldState: WouldTool, FindingsState: FindingsTool,
WakesState: WakesTool, "a settle's end": HealTool,
"an act on an event": HealTool, "S15's count": HealersTool + " recount=true"}