From d7247eae19e07eedefc697d008e7b3dc24e156cb Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 29 Oct 2026 21:37:43 +0100 Subject: [PATCH] Move the self-check out of the controller into mesh-self-check, so its runs and findings no longer need the controller's lease (issue 1952) The probes whose facts are on the bus run with their assertions unchanged (D6, D7, D15, DW, H-bus); the rest answer deferred, naming what they lack, never a pass. Findings go to its own bucket for the condition keeper; each run is kept a week and said as checked. The finding layout is a copy of mesh-sdk's S3 until that lands. --- modules/mesh-self-check/.gitignore | 2 + modules/mesh-self-check/README.md | 33 ++ .../cmd/mesh-self-check/finding.go | 149 ++++++ .../cmd/mesh-self-check/main.go | 57 +++ .../cmd/mesh-self-check/manifest_test.go | 100 ++++ .../cmd/mesh-self-check/probes.go | 448 ++++++++++++++++++ .../cmd/mesh-self-check/probes_test.go | 176 +++++++ .../cmd/mesh-self-check/reads-outside | 2 + .../cmd/mesh-self-check/selfcheck.go | 412 ++++++++++++++++ .../cmd/mesh-self-check/selfcheck_test.go | 250 ++++++++++ .../cmd/mesh-self-check/tools.go | 210 ++++++++ .../mesh-self-check/whole_integration_test.go | 290 ++++++++++++ modules/mesh-self-check/go.mod | 22 + modules/mesh-self-check/go.sum | 27 ++ modules/mesh-self-check/module.json | 67 +++ 15 files changed, 2245 insertions(+) create mode 100644 modules/mesh-self-check/.gitignore create mode 100644 modules/mesh-self-check/README.md create mode 100644 modules/mesh-self-check/cmd/mesh-self-check/finding.go create mode 100644 modules/mesh-self-check/cmd/mesh-self-check/main.go create mode 100644 modules/mesh-self-check/cmd/mesh-self-check/manifest_test.go create mode 100644 modules/mesh-self-check/cmd/mesh-self-check/probes.go create mode 100644 modules/mesh-self-check/cmd/mesh-self-check/probes_test.go create mode 100644 modules/mesh-self-check/cmd/mesh-self-check/reads-outside create mode 100644 modules/mesh-self-check/cmd/mesh-self-check/selfcheck.go create mode 100644 modules/mesh-self-check/cmd/mesh-self-check/selfcheck_test.go create mode 100644 modules/mesh-self-check/cmd/mesh-self-check/tools.go create mode 100644 modules/mesh-self-check/cmd/mesh-self-check/whole_integration_test.go create mode 100644 modules/mesh-self-check/go.mod create mode 100644 modules/mesh-self-check/go.sum create mode 100644 modules/mesh-self-check/module.json diff --git a/modules/mesh-self-check/.gitignore b/modules/mesh-self-check/.gitignore new file mode 100644 index 00000000..2c7bac8b --- /dev/null +++ b/modules/mesh-self-check/.gitignore @@ -0,0 +1,2 @@ +/mesh-self-check +/cmd/mesh-self-check/mesh-self-check diff --git a/modules/mesh-self-check/README.md b/modules/mesh-self-check/README.md new file mode 100644 index 00000000..9411474c --- /dev/null +++ b/modules/mesh-self-check/README.md @@ -0,0 +1,33 @@ +# mesh-self-check + +Holds the seat `mesh-self-check` (novox/hq ADR 0564 §6): the mesh's self-check, moved out of the controller. Go, one +binary (`cmd/mesh-self-check`), served by a runtime of its own on its own bus account (`runs-as`), so that its account +alone writes its buckets. + +## What it does + +Every five minutes (the first run a minute after it starts) it runs its registry of probes, each within thirty +seconds. A probe is one live invariant of a design, with its assertion as the controller had it, reading its facts on +the bus (the bus module's tools for the bus's own state). A probe whose facts have no home on the bus yet answers +`deferred`, naming them: never a pass. A probe that cannot run is said in the verdict at once, and written as a finding +of its own (`probe..failed`) when the next run cannot run it either. + +What a probe finds is written as a finding, one key `..` per fault, only when what it says changes, and +deleted when a run no longer sees it. The condition keeper (`mesh-conditions`) raises and clears conditions from them. +A finding one look can be wrong about (a consumer behind after a burst) is written when two runs in a row see it. + +Every run ends with the event `checked`, the body of the controller's `doctor-heartbeat`, which the watchman (S10) and +mesh-watcher on another machine hear: a self-check that stops checking is said through a path that does not pass +through it. + +## What it keeps + +- `mesh-self-check_findings`: what it finds wrong now. +- `mesh-self-check_verdict`: the key `last`, the last run whole. +- `mesh-self-check_verdicts`: every run (each probe's findings at most twenty), kept a week by the bucket's age. + +## Its verbs and tools + +- `mesh-self-check.doctor`: the last verdict; `run=true` runs now; `probe=` one probe whole. +- `mesh-self-check.probes`: the registry, with what each deferred probe lacks. +- `self_check_findings`, `self_check_verdicts`: the reads of its buckets. diff --git a/modules/mesh-self-check/cmd/mesh-self-check/finding.go b/modules/mesh-self-check/cmd/mesh-self-check/finding.go new file mode 100644 index 00000000..9a20352e --- /dev/null +++ b/modules/mesh-self-check/cmd/mesh-self-check/finding.go @@ -0,0 +1,149 @@ +package main + +// A finding: what an observer writes for the condition keeper (novox/hq ADR 0564 §2). **A copy of the contract +// mesh-sdk holds (issue 1952, S3)**: its layout is the controller's observation (mesh-controller +// internal/conditions/condition.go), its key the controller's key rule, so a finding written here and the condition +// the controller raised for the same fault share one key. When mesh-sdk's go/health carries the finding, this file +// is replaced by it and nothing else changes. + +import ( + "encoding/json" + "regexp" + "strings" +) + +// The two severities a condition has (to-be 45 §2). +const ( + Urgent = "urgent" + Warning = "warning" +) + +// The scopes a key starts with, as the controller names them. +const ( + ScopeBus = "bus" + ScopeCore = "core" + ScopeProbe = "probe" + ScopeMesh = "mesh" +) + +// Subject is what a finding is about: its scope, its id within the scope, the machine it concerns. +type Subject struct { + Scope string `json:"scope"` + ID string `json:"id"` + Machine string `json:"machine,omitempty"` + Also []string `json:"also,omitempty"` +} + +// Finding is one thing an observer sees wrong now, kept under its key in the observer's bucket `findings` and +// deleted when it is no longer seen. Token and Confirm are the observer's own and never written. +type Finding struct { + Kind string `json:"kind"` + Subject Subject `json:"subject"` + Severity string `json:"severity"` + // Summary is one line in the mesh's words; Said the evidence (an address or a name stays here, never in the + // summary, ADR 0234 §6). + Summary string `json:"summary"` + Said string `json:"said,omitempty"` + // Headline, Explanation, Resolved and Needs are what the operator reads (ADR 0253). + Headline string `json:"headline"` + Explanation string `json:"explanation"` + Resolved string `json:"resolved"` + Needs string `json:"needs,omitempty"` + // Source is what found it: a probe's id, or the observer itself for its own blindness. + Source string `json:"source"` + // Observer is the module that wrote it. + Observer string `json:"observer"` + + Token string `json:"-"` + Confirm bool `json:"-"` +} + +// Key is where the finding is kept: `..`, the controller's key for the same condition. +func (f Finding) Key() string { + token := f.Token + if token == "" { + token = f.Kind + } + return Key(f.Subject.Scope, f.Subject.ID, token) +} + +// Says is what the finding says, as written: two findings that say the same are one write (ADR 0289 §1). +func (f Finding) Says() string { + raw, _ := json.Marshal(f) + return string(raw) +} + +var unsafeKey = regexp.MustCompile(`[^A-Za-z0-9_=/-]`) + +// Key composes a key from its parts, each token made safe for the bus, as the controller's conditions.Key does. +func Key(scope, id, token string) string { + var parts []string + for _, p := range append(append([]string{scope}, strings.Split(id, ".")...), token) { + p = unsafeKey.ReplaceAllString(strings.TrimSpace(p), "_") + if p == "" { + p = "_" + } + parts = append(parts, p) + } + return strings.Join(parts, ".") +} + +// worded gives a finding the operator's words registered for its kind, where it says none of its own. +func worded(f Finding) Finding { + w, ok := words[f.Kind] + if !ok { + return f + } + if f.Headline == "" { + f.Headline = w.Headline + } + if f.Explanation == "" { + f.Explanation = w.Explanation + } + if f.Resolved == "" { + f.Resolved = w.Resolved + } + if f.Needs == "" { + f.Needs = w.Needs + } + return f +} + +type plain struct{ Headline, Explanation, Resolved, Needs string } + +// words are the controller's words for each kind the self-check raises (mesh-controller plain_words.go), the +// controller's name for itself replaced by the self-check's where the words spoke of it. +var words = map[string]plain{ + "consumer-lost": {Headline: "A listener on the bus is gone", + Explanation: "One of the mesh's listeners is no longer on the bus; the next send sets it up again.", + Resolved: "The listener is back"}, + "consumer-wrong": {Headline: "A listener on the bus is set up wrong", + Explanation: "One of the mesh's listeners on the bus is missing or not as defined; the next send sets it up again.", + Resolved: "The listener is set up right again"}, + "consumer-behind": {Headline: "Messages pile up for a listener", + Explanation: "One of the mesh's listeners on the bus is far behind, so what it handles happens late. A healer " + + "restarts it if it stays.", + Resolved: "The listener caught up"}, + "stream-wrong": {Headline: "Part of the bus's storage is wrong", + Needs: "check the machine the bus runs on; the details say what is missing.", + Explanation: "A store the mesh keeps on the bus is missing or not as defined, so what it keeps there may be lost.", + Resolved: "The bus's storage is right again"}, + "bus-reconnects": {Headline: "A client keeps losing the bus", + Explanation: "One of the mesh's clients lost its connection to the bus again and again in the last hour. While " + + "it reconnects, what it says and what it is asked waits.", + Resolved: "Resolved: the client stays connected"}, + "core-unhealthy": {Headline: "The bus is not healthy", + Needs: "check the machine the bus runs on; the details say what fails.", + Explanation: "The bus, which carries every message in the mesh, is not healthy. Instructions, builds and tool " + + "calls may fail until it recovers.", + Resolved: "The bus is healthy again"}, + "watchdogs-silent": {Headline: "The mesh's watchdogs stopped", + Needs: "restart the watchman if this stays, from the mesh MCP server.", + Explanation: "The watchdogs, which notice when something expected does not happen, do not answer, so missed " + + "signals are not noticed.", + Resolved: "The watchdogs run again"}, + "probe-failed": {Headline: "A check of the mesh could not run", + Explanation: "One of the mesh's own checks could not run, so what it watches is unknown for now. It runs again " + + "every few minutes.", + Resolved: "The check runs again"}, +} diff --git a/modules/mesh-self-check/cmd/mesh-self-check/main.go b/modules/mesh-self-check/cmd/mesh-self-check/main.go new file mode 100644 index 00000000..c423e855 --- /dev/null +++ b/modules/mesh-self-check/cmd/mesh-self-check/main.go @@ -0,0 +1,57 @@ +// mesh-self-check: the holder of the seat mesh-self-check (novox/hq ADR 0564 §6). A Go bundle served by a runtime of +// its own, on its own bus account (`runs-as`), so that its account alone writes its buckets: `findings` (what it finds +// wrong now, which the condition keeper raises and clears conditions from), `verdict` (the last run) and `verdicts` +// (every run, kept a week). It runs the probes every five minutes and says `checked` at the end of each run. stdout is +// the MCP channel; what this module says, it says on stderr. +package main + +import ( + "context" + "encoding/json" + "fmt" + "os" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +func logf(format string, a ...any) { fmt.Fprintf(os.Stderr, format+"\n", a...) } + +// runtimeStore is the bus, through the runtime that launched this bundle. +type runtimeStore struct{} + +func (runtimeStore) Get(state, key string) ([]byte, bool, error) { + e, err := stdio.State(state).Get(key) + if err != nil || e == nil { + return nil, false, err + } + return e.Value, true, nil +} + +func (runtimeStore) Put(state, key string, value any) error { + _, err := stdio.State(state).Put(key, value) + return err +} + +func (runtimeStore) Delete(state, key string) error { return stdio.State(state).Delete(key) } + +func (runtimeStore) Keys(state string) ([]string, error) { return stdio.State(state).Keys() } + +// 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"), + Ask: func(key string, body any) (json.RawMessage, error) { return stdio.Ask(key, body) }, + Emit: func(event string, body any) error { return stdio.Emit(event, body) }} +} + +func main() { + c := NewChecker() + go func() { + <-stdio.Served() + c.Keep(context.Background()) + }() + if err := stdio.Serve("", Tools(c)); err != nil { + logf("%v", err) + os.Exit(1) + } +} diff --git a/modules/mesh-self-check/cmd/mesh-self-check/manifest_test.go b/modules/mesh-self-check/cmd/mesh-self-check/manifest_test.go new file mode 100644 index 00000000..10f33dc9 --- /dev/null +++ b/modules/mesh-self-check/cmd/mesh-self-check/manifest_test.go @@ -0,0 +1,100 @@ +package main + +import ( + "encoding/json" + "os" + "path/filepath" + "slices" + "testing" +) + +type manifest struct { + Module string `json:"module"` + RunsAs string `json:"runs-as"` + Claims []struct { + Name string `json:"name"` + Scope string `json:"scope"` + Serves []string `json:"serves"` + } `json:"claims"` + Emits []string `json:"emits"` + Invokes []string `json:"invokes"` + Reads []string `json:"reads"` + State []json.RawMessage `json:"state"` +} + +func readManifest(t *testing.T) manifest { + t.Helper() + raw, err := os.ReadFile(filepath.Join("..", "..", "module.json")) + if err != nil { + t.Fatal(err) + } + var m manifest + if err := json.Unmarshal(raw, &m); err != nil { + t.Fatal(err) + } + return m +} + +// The manifest says what the code does: it claims the seat mesh-self-check once for the mesh with the verbs it serves +// (ADR 0564 §1, §6), runs on its own account, says checked, asks only what its probes ask, keeps a week of verdicts, +// and reads no bucket of the controller's. Every state it keeps has a tool that reads it, and every act it takes by +// itself a tool that starts it by hand (ADR 0564 §12). +func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) { + m := readManifest(t) + if m.Module != Module || m.RunsAs != Module { + t.Errorf("module %q runs as %q", m.Module, m.RunsAs) + } + if len(m.Claims) != 1 || m.Claims[0].Name != Seat || m.Claims[0].Scope != "mesh" || !slices.Equal(m.Claims[0].Serves, Verbs) { + t.Errorf("claims %+v, want %s at mesh scope serving %v", m.Claims, Seat, Verbs) + } + if !slices.Equal(m.Emits, []string{CheckedEvent}) { + t.Errorf("emits %v", m.Emits) + } + if !slices.Equal(m.Invokes, []string{StreamsTool, ClosedTool, SignalsVerb}) { + t.Errorf("invokes %v, want what the probes ask", m.Invokes) + } + if len(m.Reads) != 0 { + t.Errorf("reads %v: the self-check reads its facts through the bus module's tools", m.Reads) + } + states := map[string]int{} + for _, s := range m.State { + var d struct { + Name string `json:"name"` + TTL int `json:"ttl-seconds"` + } + if json.Unmarshal(s, &d.Name) != nil { + _ = json.Unmarshal(s, &d) + } + states[d.Name] = d.TTL + } + if states[VerdictsState] != 7*24*60*60 { + t.Errorf("verdicts are kept %d s, want a week", states[VerdictsState]) + } + // ADR 0564 §12: each state with the tool that reads it, each act with the tool that starts it. + readBy := map[string]string{FindingsState: FindingsTool, VerdictState: DoctorTool, VerdictsState: VerdictsTool, + "the run": DoctorTool + " run=true"} + tools := map[string]bool{} + for _, tool := range Tools(&Checker{}) { + tools[tool.Name] = true + } + for state := range states { + if _, ok := readBy[state]; !ok { + t.Errorf("the state %s has no tool that reads it", state) + } + } + for what, tool := range readBy { + name, _, _ := cutSpace(tool) + if !tools[name] { + t.Errorf("%s is read or started by %s, which is not served", what, tool) + } + } +} + +func cutSpace(s string) (string, string, bool) { + for i, r := range s { + if r == ' ' { + return s[:i], s[i+1:], true + } + } + return s, "", false +} diff --git a/modules/mesh-self-check/cmd/mesh-self-check/probes.go b/modules/mesh-self-check/cmd/mesh-self-check/probes.go new file mode 100644 index 00000000..9f4e9882 --- /dev/null +++ b/modules/mesh-self-check/cmd/mesh-self-check/probes.go @@ -0,0 +1,448 @@ +package main + +// The registry (novox/hq ADR 0564 §6 and the probe table of its Context): the probes that move from the controller, +// each with its assertion unchanged, reading its facts where they live on the bus. A probe whose facts have no home +// on the bus yet answers deferred, naming them, and is never a pass (tracked by issue 1952). Not registered, as the +// table says: D1 (nodes compose their own; a refusal is the watchman's S21), D10 (a want ahead of its have is the +// watchman's pulse), D5, D9 and H-controller (they end with the controller), stuck (the watchman's own finding) and +// the bus step (its home is issue 1953's). + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "slices" + "sort" + "strings" + + "git.novox.be/novox/mesh-sdk/go/buscompose" +) + +// The tools a probe asks: the bus module's reads of its own server, and the watchdogs' seat. +const ( + StreamsTool = "nats.nats_streams" + ClosedTool = "nats.nats_closed_connections" + SignalsVerb = "seat:mesh-watchdogs.signals" + notOnTheBus = "the bus module's tools (nats_streams) answer this only once asked; " + storeOnly = "held only in the controller's store today" + behindBound = 1000 + reconnectMax = 12 +) + +// Registry is the self-check's registry, in to-be 45's order. **The registry is the design's live form**: a probe +// that moves here is a row added, and the registry test holds it to ADR 0564's table. +func Registry() []Probe { + return []Probe{ + {ID: "D2", Asserts: "every holder of the mesh's resolver answers a machine name for IPv4, and NODATA for IPv6", + From: "issue 262", Kind: "resolver-wrong", + Deferred: "the machines' names and the resolver holders' addresses to ask: the overlay's addresses are " + storeOnly}, + {ID: "D3", Asserts: "every seat on record that serves verbs has a live holder that answers, on every machine that " + + "is heard from", From: "issues 208, 218", Kind: "holder-silent", + Deferred: "the seats on record with their holders per machine, and which machines are heard from: the seat " + + "records are " + storeOnly + ", and the watchman's presence has no read granted to this module"}, + {ID: "D4", Asserts: "every kept archive is held by a manifest", From: "issue 253", Kind: "archives-unheld", + Deferred: "the artifact store's kept archives with the manifests that hold them: the archive index is " + storeOnly}, + {ID: "D6", Asserts: "every durable consumer the mesh expects exists with its definition, and is near its " + + "stream's head", From: "issues 248, 266", Kind: "consumer-wrong", + Raises: []string{"consumer-lost", "consumer-behind"}, Run: probeConsumers}, + {ID: "D7", Asserts: "every stream the mesh defines exists with its definition", From: "issue 208", + Kind: "stream-wrong", Run: probeStreams}, + {ID: "D8", Asserts: "no address the mesh owns — a machine's private address or its endpoint — is in a ban list", + From: "issue 238", Kind: "own-address-banned", + Deferred: "the machines' private addresses and endpoints: the overlay is " + storeOnly}, + {ID: "D11", Asserts: "no provider holds a consumer retired more than thirty days without a person deciding its " + + "cleanup", From: "ADR 0230", Kind: "cleanup-waiting", + Deferred: "the retirement records of providers' consumers: " + storeOnly}, + {ID: "D12", Asserts: "every consumer of a provision that keeps its data is bound where it was last sent, or moves " + + "by a pin", From: "issue 273, ADR 0232", Kind: "binding-moved", + Deferred: "the bindings of provisions to their consumers: " + storeOnly}, + {ID: "D13", Asserts: "every item of data a machine declares is measured, is there, holds what it held, and is " + + "backed up within its bound", From: "issue 273, ADR 0233", Kind: "data-shrank", + Deferred: "the data items' measurements and their history: " + storeOnly}, + {ID: "D15", Asserts: "no user of the bus had its connection dropped more than twelve times in the last hour: the " + + "bus module's nats_closed_connections", From: "issue 327", Kind: "bus-reconnects", Run: probeReconnects}, + {ID: "DA", Asserts: "every machine that names an agent account has it judged, on its node-engine's newest " + + "statement, unable to become root without a person", From: "ADR 0266, ADR 0259 §8", + Kind: "agent-can-become-root", + Deferred: "the node-engines' statements of the agent account's judgement: " + storeOnly}, + {ID: "D-root", Asserts: "no agent can become root without a person on a machine where the router or a channel " + + "proving its sender runs", From: "ADR 0259 §8", Kind: "root-not-free", + Deferred: "the node-engines' statements of root and the tools that run as root: " + storeOnly}, + {ID: "D-fill", Asserts: "every module's bucket and log holds less than three quarters of its cap; one said " + + "filling is cleared once it holds less than seven tenths", From: "ADR 0297, issue 501", + Kind: "bucket-or-log-filling", Raises: []string{"bucket-or-log-unread"}, + Deferred: "each bucket's and log's cap: the bus module's nats_buckets answers bytes and no cap, and the " + + "declared caps are " + storeOnly}, + {ID: "D-heard", Asserts: "no grant of a restricted stream names a module that composes nothing for it", + From: "ADR 0376 §2", Kind: "grant-composes-nothing", + Deferred: "the grants of restricted streams with what each machine composed for them: " + storeOnly}, + {ID: "DW", Asserts: "the watchdogs answer: the watchman holds the seat mesh-watchdogs and answers signals", + From: "ADR 0227 rule 6: the watchers are watched", Kind: "watchdogs-silent", Run: probeWatchdogs}, + {ID: "H-bus", Asserts: "every stream and durable consumer the mesh defines is on the bus, and a request crosses it " + + "to the bus's machine and back", From: "ADR 0236, to-be 45 §8", Kind: "core-unhealthy", Run: probeBusHealth}, + } +} + +// --- the bus's own state, as its module answers it --------------------------------------------------------------- + +// streamsAnswer is what nats_streams answers. +type streamsAnswer struct { + Detail []streamRow `json:"detail"` +} + +type streamRow struct { + Name string `json:"name"` + Subjects []string `json:"subjects"` + Retention string `json:"retention"` + MaxPerSubject int64 `json:"max_per_subject"` + Consumers []consumerRow `json:"consumer_detail"` +} + +type consumerRow struct { + Name string `json:"name"` + Kind string `json:"kind"` + Filters []string `json:"filters"` + Pending int64 `json:"pending"` + Unacked int64 `json:"unacknowledged"` + MaxDeliver int64 `json:"max_deliver"` +} + +// errNoStream is a stream the bus module says it does not hold. +var errNoStream = errors.New("no such stream") + +// streams asks the bus module for every stream, or for one with its consumers. +func (c *Checker) streams(stream string) ([]streamRow, error) { + args := map[string]any{} + if stream != "" { + args["stream"] = stream + } + raw, err := c.Ask(StreamsTool, args) + if err != nil { + if stream != "" && strings.Contains(err.Error(), "no stream called") { + return nil, errNoStream + } + return nil, fmt.Errorf("the bus module did not answer %s: %w", StreamsTool, err) + } + var a streamsAnswer + if err := json.Unmarshal(raw, &a); err != nil { + return nil, fmt.Errorf("the bus module answered %s with something unreadable: %w", StreamsTool, err) + } + return a.Detail, nil +} + +// consumerWords is a consumer as the mesh says it, with its name. +func consumerWords(c buscompose.Consumer) string { + return fmt.Sprintf("the consumer %s on %s", c.Name, c.Stream) +} + +// expectedConsumers are the durable consumers the mesh defines, by stream. +func expectedConsumers() map[string][]buscompose.Consumer { + out := map[string][]buscompose.Consumer{} + for _, c := range buscompose.MeshConsumers() { + out[c.Stream] = append(out[c.Stream], c) + } + return out +} + +// probeConsumers is D6: every durable consumer the mesh defines exists as defined, and is near its stream's head. +func probeConsumers(ctx context.Context, c *Checker) ([]Finding, error) { + var out []Finding + byStream := expectedConsumers() + streams := make([]string, 0, len(byStream)) + for s := range byStream { + streams = append(streams, s) + } + sort.Strings(streams) + for _, stream := range streams { + if ctx.Err() != nil { + return nil, ctx.Err() + } + rows, err := c.streams(stream) + held := map[string]consumerRow{} + switch { + case errors.Is(err, errNoStream): + case err != nil: + return nil, err + default: + for _, r := range rows { + for _, cr := range r.Consumers { + held[cr.Name] = cr + } + } + } + for _, want := range byStream[stream] { + who := want.Stream + "." + want.Name + have, ok := held[want.Name] + if !ok { + out = append(out, Finding{Kind: "consumer-lost", Token: "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, Finding{Token: "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, Finding{Kind: "consumer-behind", Token: "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)}) + } + } + } + return sorted(out), nil +} + +// consumerDiffers is what about a consumer on the bus is not as defined, of what the bus module says of it: its +// filters, its shape (push or pull) and how often it hands a message over. Empty when nothing is. +func consumerDiffers(want buscompose.Consumer, have consumerRow) string { + var differs []string + haveFilters := slices.Clone(have.Filters) + wantFilters := slices.Clone(want.Filters) + sort.Strings(haveFilters) + sort.Strings(wantFilters) + if !slices.Equal(haveFilters, wantFilters) { + differs = append(differs, fmt.Sprintf("filters %v, defined %v", haveFilters, wantFilters)) + } + if want.MaxDeliver != 0 && have.MaxDeliver != int64(want.MaxDeliver) { + differs = append(differs, fmt.Sprintf("hands a message over %d times, defined %d", have.MaxDeliver, want.MaxDeliver)) + } + if wantPush, havePush := want.Push || want.Queue != "", have.Kind == "push"; wantPush != havePush { + differs = append(differs, "delivers by the other shape (push or pull) than defined") + } + return strings.Join(differs, "; ") +} + +// probeStreams is D7: every stream the mesh defines exists as defined. +func probeStreams(ctx context.Context, c *Checker) ([]Finding, error) { + rows, err := c.streams("") + if err != nil { + return nil, err + } + held := map[string]streamRow{} + for _, r := range rows { + held[r.Name] = r + } + var out []Finding + for _, s := range buscompose.MeshStreams() { + have, ok := held[s.Name] + if !ok { + out = append(out, Finding{Token: "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, Finding{Token: "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}) + } + } + return sorted(out), nil +} + +// streamDiffers is what about a stream on the bus is not as defined, of what the bus module says of it: its +// subjects, its retention and how many it keeps per subject. Empty when nothing is. +func streamDiffers(want buscompose.Stream, have streamRow) string { + var differs []string + haveSubjects := slices.Clone(have.Subjects) + wantSubjects := slices.Clone(want.Subjects) + sort.Strings(haveSubjects) + sort.Strings(wantSubjects) + if !slices.Equal(haveSubjects, wantSubjects) { + differs = append(differs, fmt.Sprintf("subjects %v, defined %v", haveSubjects, wantSubjects)) + } + retention, perSubject := "limits", int64(want.MaxMsgsPerSubject) + switch want.Retention { + case buscompose.RetentionWorkQueue: + retention = "workqueue" + case buscompose.RetentionLastPerSubject: + perSubject = 1 + } + if have.Retention != "" && have.Retention != retention { + differs = append(differs, fmt.Sprintf("keeps by %s, defined %s", have.Retention, retention)) + } + if perSubject != 0 && have.MaxPerSubject != perSubject { + differs = append(differs, fmt.Sprintf("keeps %d per subject, defined %d", have.MaxPerSubject, perSubject)) + } + return strings.Join(differs, "; ") +} + +// --- D15 ----------------------------------------------------------------------------------------------------- + +// closedAnswer is what nats_closed_connections answers. +type closedAnswer struct { + Hours float64 `json:"hours"` + Reaches bool `json:"reaches"` + Users []struct { + User string `json:"user"` + Closed int `json:"closed"` + Dropped int `json:"dropped"` + Names []struct { + Name string `json:"name"` + Dropped int `json:"dropped"` + Reasons map[string]int `json:"reasons"` + } `json:"names"` + } `json:"users"` +} + +// probeReconnects is D15: each user of the bus whose connections were dropped more than twelve times an hour. +func probeReconnects(_ context.Context, c *Checker) ([]Finding, error) { + raw, err := c.Ask(ClosedTool, map[string]any{"hours": 1}) + if err != nil { + return nil, fmt.Errorf("the bus module did not answer %s: %w", ClosedTool, err) + } + var read closedAnswer + if err := json.Unmarshal(raw, &read); err != nil { + return nil, fmt.Errorf("the bus module answered %s with something unreadable: %w", ClosedTool, err) + } + hours := read.Hours + if hours <= 0 { + hours = 1 + } + var out []Finding + for _, u := range read.Users { + if u.User == "" || strings.HasPrefix(u.User, "(") || float64(u.Dropped)/hours <= reconnectMax { + continue // refused before it logged in is nobody's reconnect loop + } + reasons := map[string]int{} + var names []string + for _, n := range u.Names { + if n.Dropped == 0 { + continue + } + names = append(names, fmt.Sprintf("%q ×%d", n.Name, n.Dropped)) + for r, k := range n.Reasons { + if r != "Client Closed" { + reasons[r] += k + } + } + } + var why []string + for r, k := range reasons { + why = append(why, fmt.Sprintf("%s ×%d", r, k)) + } + sort.Strings(why) + partial := "" + if !read.Reaches { + partial = " (at least: the server's record of closed connections does not reach back the whole hour)" + } + who, machine := busUserWords(u.User) + out = append(out, Finding{Kind: "bus-reconnects", Severity: Warning, + Subject: Subject{Scope: ScopeBus, ID: u.User, Machine: machine}, + Summary: fmt.Sprintf("the bus dropped %s's connection %d times in the last hour%s, more than %d: a client "+ + "reconnecting in a loop", u.User, u.Dropped, partial, reconnectMax), + Said: fmt.Sprintf("%d of %d closed connections dropped in %.0f h; by name %s; why %s", u.Dropped, u.Closed, + hours, strings.Join(names, ", "), strings.Join(why, ", ")), + Headline: capital(who) + " keeps losing the bus", + Explanation: capital(fmt.Sprintf("%s lost its connection to the bus %d times in the last hour and connected "+ + "again each time. While it reconnects, what it says and what it is asked waits.", who, u.Dropped)), + Resolved: "Resolved: " + who + " stays connected"}) + } + return sorted(out), nil +} + +// busUserWords is a bus user as the operator says it, and the machine it is on. +func busUserWords(user string) (string, string) { + switch { + case user == "controller": + return "the controller", "" + case strings.HasPrefix(user, "node."): + m := strings.TrimPrefix(user, "node.") + return "the node-engine on " + m, m + } + if m, module, ok := strings.Cut(user, "."); ok { + if module == "node-tools" { + return "the tool runner on " + m, m + } + return module + " on " + m, m + } + return "a client of the bus", "" +} + +// --- DW and H-bus -------------------------------------------------------------------------------------------- + +// probeWatchdogs is DW: the watchdogs answer. The self-check and the watchdogs watch each other: S10, the +// watchman's pulse on `checked`, is the other half. +func probeWatchdogs(_ context.Context, c *Checker) ([]Finding, error) { + if _, err := c.Ask(SignalsVerb, map[string]any{}); err != nil { + return []Finding{{Token: "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 + } + return nil, nil +} + +// probeBusHealth is H-bus: every stream and durable consumer the mesh defines is on the bus, and a request crosses it +// to the bus module's machine and back. The question to the bus module is that request: an answer is its round trip, +// and no answer is no pass. +func probeBusHealth(ctx context.Context, c *Checker) ([]Finding, error) { + rows, err := c.streams("") + if err != nil { + return nil, fmt.Errorf("no request crossed the bus to its module and came back: %w", err) + } + held := map[string]bool{} + for _, r := range rows { + held[r.Name] = true + } + var problems []string + for _, s := range buscompose.MeshStreams() { + if !held[s.Name] { + problems = append(problems, "the stream "+s.Name+" is missing") + } + } + byStream := expectedConsumers() + streams := make([]string, 0, len(byStream)) + for s := range byStream { + streams = append(streams, s) + } + sort.Strings(streams) + for _, stream := range streams { + if !held[stream] { + continue // said above + } + if ctx.Err() != nil { + return nil, ctx.Err() + } + detail, err := c.streams(stream) + if err != nil && !errors.Is(err, errNoStream) { + return nil, err + } + have := map[string]bool{} + for _, r := range detail { + for _, cr := range r.Consumers { + have[cr.Name] = true + } + } + for _, want := range byStream[stream] { + if !have[want.Name] { + problems = append(problems, consumerWords(want)+" is missing") + } + } + } + if len(problems) == 0 { + return nil, nil + } + return []Finding{{Token: "unhealthy", Severity: Urgent, Subject: Subject{Scope: ScopeBus, ID: "mesh"}, + Summary: "the bus is not healthy: " + strings.Join(problems, "; "), Said: strings.Join(problems, "; ")}}, nil +} + +// sorted is findings in a stated order, so two runs over one mesh say the same. +func sorted(out []Finding) []Finding { + sort.SliceStable(out, func(i, j int) bool { return out[i].Key() < out[j].Key() }) + return out +} + +func capital(s string) string { + if s == "" { + return s + } + return strings.ToUpper(s[:1]) + s[1:] +} 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 new file mode 100644 index 00000000..7b918757 --- /dev/null +++ b/modules/mesh-self-check/cmd/mesh-self-check/probes_test.go @@ -0,0 +1,176 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "slices" + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/buscompose" +) + +// bus is the bus module's answers as a probe asks them: every mesh stream and consumer as defined, unless changed. +type bus struct { + drop map[string]bool // stream or stream.consumer missing + change func(rows map[string]*streamRow) + closed string + signals error + asksFail error +} + +func (b bus) rows() map[string]*streamRow { + rows := map[string]*streamRow{} + for _, s := range buscompose.MeshStreams() { + if b.drop[s.Name] { + continue + } + retention := "limits" + per := int64(s.MaxMsgsPerSubject) + switch s.Retention { + case buscompose.RetentionWorkQueue: + retention = "workqueue" + case buscompose.RetentionLastPerSubject: + per = 1 + } + rows[s.Name] = &streamRow{Name: s.Name, Subjects: s.Subjects, Retention: retention, MaxPerSubject: per} + } + for _, c := range buscompose.MeshConsumers() { + r := rows[c.Stream] + if r == nil || b.drop[c.Stream+"."+c.Name] { + continue + } + kind := "pull" + if c.Push || c.Queue != "" { + kind = "push" + } + r.Consumers = append(r.Consumers, consumerRow{Name: c.Name, Kind: kind, Filters: c.Filters, MaxDeliver: int64(c.MaxDeliver)}) + } + if b.change != nil { + b.change(rows) + } + return rows +} + +func (b bus) ask(key string, body any) (json.RawMessage, error) { + if b.asksFail != nil { + return nil, b.asksFail + } + switch key { + case StreamsTool: + stream, _ := body.(map[string]any)["stream"].(string) + var detail []streamRow + for name, r := range b.rows() { + if stream == "" { + row := *r + row.Consumers = nil + detail = append(detail, row) + } else if name == stream { + detail = append(detail, *r) + } + } + if stream != "" && len(detail) == 0 { + return nil, errors.New("no stream called " + stream) + } + return json.Marshal(map[string]any{"detail": detail}) + case ClosedTool: + if b.closed == "" { + return json.RawMessage(`{"hours":1,"reaches":true,"users":[]}`), nil + } + return json.RawMessage(b.closed), nil + case SignalsVerb: + if b.signals != nil { + return nil, b.signals + } + return json.RawMessage(`{"signals":[]}`), nil + } + return nil, errors.New("nats: no responders available for request") +} + +func keysOf(found []Finding) []string { + var out []string + for _, f := range found { + out = append(out, f.Key()) + } + return out +} + +// Each probe that moved judges the bus module's answers with its assertion unchanged (ADR 0564 §6; issue 1952 SC2). +func TestTheBusProbes(t *testing.T) { + cases := []struct { + name string + probe func(context.Context, *Checker) ([]Finding, error) + bus bus + want []string + err string + }{ + {"D6 all as defined", probeConsumers, bus{}, nil, ""}, + {"D6 a consumer gone", probeConsumers, bus{drop: map[string]bool{"DEAD_LETTER_NOTICES.controller": true}}, + []string{"bus.DEAD_LETTER_NOTICES.controller.missing"}, ""}, + {"D6 a stream gone takes its consumers", probeConsumers, bus{drop: map[string]bool{"CONTROL": true}}, + []string{"bus.CONTROL.controller.missing"}, ""}, + {"D6 redefined and far behind", probeConsumers, bus{change: func(rows map[string]*streamRow) { + for i, c := range rows["EVENTS"].Consumers { + if c.Name == "controller" { + rows["EVENTS"].Consumers[i].MaxDeliver, rows["EVENTS"].Consumers[i].Pending = 9, 5000 + } + } + }}, []string{"bus.EVENTS.controller.behind", "bus.EVENTS.controller.redefined"}, ""}, + {"D6 the bus module silent", probeConsumers, bus{asksFail: errors.New("no responders")}, nil, "did not answer"}, + {"D7 all as defined", probeStreams, bus{}, nil, ""}, + {"D7 missing and redefined", probeStreams, bus{drop: map[string]bool{"NODES": true}, change: func(rows map[string]*streamRow) { + rows["EVENTS"].Retention = "workqueue" + }}, []string{"bus.EVENTS.stream-redefined", "bus.NODES.stream-missing"}, ""}, + {"D15 quiet", probeReconnects, bus{}, nil, ""}, + {"D15 a loop", probeReconnects, bus{closed: `{"hours":1,"reaches":true,"users":[ + {"user":"g14.node-tools","closed":30,"dropped":20,"names":[{"name":"node-tools","dropped":20,"reasons":{"Stale Connection":20}}]}, + {"user":"ace.records","closed":5,"dropped":3},{"user":"(unauthenticated)","closed":90,"dropped":90}]}`}, + []string{"bus.g14.node-tools.bus-reconnects"}, ""}, + {"DW answers", probeWatchdogs, bus{}, nil, ""}, + {"DW silent", probeWatchdogs, bus{signals: errors.New("no responders")}, []string{"core.watchdogs.silent"}, ""}, + {"H-bus healthy", probeBusHealth, bus{}, nil, ""}, + {"H-bus a stream and a consumer missing", probeBusHealth, + bus{drop: map[string]bool{"NODES": true, "EVENTS.controller-merges": true}}, []string{"bus.mesh.unhealthy"}, ""}, + {"H-bus no round trip", probeBusHealth, bus{asksFail: errors.New("timeout")}, nil, "no request crossed"}, + } + for _, tc := range cases { + c := &Checker{Ask: tc.bus.ask} + found, err := tc.probe(context.Background(), c) + if tc.err != "" { + if err == nil || !strings.Contains(err.Error(), tc.err) { + t.Errorf("%s: error %v, want %q", tc.name, err, tc.err) + } + continue + } + if err != nil { + t.Errorf("%s: %v", tc.name, err) + continue + } + if got := keysOf(found); !slices.Equal(got, tc.want) { + t.Errorf("%s: found %v, want %v", tc.name, got, tc.want) + } + } +} + +// The registry holds exactly the probes ADR 0564's table moves here, each with a kind, and a deferred one names what +// it lacks; the probes that end with the controller or moved to the watchman are not in it (issue 1952 SC4). +func TestTheRegistryIsADR0564sTable(t *testing.T) { + moved := []string{"D2", "D3", "D4", "D6", "D7", "D8", "D11", "D12", "D13", "D15", "DA", "D-root", "D-fill", "D-heard", + "DW", "H-bus"} + var ids []string + for _, p := range Registry() { + ids = append(ids, p.ID) + if p.Kind == "" || p.Asserts == "" { + t.Errorf("%s has no kind or assertion", p.ID) + } + if (p.Run == nil) == (p.Deferred == "") { + t.Errorf("%s is neither run nor deferred naming what it lacks", p.ID) + } + } + slices.Sort(ids) + slices.Sort(moved) + if !slices.Equal(ids, moved) { + t.Errorf("the registry holds %v, ADR 0564 moves %v", ids, moved) + } +} diff --git a/modules/mesh-self-check/cmd/mesh-self-check/reads-outside b/modules/mesh-self-check/cmd/mesh-self-check/reads-outside new file mode 100644 index 00000000..cb8f579b --- /dev/null +++ b/modules/mesh-self-check/cmd/mesh-self-check/reads-outside @@ -0,0 +1,2 @@ +# What these tests read outside their package, as paths from the repository's root (issue 1936). +modules/mesh-self-check/module.json diff --git a/modules/mesh-self-check/cmd/mesh-self-check/selfcheck.go b/modules/mesh-self-check/cmd/mesh-self-check/selfcheck.go new file mode 100644 index 00000000..aa670108 --- /dev/null +++ b/modules/mesh-self-check/cmd/mesh-self-check/selfcheck.go @@ -0,0 +1,412 @@ +package main + +// The self-check (novox/hq ADR 0564 §6): the registry of probes, each a live invariant of a design, run every five +// minutes, each probe within thirty seconds. What a probe finds is written as a finding in this module's bucket +// `findings`, which the condition keeper raises and clears conditions from (ADR 0564 §2); every run is kept in the +// bucket `verdicts` for a week and the last as the key `last` of `verdict`, and is said as `checked`, which the +// watchman and mesh-watcher hear (S10, ADR 0227 §6). A probe that errors or does not finish is never a pass: it is +// said in the verdict at once and written as a finding of its own when the next run cannot run it either. + +import ( + "context" + "encoding/json" + "fmt" + "sort" + "strings" + "sync" + "time" +) + +// Module and Seat are this module and the seat it holds once for the mesh. +const ( + Module = "mesh-self-check" + Seat = "mesh-self-check" +) + +// The states it keeps, and the event that ends every run. +const ( + FindingsState = "findings" + VerdictState = "verdict" + VerdictsState = "verdicts" + LastKey = "last" + CheckedEvent = "checked" +) + +// The clocks: the self-check's run is the one timer among the four holders (ADR 0564 §1). +var ( + Every = 5 * time.Minute + FirstAfter = time.Minute + Within = 30 * time.Second +) + +// The verdicts a probe can have. +const ( + VerdictPass = "pass" + VerdictFail = "fail" + VerdictFailedToRun = "failed-to-run" + VerdictDeferred = "deferred" +) + +// sourceSelf is what writes a probe's failure to run: the self-check itself. +const sourceSelf = "doctor" + +// Store is the bus's buckets as this module reaches them, through its runtime. +type Store interface { + Get(state, key string) ([]byte, bool, error) + Put(state, key string, value any) error + Delete(state, key string) error + Keys(state string) ([]string, error) +} + +// Probe is one live invariant: its id, what it asserts, the kind it raises, and the run; or why it waits. +type Probe struct { + ID string + Asserts string + From string + Kind string + Raises []string + // Deferred names the facts it lacks on the bus, for a probe whose facts have no home there yet (ADR 0564 §6): + // answered deferred, never a pass. + Deferred string + Run func(ctx context.Context, c *Checker) ([]Finding, error) +} + +// ProbeVerdict is one probe's outcome in a run, as the controller's doctor says it. +type ProbeVerdict struct { + ID string `json:"id"` + Verdict string `json:"verdict"` + Found []string `json:"found,omitempty"` + Unconfirmed []string `json:"unconfirmed,omitempty"` + Error string `json:"error,omitempty"` + Took string `json:"took,omitempty"` + FoundCount int `json:"found-count,omitempty"` + More string `json:"more,omitempty"` +} + +// Counts are a run's verdicts, counted. +type Counts struct { + Passed int `json:"passed"` + Failed int `json:"failed"` + FailedToRun int `json:"failed-to-run"` + Deferred int `json:"deferred"` +} + +// Run is one run of the registry, and the body of `checked`: the controller's doctor-heartbeat body unchanged +// (ADR 0564 §6), Controller naming the machine that ran it. +type Run struct { + Run string `json:"run"` + At time.Time `json:"at"` + Started time.Time `json:"started"` + Took string `json:"took"` + IntervalSeconds int `json:"interval-seconds"` + Counts Counts `json:"counts"` + Probes []ProbeVerdict `json:"probes"` + Controller string `json:"controller"` + Why string `json:"why"` + Unsaid int `json:"unsaid,omitempty"` +} + +// Checker is the self-check: the registry and what its probes reach. +type Checker struct { + Store Store + Ask func(key string, body any) (json.RawMessage, error) + Emit func(event string, body any) error + Now func() time.Time + Logf func(format string, args ...any) + Host string + Registry []Probe + + running sync.Mutex + mu sync.Mutex + n uint64 + last *Run + // seen are the keys the run before saw, confirmed or not: a finding a single look can be wrong about is + // written only when two runs in a row see it, or while it is already written (novox/hq issue 277). + seen map[string]bool + // unsaid counts the runs whose `checked` could not be said since the start. + unsaid int +} + +// LastRun is the last run, nil before the first. +func (c *Checker) LastRun() *Run { + c.mu.Lock() + defer c.mu.Unlock() + return c.last +} + +type result struct { + found []Finding + err error + took time.Duration +} + +// RunOnce runs every probe of the registry, writes what changed in `findings`, keeps the verdict and says `checked`. +// One run at a time: a person's run during a scheduled one waits for it. +func (c *Checker) RunOnce(ctx context.Context, why string) Run { + c.running.Lock() + defer c.running.Unlock() + c.mu.Lock() + c.n++ + n := c.n + c.mu.Unlock() + started := c.Now() + run := Run{Run: fmt.Sprintf("doctor-%d-%d", started.Unix(), n), Started: started.UTC(), + IntervalSeconds: int(Every / time.Second), Controller: c.Host, Why: why} + + results := make([]result, len(c.Registry)) + var wg sync.WaitGroup + for i, p := range c.Registry { + if p.Run == nil { + continue + } + wg.Add(1) + go func(i int, p Probe) { + defer wg.Done() + results[i] = c.probe(ctx, p) + }(i, p) + } + wg.Wait() + + written, readErr := c.written() + seen := map[string]bool{} + want := map[string]Finding{} // every finding this run writes, by key + ran := map[string]bool{} // the sources whose findings this run judged whole + var blind []Finding + for i, p := range c.Registry { + v := ProbeVerdict{ID: p.ID} + r := results[i] + switch { + case p.Run == nil: + v.Verdict = VerdictDeferred + v.Error = "deferred: " + p.Deferred + run.Counts.Deferred++ + case r.err != nil: + v.Verdict, v.Error = VerdictFailedToRun, r.err.Error() + run.Counts.FailedToRun++ + blind = append(blind, Finding{Kind: "probe-failed", Token: "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())}) + default: + ran[p.ID] = true + for _, f := range r.found { + if f.Kind == "" { + f.Kind = p.Kind + } + f.Source = p.ID + f = worded(f) + f.Observer = Module + key := f.Key() + seen[key] = true + if f.Confirm && !c.seenBefore(key) && !written.has(key) { + v.Unconfirmed = append(v.Unconfirmed, f.Summary) + continue + } + want[key] = f + v.Found = append(v.Found, f.Summary) + } + if len(v.Found) == 0 { + v.Verdict = VerdictPass + run.Counts.Passed++ + } else { + v.Verdict = VerdictFail + run.Counts.Failed++ + } + } + if p.Run != nil { + v.Took = r.took.Round(time.Millisecond).String() + } + run.Probes = append(run.Probes, v) + } + ran[sourceSelf] = true + for _, f := range blind { + f = worded(f) + f.Observer = Module + key := f.Key() + seen[key] = true + if c.seenBefore(key) || written.has(key) { + want[key] = f + } + } + if readErr != nil { + // Blind to its own bucket: nothing is written or deleted (ADR 0227 §4), and the log says so. + c.Logf("[%s] its findings cannot be read, so none is written or cleared this run: %v", Module, readErr) + } else if err := c.write(written, want, ran); err != nil { + c.Logf("[%s] what this run found is not all written: %v", Module, err) + } + c.mu.Lock() + c.seen = seen + c.mu.Unlock() + + run.At = c.Now().UTC() + run.Took = run.At.Sub(started).Round(time.Millisecond).String() + c.mu.Lock() + run.Unsaid = c.unsaid + c.mu.Unlock() + c.keep(run) + return run +} + +// probe runs one probe within its bound; a panic or an overrun is an error, never a pass. +func (c *Checker) probe(ctx context.Context, p Probe) result { + probing, cancel := context.WithTimeout(ctx, Within) + defer cancel() + began := c.Now() + done := make(chan result, 1) + go func() { + defer func() { + if r := recover(); r != nil { + done <- result{err: fmt.Errorf("the probe panicked: %v", r)} + } + }() + found, err := p.Run(probing, c) + done <- result{found: found, err: err} + }() + select { + case r := <-done: + r.took = c.Now().Sub(began) + return r + case <-probing.Done(): + return result{err: fmt.Errorf("it did not finish within %s", Within), took: c.Now().Sub(began)} + } +} + +func (c *Checker) seenBefore(key string) bool { + c.mu.Lock() + defer c.mu.Unlock() + return c.seen[key] +} + +// writtenFindings are the findings in the bucket, by key, with the source that wrote each and what it says. +type writtenFindings map[string]Finding + +func (w writtenFindings) has(key string) bool { _, ok := w[key]; return ok } + +// written reads every finding this module has written. +func (c *Checker) written() (writtenFindings, error) { + keys, err := c.Store.Keys(FindingsState) + if err != nil { + return writtenFindings{}, err + } + out := writtenFindings{} + for _, k := range keys { + raw, found, err := c.Store.Get(FindingsState, k) + if err != nil { + return out, err + } + if !found { + continue + } + var f Finding + if err := json.Unmarshal(raw, &f); err != nil { + return out, fmt.Errorf("the finding %s is not readable: %w", k, err) + } + out[k] = f + } + return out, nil +} + +// write puts each finding whose words changed and deletes each one a source that ran whole no longer sees. A finding +// of a source that could not run is left as it is: a blind observer clears nothing (ADR 0227 §4). +func (c *Checker) write(written writtenFindings, want map[string]Finding, ran map[string]bool) error { + var errs []string + keys := make([]string, 0, len(want)) + for k := range want { + keys = append(keys, k) + } + sort.Strings(keys) + for _, k := range keys { + f := want[k] + if was, ok := written[k]; ok && was.Says() == f.Says() { + continue + } + if err := c.Store.Put(FindingsState, k, f); err != nil { + errs = append(errs, k+": "+err.Error()) + } + } + for k, f := range written { + if _, still := want[k]; still || !ran[f.Source] { + continue + } + if err := c.Store.Delete(FindingsState, k); err != nil { + errs = append(errs, k+": "+err.Error()) + } + } + if len(errs) > 0 { + return fmt.Errorf("%s", strings.Join(errs, "; ")) + } + return nil +} + +// keep keeps the run as the last verdict and in the week's verdicts, and says it as `checked`. Not said is said in +// the log: the watchman's S10 says the self-check silent when `checked` stops. +func (c *Checker) keep(run Run) { + c.mu.Lock() + c.last = &run + c.mu.Unlock() + if err := c.Store.Put(VerdictState, LastKey, run); err != nil { + c.Logf("[%s] the run %s is not kept as the last verdict: %v", Module, run.Run, err) + } + if err := c.Store.Put(VerdictsState, run.Run, Overview(run)); err != nil { + c.Logf("[%s] the run %s is not kept among the week's verdicts: %v", Module, run.Run, err) + } + if c.Emit == nil { + return + } + if err := c.Emit(CheckedEvent, run); err != nil { + c.mu.Lock() + c.unsaid++ + c.mu.Unlock() + c.Logf("[%s] the run %s could NOT be said as checked, so the watchers will say the self-check is silent: %v", + Module, run.Run, err) + } +} + +// overviewFindings is how many findings of one probe an overview carries (novox/hq issue 314). +const overviewFindings = 20 + +// Overview is a run with each probe's findings at most overviewFindings, with their count and how to read them all. +func Overview(run Run) Run { + probes := make([]ProbeVerdict, len(run.Probes)) + for i, p := range run.Probes { + total := len(p.Found) + len(p.Unconfirmed) + if total > overviewFindings { + p.FoundCount = len(p.Found) + if len(p.Found) > overviewFindings { + p.Found = p.Found[:overviewFindings:overviewFindings] + } + if room := overviewFindings - len(p.Found); len(p.Unconfirmed) > room { + p.Unconfirmed = p.Unconfirmed[:room:room] + } + p.More = fmt.Sprintf("%d found and %d unconfirmed, %d of them shown — `%s.doctor probe=%s` gives them all", + p.FoundCount, total-p.FoundCount, len(p.Found)+len(p.Unconfirmed), Seat, p.ID) + } + probes[i] = p + } + run.Probes = probes + return run +} + +// Keep runs the registry on its schedule until ctx ends: the first run FirstAfter the start, then every Every. +func (c *Checker) Keep(ctx context.Context) { + select { + case <-ctx.Done(): + return + case <-time.After(FirstAfter): + } + tick := time.NewTicker(Every) + defer tick.Stop() + for { + c.RunOnce(ctx, "the schedule") + select { + case <-ctx.Done(): + return + case <-tick.C: + } + } +} + +func firstLine(s string) string { + line, _, _ := strings.Cut(s, "\n") + return line +} 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 new file mode 100644 index 00000000..42793250 --- /dev/null +++ b/modules/mesh-self-check/cmd/mesh-self-check/selfcheck_test.go @@ -0,0 +1,250 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "slices" + "sort" + "strings" + "sync" + "testing" + "time" +) + +// memStore is the bus's buckets in memory, counting the writes. +type memStore struct { + mu sync.Mutex + kv map[string]map[string][]byte + puts map[string]int + keysErr error +} + +func newStore() *memStore { + return &memStore{kv: map[string]map[string][]byte{}, puts: map[string]int{}} +} + +func (s *memStore) Get(state, key string) ([]byte, bool, error) { + s.mu.Lock() + defer s.mu.Unlock() + v, ok := s.kv[state][key] + return v, ok, nil +} + +func (s *memStore) Put(state, key string, value any) error { + raw, err := json.Marshal(value) + if err != nil { + return err + } + s.mu.Lock() + defer s.mu.Unlock() + if s.kv[state] == nil { + s.kv[state] = map[string][]byte{} + } + s.kv[state][key] = raw + s.puts[state+"/"+key]++ + return nil +} + +func (s *memStore) Delete(state, key string) error { + s.mu.Lock() + defer s.mu.Unlock() + delete(s.kv[state], key) + return nil +} + +func (s *memStore) Keys(state string) ([]string, error) { + s.mu.Lock() + defer s.mu.Unlock() + if s.keysErr != nil && state == FindingsState { + return nil, s.keysErr + } + var out []string + for k := range s.kv[state] { + out = append(out, k) + } + sort.Strings(out) + return out, nil +} + +func (s *memStore) findingKeys() []string { + keys, _ := s.Keys(FindingsState) + return keys +} + +// checker is a self-check over memory, with a registry the test drives. +func checker(store *memStore, registry ...Probe) (*Checker, *[]string) { + var said []string + now := time.Date(2026, 10, 29, 21, 0, 0, 0, time.UTC) + c := &Checker{Store: store, Registry: registry, Host: "novox", Logf: func(string, ...any) {}, + Now: func() time.Time { now = now.Add(time.Second); return now }, + Emit: func(event string, body any) error { + raw, _ := json.Marshal(body) + said = append(said, event+" "+string(raw)) + return nil + }} + return c, &said +} + +// A run writes what a probe finds, writes nothing it already wrote, and deletes what it no longer sees; a probe that +// could not run clears nothing and is written as its own finding at the second run in a row; a finding one look can +// be wrong about waits for the second run; a deferred probe is never a pass (ADR 0564 §2, §6; issue 1952 SC1). +func TestARunWritesFindingsAsTheyChange(t *testing.T) { + var found []Finding + var fail error + steady := Probe{ID: "P1", Kind: "stream-wrong", Run: func(context.Context, *Checker) ([]Finding, error) { return found, fail }} + deferred := Probe{ID: "P2", Kind: "resolver-wrong", Deferred: "the overlay's addresses"} + missing := Finding{Token: "stream-missing", Severity: Urgent, Subject: Subject{Scope: ScopeBus, ID: "EVENTS"}, + Summary: "the stream EVENTS is not on the bus"} + behind := Finding{Kind: "consumer-behind", Token: "behind", Severity: Warning, Confirm: true, + Subject: Subject{Scope: ScopeBus, ID: "EVENTS.controller"}, Summary: "far behind"} + store := newStore() + c, said := checker(store, steady, deferred) + + steps := []struct { + name string + found []Finding + fail error + want []string // finding keys written after the run + verdict string + missingPs int // puts of the missing finding so far + }{ + {"a finding is written", []Finding{missing}, nil, []string{"bus.EVENTS.stream-missing"}, VerdictFail, 1}, + {"the same again is no write", []Finding{missing}, nil, []string{"bus.EVENTS.stream-missing"}, VerdictFail, 1}, + {"one look can be wrong: held", []Finding{missing, behind}, nil, []string{"bus.EVENTS.stream-missing"}, VerdictFail, 1}, + {"seen twice: written", []Finding{missing, behind}, nil, + []string{"bus.EVENTS.controller.behind", "bus.EVENTS.stream-missing"}, VerdictFail, 1}, + {"blind once: clears nothing, held", nil, errors.New("timed out"), + []string{"bus.EVENTS.controller.behind", "bus.EVENTS.stream-missing"}, VerdictFailedToRun, 1}, + {"blind twice: its own finding", nil, errors.New("timed out"), + []string{"bus.EVENTS.controller.behind", "bus.EVENTS.stream-missing", "probe.P1.failed"}, VerdictFailedToRun, 1}, + {"no longer seen: cleared", nil, nil, nil, VerdictPass, 1}, + } + for _, s := range steps { + found, fail = s.found, s.fail + run := c.RunOnce(context.Background(), "the test") + if got := store.findingKeys(); !slices.Equal(got, s.want) { + t.Errorf("%s: findings %v, want %v", s.name, got, s.want) + } + if run.Probes[0].Verdict != s.verdict { + t.Errorf("%s: P1 %s, want %s", s.name, run.Probes[0].Verdict, s.verdict) + } + if run.Probes[1].Verdict != VerdictDeferred || !strings.Contains(run.Probes[1].Error, "overlay") { + t.Errorf("%s: a deferred probe answered %+v", s.name, run.Probes[1]) + } + if n := store.puts[FindingsState+"/bus.EVENTS.stream-missing"]; n != s.missingPs { + t.Errorf("%s: the unchanged finding was written %d times", s.name, n) + } + } + raw, _, _ := store.Get(FindingsState, "probe.P1.failed") + if raw != nil { + t.Errorf("the blind finding outlived the probe running again: %s", raw) + } + // Every run is kept, the last as `last`, and said as checked with the controller's heartbeat body. + if keys, _ := store.Keys(VerdictsState); len(keys) != len(steps) { + t.Errorf("verdicts kept %d runs, want %d", len(keys), len(steps)) + } + if len(*said) != len(steps) || !strings.HasPrefix((*said)[0], CheckedEvent+" ") { + t.Fatalf("said %v", *said) + } + var body map[string]any + _ = json.Unmarshal([]byte(strings.TrimPrefix((*said)[0], CheckedEvent+" ")), &body) + for _, field := range []string{"run", "at", "started", "took", "interval-seconds", "counts", "probes", "controller", "why"} { + if _, ok := body[field]; !ok { + t.Errorf("checked's body lacks %s, which doctor-heartbeat carries: %v", field, body) + } + } + last, _, _ := store.Get(VerdictState, LastKey) + var r Run + _ = json.Unmarshal(last, &r) + if r.Counts.Passed != 1 || r.Counts.Deferred != 1 { + t.Errorf("the last verdict counts %+v", r.Counts) + } + // A finding carries the operator's words of its kind, the probe that found it and who observed it. + found, fail = []Finding{missing}, nil + c.RunOnce(context.Background(), "the test") + raw, _, _ = store.Get(FindingsState, "bus.EVENTS.stream-missing") + var f Finding + _ = json.Unmarshal(raw, &f) + if f.Kind != "stream-wrong" || f.Source != "P1" || f.Observer != Module || f.Headline == "" || f.Explanation == "" { + t.Errorf("the written finding is %+v", f) + } + // Blind to its own bucket, a run writes and clears nothing. + store.keysErr = errors.New("the bus is away") + found = nil + c.RunOnce(context.Background(), "the test") + store.keysErr = nil + if got := store.findingKeys(); !slices.Equal(got, []string{"bus.EVENTS.stream-missing"}) { + t.Errorf("a run blind to its findings changed them: %v", got) + } +} + +// A probe that panics or overruns its bound could not run: never a pass. +func TestAProbeThatPanicsOrOverrunsCouldNotRun(t *testing.T) { + within := Within + Within = 50 * time.Millisecond + t.Cleanup(func() { Within = within }) + for name, run := range map[string]func(context.Context, *Checker) ([]Finding, error){ + "panic": func(context.Context, *Checker) ([]Finding, error) { panic("boom") }, + "overrun": func(ctx context.Context, _ *Checker) ([]Finding, error) { time.Sleep(time.Second); return nil, nil }, + } { + c, _ := checker(newStore(), Probe{ID: "P", Kind: "k", Run: run}) + c.Now = time.Now + r := c.RunOnce(context.Background(), "the test") + if r.Probes[0].Verdict != VerdictFailedToRun || r.Counts.Passed != 0 { + t.Errorf("%s: %+v", name, r.Probes[0]) + } + } +} + +// The verb answers the last run, one probe whole, a run now, and before any run says when one comes; the reads answer +// every state the module keeps (ADR 0564 §12). +func TestTheVerbsAndReads(t *testing.T) { + store := newStore() + many := make([]Finding, 25) + for i := range many { + many[i] = Finding{Severity: Warning, Subject: Subject{Scope: ScopeBus, ID: fmt.Sprintf("S%02d", i)}, Summary: "x"} + } + c, _ := checker(store, Probe{ID: "D7", Kind: "stream-wrong", + Run: func(context.Context, *Checker) ([]Finding, error) { return many, nil }}) + if _, err := c.Doctor(false, "", "x"); err == nil || !strings.Contains(err.Error(), "first run") { + t.Fatalf("before a run: %v", err) + } + if _, err := c.Doctor(true, "", "asked by person.jochen"); err != nil { + t.Fatal(err) + } + over, _ := c.Doctor(false, "", "") + if v := over["run"].(Run).Probes[0]; len(v.Found) != overviewFindings || v.More == "" { + t.Errorf("the overview carries %d findings, more %q", len(v.Found), v.More) + } + whole, _ := c.Doctor(false, "d7", "") + if v := whole["run"].(Run).Probes[0]; len(v.Found) != 25 { + t.Errorf("one probe whole carries %d findings", len(v.Found)) + } + if _, err := c.Doctor(false, "D99", ""); err == nil { + t.Error("an unknown probe answered") + } + // A fresh checker over the same bucket answers the last verdict from it. + fresh, _ := checker(store) + if _, err := fresh.Doctor(false, "", ""); err != nil { + t.Errorf("the last verdict is not read from its bucket: %v", err) + } + fs, err := c.Findings("") + if err != nil || fs.(map[string]any)["count"] != 25 { + t.Errorf("findings answered %v, %v", fs, err) + } + if one, err := c.Findings("bus.S03.stream-wrong"); err != nil || !strings.Contains(string(one.(json.RawMessage)), "S03") { + t.Errorf("one finding answered %v, %v", one, err) + } + vs, err := c.Verdicts("", 0) + if err != nil || len(vs.(map[string]any)["verdicts"].([]map[string]any)) != 1 { + t.Errorf("verdicts answered %v, %v", vs, err) + } + if _, err := c.Verdicts("doctor-1-1", 0); err == nil { + t.Error("an unknown run answered") + } + if p := c.Probes()["probes"].([]map[string]any); len(p) != 1 || p[0]["id"] != "D7" { + t.Errorf("probes answered %v", p) + } +} diff --git a/modules/mesh-self-check/cmd/mesh-self-check/tools.go b/modules/mesh-self-check/cmd/mesh-self-check/tools.go new file mode 100644 index 00000000..b38fa830 --- /dev/null +++ b/modules/mesh-self-check/cmd/mesh-self-check/tools.go @@ -0,0 +1,210 @@ +package main + +// The seat's verbs and the module's own tools (novox/hq ADR 0564 §6, §12): `doctor` answers the last verdict at once, +// runs now, or gives one probe's verdict whole; `probes` is the registry; and a read of each state it keeps — the +// findings, the last verdict and the week's verdicts — so a person sees all it holds and a recovery is a call. + +import ( + "context" + "encoding/json" + "fmt" + "sort" + "strings" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// Verbs are the seat's verbs, as the manifest claims them. +var Verbs = []string{"doctor", "probes"} + +// Tool names: the seat's verbs, and the module's own reads. +const ( + DoctorTool = Seat + ".doctor" + ProbesTool = Seat + ".probes" + FindingsTool = "self_check_findings" + VerdictsTool = "self_check_verdicts" +) + +func obj(props map[string]any) map[string]any { + return map[string]any{"type": "object", "properties": props} +} + +// Tools are what the runtime serves. +func Tools(c *Checker) []stdio.Tool { + return []stdio.Tool{ + {Name: DoctorTool, + Description: "The mesh's self-check (novox/hq ADR 0564 §6): the last run's verdict at once, each probe passed, " + + "failed with what it found, could not run (never a pass) or deferred with the facts it lacks; run=true runs " + + "every probe now and answers that run; probe= gives one probe's verdict whole, every finding.", + Input: obj(map[string]any{ + "run": map[string]any{"type": "boolean", "description": "run every probe now"}, + "probe": map[string]any{"type": "string", "description": "one probe's verdict, whole, e.g. D6"}}), + Run: func(a map[string]any) (any, error) { + return c.Doctor(boolArg(a, "run"), strArg(a, "probe"), "asked by "+stdio.CallerOf(a)) + }}, + {Name: ProbesTool, + Description: "The self-check's registry: each probe's id, what it asserts, where it came from, the condition kinds " + + "it raises, and for one deferred the facts it lacks on the bus.", + Run: func(map[string]any) (any, error) { return c.Probes(), nil }}, + {Name: FindingsTool, + Description: "What the self-check finds wrong now, as the condition keeper reads it (its bucket findings): every " + + "finding's key, kind, severity, the probe that found it and its summary; or one finding whole by key.", + Input: obj(map[string]any{"key": map[string]any{"type": "string", "description": "one finding's key"}}), + Run: func(a map[string]any) (any, error) { return c.Findings(strArg(a, "key")) }}, + {Name: VerdictsTool, + Description: "The self-check's runs of the last week (its bucket verdicts, each kept a week) newest first, each " + + "with its counts; or one run whole by its id. The last run is the doctor verb's.", + Input: obj(map[string]any{ + "run": map[string]any{"type": "string", "description": "one run's id, e.g. doctor-1793284800-12"}, + "limit": map[string]any{"type": "number", "description": "at most this many, newest (default 50)"}}), + Run: func(a map[string]any) (any, error) { return c.Verdicts(strArg(a, "run"), intArg(a, "limit", 50)) }}, + } +} + +// Doctor is the verb: the last verdict, a run now, or one probe of either. +func (c *Checker) Doctor(run bool, probe, why string) (map[string]any, error) { + var r Run + switch { + case run: + r = c.RunOnce(context.Background(), why) + case c.LastRun() != nil: + r = *c.LastRun() + default: + raw, found, err := c.Store.Get(VerdictState, LastKey) + if err != nil { + return nil, fmt.Errorf("the last verdict cannot be read: %w", err) + } + if !found { + return nil, fmt.Errorf("the self-check has not finished its first run yet: it runs %s after it starts, then "+ + "every %s — run=true runs it now", FirstAfter, Every) + } + if err := json.Unmarshal(raw, &r); err != nil { + return nil, fmt.Errorf("the last verdict is not readable: %w", err) + } + } + if probe != "" { + one, ok := oneProbe(r, probe) + if !ok { + return nil, fmt.Errorf("the run %s has no probe %q — %s lists them", r.Run, probe, ProbesTool) + } + r = one + } else { + r = Overview(r) + } + return map[string]any{"run": r, "age": c.Now().Sub(r.At).Round(time.Second).String(), + "note": "a probe that could not run is never a pass; each finding is a condition the keeper raises until a run no " + + "longer sees it, and one a single look can be wrong about is written when two runs in a row see it"}, nil +} + +func oneProbe(r Run, id string) (Run, bool) { + for _, p := range r.Probes { + if strings.EqualFold(p.ID, id) { + r.Probes = []ProbeVerdict{p} + return r, true + } + } + return Run{}, false +} + +// Probes is the registry as the verb answers it. +func (c *Checker) Probes() map[string]any { + var out []map[string]any + 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 p.Run == nil { + row["deferred"] = p.Deferred + } + out = append(out, row) + } + return map[string]any{"probes": out, "every": Every.String(), "each within": Within.String()} +} + +// Findings are the findings written now, or one whole. +func (c *Checker) Findings(key string) (any, error) { + if key != "" { + raw, found, err := c.Store.Get(FindingsState, key) + if err != nil { + return nil, err + } + if !found { + return nil, fmt.Errorf("no finding is written under %s now", key) + } + return json.RawMessage(raw), nil + } + written, err := c.written() + if err != nil { + return nil, fmt.Errorf("the findings cannot be read: %w", err) + } + rows := []map[string]any{} + for k, f := range written { + rows = append(rows, map[string]any{"key": k, "kind": f.Kind, "severity": f.Severity, "probe": f.Source, + "summary": f.Summary}) + } + sort.Slice(rows, func(i, j int) bool { return rows[i]["key"].(string) < rows[j]["key"].(string) }) + return map[string]any{"findings": rows, "count": len(rows)}, nil +} + +// Verdicts are the week's runs newest first, or one whole. +func (c *Checker) Verdicts(run string, limit int) (any, error) { + if run != "" { + raw, found, err := c.Store.Get(VerdictsState, run) + if err != nil { + return nil, err + } + if !found { + return nil, fmt.Errorf("no run %s is kept: runs are kept a week", run) + } + return json.RawMessage(raw), nil + } + keys, err := c.Store.Keys(VerdictsState) + if err != nil { + return nil, fmt.Errorf("the week's verdicts cannot be read: %w", err) + } + sort.Slice(keys, func(i, j int) bool { return runOrder(keys[i]) > runOrder(keys[j]) }) + total := len(keys) + if limit > 0 && len(keys) > limit { + keys = keys[:limit] + } + rows := []map[string]any{} + for _, k := range keys { + raw, found, err := c.Store.Get(VerdictsState, k) + if err != nil || !found { + continue + } + var r Run + if json.Unmarshal(raw, &r) == nil { + rows = append(rows, map[string]any{"run": r.Run, "at": r.At, "why": r.Why, "counts": r.Counts}) + } + } + return map[string]any{"verdicts": rows, "kept": total, "kept-for": "a week"}, nil +} + +// runOrder orders run ids `doctor--` by time, then number. +func runOrder(id string) string { + parts := strings.Split(id, "-") + if len(parts) != 3 { + return id + } + return fmt.Sprintf("%020s-%020s", parts[1], parts[2]) +} + +func strArg(a map[string]any, k string) string { + s, _ := a[k].(string) + return strings.TrimSpace(s) +} + +func boolArg(a map[string]any, k string) bool { + b, _ := a[k].(bool) + return b +} + +func intArg(a map[string]any, k string, otherwise int) int { + if n, ok := a[k].(float64); ok && n > 0 { + return int(n) + } + return otherwise +} diff --git a/modules/mesh-self-check/cmd/mesh-self-check/whole_integration_test.go b/modules/mesh-self-check/cmd/mesh-self-check/whole_integration_test.go new file mode 100644 index 00000000..8b82b408 --- /dev/null +++ b/modules/mesh-self-check/cmd/mesh-self-check/whole_integration_test.go @@ -0,0 +1,290 @@ +//go:build integration + +package main + +// The self-check whole against a real bus (novox/hq ADR 0564 §6, issue 1952 SC1): the built binary, played over its +// real stdin and stdout as the node's tool runtime plays it (tools/call; mesh/state.* answered from a real +// nats-server's key-value buckets, mesh/publish onto its subjects, and the bus module's nats_streams answered from +// that server's own streams). One run asked through the seat's verb finds the mesh's streams missing on a bare bus, +// writes them as findings into its bucket on the bus, keeps the verdict, and says checked. + +import ( + "bufio" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "os" + "os/exec" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/testbus" +) + +type wire struct { + ID json.RawMessage `json:"id,omitempty"` + Method string `json:"method,omitempty"` + Params json.RawMessage `json:"params,omitempty"` + Result json.RawMessage `json:"result,omitempty"` + Error *struct { + Message string `json:"message"` + } `json:"error,omitempty"` +} + +// player is the node's tool runtime, as far as the self-check needs it, over a real bus. +type player struct { + ctx context.Context + nc *nats.Conn + js jetstream.JetStream + mu sync.Mutex + in io.Writer + next int + pending map[string]chan wire +} + +func (p *player) send(m map[string]any) { + raw, _ := json.Marshal(m) + p.mu.Lock() + defer p.mu.Unlock() + _, _ = p.in.Write(append(raw, '\n')) +} + +func (p *player) answer(id json.RawMessage, result any, err error) { + if err != nil { + p.send(map[string]any{"id": id, "error": map[string]any{"code": -32000, "message": err.Error()}}) + return + } + p.send(map[string]any{"id": id, "result": result}) +} + +func (p *player) serve(out io.Reader) { + scanner := bufio.NewScanner(out) + scanner.Buffer(make([]byte, 1<<20), 1<<24) + for scanner.Scan() { + var m wire + if json.Unmarshal(scanner.Bytes(), &m) != nil { + continue + } + if m.Method == "" { + var key string + _ = json.Unmarshal(m.ID, &key) + p.mu.Lock() + ch := p.pending[key] + delete(p.pending, key) + p.mu.Unlock() + if ch != nil { + ch <- m + } + continue + } + go p.handle(m) + } +} + +func (p *player) kv(state string) (jetstream.KeyValue, error) { + return p.js.KeyValue(p.ctx, Module+"_"+state) +} + +func (p *player) handle(m wire) { + var a struct { + State string `json:"state"` + Key string `json:"key"` + Value json.RawMessage `json:"value"` + Body json.RawMessage `json:"body"` + } + _ = json.Unmarshal(m.Params, &a) + switch m.Method { + case "mesh/state.get": + kv, err := p.kv(a.State) + if err != nil { + p.answer(m.ID, nil, err) + return + } + e, err := kv.Get(p.ctx, a.Key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + p.answer(m.ID, nil, nil) + return + } + if err != nil { + p.answer(m.ID, nil, err) + return + } + p.answer(m.ID, map[string]any{"key": e.Key(), "value": json.RawMessage(e.Value()), "revision": e.Revision()}, nil) + case "mesh/state.put": + kv, err := p.kv(a.State) + if err == nil { + var rev uint64 + if rev, err = kv.Put(p.ctx, a.Key, a.Value); err == nil { + p.answer(m.ID, map[string]any{"revision": rev}, nil) + return + } + } + p.answer(m.ID, nil, err) + case "mesh/state.keys": + kv, err := p.kv(a.State) + if err != nil { + p.answer(m.ID, nil, err) + return + } + keys, err := kv.Keys(p.ctx) + if errors.Is(err, jetstream.ErrNoKeysFound) { + keys, err = []string{}, nil + } + p.answer(m.ID, keys, err) + case "mesh/state.delete": + kv, err := p.kv(a.State) + if err == nil { + err = kv.Delete(p.ctx, a.Key) + } + p.answer(m.ID, map[string]any{}, err) + case "mesh/publish": + var e struct { + Key string `json:"key"` + Body json.RawMessage `json:"body"` + } + _ = json.Unmarshal(m.Params, &e) + p.answer(m.ID, map[string]any{}, p.nc.Publish("mesh.mod."+Module+".event."+e.Key, e.Body)) + case "mesh/ask": + var ask struct { + Key string `json:"key"` + } + _ = json.Unmarshal(m.Params, &ask) + if ask.Key != StreamsTool { + p.answer(m.ID, nil, errors.New("nats: no responders available for request")) + return + } + // The bus module's answer, from the server's own streams. + var detail []map[string]any + for name := range p.js.StreamNames(p.ctx).Name() { + detail = append(detail, map[string]any{"name": name}) + } + p.answer(m.ID, map[string]any{"detail": detail}, nil) + default: + p.answer(m.ID, nil, errors.New("nats: no responders available for request")) + } +} + +func (p *player) call(name string, args map[string]any) (json.RawMessage, error) { + p.mu.Lock() + p.next++ + id := fmt.Sprintf("c-%d", p.next) + ch := make(chan wire, 1) + p.pending[id] = ch + p.mu.Unlock() + p.send(map[string]any{"id": id, "method": "tools/call", "params": map[string]any{"name": name, "arguments": args, + "_meta": map[string]any{"mesh/caller": "person.jochen"}}}) + select { + case m := <-ch: + if m.Error != nil { + return nil, errors.New(m.Error.Message) + } + var r struct { + Content []struct { + Text string `json:"text"` + } `json:"content"` + IsError bool `json:"isError"` + } + if err := json.Unmarshal(m.Result, &r); err != nil || len(r.Content) != 1 { + return nil, fmt.Errorf("%s answered %s", name, m.Result) + } + if r.IsError { + return nil, errors.New(r.Content[0].Text) + } + return json.RawMessage(r.Content[0].Text), nil + case <-time.After(60 * time.Second): + return nil, fmt.Errorf("%s did not answer in 60s", name) + } +} + +// 1952/SC1: one run of the self-check whole ends with checked and a verdict, its findings on the bus. +func TestTheSelfCheckWholeRunsOnTheBus(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + dir := t.TempDir() + bin := filepath.Join(dir, Module) + build := exec.Command("go", "build", "-o", bin, ".") + build.Stderr = os.Stderr + if err := build.Run(); err != nil { + t.Fatalf("the self-check does not build: %v", err) + } + nc, err := nats.Connect(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(nc.Close) + js, err := jetstream.New(nc) + if err != nil { + t.Fatal(err) + } + for _, s := range []string{FindingsState, VerdictState, VerdictsState} { + if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: Module + "_" + s}); err != nil { + t.Fatal(err) + } + } + checked, err := nc.SubscribeSync("mesh.mod." + Module + ".event." + CheckedEvent) + if err != nil { + t.Fatal(err) + } + cmd := exec.CommandContext(ctx, bin) + cmd.Env = append(os.Environ(), "MESH_NODE=ace", "MESH_MODULE="+Module) + stdin, _ := cmd.StdinPipe() + stdout, _ := cmd.StdoutPipe() + cmd.Stderr = os.Stderr + if err := cmd.Start(); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = stdin.Close(); cancel(); _ = cmd.Wait() }) + p := &player{ctx: ctx, nc: nc, js: js, in: stdin, pending: map[string]chan wire{}} + go p.serve(stdout) + + raw, err := p.call(DoctorTool, map[string]any{"run": true}) + if err != nil { + t.Fatal(err) + } + var answer struct { + Run Run `json:"run"` + } + if err := json.Unmarshal(raw, &answer); err != nil { + t.Fatal(err) + } + verdicts := map[string]string{} + for _, v := range answer.Run.Probes { + verdicts[v.ID] = v.Verdict + } + if verdicts["D7"] != VerdictFail || verdicts["DW"] != VerdictFail || verdicts["D15"] != VerdictFailedToRun || + verdicts["D2"] != VerdictDeferred || answer.Run.Controller != "ace" || !strings.Contains(answer.Run.Why, "person.jochen") { + t.Errorf("the run answered %+v", answer.Run) + } + msg, err := checked.NextMsg(10 * time.Second) + if err != nil { + t.Fatalf("checked was not said: %v", err) + } + var said Run + if err := json.Unmarshal(msg.Data, &said); err != nil || said.Run != answer.Run.Run { + t.Errorf("checked said %s", msg.Data) + } + findings, _ := js.KeyValue(ctx, Module+"_"+FindingsState) + e, err := findings.Get(ctx, "bus.EVENTS.stream-missing") + if err != nil { + t.Fatalf("the missing stream is not a finding on the bus: %v", err) + } + var f Finding + if err := json.Unmarshal(e.Value(), &f); err != nil || f.Kind != "stream-wrong" || f.Severity != Urgent || f.Source != "D7" { + t.Errorf("the finding is %s", e.Value()) + } + verdict, _ := js.KeyValue(ctx, Module+"_"+VerdictState) + if _, err := verdict.Get(ctx, LastKey); err != nil { + t.Errorf("the last verdict is not kept: %v", err) + } + if raw, err := p.call(VerdictsTool, map[string]any{}); err != nil || !strings.Contains(string(raw), answer.Run.Run) { + t.Errorf("the week's verdicts answered %s, %v", raw, err) + } +} diff --git a/modules/mesh-self-check/go.mod b/modules/mesh-self-check/go.mod new file mode 100644 index 00000000..abaa962a --- /dev/null +++ b/modules/mesh-self-check/go.mod @@ -0,0 +1,22 @@ +module mesh-self-check + +go 1.26.0 + +require ( + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029200316-22c9ddd32b6a + github.com/nats-io/nats.go v1.54.0 +) + +require ( + github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect + github.com/google/go-tpm v0.9.8 // indirect + github.com/klauspost/compress v1.20.0 // indirect + github.com/minio/highwayhash v1.0.4 // indirect + github.com/nats-io/jwt/v2 v2.8.1 // indirect + github.com/nats-io/nats-server/v2 v2.11.17 // indirect + github.com/nats-io/nkeys v0.4.16 // indirect + github.com/nats-io/nuid v1.0.1 // indirect + golang.org/x/crypto v0.57.0 // indirect + golang.org/x/sys v0.48.0 // indirect + golang.org/x/time v0.15.0 // indirect +) diff --git a/modules/mesh-self-check/go.sum b/modules/mesh-self-check/go.sum new file mode 100644 index 00000000..043e05f0 --- /dev/null +++ b/modules/mesh-self-check/go.sum @@ -0,0 +1,27 @@ +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029200316-22c9ddd32b6a h1:nhRFmtdcI4VkghGUdJhpt1BnPuTQcekioOIRzW8fcG0= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029200316-22c9ddd32b6a/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= +github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= +github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= +github.com/google/go-tpm v0.9.8/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY= +github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA= +github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= +github.com/minio/highwayhash v1.0.4 h1:asJizugGgchQod2ja9NJlGOWq4s7KsAWr5XUc9Clgl4= +github.com/minio/highwayhash v1.0.4/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ= +github.com/nats-io/jwt/v2 v2.8.1 h1:V0xpGuD/N8Mi+fQNDynXohVvp7ZztevW5io8CUWlPmU= +github.com/nats-io/jwt/v2 v2.8.1/go.mod h1:nWnOEEiVMiKHQpnAy4eXlizVEtSfzacZ1Q43LIRavZg= +github.com/nats-io/nats-server/v2 v2.11.17 h1:GKEghcFK6A+aFx11Yf1LjgLC3txAwvyhnYzhBIQZA8I= +github.com/nats-io/nats-server/v2 v2.11.17/go.mod h1:B1sFVz4StNosQ903ak4N1G01Fl/9f8e06mXpFIE2K24= +github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA= +github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k= +github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg= +github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs= +github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= +github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= +golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= +golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA= +golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= +golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= +golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= +golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= diff --git a/modules/mesh-self-check/module.json b/modules/mesh-self-check/module.json new file mode 100644 index 00000000..b0e732fe --- /dev/null +++ b/modules/mesh-self-check/module.json @@ -0,0 +1,67 @@ +{ + "module": "mesh-self-check", + "version": "1", + "runs-as": "mesh-self-check", + "claims": [ + { + "name": "mesh-self-check", + "scope": "mesh", + "serves": [ + "doctor", + "probes" + ] + } + ], + "emits": [ + "checked" + ], + "invokes": [ + "nats.nats_streams", + "nats.nats_closed_connections", + "seat:mesh-watchdogs.signals" + ], + "state": [ + "findings", + "verdict", + { + "name": "verdicts", + "ttl-seconds": 604800 + } + ], + "own-secrets": { + "broker": "${dir:state}/broker" + }, + "secrets-owner": "mesh-self-check", + "resources": [ + { + "id": "account", + "type": "user", + "name": "mesh-self-check", + "shell": "/usr/bin/nologin", + "home": "/var/lib/mesh-self-check" + }, + { + "id": "state", + "type": "directory", + "mode": "0700", + "owner": "mesh-self-check", + "place": ".", + "unknown": "program" + } + ], + "build": { + "artifacts": [ + { + "name": "tools", + "kind": "bundle", + "language": "go", + "system": "arch", + "from": "cmd/mesh-self-check", + "binary": "mesh-self-check", + "loads": [ + "mesh-self-check" + ] + } + ] + } +}