diff --git a/modules/mesh-conditions/cmd/mesh-conditions/catalog.go b/modules/mesh-conditions/cmd/mesh-conditions/catalog.go index 4b9cb39dc..e8b6b2fcd 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/catalog.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/catalog.go @@ -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() } } diff --git a/modules/mesh-conditions/cmd/mesh-conditions/engine_events.go b/modules/mesh-conditions/cmd/mesh-conditions/engine_events.go new file mode 100644 index 000000000..1b50054de --- /dev/null +++ b/modules/mesh-conditions/cmd/mesh-conditions/engine_events.go @@ -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.[.[.]]`. +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 `.`: 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) +} diff --git a/modules/mesh-conditions/cmd/mesh-conditions/events.go b/modules/mesh-conditions/cmd/mesh-conditions/events.go index 458469f9f..4a22ca0b5 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/events.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/events.go @@ -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..event..`, which the runtime delivers as +// `..` (or as `.` with the node in the envelope), and the template form +// `mesh.bus..finding-events..`. 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) diff --git a/modules/mesh-conditions/cmd/mesh-conditions/hold.go b/modules/mesh-conditions/cmd/mesh-conditions/hold.go index 14d787778..22b026b8b 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/hold.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/hold.go @@ -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() diff --git a/modules/mesh-conditions/cmd/mesh-conditions/hold_test.go b/modules/mesh-conditions/cmd/mesh-conditions/hold_test.go index 2bd970e4a..b9e7a82c4 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/hold_test.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/hold_test.go @@ -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}) + } +} diff --git a/modules/mesh-conditions/cmd/mesh-conditions/keeper.go b/modules/mesh-conditions/cmd/mesh-conditions/keeper.go index f1aeb4a3a..179e30b70 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/keeper.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/keeper.go @@ -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 `.`. 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 `.`, 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: (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:. +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) { diff --git a/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go b/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go index a4a3728a0..2c69ba789 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go @@ -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) diff --git a/modules/mesh-conditions/cmd/mesh-conditions/main.go b/modules/mesh-conditions/cmd/mesh-conditions/main.go index 2a72170f7..859732bdc 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/main.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/main.go @@ -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 { diff --git a/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go b/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go index 73d3b0920..62f46aa07 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go @@ -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 diff --git a/modules/mesh-conditions/cmd/mesh-conditions/model.go b/modules/mesh-conditions/cmd/mesh-conditions/model.go index cc229449a..b1c4ef392 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/model.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/model.go @@ -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 } diff --git a/modules/mesh-conditions/cmd/mesh-conditions/owner_test.go b/modules/mesh-conditions/cmd/mesh-conditions/owner_test.go new file mode 100644 index 000000000..d590f9b13 --- /dev/null +++ b/modules/mesh-conditions/cmd/mesh-conditions/owner_test.go @@ -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" +) diff --git a/modules/mesh-conditions/cmd/mesh-conditions/settings.go b/modules/mesh-conditions/cmd/mesh-conditions/settings.go index b29fe7596..46da63b7d 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/settings.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/settings.go @@ -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) } diff --git a/modules/mesh-conditions/cmd/mesh-conditions/tools.go b/modules/mesh-conditions/cmd/mesh-conditions/tools.go index 621fe6f46..6185ffb64 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/tools.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/tools.go @@ -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:; " + + "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") diff --git a/modules/mesh-conditions/module.json b/modules/mesh-conditions/module.json index a59a68797..7079970a7 100644 --- a/modules/mesh-conditions/module.json +++ b/modules/mesh-conditions/module.json @@ -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.", diff --git a/modules/mesh-healers/cmd/mesh-healers/healers.go b/modules/mesh-healers/cmd/mesh-healers/healers.go index f1ad1caf6..b181959ca 100644 --- a/modules/mesh-healers/cmd/mesh-healers/healers.go +++ b/modules/mesh-healers/cmd/mesh-healers/healers.go @@ -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) diff --git a/modules/mesh-healers/cmd/mesh-healers/healers_test.go b/modules/mesh-healers/cmd/mesh-healers/healers_test.go index 8b1413991..b3c6a811c 100644 --- a/modules/mesh-healers/cmd/mesh-healers/healers_test.go +++ b/modules/mesh-healers/cmd/mesh-healers/healers_test.go @@ -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). diff --git a/modules/mesh-healers/cmd/mesh-healers/manifest_test.go b/modules/mesh-healers/cmd/mesh-healers/manifest_test.go index 83f59216b..e1c1989cf 100644 --- a/modules/mesh-healers/cmd/mesh-healers/manifest_test.go +++ b/modules/mesh-healers/cmd/mesh-healers/manifest_test.go @@ -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"}