novox/mesh-delivery
10
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

GR14: the keeper hears either finding stream form, each module publishing its own node's

Browse Source 
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
This commit is contained in:
jschoubben
2026-11-12 06:05:07 +01:00
parent a4fcf3de51
commit eb45c3c67c
5 changed files with 98 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-grants/internal/buscompose/busstate.go
+15 -1
View File
@@ -239,6 +239,16 @@ type boundGrants struct {
// the bus being the mesh's `<module>_<local>` (BusStateStreams) and a stream's subjects under `mesh.bus.<module>.<local>`;
// 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.<module>.finding-events.<event>.<node>`), 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+".>")
}
modules/mesh-grants/internal/buscompose/nats.go
+2 -2
View File
@@ -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)...)
modules/mesh-grants/internal/buscompose/state_objects_test.go
+20
View File
@@ -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)
}
}
}
modules/mesh-grants/internal/grants/findings_hear_test.go
+34
View File
@@ -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)
}
}
modules/mesh-grants/internal/grants/records.go
+27
View File
@@ -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.<module>.<local>.<event>`.
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): `<event>.<node>`.
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
}