Assert the bus's objects on every send, not only at start (hq issue 208)

A module or seat holder assigned after the controller started was sent its declaration and found nothing to bind: messenger on novox ("consumer novox_messenger not found", 2026-10-06) and every first build-agent holder (2026-10-03). assertBusObjects ran only in the start raise; a push ensured a module's consumer only when a bus credential was minted, which a module carried by the runtime never is. The send's grant now runs the same derivation (assertOnSend) before the memberships, on every push, cascade, plan send and rotation. A failure is said in the send's output and raised as bus.objects.unasserted, which the next send that asserts everything clears; the send itself goes on, because the objects are the mesh's and holding every machine back for one would turn one fault into all. Module consumers are each tried and every failure named. The start raise stays as it was.
2026-10-06 11:45:51 +02:00
parent 9cf47f4429
commit b853439792
3 changed files with 265 additions and 1 deletions
@@ -0,0 +1,199 @@
+package main
+
+import (
+	"errors"
+	"os"
+	"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"
+)
+
+// 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 := os.Getenv("MESH_TEST_NATS")
+	if url == "" {
+		t.Skip("MESH_TEST_NATS unset")
+	}
+	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")
+	}
+}