mesh-catalog/modules/mesh-self-check/cmd/mesh-self-check/selfcheck_test.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 and the reads one key at a time, and handing every
// change to its watches.
type memStore struct {
	mu      sync.Mutex
	kv      map[string]map[string][]byte
	puts    map[string]int
	gets    map[string]int
	watches map[string][]func(key string, value []byte, gone bool)
	keysErr error
}

func newStore() *memStore {
	return &memStore{kv: map[string]map[string][]byte{}, puts: map[string]int{}, gets: map[string]int{},
		watches: map[string][]func(string, []byte, bool){}}
}

func (s *memStore) Get(state, key string) ([]byte, bool, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.gets[state]++
	v, ok := s.kv[state][key]
	return v, ok, nil
}

func (s *memStore) Watch(state string, handle func(key string, value []byte, gone bool)) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	for k, v := range s.kv[state] {
		handle(k, v, false)
	}
	s.watches[state] = append(s.watches[state], handle)
	return nil
}

func (s *memStore) Put(state, key string, value any) error {
	raw, err := json.Marshal(value)
	if err != nil {
		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]++
	for _, w := range s.watches[state] {
		w(key, raw, false)
	}
	return nil
}

func (s *memStore) Delete(state, key string) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	delete(s.kv[state], key)
	for _, w := range s.watches[state] {
		w(key, nil, true)
	}
	return nil
}

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{Ref: "stream-missing", Severity: Urgent, Subject: Subject{Scope: ScopeBus, ID: "EVENTS"},
		Summary: "the stream EVENTS is not on the bus"}
	behind := Observation{Kind: "consumer-behind", Ref: "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.Ref != "stream-missing" || len(f.Evidence) != 1 || f.Run == "" {
		t.Errorf("the written finding is %+v", f)
	}
	// Blind to its own bucket, a run writes and clears nothing.
	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{Ref: "stream-missing", Severity: Urgent, Subject: Subject{Scope: ScopeBus, ID: "EVENTS"}, Summary: "gone"}
	probe := func(id string) Probe {
		return Probe{ID: id, Kind: "stream-wrong", Run: func(context.Context, *Checker) ([]Observation, error) {
			ran[id]++
			return []Observation{finds}, nil
		}}
	}
	store := newStore()
	c, _ := checker(store, probe("D7"), probe("D13"))
	read := 0
	c.Enabled = func() []string { read++; return []string{"D7", "D13"} }
	run := c.RunOnce(context.Background(), "test")
	if ran["D7"] != 1 || ran["D13"] != 1 || len(run.Probes) != 2 || len(store.findingKeys()) != 1 {
		t.Fatalf("both listed: ran %v, %d verdicts, findings %v", ran, len(run.Probes), store.findingKeys())
	}
	// The setting is read once a run, and once an answer of the verb, never once a probe (issue 1952).
	if c.Probes(); read != 2 {
		t.Fatalf("a run and the verb read the setting %d times, want 2", read)
	}
	c.Enabled = func() []string { return []string{"D13"} }
	run = c.RunOnce(context.Background(), "test")
	if ran["D7"] != 1 || ran["D13"] != 2 || len(run.Probes) != 1 || run.Probes[0].ID != "D13" {
		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")
	}
}