diff --git a/modules/mesh-conditions/cmd/mesh-conditions/homes.go b/modules/mesh-conditions/cmd/mesh-conditions/homes.go index 06ad8187..61e5acbd 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/homes.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/homes.go @@ -86,8 +86,7 @@ var Homes = map[string]Home{ "healer-wanted": {Source: "S15", Controller: true, Observer: healers}, "healers-braked": {Source: "healers", Controller: true, Observer: healers}, // The assigner's, as its own findings (the bus step, ADR 0565 §9). - "bus-maintenance": {Source: "bus", Controller: true, Observer: assigner}, - "bus-upgrade-failed": {Source: "DB", Controller: true, Observer: assigner}, + "bus-maintenance": {Source: "bus", Controller: true, Observer: assigner}, // The provisioner's (ADR 0570 §5): its own ask replaces retire-waiting; a declined key stays its finding. "retire-waiting": {Source: "retire", Controller: true, Observer: provisioner}, "retire-rejected": {Source: "retire", Controller: true, Observer: provisioner}, @@ -116,6 +115,10 @@ var Homes = map[string]Home{ "declaration-refused": {Source: "D1", Controller: true, Carried: true}, "unknown-field": {Source: "read", Controller: true, Carried: true}, "bus-objects-unasserted": {Source: "send", Controller: true, Carried: true}, + // The controller's own bus upgrade (`bus.nats.upgrade-failed`, its DB step): it ends with the controller's lease and + // is carried until then, approved by the operator on 2026-10-31 (issue 1952); ADR 0564's table had homed it with the + // assigner, which writes no such finding, and the lead records the change of the record. + "bus-upgrade-failed": {Source: "DB", Controller: true, Carried: true}, // The probes the operator dropped from this release (D2 resolvers, D3 holders, D4 archives, D12 bindings, D-heard // withheld hearings, D-root, D11 retired data, and D13's data held twice): their kinds are carried until the lease // write. diff --git a/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go b/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go index 475f8ec5..6d2b16f3 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go @@ -281,6 +281,7 @@ func TestCompareCountsOnlyWhatBothSidesRaise(t *testing.T) { _ = r.k.Controller("delivery.x.mystery", put("delivery.x.mystery", "mystery"), false) _ = r.k.Controller("bus.x.consumer-wrong", put("bus.x.consumer-wrong", "consumer-wrong"), false) _ = r.k.Controller("seat.vault.ace.holder-silent", put("seat.vault.ace.holder-silent", "holder-silent"), false) + _ = r.k.Controller("bus.nats.upgrade-failed", put("bus.nats.upgrade-failed", "bus-upgrade-failed"), false) cmp, err := r.k.Compare() if err != nil { t.Fatal(err) @@ -296,6 +297,8 @@ func TestCompareCountsOnlyWhatBothSidesRaise(t *testing.T) { "bus.x.consumer-wrong": "mesh-self-check has not written its finding", // A kind of a probe the operator dropped (2026-10-30) is carried, never missing. "seat.vault.ace.holder-silent": "carried from its bucket until the lease write (dropped 2026-10-30 by the operator", + // The controller's own bus upgrade ends with its lease (the operator, 2026-10-31; issue 1952). + "bus.nats.upgrade-failed": "ends with the controller", } { if !strings.Contains(reasons[key], says) { t.Errorf("%s: %q, want it to say %q", key, reasons[key], says) @@ -313,7 +316,7 @@ func TestCompareCountsOnlyWhatBothSidesRaise(t *testing.T) { } } for kind, observer := range map[string]string{"needs-operator": watchman, "machine-network": watchman, "lease-split": selfCheck, - "data-shrank": selfCheck, "bus-upgrade-failed": assigner, "retire-waiting": provisioner, "asks-undelivered": messenger, + "data-shrank": selfCheck, "retire-waiting": provisioner, "asks-undelivered": messenger, "image-missing": artifactStore} { if h, ok := Homes[kind]; !ok || h.Observer != observer || h.Carried { t.Errorf("%s's home is %+v", kind, h) @@ -325,7 +328,8 @@ func TestCompareCountsOnlyWhatBothSidesRaise(t *testing.T) { t.Errorf("the dropped probe's kind %s is %+v", kind, h) } } - if len(Homes) != 84 || carried != 33 { // 19 of ADR 0564, the 63 of the table, plan-failed and stalled + // And the controller's own bus upgrade, carried (the operator, 2026-10-31): 34. + if len(Homes) != 84 || carried != 34 { // 19 of ADR 0564, the 63 of the table, plan-failed and stalled t.Errorf("the table holds %d kinds, %d carried", len(Homes), carried) } if _, err := r.k.TakeOver(nil, "jochen", "now"); err == nil || !strings.Contains(err.Error(), "2 difference") { diff --git a/modules/mesh-conditions/go.mod b/modules/mesh-conditions/go.mod index 49955684..02410641 100644 --- a/modules/mesh-conditions/go.mod +++ b/modules/mesh-conditions/go.mod @@ -3,7 +3,7 @@ module mesh-conditions go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031084708-acbf80ad910a + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031203031-cbbfbf3d2d55 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-conditions/go.sum b/modules/mesh-conditions/go.sum index 8e0c33a9..5a992d56 100644 --- a/modules/mesh-conditions/go.sum +++ b/modules/mesh-conditions/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031084708-acbf80ad910a h1:DcSvfEmdPBKqCg5OmXRNb269EkkPe7BsEp+RykEhkgE= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031084708-acbf80ad910a/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031203031-cbbfbf3d2d55 h1:OcKTrlH72cPwAHuyAwyRLjGz0qt4VHTeal4MeahPnBM= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031203031-cbbfbf3d2d55/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-healers/cmd/mesh-healers/healers.go b/modules/mesh-healers/cmd/mesh-healers/healers.go index 425eba47..3d7d5ff4 100644 --- a/modules/mesh-healers/cmd/mesh-healers/healers.go +++ b/modules/mesh-healers/cmd/mesh-healers/healers.go @@ -544,7 +544,7 @@ func brakeFinding(acts []Attempt, said string) Observation { most = append(most, fmt.Sprintf("%s ×%d", what, n)) } sort.Strings(most) - return Observation{Kind: "healers-braked", Token: "braked", Severity: Urgent, Subject: Subject{Scope: ScopeMesh, ID: "healers"}, + return Observation{Kind: "healers-braked", Ref: "braked", Severity: Urgent, Subject: Subject{Scope: ScopeMesh, ID: "healers"}, Summary: "every healer has stopped: the healers acted more often in an hour than the mesh allows, which is a " + "healer looping, not repairing — " + said, Said: strings.Join(most, ", "), @@ -557,7 +557,7 @@ func brakeFinding(acts []Attempt, said string) Observation { // blind writes the healers' own blindness: they cannot read what they did. func (h *Healers) blind(err error) { - h.see("blind", []Observation{{Kind: "probe-failed", Token: "failed", Severity: Warning, + h.see("blind", []Observation{{Kind: "probe-failed", Ref: "failed", Severity: Warning, Subject: Subject{Scope: ScopeProbe, ID: "healers"}, Summary: "the healers cannot read what they did, so they count no budget and act on nothing", Said: firstLine(err.Error()), Headline: "A check of the mesh could not run", diff --git a/modules/mesh-healers/cmd/mesh-healers/observation.go b/modules/mesh-healers/cmd/mesh-healers/observation.go index 853550b5..b2ef411d 100644 --- a/modules/mesh-healers/cmd/mesh-healers/observation.go +++ b/modules/mesh-healers/cmd/mesh-healers/observation.go @@ -25,7 +25,7 @@ type Subject = health.Subject // Observation is one thing a probe sees wrong now. type Observation struct { Kind string - Token string + Ref string Subject Subject Severity health.Severity Summary string @@ -40,7 +40,7 @@ type Observation struct { // Key is where it is kept: `..`, the controller's key for the same condition. func (o Observation) Key() string { - token := o.Token + token := o.Ref if token == "" { token = o.Kind } @@ -52,8 +52,8 @@ func (o Observation) Finding(run string, at time.Time) health.Finding { f := health.Finding{Key: o.Key(), Kind: o.Kind, Subject: o.Subject, Severity: o.Severity, Summary: o.Summary, Headline: o.Headline, Explanation: o.Explanation, Resolved: o.Resolved, Needs: o.Needs, Source: o.Source, Observer: Module, Run: run, At: at.UTC(), Evidence: []health.Evidence{}} - if o.Token != "" && o.Token != o.Kind { - f.Token = o.Token + if o.Ref != "" && o.Ref != o.Kind { + f.Ref = o.Ref } said := o.Said if said == "" { diff --git a/modules/mesh-healers/cmd/mesh-healers/wanted.go b/modules/mesh-healers/cmd/mesh-healers/wanted.go index 065f2c41..84e30161 100644 --- a/modules/mesh-healers/cmd/mesh-healers/wanted.go +++ b/modules/mesh-healers/cmd/mesh-healers/wanted.go @@ -135,7 +135,7 @@ func (h *Healers) Wanted() []Observation { } } words := strings.NewReplacer(".", " ", "_", " ").Replace(cause) - found = append(found, Observation{Kind: "healer-wanted", Token: "healer-wanted", Severity: Warning, + found = append(found, Observation{Kind: "healer-wanted", Ref: "healer-wanted", Severity: Warning, Subject: Subject{Scope: ScopeMesh, ID: "hand-acts." + cause}, Summary: fmt.Sprintf("%q was repaired by hand %d times in %d days, the last by %s: %s", cause, len(acts), int(Within.Hours()/24), newest.By, wanted), diff --git a/modules/mesh-healers/go.mod b/modules/mesh-healers/go.mod index d0ba8bc3..051852ac 100644 --- a/modules/mesh-healers/go.mod +++ b/modules/mesh-healers/go.mod @@ -3,7 +3,7 @@ module mesh-healers go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030174137-a37181261c24 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031203031-cbbfbf3d2d55 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-healers/go.sum b/modules/mesh-healers/go.sum index 61230834..5a992d56 100644 --- a/modules/mesh-healers/go.sum +++ b/modules/mesh-healers/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030174137-a37181261c24 h1:s0XXZtR7CFPBbnaDygWnE4/Q827iEFt0YfEzujk4Q9g= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030174137-a37181261c24/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031203031-cbbfbf3d2d55 h1:OcKTrlH72cPwAHuyAwyRLjGz0qt4VHTeal4MeahPnBM= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031203031-cbbfbf3d2d55/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-self-check/cmd/mesh-self-check/main.go b/modules/mesh-self-check/cmd/mesh-self-check/main.go index 6d3d53f8..71fefc4b 100644 --- a/modules/mesh-self-check/cmd/mesh-self-check/main.go +++ b/modules/mesh-self-check/cmd/mesh-self-check/main.go @@ -37,6 +37,13 @@ func (runtimeStore) Delete(state, key string) error { return stdio.State(state). func (runtimeStore) Keys(state string) ([]string, error) { return stdio.State(state).Keys() } +func (runtimeStore) Watch(state string, handle func(key string, value []byte, gone bool)) error { + return stdio.State(state).Watch("", func(ch stdio.StateChange) error { + handle(ch.Key, ch.Value, ch.Op != stdio.OpPut) + return nil + }) +} + // NewChecker is the self-check over the runtime. func NewChecker() *Checker { return &Checker{Store: runtimeStore{}, Registry: Registry(), Now: time.Now, Logf: logf, Host: os.Getenv("MESH_NODE"), @@ -49,6 +56,7 @@ func main() { c := NewChecker() go func() { <-stdio.Served() + c.WatchKept(context.Background()) c.Keep(context.Background()) }() if err := stdio.Serve("", Tools(c)); err != nil { @@ -57,7 +65,7 @@ func main() { } } -// enabledProbes is the setting `probes`, read at each run so a change applies at the next; with it unreadable, the +// enabledProbes is the setting `probes`, read once at each run so a change applies at the next; with it unreadable, the // default list. func enabledProbes() []string { raw, err := stdio.Config("probes") diff --git a/modules/mesh-self-check/cmd/mesh-self-check/observation.go b/modules/mesh-self-check/cmd/mesh-self-check/observation.go index 71209ce9..ac535f44 100644 --- a/modules/mesh-self-check/cmd/mesh-self-check/observation.go +++ b/modules/mesh-self-check/cmd/mesh-self-check/observation.go @@ -28,7 +28,7 @@ type Subject = health.Subject // Observation is one thing a probe sees wrong now. type Observation struct { Kind string - Token string + Ref string Subject Subject Severity health.Severity Summary string @@ -45,7 +45,7 @@ type Observation struct { // Key is where it is kept: `..`, the controller's key for the same condition. func (o Observation) Key() string { - token := o.Token + token := o.Ref if token == "" { token = o.Kind } @@ -57,8 +57,8 @@ func (o Observation) Finding(run string, at time.Time) health.Finding { f := health.Finding{Key: o.Key(), Kind: o.Kind, Subject: o.Subject, Severity: o.Severity, Summary: o.Summary, Headline: o.Headline, Explanation: o.Explanation, Resolved: o.Resolved, Needs: o.Needs, Source: o.Source, Observer: Module, Run: run, At: at.UTC(), Evidence: []health.Evidence{}} - if o.Token != "" && o.Token != o.Kind { - f.Token = o.Token + if o.Ref != "" && o.Ref != o.Kind { + f.Ref = o.Ref } said := o.Said if said == "" { diff --git a/modules/mesh-self-check/cmd/mesh-self-check/probes.go b/modules/mesh-self-check/cmd/mesh-self-check/probes.go index e4dbce45..5e0d3db3 100644 --- a/modules/mesh-self-check/cmd/mesh-self-check/probes.go +++ b/modules/mesh-self-check/cmd/mesh-self-check/probes.go @@ -180,21 +180,21 @@ func probeConsumers(ctx context.Context, c *Checker) ([]Observation, error) { who := want.Stream + "." + want.Name have, ok := held[want.Name] if !ok { - out = append(out, Observation{Kind: "consumer-lost", Token: "missing", Severity: Warning, + out = append(out, Observation{Kind: "consumer-lost", Ref: "missing", Severity: Warning, Subject: Subject{Scope: ScopeBus, ID: who}, Summary: fmt.Sprintf("%s is not on the bus: what it delivers reaches nobody", consumerWords(want)), Said: "no consumer " + want.Name + " on " + want.Stream}) continue } if differs := consumerDiffers(want, have); differs != "" { - out = append(out, Observation{Token: "redefined", Severity: Warning, Subject: Subject{Scope: ScopeBus, ID: who}, + out = append(out, Observation{Ref: "redefined", Severity: Warning, Subject: Subject{Scope: ScopeBus, ID: who}, Summary: fmt.Sprintf("%s is not as the mesh defines it: %s", consumerWords(want), differs), Said: differs}) } if want.Stream != "NODES" && have.Pending > behindBound { // One look at a consumer a burst put behind is not one that stays behind: held for the next run, // so healer H4 resets nothing on a single sample (novox/hq issue 277). - out = append(out, Observation{Kind: "consumer-behind", Token: "behind", Severity: Warning, Confirm: true, + out = append(out, Observation{Kind: "consumer-behind", Ref: "behind", Severity: Warning, Confirm: true, Subject: Subject{Scope: ScopeBus, ID: who}, Summary: fmt.Sprintf("%s is %d message(s) behind its stream's head", consumerWords(want), have.Pending), Said: fmt.Sprintf("%d pending, %d handed out and not settled", have.Pending, have.Unacked)}) @@ -238,13 +238,13 @@ func probeStreams(ctx context.Context, c *Checker) ([]Observation, error) { for _, s := range buscompose.MeshStreams() { have, ok := held[s.Name] if !ok { - out = append(out, Observation{Token: "stream-missing", Severity: Urgent, Subject: Subject{Scope: ScopeBus, ID: s.Name}, + out = append(out, Observation{Ref: "stream-missing", Severity: Urgent, Subject: Subject{Scope: ScopeBus, ID: s.Name}, Summary: fmt.Sprintf("the stream %s is not on the bus: %s", s.Name, firstLine(s.Why)), Said: "no stream " + s.Name}) continue } if differs := streamDiffers(s, have); differs != "" { - out = append(out, Observation{Token: "stream-redefined", Severity: Warning, Subject: Subject{Scope: ScopeBus, ID: s.Name}, + out = append(out, Observation{Ref: "stream-redefined", Severity: Warning, Subject: Subject{Scope: ScopeBus, ID: s.Name}, Summary: fmt.Sprintf("the stream %s is not as the mesh defines it: %s", s.Name, differs), Said: differs}) } @@ -377,7 +377,7 @@ func busUserWords(user string) (string, string) { // watchman's pulse on `checked`, is the other half. func probeWatchdogs(_ context.Context, c *Checker) ([]Observation, error) { if _, err := c.Ask(SignalsVerb, map[string]any{}); err != nil { - return []Observation{{Token: "silent", Severity: Urgent, Subject: Subject{Scope: ScopeCore, ID: "watchdogs"}, + return []Observation{{Ref: "silent", Severity: Urgent, Subject: Subject{Scope: ScopeCore, ID: "watchdogs"}, Summary: "the watchdogs do not answer: the watchman holding mesh-watchdogs did not answer signals, so no " + "late signal is being said", Said: firstLine(err.Error())}}, nil @@ -435,7 +435,7 @@ func probeBusHealth(ctx context.Context, c *Checker) ([]Observation, error) { if len(problems) == 0 { return nil, nil } - return []Observation{{Token: "unhealthy", Severity: Urgent, Subject: Subject{Scope: ScopeBus, ID: "mesh"}, + return []Observation{{Ref: "unhealthy", Severity: Urgent, Subject: Subject{Scope: ScopeBus, ID: "mesh"}, Summary: "the bus is not healthy: " + strings.Join(problems, "; "), Said: strings.Join(problems, "; ")}}, nil } diff --git a/modules/mesh-self-check/cmd/mesh-self-check/probes_data.go b/modules/mesh-self-check/cmd/mesh-self-check/probes_data.go index 98d8146f..cd0c025a 100644 --- a/modules/mesh-self-check/cmd/mesh-self-check/probes_data.go +++ b/modules/mesh-self-check/cmd/mesh-self-check/probes_data.go @@ -12,8 +12,11 @@ import ( "context" "encoding/json" "fmt" + "maps" + "slices" "sort" "strings" + "sync" "time" "git.novox.be/novox/mesh-sdk/go/catalogue" @@ -57,51 +60,98 @@ type dataFacts struct { now time.Time } -// dataFacts reads them from the bus. -func (c *Checker) dataFacts() (dataFacts, error) { +// keptStates are the states D13 judges from memory (issue 1952): read one key at a time through the runtime, each +// read reopening its bucket, a run of D13 never finished within its bound. Each is watched once and kept as it changes. +var keptStates = []string{ReadingsRead, GoodRead, ModulesRead, ManifestsRead} + +// view is one state kept in memory from its watch: ready once the watch handed over every current value, err when it +// was refused (then the next read asks again). +type view struct { + mu sync.Mutex + values map[string][]byte + ready chan struct{} + err error +} + +// WatchKept starts watching every state D13 reads, so the first run judges from memory too. Called once the bundle serves. +func (c *Checker) WatchKept(ctx context.Context) { + for _, s := range keptStates { + go func(s string) { _, _ = c.kept(ctx, s) }(s) + } +} + +// kept is every current value of the state, from its watch, which it starts on the first ask. It waits for the watch's +// current values within ctx; a state not yet kept, or one whose watch was refused, is an error, never an empty state. +func (c *Checker) kept(ctx context.Context, state string) (map[string][]byte, error) { + c.viewsMu.Lock() + v := c.views[state] + if v == nil { + v = &view{values: map[string][]byte{}, ready: make(chan struct{})} + if c.views == nil { + c.views = map[string]*view{} + } + c.views[state] = v + go func() { + err := c.Store.Watch(state, func(key string, value []byte, gone bool) { + v.mu.Lock() + defer v.mu.Unlock() + if gone { + delete(v.values, key) + } else { + v.values[key] = value + } + }) + if err != nil { + c.viewsMu.Lock() + if c.views[state] == v { + delete(c.views, state) + } + c.viewsMu.Unlock() + v.err = err + } + close(v.ready) + }() + } + c.viewsMu.Unlock() + select { + case <-v.ready: + case <-ctx.Done(): + return nil, fmt.Errorf("the state %s is not kept yet: its watch has not handed over every value (%v)", state, ctx.Err()) + } + if v.err != nil { + return nil, fmt.Errorf("the state %s cannot be watched: %w", state, v.err) + } + v.mu.Lock() + defer v.mu.Unlock() + return maps.Clone(v.values), nil +} + +// dataFacts reads them from the states the self-check keeps by watching them. +func (c *Checker) dataFacts(ctx context.Context) (dataFacts, error) { f := dataFacts{good: map[string]time.Time{}, manifests: map[string]catalogue.Manifest{}, now: c.Now()} - keys, err := c.Store.Keys(ReadingsRead) - if err != nil { - return f, fmt.Errorf("disk-load's readings cannot be read: %w", err) + states := map[string]map[string][]byte{} + for _, s := range keptStates { + values, err := c.kept(ctx, s) + if err != nil { + return f, err + } + states[s] = values } byNode := map[string]int{} - for _, k := range keys { - raw, ok, err := c.Store.Get(ReadingsRead, k) - if err != nil { - return f, fmt.Errorf("the reading %s cannot be read: %w", k, err) - } - if !ok { - continue - } - r, err := health.DecodeReading(raw) + for _, k := range slices.Sorted(maps.Keys(states[ReadingsRead])) { + r, err := health.DecodeReading(states[ReadingsRead][k]) if err != nil { continue } f.readings = append(f.readings, r) byNode[r.Node]++ } - gkeys, err := c.Store.Keys(GoodRead) - if err != nil { - return f, fmt.Errorf("restic's last good nights cannot be read: %w", err) - } - for _, k := range gkeys { - raw, ok, err := c.Store.Get(GoodRead, k) - if err != nil { - return f, fmt.Errorf("the last good night %s cannot be read: %w", k, err) - } - if g, derr := health.DecodeGoodRow(raw); ok && derr == nil && !g.LastGood.IsZero() { + for _, raw := range states[GoodRead] { + if g, derr := health.DecodeGoodRow(raw); derr == nil && !g.LastGood.IsZero() { f.good[health.GoodKey(g.Node, g.Module, g.Item)] = g.LastGood } } - mods, err := c.Store.Keys(ModulesRead) - if err != nil { - return f, fmt.Errorf("the indexer's modules cannot be read: %w", err) - } - for _, name := range mods { - raw, ok, err := c.Store.Get(ModulesRead, name) - if err != nil || !ok { - continue - } + for name, raw := range states[ModulesRead] { // Only the manifest's digest of the indexer's module, and the manifest's text: what the bounds are read from. var m struct { Manifest string `json:"manifest"` @@ -109,8 +159,8 @@ func (c *Checker) dataFacts() (dataFacts, error) { if json.Unmarshal(raw, &m) != nil || m.Manifest == "" { continue } - mraw, ok, err := c.Store.Get(ManifestsRead, m.Manifest) - if err != nil || !ok { + mraw, ok := states[ManifestsRead][m.Manifest] + if !ok { continue } var doc struct { @@ -142,7 +192,7 @@ func (c *Checker) dataFacts() (dataFacts, error) { // probeData is D13. func probeData(ctx context.Context, c *Checker) ([]Observation, error) { - f, err := c.dataFacts() + f, err := c.dataFacts(ctx) if err != nil { return nil, err } @@ -213,7 +263,7 @@ func firstSeen(r health.Reading, now time.Time) time.Time { } func obs(kind string, id, machine string, sev health.Severity, summary, said string) Observation { - return Observation{Kind: kind, Token: kind, Severity: sev, Summary: summary, Said: said, + return Observation{Kind: kind, Ref: kind, Severity: sev, Summary: summary, Said: said, Subject: Subject{Scope: ScopeMachine, ID: id, Machine: machine}} } @@ -251,7 +301,7 @@ func dataFindings(f dataFacts) []Observation { "`cleanup delete %s %s %s --why …` once a person has decided, or assign %s there again", r.ID, r.Module, r.Node, r.Class, sizeWords(r.SizeBytes), int(f.now.Sub(r.RetiredAt).Hours()/24), r.RetiredWhy, r.Node, r.Module, r.ID, r.Module), "kept at "+orUnknownPath(r.Path))) - out[len(out)-1].Token = "cleanup" + out[len(out)-1].Ref = "cleanup" } continue } diff --git a/modules/mesh-self-check/cmd/mesh-self-check/probes_fill.go b/modules/mesh-self-check/cmd/mesh-self-check/probes_fill.go index a868ab33..edeaba27 100644 --- a/modules/mesh-self-check/cmd/mesh-self-check/probes_fill.go +++ b/modules/mesh-self-check/cmd/mesh-self-check/probes_fill.go @@ -54,7 +54,7 @@ func mibWords(b uint64) string { return fmt.Sprintf("%.1f MiB", float64(b)/(1024 // fillKey is where a bucket's or log's fill finding is kept. func fillKey(stream string) string { - return Observation{Subject: Subject{Scope: ScopeBus, ID: stream}, Token: "filling"}.Key() + return Observation{Subject: Subject{Scope: ScopeBus, ID: stream}, Ref: "filling"}.Key() } // fillObservation is what one bucket's or log's fill says: a finding at fillRaiseAt or above, and, while its finding @@ -71,7 +71,7 @@ func fillObservation(stream string, r fillRow, open bool) (Observation, bool) { return Observation{}, false } pct := r.Bytes * 100 / r.MaxBytes - return Observation{Kind: "bucket-or-log-filling", Token: "filling", Severity: Warning, + return Observation{Kind: "bucket-or-log-filling", Ref: "filling", Severity: Warning, Subject: Subject{Scope: ScopeBus, ID: stream}, Summary: fmt.Sprintf("%s's %s %s holds %s of %s, %d%%: at its cap the bus refuses the module's next write", module, kind, name, mibWords(r.Bytes), mibWords(r.MaxBytes), pct), diff --git a/modules/mesh-self-check/cmd/mesh-self-check/probes_mesh.go b/modules/mesh-self-check/cmd/mesh-self-check/probes_mesh.go index efb40831..af57290a 100644 --- a/modules/mesh-self-check/cmd/mesh-self-check/probes_mesh.go +++ b/modules/mesh-self-check/cmd/mesh-self-check/probes_mesh.go @@ -197,7 +197,7 @@ func probeBans(ctx context.Context, c *Checker) ([]Observation, error) { } } if len(banned) > 0 { - out = append(out, Observation{Kind: "own-address-banned", Token: "bans-the-mesh", Severity: Urgent, + out = append(out, Observation{Kind: "own-address-banned", Ref: "bans-the-mesh", Severity: Urgent, Subject: Subject{Scope: ScopeMachine, ID: node, Machine: node}, Summary: fmt.Sprintf("%s's ban list holds %d of the mesh's own address(es) — %s: the mesh is locked out "+ "of itself there (ADR 0186)", node, len(banned), strings.Join(whose, ", ")), diff --git a/modules/mesh-self-check/cmd/mesh-self-check/probes_root.go b/modules/mesh-self-check/cmd/mesh-self-check/probes_root.go index 2d46fba4..c52c72ff 100644 --- a/modules/mesh-self-check/cmd/mesh-self-check/probes_root.go +++ b/modules/mesh-self-check/cmd/mesh-self-check/probes_root.go @@ -230,7 +230,7 @@ func probeAgentAccounts(_ context.Context, c *Checker) ([]Observation, error) { if v.Confined || v.Quiet { continue } - out = append(out, Observation{Kind: "agent-can-become-root", Token: "agent-root", Severity: Urgent, + out = append(out, Observation{Kind: "agent-can-become-root", Ref: "agent-root", Severity: Urgent, Subject: Subject{Scope: ScopeMachine, ID: n, Machine: n}, Summary: fmt.Sprintf("on %s, %s (ADR 0266): an agent there may become root without a person, and no "+ "answer from a channel authorises an act there (ADR 0259 §8)", n, v.Why), diff --git a/modules/mesh-self-check/cmd/mesh-self-check/probes_test.go b/modules/mesh-self-check/cmd/mesh-self-check/probes_test.go index cb083b0f..e1789f93 100644 --- a/modules/mesh-self-check/cmd/mesh-self-check/probes_test.go +++ b/modules/mesh-self-check/cmd/mesh-self-check/probes_test.go @@ -236,6 +236,15 @@ func TestTheBusProbes(t *testing.T) { c.Store = newStore() } found, err := tc.probe(context.Background(), c) + // D13 judges the states it reads from memory, kept by watching them, never one key at a time through the runtime, + // which did not finish within its bound (issue 1952). + if strings.HasPrefix(tc.name, "D13") { + for _, s := range []string{ReadingsRead, GoodRead, ModulesRead, ManifestsRead} { + if n := tc.mesh.gets[s]; n > 0 { + t.Errorf("%s: D13 read %s one key at a time %d times", tc.name, s, n) + } + } + } if tc.err != "" { if err == nil || !strings.Contains(err.Error(), tc.err) { t.Errorf("%s: error %v, want %q", tc.name, err, tc.err) @@ -250,6 +259,19 @@ func TestTheBusProbes(t *testing.T) { t.Errorf("%s: found %v, want %v", tc.name, got, tc.want) } } + // What changes after the watch began is judged at the next run, from memory: every reading gone is seen. + mesh := dataMesh(func(*dataRig) {}) + c := &Checker{Store: mesh, Now: func() time.Time { return dataNow }} + if found, err := probeData(context.Background(), c); err != nil || len(found) != 0 { + t.Fatalf("D13 on a held mesh: %v, %v", keysOf(found), err) + } + keys, _ := mesh.Keys(ReadingsRead) + for _, k := range keys { + _ = mesh.Delete(ReadingsRead, k) + } + if found, err := probeData(context.Background(), c); err != nil || !slices.Contains(keysOf(found), "machine.ace.data-unmeasured") { + t.Errorf("D13 after every reading went found %v, %v", keysOf(found), err) + } } // machines is a mesh of three: novox the hub with its endpoint, ace and g14 on the private network; ace and novox @@ -302,7 +324,7 @@ func placed(s *memStore) map[string]delivery.Placement { // filling is a store holding the self-check's written fill finding of a stream: the condition open. func filling(stream string) *memStore { s := newStore() - o := Observation{Kind: "bucket-or-log-filling", Token: "filling", Severity: Warning, Subject: Subject{Scope: ScopeBus, ID: stream}, + o := Observation{Kind: "bucket-or-log-filling", Ref: "filling", Severity: Warning, Subject: Subject{Scope: ScopeBus, ID: stream}, Summary: "filling", Source: "D-fill"} raw, _ := health.EncodeFinding(o.Finding("run", time.Now())) _ = s.Put(FindingsState, o.Key(), json.RawMessage(raw)) diff --git a/modules/mesh-self-check/cmd/mesh-self-check/selfcheck.go b/modules/mesh-self-check/cmd/mesh-self-check/selfcheck.go index 174ff884..12e77dfb 100644 --- a/modules/mesh-self-check/cmd/mesh-self-check/selfcheck.go +++ b/modules/mesh-self-check/cmd/mesh-self-check/selfcheck.go @@ -62,14 +62,22 @@ type Store interface { Put(state, key string, value any) error Delete(state, key string) error Keys(state string) ([]string, error) + // Watch hands every current value of the state to handle, returning once all were handed, and then every change + // for as long as the process runs; gone is a key deleted or expired. + Watch(state string, handle func(key string, value []byte, gone bool)) error } // On says whether the setting lists the probe. -func (c *Checker) On(id string) bool { +func (c *Checker) On(id string) bool { return c.listed()(id) } + +// listed reads the setting `probes` once and answers for each probe from that read: a run or an answer of the verb asks +// the runtime for it once, not per probe (issue 1952). +func (c *Checker) listed() func(id string) bool { if c.Enabled == nil { - return true + return func(string) bool { return true } } - return slices.Contains(c.Enabled(), id) + ids := c.Enabled() + return func(id string) bool { return slices.Contains(ids, id) } } // DefaultProbes are the probes that run unless the setting says otherwise. @@ -121,6 +129,9 @@ type Checker struct { unsaid int // pending is when each machine's agent account was first seen waiting for its setuid search (DA, D-root). pending pendingSince + // views are the states D13 keeps in memory by watching them (probes_data.go kept). + viewsMu sync.Mutex + views map[string]*view } // LastRun is the last run, nil before the first. @@ -149,10 +160,11 @@ func (c *Checker) RunOnce(ctx context.Context, why string) Run { run := Run{Run: fmt.Sprintf("doctor-%d-%d", started.Unix(), n), Started: started.UTC(), IntervalSeconds: int(Every / time.Second), Controller: c.Host, Why: why} + on := c.listed() results := make([]result, len(c.Registry)) var wg sync.WaitGroup for i, p := range c.Registry { - if p.Run == nil || !c.On(p.ID) { + if p.Run == nil || !on(p.ID) { continue } wg.Add(1) @@ -169,7 +181,7 @@ func (c *Checker) RunOnce(ctx context.Context, why string) Run { ran := map[string]bool{} // the sources whose findings this run judged whole var blind []Observation for i, p := range c.Registry { - if !c.On(p.ID) { + if !on(p.ID) { // Off by the setting: it ran nothing and sees nothing, so what it raised before is cleared, and it has no // verdict in the run. ran[p.ID] = true @@ -185,7 +197,7 @@ func (c *Checker) RunOnce(ctx context.Context, why string) Run { case r.err != nil: v.Verdict, v.Error = VerdictFailedToRun, r.err.Error() run.Counts.FailedToRun++ - blind = append(blind, Observation{Kind: "probe-failed", Token: "failed", Severity: Warning, Confirm: true, + blind = append(blind, Observation{Kind: "probe-failed", Ref: "failed", Severity: Warning, Confirm: true, Subject: Subject{Scope: ScopeProbe, ID: p.ID}, Source: sourceSelf, Summary: fmt.Sprintf("the probe %s (%s) could not run: what it checks is not known — never a pass", p.ID, p.Asserts), Said: firstLine(r.err.Error())}) diff --git a/modules/mesh-self-check/cmd/mesh-self-check/selfcheck_test.go b/modules/mesh-self-check/cmd/mesh-self-check/selfcheck_test.go index 50557340..09fa8f5c 100644 --- a/modules/mesh-self-check/cmd/mesh-self-check/selfcheck_test.go +++ b/modules/mesh-self-check/cmd/mesh-self-check/selfcheck_test.go @@ -15,25 +15,40 @@ import ( "git.novox.be/novox/mesh-sdk/go/health" ) -// memStore is the bus's buckets in memory, counting the writes. +// memStore is the bus's buckets in memory, counting the writes and the reads one key at a time, and handing every +// change to its watches. type memStore struct { mu sync.Mutex kv map[string]map[string][]byte puts map[string]int + gets map[string]int + watches map[string][]func(key string, value []byte, gone bool) keysErr error } func newStore() *memStore { - return &memStore{kv: map[string]map[string][]byte{}, puts: map[string]int{}} + return &memStore{kv: map[string]map[string][]byte{}, puts: map[string]int{}, gets: map[string]int{}, + watches: map[string][]func(string, []byte, bool){}} } func (s *memStore) Get(state, key string) ([]byte, bool, error) { s.mu.Lock() defer s.mu.Unlock() + s.gets[state]++ v, ok := s.kv[state][key] return v, ok, nil } +func (s *memStore) Watch(state string, handle func(key string, value []byte, gone bool)) error { + s.mu.Lock() + defer s.mu.Unlock() + for k, v := range s.kv[state] { + handle(k, v, false) + } + s.watches[state] = append(s.watches[state], handle) + return nil +} + func (s *memStore) Put(state, key string, value any) error { raw, err := json.Marshal(value) if err != nil { @@ -46,6 +61,9 @@ func (s *memStore) Put(state, key string, value any) error { } s.kv[state][key] = raw s.puts[state+"/"+key]++ + for _, w := range s.watches[state] { + w(key, raw, false) + } return nil } @@ -53,6 +71,9 @@ func (s *memStore) Delete(state, key string) error { s.mu.Lock() defer s.mu.Unlock() delete(s.kv[state], key) + for _, w := range s.watches[state] { + w(key, nil, true) + } return nil } @@ -97,9 +118,9 @@ func TestARunWritesFindingsAsTheyChange(t *testing.T) { var fail error steady := Probe{ID: "P1", Kind: "stream-wrong", Run: func(context.Context, *Checker) ([]Observation, error) { return found, fail }} deferred := Probe{ID: "P2", Kind: "resolver-wrong", Deferred: "the overlay's addresses"} - missing := Observation{Token: "stream-missing", Severity: Urgent, Subject: Subject{Scope: ScopeBus, ID: "EVENTS"}, + missing := Observation{Ref: "stream-missing", Severity: Urgent, Subject: Subject{Scope: ScopeBus, ID: "EVENTS"}, Summary: "the stream EVENTS is not on the bus"} - behind := Observation{Kind: "consumer-behind", Token: "behind", Severity: Warning, Confirm: true, + behind := Observation{Kind: "consumer-behind", Ref: "behind", Severity: Warning, Confirm: true, Subject: Subject{Scope: ScopeBus, ID: "EVENTS.controller"}, Summary: "far behind"} store := newStore() c, said := checker(store, steady, deferred) @@ -172,7 +193,7 @@ func TestARunWritesFindingsAsTheyChange(t *testing.T) { raw, _, _ = store.Get(FindingsState, "bus.EVENTS.stream-missing") f, err := health.DecodeFinding(raw) if err != nil || f.Kind != "stream-wrong" || f.Source != "P1" || f.Observer != Module || f.Headline == "" || - f.Explanation == "" || f.Token != "stream-missing" || len(f.Evidence) != 1 || f.Run == "" { + f.Explanation == "" || f.Ref != "stream-missing" || len(f.Evidence) != 1 || f.Run == "" { t.Errorf("the written finding is %+v", f) } // Blind to its own bucket, a run writes and clears nothing. @@ -258,7 +279,7 @@ func TestTheVerbsAndReads(t *testing.T) { // and is reported by the verb as off by setting; the default list is D13 and D-fill (the operator, 2026-10-30). func TestOnlyTheProbesTheSettingListsRun(t *testing.T) { ran := map[string]int{} - finds := Observation{Token: "stream-missing", Severity: Urgent, Subject: Subject{Scope: ScopeBus, ID: "EVENTS"}, Summary: "gone"} + finds := Observation{Ref: "stream-missing", Severity: Urgent, Subject: Subject{Scope: ScopeBus, ID: "EVENTS"}, Summary: "gone"} probe := func(id string) Probe { return Probe{ID: id, Kind: "stream-wrong", Run: func(context.Context, *Checker) ([]Observation, error) { ran[id]++ @@ -267,11 +288,16 @@ func TestOnlyTheProbesTheSettingListsRun(t *testing.T) { } store := newStore() c, _ := checker(store, probe("D7"), probe("D13")) - c.Enabled = func() []string { return []string{"D7", "D13"} } + read := 0 + c.Enabled = func() []string { read++; return []string{"D7", "D13"} } run := c.RunOnce(context.Background(), "test") if ran["D7"] != 1 || ran["D13"] != 1 || len(run.Probes) != 2 || len(store.findingKeys()) != 1 { t.Fatalf("both listed: ran %v, %d verdicts, findings %v", ran, len(run.Probes), store.findingKeys()) } + // The setting is read once a run, and once an answer of the verb, never once a probe (issue 1952). + if c.Probes(); read != 2 { + t.Fatalf("a run and the verb read the setting %d times, want 2", read) + } c.Enabled = func() []string { return []string{"D13"} } run = c.RunOnce(context.Background(), "test") if ran["D7"] != 1 || ran["D13"] != 2 || len(run.Probes) != 1 || run.Probes[0].ID != "D13" { diff --git a/modules/mesh-self-check/cmd/mesh-self-check/tools.go b/modules/mesh-self-check/cmd/mesh-self-check/tools.go index a4377b73..80e256d1 100644 --- a/modules/mesh-self-check/cmd/mesh-self-check/tools.go +++ b/modules/mesh-self-check/cmd/mesh-self-check/tools.go @@ -110,12 +110,13 @@ func oneProbe(r Run, id string) (Run, bool) { // Probes is the registry as the verb answers it. func (c *Checker) Probes() map[string]any { var out []map[string]any + on := c.listed() for _, p := range c.Registry { row := map[string]any{"id": p.ID, "asserts": p.Asserts, "from": p.From, "kind": p.Kind} if len(p.Raises) > 0 { row["raises"] = p.Raises } - if !c.On(p.ID) { + if !on(p.ID) { row["off"] = "off by setting" } else if p.Run == nil { row["deferred"] = p.Deferred diff --git a/modules/mesh-self-check/go.mod b/modules/mesh-self-check/go.mod index d345cd2f..f0ab5377 100644 --- a/modules/mesh-self-check/go.mod +++ b/modules/mesh-self-check/go.mod @@ -3,7 +3,7 @@ module mesh-self-check go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031084708-acbf80ad910a + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031203031-cbbfbf3d2d55 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-self-check/go.sum b/modules/mesh-self-check/go.sum index 8e0c33a9..5a992d56 100644 --- a/modules/mesh-self-check/go.sum +++ b/modules/mesh-self-check/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031084708-acbf80ad910a h1:DcSvfEmdPBKqCg5OmXRNb269EkkPe7BsEp+RykEhkgE= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031084708-acbf80ad910a/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031203031-cbbfbf3d2d55 h1:OcKTrlH72cPwAHuyAwyRLjGz0qt4VHTeal4MeahPnBM= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031203031-cbbfbf3d2d55/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=