mesh-controller/cmd/mesh-controller/healers_test.go

package main

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

	"github.com/nats-io/nats.go"

	"github.com/novox/mesh-controller/internal/broker"
	"github.com/novox/mesh-controller/internal/conditions"
	"github.com/novox/mesh-controller/internal/inventory"
	"github.com/novox/mesh-controller/internal/link"
)

// The test generated from the healer registry (novox/hq to-be 45 §7, ADR 0227 rule 7 "how it is
// checked"): **every row is walked.** Its kinds are ones the mesh raises — a row of the signals table, a
// probe of the self-check, or a provider's event — it has a budget, a window and a settle inside it,
// what happens when the budget is spent, and the event each act is said as; a healer the controller runs
// has its applies and its repair, and an induced failure below that sees it act, say so and brake. A
// row added without one fails, so the registry cannot grow a healer nobody has seen act.

// raisedKinds is every condition kind the mesh raises, and what raises it.
func raisedKinds() map[string]string {
	out := map[string]string{kindProviderFailing: "the provisioner.failing event (ADR 0224)"}
	for _, r := range signalsTable {
		for _, k := range kindsOf(r) {
			out[k] = r.Row
		}
	}
	for _, p := range probeRegistry {
		out[p.Kind] = p.ID
		for _, k := range p.Raises {
			out[k] = p.ID
		}
	}
	return out
}

// inducedFailures are the healers seen acting in this file, by id: a row without one fails.
var inducedFailures = map[string]string{
	"H1": "TestH1AsksAMachineToReportAndSendsItAgain",
	"H2": "TestH2ClosesAPlanAnotherHasTakenOver",
	"H3": "TestNatsH3AssertsAMissingConsumerAgainAndBrakesAfterItsBudget",
	"H4": "TestNatsH4ResetsTheControllersEventsConsumerAndItStillDelivers",
}

func TestEveryHealerAnswersAKindTheMeshRaisesWithABudgetABrakeAndItsEvent(t *testing.T) {
	kinds := raisedKinds()
	source, err := os.ReadFile("healers_test.go")
	if err != nil {
		t.Fatal(err)
	}
	seen := map[string]bool{}
	answered := map[string]string{}
	for _, r := range healerRegistry {
		t.Run(r.ID, func(t *testing.T) {
			if seen[r.ID] {
				t.Fatalf("%s is in the registry twice", r.ID)
			}
			seen[r.ID] = true
			if len(r.Kinds) == 0 || r.Condition == "" || r.Repair == "" || r.Then == "" || r.From == "" || r.ActsIn == "" {
				t.Fatalf("%s does not say what it answers, what it does, what then, and which hand act it replaces: %+v", r.ID, r)
			}
			for _, k := range r.Kinds {
				if _, ok := kinds[k]; !ok {
					t.Errorf("%s answers %q, which nothing in the mesh raises", r.ID, k)
				}
				if other, twice := answered[k]; twice {
					t.Errorf("%q is answered by %s and %s: one healer per kind", k, other, r.ID)
				}
				answered[k] = r.ID
			}
			if r.Budget <= 0 || r.Window <= 0 || r.Settle <= 0 || r.Settle > r.Window {
				t.Errorf("%s has no budget it can spend: %d within %s, settled after %s", r.ID, r.Budget, r.Window, r.Settle)
			}
			if r.Event == "" {
				t.Errorf("%s says nothing when it acts", r.ID)
			}
			if r.ActsIn != actsInController {
				if r.repair != nil || r.applies != nil {
					t.Errorf("%s acts in %s and the controller would act for it too", r.ID, r.ActsIn)
				}
				return
			}
			if r.repair == nil || r.applies == nil {
				t.Fatalf("%s acts in the controller and has no repair or no applies", r.ID)
			}
			if r.Event != link.KeyHealerActed || !slices.Contains(broker.ControllerStates, r.Event) {
				t.Errorf("%s is said as %q, which the controller's grant does not permit", r.ID, r.Event)
			}
			if inducedFailures[r.ID] == "" {
				t.Errorf("%s has no induced failure: a healer nobody has seen act", r.ID)
			} else if !strings.Contains(string(source), "func "+inducedFailures[r.ID]+"(t *testing.T)") {
				t.Errorf("%s's induced failure %s is not a test in this file", r.ID, inducedFailures[r.ID])
			}
		})
	}
	for id := range inducedFailures {
		if !seen[id] {
			t.Errorf("an induced failure for %s, which the registry does not have", id)
		}
	}
	if healBrakeLimit <= 0 || healBrakeWindow <= 0 {
		t.Error("the mesh-wide brake holds nothing")
	}
}

// heard keeps the healer-acted events said.
type heard struct {
	mu   sync.Mutex
	acts []healerActed
}

func (h *heard) PublishSeatEvent(_ context.Context, seat, event string, body []byte) error {
	if seat != conditions.Seat || event != link.KeyHealerActed {
		return errors.New("said under the wrong seat or name: " + seat + " " + event)
	}
	var e healerActed
	if err := json.Unmarshal(body, &e); err != nil {
		return err
	}
	h.mu.Lock()
	defer h.mu.Unlock()
	h.acts = append(h.acts, e)
	return nil
}

func (h *heard) said() []healerActed {
	h.mu.Lock()
	defer h.mu.Unlock()
	return append([]healerActed(nil), h.acts...)
}

// testClock is a moment a test moves by hand.
type testClock struct {
	mu sync.Mutex
	at time.Time
}

func (c *testClock) now() time.Time {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.at
}

func (c *testClock) pass(d time.Duration) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.at = c.at.Add(d)
}

// healingOn is a runner over a mesh's stores and its condition store, under epoch 57, every act a
// fake that fails the test unless the test gives it.
func healingOn(t *testing.T, open *stores) (*healing, *heard, *testClock) {
	t.Helper()
	if conditionsFrom == nil {
		t.Fatal("the mesh has no condition store")
	}
	told, clock := &heard{}, &testClock{at: time.Now()}
	// The store's gate, as the serving controller's is the lease's (stores.go).
	open.inventory.ActsUnder(func(context.Context) (uint64, error) { return 57, nil })
	h := &healing{open: open, keeper: conditionsFrom, teller: told,
		epoch: func(context.Context) (uint64, error) { return 57, nil }, now: clock.now,
		say: func(f string, a ...any) { t.Logf(f, a...) }, reportWait: 300 * time.Millisecond,
		declined: map[string]string{}}
	h.askReport = func(context.Context, string) error { t.Error("asked a machine to report"); return nil }
	h.sendAgain = func(context.Context, string) error { t.Error("sent a machine again"); return nil }
	h.assertObjects = func(context.Context) error { t.Error("asserted the bus's objects"); return nil }
	h.resetConsumer = func(string, string) (string, error) { t.Error("reset a consumer"); return "", nil }
	return h, told, clock
}

// sentNotReported raises S2 for a machine, as the watchdog does.
func sentNotReported(t *testing.T, node string) string {
	t.Helper()
	o := conditions.Observation{Scope: conditions.ScopeMachine, ID: node, Kind: "sent-not-reported", Machine: node,
		Severity: conditions.Warning, Summary: node + " was sent a declaration and has not reported it", Source: "S2"}
	if _, err := conditionsFrom.Observe(t.Context(), o); err != nil {
		t.Fatal(err)
	}
	return o.Key()
}

// **H1, the commonest hand act** (031/01 §f: a push by hand to unstick a plan waiting on a report, four
// times): the machine is asked to report; a report that names what it was sent is all it takes, and one
// that does not — or none — is a send of its current declaration again. Each act kept in `tried` as
// `healer H1`, said as healer-acted, counted; twice, then the operator's, urgent; then nothing more.
func TestH1AsksAMachineToReportAndSendsItAgain(t *testing.T) {
	open := aMesh(t)
	ctx := t.Context()
	h, told, clock := healingOn(t, open)
	record, err := open.inventory.NodeByName(ctx, "laptop")
	if err != nil {
		t.Fatal(err)
	}
	if err := open.inventory.RecordSent(ctx, record.ID, "d2", nil); err != nil {
		t.Fatal(err)
	}
	key := sentNotReported(t, "laptop")

	// First: the report had not reached the mesh, and asking for it brings it.
	asked := 0
	h.askReport = func(ctx context.Context, node string) error {
		asked++
		if node != "laptop" {
			t.Errorf("asked %s", node)
		}
		_, err := open.inventory.RecordDoing(ctx, record.ID, inventory.Doing{Outcome: inventory.OutcomeApplied,
			At: time.Now().Add(time.Second), Declared: "d2"})
		return err
	}
	h.tick(ctx)
	c, found, err := conditionsFrom.Get(ctx, key)
	if err != nil || !found {
		t.Fatalf("the condition is gone after the act — a healer cleared it, not an observation: %v", err)
	}
	if asked != 1 || len(c.Tried) != 1 || c.Tried[0].By != "healer H1" || !strings.Contains(c.Tried[0].Outcome, "acted") ||
		c.Resolver != "healer:H1" {
		t.Fatalf("after the first act: asked %d, %+v", asked, c)
	}
	if acts := told.said(); len(acts) != 1 || acts[0].Healer != "H1" || acts[0].Outcome != inventory.HealActed ||
		acts[0].Condition != key || acts[0].By != "healer H1" || acts[0].Epoch != 57 {
		t.Fatalf("the act was not said as healer-acted: %+v", acts)
	}

	// Within its settle, nothing more: the observation has its turn.
	clock.pass(time.Minute)
	h.tick(ctx)
	if asked != 1 {
		t.Fatalf("acted again inside the settle: asked %d", asked)
	}

	// Second: the machine says nothing, so it is sent again.
	clock.pass(3 * time.Minute)
	sent := 0
	h.askReport = func(context.Context, string) error { asked++; return nil }
	h.sendAgain = func(_ context.Context, node string) error { sent++; return nil }
	h.tick(ctx)
	if asked != 2 || sent != 1 {
		t.Fatalf("the second act: asked %d, sent %d", asked, sent)
	}
	c, _, _ = conditionsFrom.Get(ctx, key)
	if len(c.Tried) != 2 || !strings.Contains(c.Tried[1].Outcome, "sent again") || !strings.Contains(c.Tried[1].Outcome, "act 2 of 2") {
		t.Fatalf("tried %+v", c.Tried)
	}

	// The budget is spent: the operator's, urgent, said; and no healer touches it again.
	clock.pass(4 * time.Minute)
	h.tick(ctx)
	c, _, _ = conditionsFrom.Get(ctx, key)
	if !c.Escalated() || c.Severity != conditions.Urgent || len(c.Tried) != 3 ||
		!strings.Contains(c.Tried[2].Outcome, "budget of 2") {
		t.Fatalf("not handed to the operator: %+v", c)
	}
	if acts := told.said(); len(acts) != 3 || acts[2].Outcome != inventory.HealEscalated {
		t.Fatalf("the escalation was not said: %+v", acts)
	}
	clock.pass(time.Hour)
	h.tick(ctx)
	if asked != 2 || sent != 1 || len(told.said()) != 3 {
		t.Fatalf("a healer acted on a condition the operator holds: asked %d sent %d said %d", asked, sent, len(told.said()))
	}
	heals, err := open.inventory.HealsSince(ctx, time.Now().Add(-time.Hour))
	if err != nil || len(heals) != 3 || heals[0].Outcome != inventory.HealActed || heals[1].Outcome != inventory.HealActed ||
		heals[2].Outcome != inventory.HealEscalated || heals[0].Epoch != 57 {
		t.Fatalf("the heals kept: %+v %v", heals, err)
	}
	// And a heal is not a hand act: what S15 counts never sees it.
	for _, x := range heals {
		if x.Healer == "" || strings.HasPrefix(x.Act, "hand-act") {
			t.Errorf("a heal reads as a hand act: %+v", x)
		}
	}
}

// **Only the controller holding the lease heals** (to-be 45 §6): unleased, or standing by, nothing.
func TestNoHealerActsWithoutTheLease(t *testing.T) {
	open := aMesh(t)
	ctx := t.Context()
	h, told, _ := healingOn(t, open)
	sentNotReported(t, "laptop")
	h.epoch = func(context.Context) (uint64, error) { return 0, nil }
	h.tick(ctx)
	h.epoch = func(context.Context) (uint64, error) { return 0, errors.New("the lease is not held") }
	h.tick(ctx)
	h.epoch = func(context.Context) (uint64, error) { return 57, nil }
	h.acting = func() bool { return false }
	h.tick(ctx)
	if len(told.said()) != 0 {
		t.Fatalf("a healer acted without the lease: %+v", told.said())
	}
}

// **The mesh-wide brake**: a dozen acts in an hour and every healer stops, said urgently, until an hour
// after the last.
func TestTheBrakeStopsEveryHealerAndSaysSo(t *testing.T) {
	open := aMesh(t)
	ctx := t.Context()
	h, told, clock := healingOn(t, open)
	for i := 0; i < healBrakeLimit; i++ {
		if _, err := open.inventory.BeginHeal(ctx, inventory.Heal{Healer: "H3", ConditionKey: "seat.x.anchor.silent",
			Kind: "holder-silent", Act: "asserted", Outcome: inventory.HealActed,
			At: clock.now().Add(-time.Duration(healBrakeLimit-i) * time.Minute)}); err != nil {
			t.Fatal(err)
		}
	}
	sentNotReported(t, "laptop")
	h.tick(ctx) // the fakes fail the test if anything acts
	braked, found, err := conditionsFrom.Get(ctx, "mesh.healers.braked")
	if err != nil || !found || braked.Severity != conditions.Urgent || !strings.Contains(braked.Evidence[0].Said, "H3 on seat.x.anchor.silent ×12") {
		t.Fatalf("the brake was not said: %+v %v", braked, err)
	}
	if len(told.said()) != 0 {
		t.Fatalf("a healer acted under the brake: %+v", told.said())
	}
	// Past the hour of the first act, still held: it lets go an hour after the last.
	clock.pass(30 * time.Minute)
	h.tick(ctx)
	if _, held, _ := conditionsFrom.Get(ctx, "mesh.healers.braked"); !held {
		t.Fatal("the brake let go before an hour had passed since the last act")
	}
	clock.pass(31 * time.Minute)
	asked := 0
	h.askReport = func(context.Context, string) error { asked++; return nil }
	h.sendAgain = func(context.Context, string) error { return nil }
	h.tick(ctx)
	if _, held, _ := conditionsFrom.Get(ctx, "mesh.healers.braked"); held {
		t.Fatal("the brake held an hour after the last act")
	}
	if asked != 1 {
		t.Fatalf("the healers did not act again after the brake let go: asked %d", asked)
	}
}

// **H2 closes a plan another has taken over**, with its note — and leaves a plan alone whose wait is
// still to come: that one is its own signal's, not a healer's.
func TestH2ClosesAPlanAnotherHasTakenOver(t *testing.T) {
	open := aMesh(t)
	ctx := t.Context()
	h, told, _ := healingOn(t, open)
	created := time.Now().UTC().Add(-time.Hour)
	older := inventory.Plan{ID: "plan-old", Repository: "novox/app", Branch: "main", Commit: "0ld0ld0", Created: created,
		State: inventory.PlanBuilding, Tiers: [][]string{{"a"}}, Modules: map[string]*inventory.PlanModule{"a": {State: "asked"}}}
	waiting := inventory.Plan{ID: "plan-other", Repository: "novox/other", Branch: "main", Commit: "07he707", Created: created,
		State: inventory.PlanBuilding, Tiers: [][]string{{"b"}}, Modules: map[string]*inventory.PlanModule{"b": {State: "asked"}}}
	newer := inventory.Plan{ID: "plan-new", Repository: "novox/app", Branch: "main", Commit: "new0new", Created: created.Add(time.Minute),
		State: inventory.PlanDone, Tiers: [][]string{{"a"}}, Modules: map[string]*inventory.PlanModule{"a": {State: "built"}}}
	for _, p := range []*inventory.Plan{&older, &waiting, &newer} {
		if err := open.inventory.SavePlan(ctx, p); err != nil {
			t.Fatal(err)
		}
	}
	for _, id := range []string{"plan-old", "plan-other"} {
		if _, err := conditionsFrom.Observe(ctx, conditions.Observation{Scope: conditions.ScopePlan, ID: id, Kind: "stalled",
			Severity: conditions.Warning, Summary: id + " is stalled", Source: "S3"}); err != nil {
			t.Fatal(err)
		}
	}
	h.tick(ctx)
	closed, err := open.inventory.PlanByID(ctx, "plan-old")
	if err != nil || closed.State != inventory.PlanSuperseded || !strings.Contains(closed.Note, "closed by healer H2") ||
		!strings.Contains(closed.Note, "plan-new") {
		t.Fatalf("the superseded plan: %+v %v", closed, err)
	}
	if other, _ := open.inventory.PlanByID(ctx, "plan-other"); other.State != inventory.PlanBuilding {
		t.Fatalf("a plan still waiting was closed: %+v", other)
	}
	if c, _, _ := conditionsFrom.Get(ctx, "plan.plan-other.stalled"); len(c.Tried) != 0 || c.Resolver != conditions.ResolverSelf {
		t.Fatalf("a plan H2 does not repair was touched: %+v", c)
	}
	if acts := told.said(); len(acts) != 1 || acts[0].Healer != "H2" || acts[0].Condition != "plan.plan-old.stalled" {
		t.Fatalf("said %+v", acts)
	}
	// Finished: every module built, none rolled out unsent — closed as done.
	done := inventory.Plan{ID: "plan-done", Repository: "novox/third", Branch: "main", Commit: "d0ned0n", Created: created,
		State: inventory.PlanRolling, Tier: 0, Tiers: [][]string{{"c"}}, Modules: map[string]*inventory.PlanModule{"c": {State: "built"}}}
	if err := open.inventory.SavePlan(ctx, &done); err != nil {
		t.Fatal(err)
	}
	if state, why, err := planStale(ctx, open.inventory, done); err != nil || state != inventory.PlanDone || !strings.Contains(why, "finished") {
		t.Fatalf("a finished plan reads %q %q %v", state, why, err)
	}
}

// **H4 only for a consumer the stream table marks resettable**: a module's consumer far behind is said
// and left — what a reset drops, nothing would catch up for it.
func TestH4ResetsOnlyWhatTheTableMarksResettable(t *testing.T) {
	open := aMesh(t)
	ctx := t.Context()
	h, _, _ := healingOn(t, open)
	marked := 0
	for _, c := range broker.MeshConsumers() {
		if c.Resettable != "" {
			marked++
			if c.Stream != broker.EventsStream || c.Name != broker.ControllerName {
				t.Errorf("%s on %s is marked resettable: only the controller's own events consumer is", c.Name, c.Stream)
			}
		}
	}
	if marked != 1 {
		t.Fatalf("%d consumers are marked resettable, want the controller's events consumer alone", marked)
	}
	for _, who := range []string{"EVENTS.anchor_shop", "CONTROL.controller"} {
		if _, err := conditionsFrom.Observe(ctx, conditions.Observation{Scope: conditions.ScopeBus, ID: who, Token: "behind",
			Kind: kindConsumerBehind, Severity: conditions.Warning, Summary: who + " is far behind", Source: "D6"}); err != nil {
			t.Fatal(err)
		}
	}
	h.tick(ctx) // the fake reset fails the test if it is called
	reset := ""
	h.resetConsumer = func(stream, name string) (string, error) {
		reset = stream + "." + name
		return "it was 1500 behind", nil
	}
	if _, err := conditionsFrom.Observe(ctx, conditions.Observation{Scope: conditions.ScopeBus, ID: "EVENTS.controller",
		Token: "behind", Kind: kindConsumerBehind, Severity: conditions.Warning, Summary: "far behind", Source: "D6"}); err != nil {
		t.Fatal(err)
	}
	h.tick(ctx)
	if reset != "EVENTS.controller" {
		t.Fatalf("reset %q", reset)
	}
}

// **H3 against a real bus** (issue 208's note): a consumer the mesh expects, deleted; D6 says so; H3
// asserts the bus's objects the way a send does, and D6's next run clears it — the healer never does.
// Deleted again within the hour, the budget is spent: the operator's, urgent, and H3 stops.
func TestNatsH3AssertsAMissingConsumerAgainAndBrakesAfterItsBudget(t *testing.T) {
	url := os.Getenv("MESH_TEST_NATS")
	if url == "" {
		t.Skip("MESH_TEST_NATS unset")
	}
	open := aMesh(t)
	ctx := t.Context()
	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(ctx, open.inventory, js); err != nil {
		t.Fatal(err)
	}
	h, told, clock := healingOn(t, open)
	h.js = js
	h.assertObjects = func(ctx context.Context) error { return assertOnSend(ctx, open.inventory, js, "  ") }
	d := &doctor{open: open, js: js}
	probe := func() {
		t.Helper()
		found, err := probeConsumers(ctx, d)
		if err != nil {
			t.Fatal(err)
		}
		if err := conditionsFrom.Reconcile(ctx, "D6", kindedAs(found, "consumer-wrong")); err != nil {
			t.Fatal(err)
		}
	}
	const key = "bus.NODES.laptop.missing"

	if err := js.Context().DeleteConsumer("NODES", "laptop"); err != nil {
		t.Fatal(err)
	}
	probe()
	if c, raised, _ := conditionsFrom.Get(ctx, key); !raised || c.Kind != "consumer-lost" {
		t.Fatalf("the deleted consumer was not raised: %+v", c)
	}
	h.tick(ctx)
	if _, err := js.Context().ConsumerInfo("NODES", "laptop"); err != nil {
		t.Fatalf("H3 did not make the consumer again: %v", err)
	}
	c, still, _ := conditionsFrom.Get(ctx, key)
	if !still || len(c.Tried) != 1 || c.Tried[0].By != "healer H3" || c.Resolver != "healer:H3" {
		t.Fatalf("the act is not in the condition, or the healer cleared it: %+v", c)
	}
	probe()
	if _, still, _ := conditionsFrom.Get(ctx, key); still {
		t.Fatal("the probe's next run did not clear what H3 repaired")
	}
	// Kept in the history as the keeper says it, a moment later.
	var cleared *conditions.Event
	for wait := time.Now().Add(5 * time.Second); cleared == nil && time.Now().Before(wait); time.Sleep(20 * time.Millisecond) {
		history, err := conditionsFrom.HistorySince(ctx, time.Now().Add(-time.Minute))
		if err != nil {
			t.Fatal(err)
		}
		for i := range history {
			if history[i].Key == key && history[i].Change == conditions.ChangeCleared {
				cleared = &history[i]
			}
		}
	}
	if cleared == nil || !strings.Contains(cleared.Why, "D6 no longer observes it") || len(cleared.Tried) != 1 {
		t.Fatalf("the clearing is not the probe's, or forgets what was tried: %+v", cleared)
	}

	// Again within the hour: the budget is one, so after its settle the operator is told, and H3 stops.
	clock.pass(10 * time.Minute)
	if err := js.Context().DeleteConsumer("NODES", "laptop"); err != nil {
		t.Fatal(err)
	}
	probe()
	h.assertObjects = func(context.Context) error { t.Error("H3 acted past its budget"); return nil }
	h.tick(ctx)
	c, _, _ = conditionsFrom.Get(ctx, key)
	if !c.Escalated() || c.Severity != conditions.Urgent || c.Count != 2 {
		t.Fatalf("the spent budget was not handed to the operator: %+v", c)
	}
	acts := told.said()
	if len(acts) != 2 || acts[0].Outcome != inventory.HealActed || acts[1].Outcome != inventory.HealEscalated {
		t.Fatalf("said %+v", acts)
	}
	clock.pass(2 * time.Hour)
	h.tick(ctx)
	if len(told.said()) != 2 {
		t.Fatal("a healer acted on what the operator holds")
	}
}

// **H4 against a real bus** (issue 248): the controller's events consumer a long way behind — the week it
// once replayed — is re-made from now by the mesh itself, and the controller's bound subscription still
// receives what comes next: a reset that left the controller deaf would be the incident.
func TestNatsH4ResetsTheControllersEventsConsumerAndItStillDelivers(t *testing.T) {
	url := os.Getenv("MESH_TEST_NATS")
	if url == "" {
		t.Skip("MESH_TEST_NATS unset")
	}
	open := aMesh(t)
	ctx := t.Context()
	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(ctx, open.inventory, js); err != nil {
		t.Fatal(err)
	}
	// Bound as the controller binds it (link/receive_nats.go), taking one and acknowledging none.
	events := make(chan *nats.Msg, 64)
	sub, err := js.Context().ChanSubscribe("", events, nats.Bind(broker.EventsStream, broker.ControllerName))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = sub.Unsubscribe() })
	followed := broker.ControllerFollows[0]
	for i := 0; i < consumerFarBehind+200; i++ {
		if _, err := js.Context().Publish(followed, []byte(`{"n":`+strconv.Itoa(i)+`}`)); err != nil {
			t.Fatal(err)
		}
	}
	d := &doctor{open: open, js: js}
	probe := func() []conditions.Observation {
		t.Helper()
		found, err := probeConsumers(ctx, d)
		if err != nil {
			t.Fatal(err)
		}
		if err := conditionsFrom.Reconcile(ctx, "D6", kindedAs(found, "consumer-wrong")); err != nil {
			t.Fatal(err)
		}
		return found
	}
	probe()
	const key = "bus.EVENTS.controller.behind"
	if c, raised, _ := conditionsFrom.Get(ctx, key); !raised || c.Kind != kindConsumerBehind {
		t.Fatalf("a consumer %d behind was not raised: %+v", consumerFarBehind+200, c)
	}
	h, told, _ := healingOn(t, open)
	h.resetConsumer = func(stream, name string) (string, error) {
		before, after, err := js.ResetConsumer(stream, name)
		if err != nil {
			return "", err
		}
		return "it was " + strconv.FormatUint(before.Pending, 10) + " behind; " + strconv.FormatUint(after.Pending, 10) +
			" pending now", nil
	}
	h.tick(ctx)
	if acts := told.said(); len(acts) != 1 || acts[0].Healer != "H4" || acts[0].Outcome != inventory.HealActed {
		t.Fatalf("said %+v", acts)
	}
	if found := probe(); len(found) != 0 {
		t.Fatalf("after the reset the probe still finds %+v", found)
	}
	if _, still, _ := conditionsFrom.Get(ctx, key); still {
		t.Fatal("the probe's next run did not clear what H4 repaired")
	}
	// What comes next still reaches the controller.
	for len(events) > 0 {
		<-events
	}
	if _, err := js.Context().Publish(followed, []byte(`{"after":"the reset"}`)); err != nil {
		t.Fatal(err)
	}
	deadline := time.After(10 * time.Second)
	for {
		select {
		case m := <-events:
			if strings.Contains(string(m.Data), "after") {
				return
			}
			_ = m.Ack()
		case <-deadline:
			t.Fatal("the controller's subscription heard nothing after its consumer was reset: it would be deaf")
		}
	}
}

// **H1's question reaches the machine**, on the subject its grant lets it hear and nothing else.
func TestNatsAskToReportReachesTheMachine(t *testing.T) {
	url := os.Getenv("MESH_TEST_NATS")
	if url == "" {
		t.Skip("MESH_TEST_NATS unset")
	}
	conn, err := nats.Connect(url)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(conn.Close)
	asked, err := conn.SubscribeSync(broker.AskReportSubject("laptop"))
	if err != nil {
		t.Fatal(err)
	}
	if err := askToReport(t.Context(), conn, "laptop"); err != nil {
		t.Fatal(err)
	}
	msg, err := asked.NextMsg(5 * time.Second)
	if err != nil || !strings.Contains(string(msg.Data), "healer H1") {
		t.Fatalf("the machine heard %v, %v", msg, err)
	}
	perms, err := broker.PermissionsFor(broker.Principal{Kind: broker.KindNode, Node: "laptop", PasswordHash: "x"})
	if err != nil || !slices.Contains(perms.Subscribe, "mesh.node.laptop.ask.report") ||
		slices.Contains(perms.Subscribe, "mesh.node.anchor.ask.report") {
		t.Fatalf("a machine's grant for the question: %v %v", perms.Subscribe, err)
	}
}