From e0d7122567001387da7db1ee8b1ae9c2ff7b6d8a Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 18:09:09 +0100 Subject: [PATCH] TLS-5 integration test: the owners' own views streams, the runner's consumers on them and a membership naming them, events on views.Emit's subject Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig --- .../firstview/firstview_integration_test.go | 42 +++++++++++++++---- 1 file changed, 34 insertions(+), 8 deletions(-) diff --git a/node-tools/internal/firstview/firstview_integration_test.go b/node-tools/internal/firstview/firstview_integration_test.go index c76a5f3..035f4cc 100644 --- a/node-tools/internal/firstview/firstview_integration_test.go +++ b/node-tools/internal/firstview/firstview_integration_test.go @@ -11,6 +11,8 @@ import ( "testing" "time" + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-sdk/go/views" "github.com/nats-io/nats.go" @@ -19,13 +21,15 @@ import ( ) // issue 1966 TLS-5 [i] (ADR 0615 §2, §4, §6): against the real bus, with the owners down and then up, the runner's own -// consumer carries the manifest's four `consumes` to the owners' views events: the first views are asked until answered, +// consumers on the owners' own views streams carry the manifest's four `consumes`: the first views are asked until answered, // the unanswered ones are findings that the answer clears, and each owner's next event reaches its picture. func TestTheRunnersConsumerCarriesTheOwnersViewsFamiliesWhenTheOwnersAreDownThenUp(t *testing.T) { mesh := mt.New(t) url := mt.URL(t) - // The consumer is the mesh's, made from the manifest's `consumes` as the mesh makes it: one filter per family. + // The owners publish their views on their own streams (views.Emit: `mesh.stream..event.views.`, on + // assigner-views and indexer-views), never on EVENTS. The consumers on them are the mesh's, one filter per family the + // manifest's `consumes` names, and the runner's membership names them as the mesh issues it (ADR 0363 §8). raw, err := os.ReadFile("../../module.json") if err != nil { t.Fatal(err) @@ -36,15 +40,30 @@ func TestTheRunnersConsumerCarriesTheOwnersViewsFamiliesWhenTheOwnersAreDownThen if err := json.Unmarshal(raw, &manifest); err != nil { t.Fatal(err) } - var filters []string + streamOf := map[string]string{Assigner: catalogue.StreamBusName("assigner-views"), Indexer: catalogue.StreamBusName("indexer-views")} + filters := map[string][]string{} + families := 0 for _, c := range manifest.Consumes { emitter, event, _ := strings.Cut(c, ".views.") - filters = append(filters, "mesh.mod."+emitter+".event.views."+event) + stream := streamOf[emitter] + if stream == "" { + t.Fatalf("the manifest consumes %s, whose views stream the test does not know", c) + } + filters[stream] = append(filters[stream], process.ModuleStreamSubject(emitter, views.EventPrefix+event)) + families++ } - if len(filters) != len(Specs) { - t.Fatalf("the manifest consumes %d families, the runner holds %d first views", len(filters), len(Specs)) + if families != len(Specs) { + t.Fatalf("the manifest consumes %d families, the runner holds %d first views", families, len(Specs)) } - mesh.Consumer(t, "anchor", "node-tools", filters, 2*time.Second) + runner := mt.MembershipOf("node-tools", "anchor", false, nil) + for _, owner := range []string{Assigner, Indexer} { + stream := streamOf[owner] + mesh.ModuleStream(t, owner, stream) + mesh.ConsumerOn(t, stream, "anchor", "node-tools", filters[stream], 2*time.Second) + runner.Hears = append(runner.Hears, bus.HeardIssued{Stream: stream, Consumer: "anchor_node-tools"}) + } + mesh.Consumer(t, "anchor", "node-tools", []string{"mesh.mod.mesh-knowledge.event.reviewed"}, 2*time.Second) // its `waits`, on EVENTS + mesh.Issue(t, runner) conn, err := bus.Connect(bus.Credential{URL: url, Module: "node-tools", Node: "anchor"}) if err != nil { @@ -52,6 +71,13 @@ func TestTheRunnersConsumerCarriesTheOwnersViewsFamiliesWhenTheOwnersAreDownThen } conn.Logf = func(string, ...any) {} t.Cleanup(conn.Close) + conn.Follow("node-tools") + mt.Until(t, func() error { + if conn.Membership("node-tools") == nil { + return errFindings(-1) + } + return nil + }) var mu sync.Mutex findings := map[string]bool{} @@ -110,7 +136,7 @@ func TestTheRunnersConsumerCarriesTheOwnersViewsFamiliesWhenTheOwnersAreDownThen for _, s := range Specs { ev, _ := json.Marshal(views.Event{V: 1, Owner: s.Owner, State: s.State, Key: "next", Revision: 5, Created: time.Unix(1000, 0).UTC(), Value: json.RawMessage(`{"n":2}`)}) - mesh.Emit(t, "mesh.mod."+s.Owner+".event.views."+s.State, json.RawMessage(ev)) + mesh.Emit(t, process.ModuleStreamSubject(s.Owner, views.EventPrefix+s.State), json.RawMessage(ev)) } for _, s := range Specs { name := s.Name()