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") } }