mesh-controller/cmd/mesh-controller/doctor_integration_test.go

//go:build integration

package main

import (
	"context"
	"encoding/json"
	"errors"
	"slices"
	"strings"
	"sync/atomic"
	"testing"
	"time"

	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"
	"github.com/novox/mesh-controller/internal/broker"
	"github.com/novox/mesh-controller/internal/conditions"
	"github.com/novox/mesh-controller/internal/link"
	"github.com/novox/mesh-controller/internal/testbus"
)

func TestARunKeepsWhatEachProbeFoundAndSaysItsHeartbeat(t *testing.T) {
	failing := conditions.Observation{Scope: conditions.ScopeMachine, ID: "anchor", Token: "refused",
		Severity: conditions.Urgent, Summary: "anchor's node-engine would refuse its declaration"}
	var broken atomic.Bool
	broken.Store(true)
	withProbes(t,
		probe{ID: "P1", Asserts: "passes", Kind: "never", Phase: 1,
			run: func(context.Context, *doctor) ([]conditions.Observation, error) { return nil, nil }},
		probe{ID: "P2", Asserts: "finds a fault", Kind: "declaration-refused", Phase: 1,
			run: func(context.Context, *doctor) ([]conditions.Observation, error) {
				if broken.Load() {
					return []conditions.Observation{failing}, nil
				}
				return nil, nil
			}},
		probe{ID: "P3", Asserts: "cannot run", Kind: "x", Phase: 1,
			run: func(context.Context, *doctor) ([]conditions.Observation, error) {
				if broken.Load() {
					return nil, errors.New("the store is away")
				}
				return nil, nil
			}},
		probe{ID: "P4", Asserts: "hangs", Kind: "x", Phase: 1,
			run: func(ctx context.Context, _ *doctor) ([]conditions.Observation, error) {
				if broken.Load() {
					<-ctx.Done()
					time.Sleep(50 * time.Millisecond)
				}
				return nil, nil
			}},
		probe{ID: "P5", Asserts: "later", Kind: "x", Phase: 2, Deferred: "not yet"},
	)
	before := probeWithin
	probeWithin = 200 * time.Millisecond
	t.Cleanup(func() { probeWithin = before })

	store := conditions.NewInMemory()
	told := &conditions.Told{}
	k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store})
	defer k.Close(context.Background())
	d := &doctor{keeper: k, teller: told, host: "anchor"}
	// A probe that cannot run is said in the verdict at once, and as a condition when the next run cannot
	// run it either (novox/hq issue 277): one timed-out question is not a probe gone blind.
	first := d.runOnce(t.Context(), "a test")
	if first.Counts != (doctorCounts{Passed: 1, Failed: 1, FailedToRun: 2, Deferred: 1}) {
		t.Fatalf("counted %+v", first.Counts)
	}
	if open, _ := k.Open(t.Context()); len(open) != 1 || open[0].Key != "machine.anchor.refused" {
		t.Fatalf("after one run, open: %+v", open)
	}
	run := d.runOnce(t.Context(), "a test")
	if run.Counts != (doctorCounts{Passed: 1, Failed: 1, FailedToRun: 2, Deferred: 1}) {
		t.Fatalf("counted %+v", run.Counts)
	}
	open, err := k.Open(t.Context())
	if err != nil {
		t.Fatal(err)
	}
	var keys []string
	for _, c := range open {
		keys = append(keys, c.Key+"="+c.Kind)
	}
	for _, want := range []string{"machine.anchor.refused=declaration-refused", "probe.P3.failed=probe-failed",
		"probe.P4.failed=probe-failed"} {
		if !slices.Contains(keys, want) {
			t.Errorf("%s is not open: %v", want, keys)
		}
	}
	// The heartbeat, in the shape mesh-watcher reads (the contract with the operator's channel).
	if len(told.Names) != 2 || told.Names[1] != conditions.HeartbeatEvent {
		t.Fatalf("said %v", told.Names)
	}
	if d.lastRunEnded().IsZero() || d.lastRun().Run != run.Run {
		t.Fatal("the run is not the last verdict")
	}
	body, _ := json.Marshal(run)
	var shape map[string]any
	_ = json.Unmarshal(body, &shape)
	for _, field := range []string{"run", "at", "interval-seconds", "counts", "probes", "controller"} {
		if _, ok := shape[field]; !ok {
			t.Errorf("the heartbeat carries no %q: %s", field, body)
		}
	}

	// Mended: the next run clears every one of them.
	broken.Store(false)
	d.runOnce(t.Context(), "a test")
	if open, _ := k.Open(t.Context()); len(open) != 0 {
		t.Fatalf("a passing run left open %+v", open)
	}
}

func TestEveryMachineOfAHealthyMeshComposesAndValidates(t *testing.T) {
	open := aMesh(t)
	got, err := probeDeclarations(t.Context(), &doctor{open: open})
	if err != nil {
		t.Fatal(err)
	}
	if len(got) != 0 {
		t.Fatalf("a healthy mesh failed D1: %+v", got)
	}
}

func TestAMachineWhoseDeclarationDoesNotComposeIsSaid(t *testing.T) {
	open := aMesh(t)
	ctx := t.Context()
	one, two := rivals()
	register(t, open, one)
	register(t, open, two)
	for _, m := range []string{"rival-one", "rival-two"} {
		if _, err := assign(ctx, open, "laptop", m); err != nil && m == "rival-one" {
			t.Fatal(err)
		}
	}
	got, err := probeDeclarations(ctx, &doctor{open: open})
	linted(got)
	if err != nil {
		t.Fatal(err)
	}
	if len(got) != 1 || got[0].Key() != "machine.laptop.uncomposable" || !strings.Contains(got[0].Summary, "the-seat") {
		t.Fatalf("%+v", got)
	}
}

func TestNatsTheBusIsWhatTheControllerDefines(t *testing.T) {
	url := testbus.URL(t)
	open := aMesh(t)
	js, err := broker.Dial(url)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(js.Close)
	for _, s := range []string{"CONTROL", "NODES", "ASSIGNMENTS", "EVENTS"} {
		_ = js.Context().DeleteStream(s)
	}
	if _, err := assertBusObjects(t.Context(), open.inventory, js); err != nil {
		t.Fatal(err)
	}
	if err := js.EnsureControllerBuckets(); err != nil {
		t.Fatal(err)
	}
	d := &doctor{open: open, js: js}
	for _, p := range []func(context.Context, *doctor) ([]conditions.Observation, error){probeConsumers, probeStreams} {
		got, err := p(t.Context(), d)
		if err != nil || len(got) != 0 {
			t.Fatalf("a bus just raised fails: %+v %v", got, err)
		}
	}
	if err := js.Context().DeleteConsumer("NODES", "laptop"); err != nil {
		t.Fatal(err)
	}
	info, err := js.Context().StreamInfo("EVENTS")
	if err != nil {
		t.Fatal(err)
	}
	cfg := info.Config
	cfg.MaxMsgsPerSubject = 3
	if _, err := js.Context().UpdateStream(&cfg); err != nil {
		t.Fatal(err)
	}
	consumers, err := probeConsumers(t.Context(), d)
	if err != nil || len(consumers) != 1 || consumers[0].Key() != "bus.NODES.laptop.missing" {
		t.Fatalf("the deleted consumer: %+v %v", consumers, err)
	}
	streams, err := probeStreams(t.Context(), d)
	if err != nil || len(streams) != 1 || !strings.Contains(streams[0].Summary, "per subject") {
		t.Fatalf("the redefined stream: %+v %v", streams, err)
	}
}

func TestNatsTheBusSaysAConsumerGaveUpAndOneWasDeleted(t *testing.T) {
	url := testbus.URL(t)
	conn, err := nats.Connect(url)
	if err != nil {
		t.Fatal(err)
	}
	defer conn.Close()
	heard := make(chan *nats.Msg, 16)
	for _, subject := range broker.BusAdvisories {
		if _, err := conn.ChanSubscribe(subject, heard); err != nil {
			t.Fatal(err)
		}
	}
	api, _ := jetstream.New(conn)
	_ = api.DeleteStream(t.Context(), "SEAT_ADVISED")
	stream, err := api.CreateStream(t.Context(), jetstream.StreamConfig{Name: "SEAT_ADVISED", Subjects: []string{"advised.>"}})
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = api.DeleteStream(context.Background(), "SEAT_ADVISED") }()
	consumer, err := stream.CreateConsumer(t.Context(), jetstream.ConsumerConfig{Durable: "SEAT_ADVISED_worker",
		AckPolicy: jetstream.AckExplicitPolicy, MaxDeliver: 1, AckWait: 100 * time.Millisecond})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := api.Publish(t.Context(), "advised.x", []byte("x")); err != nil {
		t.Fatal(err)
	}
	if _, err := consumer.Fetch(1, jetstream.FetchMaxWait(time.Second)); err != nil {
		t.Fatal(err)
	}
	// A seat's worker, so the deletion is of a consumer the mesh names (link.MeshNamed).
	// Not acknowledged: after its one delivery the consumer gives up on it — on the next fetch.
	time.Sleep(300 * time.Millisecond)
	_, _ = consumer.Fetch(1, jetstream.FetchMaxWait(300*time.Millisecond))
	if err := stream.DeleteConsumer(t.Context(), "SEAT_ADVISED_worker"); err != nil {
		t.Fatal(err)
	}
	kinds := map[string]string{}
	deadline := time.After(5 * time.Second)
	for len(kinds) < 2 {
		select {
		case m := <-heard:
			if a, ok := link.ReadAdvisory(m.Subject, m.Data); ok {
				kinds[a.Kind] = a.Said
			}
		case <-deadline:
			t.Fatalf("the bus said only %v", kinds)
		}
	}
	if !strings.Contains(kinds["max-deliveries"], "gave up") || !strings.Contains(kinds["consumer-lost"], "was deleted") {
		t.Fatalf("%v", kinds)
	}
}