jschoubben
35ff67bda0
mesh/delivery-group group feat/774-l3-reland rejected: composed together: the checker's verdict at the members' heads (b48c90fea31f): integration: mesh-self-ch…
mesh/delivery closed: a newer head of the same pull request
mesh/check failed: integration: mesh-self-check/cmd/mesh-self-check.TestTheSelfCheckWholeRunsOnTheBus: the run answered {Run:doctor-1793351737-1 At:20…
One commit with exactly the tree of the reviewed head35b4509c0a, on main12c2574c9d, so two compiled programs added and removed on the branch leave no trace in history (issue 1577).
296 lines
12 KiB
Go
296 lines
12 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"slices"
|
|
"sort"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/health"
|
|
)
|
|
|
|
// 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 },
|
|
Publish: func(subject string, body any, id string) error {
|
|
raw, _ := json.Marshal(body)
|
|
said = append(said, subject+" "+id+" "+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 []Observation
|
|
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"},
|
|
Summary: "the stream EVENTS is not on the bus"}
|
|
behind := Observation{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 []Observation
|
|
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", []Observation{missing}, nil, []string{"bus.EVENTS.stream-missing"}, VerdictFail, 1},
|
|
{"the same again is no write", []Observation{missing}, nil, []string{"bus.EVENTS.stream-missing"}, VerdictFail, 1},
|
|
{"one look can be wrong: held", []Observation{missing, behind}, nil, []string{"bus.EVENTS.stream-missing"}, VerdictFail, 1},
|
|
{"seen twice: written", []Observation{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))
|
|
}
|
|
// Said on the seat's subject, where its consumers hear it, de-duplicated by the run's id.
|
|
if len(*said) != len(steps) || !strings.HasPrefix((*said)[0], "mesh.seat.mesh-self-check.event.checked doctor-") {
|
|
t.Fatalf("said %v", *said)
|
|
}
|
|
_, rest, _ := strings.Cut((*said)[0], " ")
|
|
_, rest, _ = strings.Cut(rest, " ")
|
|
var body map[string]any
|
|
_ = json.Unmarshal([]byte(rest), &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 = []Observation{missing}, nil
|
|
c.RunOnce(context.Background(), "the test")
|
|
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 == "" {
|
|
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) ([]Observation, error){
|
|
"panic": func(context.Context, *Checker) ([]Observation, error) { panic("boom") },
|
|
"overrun": func(ctx context.Context, _ *Checker) ([]Observation, 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([]Observation, 25)
|
|
for i := range many {
|
|
many[i] = Observation{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) ([]Observation, 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)
|
|
}
|
|
}
|
|
|
|
// Only the probes the setting `probes` lists run: one not listed runs nothing, has no verdict, clears what it raised before,
|
|
// 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"}
|
|
probe := func(id string) Probe {
|
|
return Probe{ID: id, Kind: "stream-wrong", Run: func(context.Context, *Checker) ([]Observation, error) {
|
|
ran[id]++
|
|
return []Observation{finds}, nil
|
|
}}
|
|
}
|
|
store := newStore()
|
|
c, _ := checker(store, probe("D7"), probe("D13"))
|
|
c.Enabled = func() []string { 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())
|
|
}
|
|
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" {
|
|
t.Fatalf("D7 unlisted: ran %v, verdicts %+v", ran, run.Probes)
|
|
}
|
|
// What the unlisted probe raised stays only as long as another listed probe sees the same key (D13 raises the same
|
|
// key here); an unlisted probe alone clears its findings.
|
|
c.Enabled = func() []string { return nil }
|
|
c.RunOnce(context.Background(), "test")
|
|
if len(store.findingKeys()) != 0 {
|
|
t.Fatalf("with nothing listed, findings are %v", store.findingKeys())
|
|
}
|
|
rows, _ := c.Probes()["probes"].([]map[string]any)
|
|
if len(rows) != 2 || rows[0]["off"] != "off by setting" || rows[1]["off"] != "off by setting" {
|
|
t.Fatalf("the verb says %v", rows)
|
|
}
|
|
c.Enabled = nil
|
|
if !c.On("anything") || !slices.Equal(DefaultProbes, []string{"D13", "D-fill"}) {
|
|
t.Fatal("no setting runs every probe, and the default list is D13 and D-fill")
|
|
}
|
|
}
|