From eb45c3c67c0ce9d8e2e4e5d33def242e5f259ee6 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 06:05:07 +0100 Subject: [PATCH] GR14: the keeper hears either finding stream form, each module publishing its own node's MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A * emitter before a finding event reaches the stream a part asked for by the template events (its finding-events) beside the old stream; a part's finding stream is published on its own node's subjects alone. Builds: design 47 G §386, §529, §515 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../internal/buscompose/busstate.go | 16 ++++++++- .../mesh-grants/internal/buscompose/nats.go | 4 +-- .../internal/buscompose/state_objects_test.go | 20 +++++++++++ .../internal/grants/findings_hear_test.go | 34 +++++++++++++++++++ .../mesh-grants/internal/grants/records.go | 27 +++++++++++++++ 5 files changed, 98 insertions(+), 3 deletions(-) create mode 100644 modules/mesh-grants/internal/grants/findings_hear_test.go diff --git a/modules/mesh-grants/internal/buscompose/busstate.go b/modules/mesh-grants/internal/buscompose/busstate.go index 7b868fea..d9874c45 100644 --- a/modules/mesh-grants/internal/buscompose/busstate.go +++ b/modules/mesh-grants/internal/buscompose/busstate.go @@ -239,6 +239,16 @@ type boundGrants struct { // the bus being the mesh's `_` (BusStateStreams) and a stream's subjects under `mesh.bus..`; // a binding naming any other is granted nothing. func boundOf(module string, bindings []membership.BusBinding) boundGrants { + return boundOn(module, "", bindings) +} + +// FindingEventsLocal is the local name a part asks its finding stream under, by the template events (design 47 G §529). +const FindingEventsLocal = "finding-events" + +// boundOn is boundOf for the account of a module on node: the module's finding stream, whose subjects end in the machine +// (`mesh.bus..finding-events..`), is published on its own node's subjects alone (§529). An empty node +// grants every machine's. +func boundOn(module, node string, bindings []membership.BusBinding) boundGrants { var g boundGrants for _, b := range bindings { base := module + "_" + b.Local @@ -263,7 +273,11 @@ func boundOf(module string, bindings []membership.BusBinding) boundGrants { if b.Stream != base || b.Subject != "mesh.bus."+module+"."+b.Local { continue } - g.Streams = append(g.Streams, b.Subject+".>", + published := b.Subject + ".>" + if node != "" && b.Local == FindingEventsLocal && safeSubject.MatchString(node) { + published = b.Subject + ".*." + node + } + g.Streams = append(g.Streams, published, "$JS.API.STREAM.INFO."+base, "$JS.API.STREAM.MSG.GET."+base, "$JS.API.DIRECT.GET."+base+".>", "$JS.API.CONSUMER.CREATE."+base+".>", "$JS.API.CONSUMER.DELETE."+base+".>", "$JS.FC."+base+".>") } diff --git a/modules/mesh-grants/internal/buscompose/nats.go b/modules/mesh-grants/internal/buscompose/nats.go index 9d9942b7..ba7035e7 100644 --- a/modules/mesh-grants/internal/buscompose/nats.go +++ b/modules/mesh-grants/internal/buscompose/nats.go @@ -1068,7 +1068,7 @@ func permissionsFor(p Principal) (Permissions, error) { // 5. Its state, and the state of others it reads (novox/hq ADR 0201): every one read and // watched, its own written too. // Its bus state asked in `requires` is granted as the old sections are (boundOf, ADR 0595 §1, issue 1966). - bound := boundOf(p.Module, p.Bound) + bound := boundOn(p.Module, p.Node, p.Bound) kept := stateAccess{Module: p.Module, Node: p.Node, Keeps: append(slices.Clone(p.State), bound.States...), PerMachine: p.PerMachine, ReadOnly: p.ReadOnly, Reads: p.Reads, KeyedReads: p.KeyedReads, Nodes: p.Nodes} pub = append(pub, stateGrants(kept)...) @@ -1178,7 +1178,7 @@ func permissionsFor(p Principal) (Permissions, error) { // each assignment is issued, as it keeps each module's events under that module's own name. for _, d := range p.Carries { // Its bus state asked in `requires` as its old sections' (boundOf, ADR 0595 §1, §9, issue 1966). - bound := boundOf(d.Module, d.Bindings) + bound := boundOn(d.Module, p.Node, d.Bindings) pub = append(pub, stateGrants(stateAccess{Module: d.Module, Node: p.Node, Keeps: append(stateNames(d.State), bound.States...), PerMachine: perMachineNames(d.State), ReadOnly: readOnlyNames(d.State), Reads: d.Reads, KeyedReads: d.KeyedReads, Nodes: p.Nodes})...) pub = append(pub, objectGrants(append(objectNames(d.Objects), bound.Objects...), d.ReadsObjects)...) diff --git a/modules/mesh-grants/internal/buscompose/state_objects_test.go b/modules/mesh-grants/internal/buscompose/state_objects_test.go index 8a3975a4..4277f2c7 100644 --- a/modules/mesh-grants/internal/buscompose/state_objects_test.go +++ b/modules/mesh-grants/internal/buscompose/state_objects_test.go @@ -36,3 +36,23 @@ func TestStateObjectsAndSentStores(t *testing.T) { t.Error("a reader of a sent store reads that one and writes none") } } + +// A part's finding stream is published on its own node's subjects alone (design 47 G §529): the machine is the subject's last +// token, and another machine's is refused. +func TestFindingEventsOnOwnNodeOnly(t *testing.T) { + bound := []membership.BusBinding{{Local: "finding-events", Storage: catalogue.StorageStream, Template: catalogue.TemplateEvents, + Stream: "shop_finding-events", Subject: "mesh.bus.shop.finding-events"}} + perms, err := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: "shop", Bound: bound}) + if err != nil { + t.Fatal(err) + } + for subject, want := range map[string]bool{ + "mesh.bus.shop.finding-events.finding-written.one": true, + "mesh.bus.shop.finding-events.finding-written.two": false, + "mesh.bus.shop.finding-events.findings-stated.two": false, + } { + if got := MayPublish(perms, subject); got != want { + t.Errorf("may publish %s: %v, want %v", subject, got, want) + } + } +} diff --git a/modules/mesh-grants/internal/grants/findings_hear_test.go b/modules/mesh-grants/internal/grants/findings_hear_test.go new file mode 100644 index 00000000..8a95cff5 --- /dev/null +++ b/modules/mesh-grants/internal/grants/findings_hear_test.go @@ -0,0 +1,34 @@ +package grants + +import ( + "slices" + "testing" + + "git.novox.be/novox/mesh-sdk/go/catalogue" +) + +// A `*` emitter before a finding event reaches both forms of a part's finding stream (design 47 G §386, §529): the old stream a +// foundation part declares and the template stream another part asks for as `finding-events`. +func TestTheKeepersConsumerOnEitherFindingForm(t *testing.T) { + events := []string{"finding-written", "finding-cleared", "findings-stated"} + declared := map[string]catalogue.Manifest{ + "old-part": {Module: "old-part", Streams: []catalogue.StreamDeclaration{{Name: "old-part-findings", Events: events}}}, + "new-part": {Module: "new-part", BusState: []catalogue.BusStateEntry{ + {Name: "findings", Storage: catalogue.StorageKeyValue, Template: catalogue.TemplateState}, + {Name: "finding-events", Storage: catalogue.StorageStream, Template: catalogue.TemplateEvents, Events: events}}}, + "quiet": {Module: "quiet", BusState: []catalogue.BusStateEntry{{Name: "other", Storage: catalogue.StorageStream, Template: catalogue.TemplateEvents, + Events: []string{"tick.*"}}}}, + } + keeper := catalogue.Manifest{Module: "keeper", Consumes: []string{"*.finding-written"}} + heard, touches, _ := hearsOf(keeper, declared, "one", nil, keeper.Consumes, func(e string) bool { return e == "finding-written" }) + got := map[string][]string{} + for _, h := range heard { + got[h.Emitter] = h.Filters + } + if !touches || len(got) != 2 || !slices.Equal(got["new-part"], []string{"mesh.bus.new-part.finding-events.finding-written.*"}) || len(got["old-part"]) != 1 { + t.Errorf("heard %+v", got) + } + if heard[len(heard)-1].Stream != "new-part_finding-events" { + t.Errorf("the template stream's bus name: %+v", heard) + } +} diff --git a/modules/mesh-grants/internal/grants/records.go b/modules/mesh-grants/internal/grants/records.go index 2fa9228b..cdf31fb4 100644 --- a/modules/mesh-grants/internal/grants/records.go +++ b/modules/mesh-grants/internal/grants/records.go @@ -501,6 +501,33 @@ func hearsOf(m catalogue.Manifest, declared map[string]catalogue.Manifest, node } out = append(out, h) } + // **Either form of a part's finding stream** (design 47 G §386, §529): a `*` emitter before an event reaches, beside each + // module's old stream, the stream a part asked for in `requires` by the template events (its `finding-events`) that carries + // the event, one durable consumer on it, heard on the binding's subject `mesh.bus...`. + for _, c := range reading.Consumes { + emitter, event, named := strings.Cut(c, ".") + if !named || emitter != catalogue.OneName || event == "" || strings.ContainsAny(event, "*>") || defined == nil || !defined(event) { + continue + } + for _, module := range slices.Sorted(maps.Keys(declared)) { + for _, e := range declared[module].BusState { + if e.Storage != catalogue.StorageStream || e.Template != catalogue.TemplateEvents { + continue + } + h := buscompose.Heard{Stream: catalogue.BusName(module, e.Name), Emitter: module} + for _, ev := range e.Events { + // An event is said on its binding's subject with the machine last (§529): `.`. + if ev == event || ev == event+".*" { + h.Filters = append(h.Filters, "mesh.bus."+module+"."+e.Name+"."+event+".*") + } + } + if len(h.Filters) > 0 { + out = append(out, h) + touches = true + } + } + } + } return out, touches, onEvents }