mesh-controller/cmd/mesh-controller/busobjects_test.go

package main

import (
	"errors"
	"strings"
	"testing"

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

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

	"github.com/novox/mesh-controller/internal/testbus"
)

// novox/hq issue 208: the bus's objects a declaration implies are asserted whenever one is sent, not
// only when the controller starts.

// aCarriedConsumer is the runtime on laptop and, carried by it, a module that consumes an event: the
// shape of messenger on 2026-10-06, assigned after the controller started.
func aCarriedConsumer(t *testing.T, open *stores) broker.Consumer {
	t.Helper()
	register(t, open, catalogue.Manifest{Module: catalogue.RuntimeModule, Version: "1",
		OwnSecrets: catalogue.OwnSecrets{"broker": {Path: "/var/lib/mesh/node-tools/broker"}}})
	register(t, open, catalogue.Manifest{Module: "messenger", Version: "1",
		Consumes: []string{"billing.order.placed"}})
	for _, m := range []string{catalogue.RuntimeModule, "messenger"} {
		if _, err := open.inventory.Assign(t.Context(), "laptop", m); err != nil {
			t.Fatal(err)
		}
	}
	consumers, err := moduleConsumers(t.Context(), open.inventory)
	if err != nil {
		t.Fatal(err)
	}
	for _, c := range consumers {
		if c.Module == "messenger" && c.Node == "laptop" {
			return c.Consumer
		}
	}
	t.Fatalf("messenger on laptop is derived no consumer: %+v", consumers)
	return broker.Consumer{}
}

// aLateHolder is a module holding the build agent's seat on anchor, assigned after the controller
// started, and the worker its seat's queue should have for it.
func aLateHolder(t *testing.T, open *stores) broker.Consumer {
	t.Helper()
	register(t, open, catalogue.Manifest{Module: "late-builder", Version: "1",
		Claims: []catalogue.Claim{{Name: "node-build-agent", Scope: catalogue.ScopeNode}}})
	if _, err := open.inventory.Assign(t.Context(), "anchor", "late-builder"); err != nil {
		t.Fatal(err)
	}
	for _, s := range inventory.MeshSeats() {
		if s.Name == "node-build-agent" {
			c, needed := broker.HolderConsumerFor("anchor", "late-builder", s)
			if !needed {
				t.Fatal("the build agent's seat needs no worker")
			}
			return c
		}
	}
	t.Fatal("the mesh declares no build agent's seat")
	return broker.Consumer{}
}

// asserted says whether a recording holds a consumer by stream and name.
func asserted(rec *recordingRaiser, want broker.Consumer) bool {
	for _, c := range rec.consumers {
		if c.Stream == want.Stream && c.Name == want.Name {
			return true
		}
	}
	return false
}

// What a send asserts includes what was assigned after the start's assertion: a carried module's
// consumer and a late holder's worker. The derivation is the start's own, so this holds without a bus.
func TestASendAssertsWhatWasAssignedAfterStart(t *testing.T) {
	open := aMesh(t)
	var atStart recordingRaiser
	if _, err := assertBusObjects(t.Context(), open.inventory, &atStart); err != nil {
		t.Fatal(err)
	}
	consumer := aCarriedConsumer(t, open)
	worker := aLateHolder(t, open)
	if asserted(&atStart, consumer) || asserted(&atStart, worker) {
		t.Fatal("the start asserted what was not yet assigned; the test proves nothing")
	}
	var onSend recordingRaiser
	if err := assertOnSend(t.Context(), open.inventory, &onSend, ""); err != nil {
		t.Fatal(err)
	}
	if !asserted(&onSend, consumer) {
		t.Fatalf("messenger's consumer %s on %s is not asserted by the send: %+v", consumer.Name, consumer.Stream, onSend.consumers)
	}
	if !asserted(&onSend, worker) {
		t.Fatalf("the late holder's worker %s on %s is not asserted by the send: %+v", worker.Name, worker.Stream, onSend.consumers)
	}
}

// failingRaiser refuses one consumer by name and records everything it was asked.
type failingRaiser struct {
	recordingRaiser
	refuse string
}

func (f *failingRaiser) EnsureConsumer(c broker.Consumer) error {
	f.consumers = append(f.consumers, c)
	if c.Name == f.refuse {
		return errors.New("nats: API error: code=503 description=insufficient resources")
	}
	return nil
}

// A send whose objects cannot be asserted says so in its output and raises a condition — and the next
// send that asserts them clears it. The consumers after the refused one are still asked for.
func TestASendThatCannotAssertTheBusSaysSoAndRaisesACondition(t *testing.T) {
	open := aMesh(t)
	consumer := aCarriedConsumer(t, open)
	worker := aLateHolder(t, open)
	failing := &failingRaiser{refuse: consumer.Name}

	var sendErr error
	out := stdoutOf(t, func() error {
		sendErr = assertOnSend(t.Context(), open.inventory, failing, "  ")
		return nil
	})
	if sendErr == nil || !strings.Contains(sendErr.Error(), "how messenger on laptop hears what it consumes") {
		t.Fatalf("the failure is not answered: %v", sendErr)
	}
	if !strings.Contains(out, "THE BUS DOES NOT HOLD WHAT THIS SEND IMPLIES") ||
		!strings.Contains(out, "insufficient resources") || !strings.Contains(out, "bus.objects.unasserted") {
		t.Fatalf("the failure is not said in the send's output:\n%s", out)
	}
	if !asserted(&failing.recordingRaiser, worker) {
		t.Fatal("the seat's worker was not asked for")
	}
	c, open1, err := conditionsFrom.Get(t.Context(), "bus.objects.unasserted")
	if err != nil || !open1 {
		t.Fatalf("no condition raised: %v", err)
	}
	if c.Kind != kindBusObjectsUnasserted || c.Source != sourceBusObjects ||
		len(c.Evidence) == 0 || !strings.Contains(c.Evidence[0].Said, "insufficient resources") {
		t.Fatalf("the condition does not say what failed: %+v", c)
	}

	// The next send asserts them, and that observation clears it.
	if err := assertOnSend(t.Context(), open.inventory, &recordingRaiser{}, ""); err != nil {
		t.Fatal(err)
	}
	if _, stillOpen, err := conditionsFrom.Get(t.Context(), "bus.objects.unasserted"); err != nil || stillOpen {
		t.Fatalf("a send that asserted everything left the condition open: %v", err)
	}
}

// Against a real bus, through the send's own grant: a module assigned after start has its consumer
// once a declaration is sent, and a holder assigned after start its seat's worker.
func TestNatsAModuleAssignedAfterStartGetsItsConsumerAtItsFirstSend(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)
	}
	// The controller starting, before either was assigned.
	if _, err := assertBusObjects(t.Context(), open.inventory, js); err != nil {
		t.Fatal(err)
	}
	consumer := aCarriedConsumer(t, open)
	worker := aLateHolder(t, open)
	_ = js.Context().DeleteConsumer(worker.Stream, worker.Name)
	for _, c := range []broker.Consumer{consumer, worker} {
		if _, err := js.Context().ConsumerInfo(c.Stream, c.Name); !errors.Is(err, nats.ErrConsumerNotFound) {
			t.Fatalf("%s on %s exists before any send; the test proves nothing: %v", c.Name, c.Stream, err)
		}
	}

	server := link.ConnectNats(js, nil, nil)
	if err := (overTheBus{open: open, server: server}).grant(t.Context(), nil); err != nil {
		t.Fatal(err)
	}
	for _, c := range []broker.Consumer{consumer, worker} {
		if _, err := js.Context().ConsumerInfo(c.Stream, c.Name); err != nil {
			t.Fatalf("%s on %s is not on the bus after a send: %v", c.Name, c.Stream, err)
		}
	}
	if _, raised, _ := conditionsFrom.Get(t.Context(), "bus.objects.unasserted"); raised {
		t.Fatal("a send that asserted everything raised a condition")
	}
}