From ee13e75bf3ef31c8b45e9f3d0a00b8cf39a49004 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 14 Oct 2026 21:52:45 +0200 Subject: [PATCH] Let a module declare a stream of its own and restrict who hears it by a trusted config value (hq issue 718, ADR 0363) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit mesh-mail's events must live on a stream of their own and reach only the modules the operator granted an account (ADR 0359 §2, §4). The controller now creates a module's declared stream, publishes its events on mesh.stream..event., and composes a consumer's durable consumer, filters and bus grants on a restricted stream only where its consumes asks and the provider's trusted value names it; a consumer that loses its grant has its consumer removed on the next assertion. Nothing changes while no manifest declares a stream. --- cmd/mesh-controller/busobjects.go | 25 + cmd/mesh-controller/push.go | 83 ++++ cmd/mesh-controller/reissue_hearing_test.go | 39 ++ internal/broker/derived.go | 25 +- internal/broker/jetstream.go | 22 + internal/broker/membership.go | 7 + internal/broker/modstreams.go | 219 +++++++++ internal/broker/modstreams_live_test.go | 62 +++ internal/broker/modstreams_test.go | 176 +++++++ internal/broker/nats.go | 23 +- internal/broker/users.go | 9 +- internal/catalogue/manifest.go | 8 + internal/catalogue/seats_declared.go | 2 + internal/catalogue/streams.go | 479 ++++++++++++++++++++ internal/catalogue/streams_test.go | 215 +++++++++ internal/inventory/busrecords.go | 116 ++++- internal/inventory/streams_test.go | 45 ++ 17 files changed, 1548 insertions(+), 7 deletions(-) create mode 100644 cmd/mesh-controller/reissue_hearing_test.go create mode 100644 internal/broker/modstreams.go create mode 100644 internal/broker/modstreams_live_test.go create mode 100644 internal/broker/modstreams_test.go create mode 100644 internal/catalogue/streams.go create mode 100644 internal/catalogue/streams_test.go create mode 100644 internal/inventory/streams_test.go diff --git a/cmd/mesh-controller/busobjects.go b/cmd/mesh-controller/busobjects.go index bb367a92..daed0d65 100644 --- a/cmd/mesh-controller/busobjects.go +++ b/cmd/mesh-controller/busobjects.go @@ -60,9 +60,19 @@ func assertBusObjects(ctx context.Context, inv *inventory.Inventory, r broker.Ra if err != nil { return nil, err } + // And every module's own stream (novox/hq ADR 0363), from the catalogue, before the consumers on it. + modStreams, err := inv.DeclaredStreams(ctx) + if err != nil { + return nil, err + } // Every one tried, and every failure named: one module's consumer the bus refuses is no reason // the modules after it in the list hear nothing (novox/hq issue 208, where this runs on each send). var failed []error + for _, s := range broker.ModuleStreams(modStreams) { + if err := r.EnsureStream(s); err != nil { + failed = append(failed, fmt.Errorf("the module stream %s: %w", s.Name, err)) + } + } for _, s := range trafficStreams { if err := r.EnsureStream(s); err != nil { failed = append(failed, fmt.Errorf("the work queue %s: %w", s.Name, err)) @@ -78,6 +88,21 @@ func assertBusObjects(ctx context.Context, inv *inventory.Inventory, r broker.Ra failed = append(failed, fmt.Errorf("how %s on %s hears what it consumes: %w", c.Module, c.Node, err)) } } + // **And a consumer no longer granted loses what it reads through** (novox/hq ADR 0363 §6): every consumer + // on a module's stream the composition no longer asks for is removed, on this very send. + if p, prunes := r.(broker.Pruner); prunes && len(modStreams) > 0 { + var wanted []broker.Consumer + for _, c := range consumers { + wanted = append(wanted, c.Consumer) + } + removed, err := broker.PruneHeard(p, modStreams, wanted) + if err != nil { + failed = append(failed, err) + } + for _, gone := range removed { + fmt.Printf(" removed the consumer %s: its module is no longer granted that stream (novox/hq ADR 0363)\n", gone) + } + } if len(failed) > 0 { return nil, errors.Join(failed...) } diff --git a/cmd/mesh-controller/push.go b/cmd/mesh-controller/push.go index 05bfc4c5..821cad0b 100644 --- a/cmd/mesh-controller/push.go +++ b/cmd/mesh-controller/push.go @@ -17,6 +17,7 @@ import ( "strings" "time" + "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" @@ -1386,6 +1387,13 @@ func issueMemberships(ctx context.Context, open *stores, server *link.Server, se if issued > 0 { fmt.Printf(" issued %d membership(s)\n", issued) } + // **And what every other machine hears of a module's stream** (novox/hq ADR 0363 §6): a grant the provider's + // setting changed reaches a consumer on a machine this send does not reach. Its membership is issued again + // as last issued, with only what it hears changed, so nothing else of it moves before its own send. + reheard := reissueHearing(ctx, bus, records, where, sent) + if reheard > 0 { + fmt.Printf(" issued %d membership(s) again for what they hear of a module's stream\n", reheard) + } if len(refused) > 0 { // Returned, never passed over (novox/hq issue 249): the declarations of the machines they are // for are not sent, and the rollout that asked is tried again rather than waiting for a push. @@ -1744,3 +1752,78 @@ func schemaReach() (int, error) { } return reach, nil } + +// reissueHearing issues again, on every machine not just sent, the membership of each assignment whose +// `consumes` names a module owning a stream, when what it hears changed: read as last issued from the bus, +// its `hears` replaced, and published. One that was never issued is left to its own send. Failures are said +// and passed over: the send that carries the grant is not the consumer's, and its own next send issues it. +func reissueHearing(ctx context.Context, bus link.OverNATS, records broker.Records, where broker.Placements, + sent []readyNode) int { + reached := map[string]bool{} + for _, s := range sent { + reached[s.node] = true + } + issued := 0 + for node, declared := range records.Assigned { + if reached[node] { + continue + } + for _, d := range declared { + if !d.HearsAStream { + continue + } + subject := broker.MembershipSubject(node, d.Module) + last, err := bus.JS.GetLastMsg(broker.AssignmentsStream, subject) + if err != nil { + if !errors.Is(err, nats.ErrMsgNotFound) { + fmt.Printf(" %s on %s: its membership could not be read to issue what it hears again: %v\n", + d.Module, node, err) + } + continue + } + body, changed, err := withHearing(last.Data, broker.HeardIssuedFor(node, d)) + if err != nil { + fmt.Printf(" %s on %s: its membership could not be read to issue what it hears again: %v\n", + d.Module, node, err) + continue + } + if !changed { + continue + } + if err := bus.PublishMembership(ctx, node, d.Module, body); err != nil { + fmt.Printf(" %s on %s: what it hears could not be issued again: %v\n", d.Module, node, err) + continue + } + issued++ + } + } + return issued +} + +// withHearing is a membership as issued with its `hears` replaced, and whether that changed it. Pure. +func withHearing(issued []byte, hears []broker.HeardIssued) ([]byte, bool, error) { + var fields map[string]json.RawMessage + if err := json.Unmarshal(issued, &fields); err != nil { + return nil, false, err + } + var had []broker.HeardIssued + if raw, ok := fields["hears"]; ok { + if err := json.Unmarshal(raw, &had); err != nil { + return nil, false, err + } + } + if slices.Equal(had, hears) { + return issued, false, nil + } + if len(hears) == 0 { + delete(fields, "hears") + } else { + raw, err := json.Marshal(hears) + if err != nil { + return nil, false, err + } + fields["hears"] = raw + } + body, err := json.Marshal(fields) + return body, true, err +} diff --git a/cmd/mesh-controller/reissue_hearing_test.go b/cmd/mesh-controller/reissue_hearing_test.go new file mode 100644 index 00000000..9375c583 --- /dev/null +++ b/cmd/mesh-controller/reissue_hearing_test.go @@ -0,0 +1,39 @@ +package main + +import ( + "encoding/json" + "testing" + + "github.com/novox/mesh-controller/internal/broker" +) + +// A membership issued again for what it hears changes nothing else of it (novox/hq ADR 0363 §6). +func TestAMembershipIssuedAgainChangesOnlyWhatItHears(t *testing.T) { + issued := []byte(`{"node":"ace","module":"reader","receives":{"x":[1]},"emits":"mesh.mod.reader.event.{event}"}`) + hears := []broker.HeardIssued{{Stream: "MAIL", Consumer: "ace_reader"}} + body, changed, err := withHearing(issued, hears) + if err != nil || !changed { + t.Fatalf("changed %v, %v", changed, err) + } + var got map[string]any + if err := json.Unmarshal(body, &got); err != nil { + t.Fatal(err) + } + if got["receives"] == nil || got["emits"] != "mesh.mod.reader.event.{event}" || got["hears"] == nil { + t.Fatalf("issued again as %s", body) + } + if _, again, _ := withHearing(body, hears); again { + t.Fatal("the same hearing issued again counts as a change") + } + gone, changed, err := withHearing(body, nil) + if err != nil || !changed { + t.Fatalf("a withdrawal: %v, %v", changed, err) + } + got = nil + if err := json.Unmarshal(gone, &got); err != nil { + t.Fatal(err) + } + if _, still := got["hears"]; still { + t.Fatalf("a withdrawn hearing is still issued: %s", gone) + } +} diff --git a/internal/broker/derived.go b/internal/broker/derived.go index 0a31ef41..1a770ecf 100644 --- a/internal/broker/derived.go +++ b/internal/broker/derived.go @@ -128,7 +128,13 @@ func ConsumerFor(p Principal) (Consumer, bool) { // connected, and its graph stayed empty with nothing anywhere reporting why. // And one that hears its own answers on a seat it uses (novox/hq ADR 0259 §3): an asker's warrants. hearsItsOwn := len(SeatTrafficOf(p.Module, nil, p.Uses, nil).Subscribe) > 0 - if p.Kind != KindModule || (len(p.Consumes) == 0 && len(p.Watches) == 0 && !hearsItsOwn) { + hearsAStream := false + for _, h := range p.Hears { + if len(h.Filters) > 0 { + hearsAStream = true + } + } + if p.Kind != KindModule || (len(p.Consumes) == 0 && len(p.Watches) == 0 && !hearsItsOwn && !hearsAStream) { return Consumer{}, false } perms, err := PermissionsFor(p) @@ -140,11 +146,14 @@ func ConsumerFor(p Principal) (Consumer, bool) { // business, which is why this is a filter and not the whole list. var filters []string for _, s := range perms.Subscribe { - if strings.Contains(s, ".event.") { + // Not a module stream's, which its own consumer on that stream hears (novox/hq ADR 0363). + if strings.Contains(s, ".event.") && !strings.HasPrefix(s, moduleStreamPrefix) { filters = append(filters, s) } } - if len(filters) == 0 { + // **A module that hears only a module's stream still has one here** (novox/hq ADR 0363 §7): what it + // gives up on there is delivered again under `mesh.again..` on EVENTS, and only this filters it. + if len(filters) == 0 && !hearsAStream { return Consumer{}, false } sort.Strings(filters) @@ -181,6 +190,13 @@ func ConsumersOf(users []Principal) []ModuleConsumer { } seen[p.Node+"/"+p.Module] = true out = append(out, ModuleConsumer{Node: p.Node, Module: p.Module, Consumer: c}) + // And one on each module stream it hears (novox/hq ADR 0363 §5). + for _, h := range p.Hears { + if len(h.Filters) == 0 { + continue + } + out = append(out, ModuleConsumer{Node: p.Node, Module: p.Module, Consumer: heardConsumer(p, h)}) + } } for _, p := range users { if p.Kind == KindModule { @@ -193,7 +209,8 @@ func ConsumersOf(users []Principal) []ModuleConsumer { } for _, d := range p.Carries { add(Principal{Kind: KindModule, Node: p.Node, Module: d.Module, Emits: d.Emits, - Consumes: d.Consumes, Serves: d.Serves, Holds: d.Holds, Uses: d.Uses, Watches: d.Watches}) + Consumes: d.Consumes, Serves: d.Serves, Holds: d.Holds, Uses: d.Uses, Watches: d.Watches, + Streams: d.Streams, Hears: d.Hears}) } } return out diff --git a/internal/broker/jetstream.go b/internal/broker/jetstream.go index 8821e286..9f9dca93 100644 --- a/internal/broker/jetstream.go +++ b/internal/broker/jetstream.go @@ -486,3 +486,25 @@ func (j *JetStream) LogStreams() ([]string, error) { } return out, lister.Err() } + +// ConsumerNames is every durable consumer on a stream; none for a stream that does not exist yet. +func (j *JetStream) ConsumerNames(stream string) ([]string, error) { + if _, err := j.js.StreamInfo(stream); errors.Is(err, nats.ErrStreamNotFound) { + return nil, nil + } else if err != nil { + return nil, err + } + var out []string + for name := range j.js.ConsumerNames(stream) { + out = append(out, name) + } + return out, nil +} + +// RemoveConsumer deletes one durable consumer (novox/hq ADR 0363 §6): one the composition no longer asks for. +func (j *JetStream) RemoveConsumer(stream, name string) error { + if err := j.js.DeleteConsumer(stream, name); err != nil && !errors.Is(err, nats.ErrConsumerNotFound) { + return err + } + return nil +} diff --git a/internal/broker/membership.go b/internal/broker/membership.go index 1f662589..fc3d6ed7 100644 --- a/internal/broker/membership.go +++ b/internal/broker/membership.go @@ -61,6 +61,11 @@ type Membership struct { // over every module on the machine, so one module's code reaching another's name or kind through // the runtime is the runtime's to refuse. SeatTraffic *SeatTraffic `json:"seat-traffic,omitempty"` + // Streams are this module's own streams (novox/hq ADR 0363 §2): an event one of them carries is published + // on that stream's subject, never on Emits. Hears are the module streams it hears, each through a durable + // consumer on that stream the runtime binds beside the one on EVENTS (§8). + Streams []StreamIssued `json:"streams,omitempty"` + Hears []HeardIssued `json:"hears,omitempty"` } // Served is one address a tool is answered on. @@ -126,6 +131,8 @@ func MembershipFor(node string, d Declared, where Placements) Membership { } m.State = stateIssuedFor(d, node) m.Logs = logsIssuedFor(d) + m.Streams = streamsIssuedFor(d) + m.Hears = HeardIssuedFor(node, d) t := SeatTrafficOf(d.Module, d.Holds, d.Uses, d.Watches) for _, s := range append(append([]Seat{}, d.Uses...), d.Watches...) { if !s.Kinded { diff --git a/internal/broker/modstreams.go b/internal/broker/modstreams.go new file mode 100644 index 00000000..fe9174a1 --- /dev/null +++ b/internal/broker/modstreams.go @@ -0,0 +1,219 @@ +package broker + +import ( + "fmt" + "sort" + "strings" +) + +// A module's own stream on the bus (novox/hq ADR 0363). +// +// A module declares a stream as it declares a log: named locally, created by the controller from the +// catalogue on every raise and every send, never removed by the mesh. Its events are published on +// `mesh.stream..event.`, a subject space the shared EVENTS stream (`mesh.mod.*.event.>`) +// does not take — the bus refuses two streams claiming one subject, so a module's stream cannot share +// EVENTS' — and a consumer hears it through a durable consumer of its own on that stream, named as on +// EVENTS. Pure; the controller's raise asserts what is composed here. + +// moduleStreamPrefix starts every subject a module's own stream takes. +const moduleStreamPrefix = "mesh.stream." + +// ModuleStreamSubject is where a module's event on its own stream lands: `mesh.stream..event.`. +// The token before `.event.` is the emitter, as on EVENTS, so a runtime reads the same key from it. +func ModuleStreamSubject(module, event string) string { + return moduleStreamPrefix + module + ".event." + event +} + +// WirePattern is a local event pattern as the bus spells it: `**` is `>`. +func WirePattern(local string) string { + parts := strings.Split(local, ".") + for i, p := range parts { + if p == "**" { + parts[i] = ">" + } + } + return strings.Join(parts, ".") +} + +// A ModuleStream is one module's declared stream as the bus holds it. +type ModuleStream struct { + Module string + // Name is its name on the bus, `MAIL`. + Name string + // Events are the events it carries, as the bus spells them (WirePattern). + Events []string + MaxAgeDays int + MaxMiB int + // Restricted says a trusted setting of the module grants who hears it (ADR 0363 §4). + Restricted bool +} + +// Subjects are the subjects the stream takes. +func (s ModuleStream) Subjects() []string { + var out []string + for _, e := range s.Events { + out = append(out, ModuleStreamSubject(s.Module, e)) + } + sort.Strings(out) + return out +} + +// Stream is the stream the controller asserts: kept by limits, its age and size the module's, dropping +// its oldest when full — a consumer that was away catches up within the age, and the module's own record +// answers for anything older. +func (s ModuleStream) Stream() Stream { + why := fmt.Sprintf("%s's own stream (novox/hq ADR 0363): its events, kept %d days and at most %d MiB so a "+ + "consumer that was away catches up", s.Module, s.MaxAgeDays, s.MaxMiB) + if s.Restricted { + why += "; heard only by the consumers its trusted setting grants" + } + return Stream{ + Name: s.Name, + Subjects: s.Subjects(), + Retention: RetentionLimits, + MaxAge: s.MaxAgeDays * 24 * 60 * 60, + MaxBytes: int64(s.MaxMiB) * 1024 * 1024, + Why: why, + } +} + +// ModuleStreams is every declared module stream as the controller asserts it, in name order. +func ModuleStreams(declared []ModuleStream) []Stream { + sorted := append([]ModuleStream(nil), declared...) + sort.Slice(sorted, func(i, j int) bool { return sorted[i].Name < sorted[j].Name }) + out := make([]Stream, 0, len(sorted)) + for _, s := range sorted { + out = append(out, s.Stream()) + } + return out +} + +// Heard is a module's stream another module hears, and the subjects it hears of it: its `consumes` +// within the stream and, for a restricted stream, within what the provider's setting grants it +// (ADR 0363 §5). Composed only with at least one subject. +type Heard struct { + Stream string + Emitter string + Filters []string +} + +// heardConsumer is the durable consumer a module hears a module's stream through: named as its consumer +// on EVENTS, on that stream, with the subjects it was granted as its filters. +func heardConsumer(p Principal, h Heard) Consumer { + filters := append([]string(nil), h.Filters...) + sort.Strings(filters) + return Consumer{ + Name: consumerDurable(p), + Stream: h.Stream, + Filters: filters, + AckWaitSeconds: 30, + MaxDeliver: 5, + Why: "what " + p.Module + " consumes of " + h.Emitter + "'s stream " + h.Stream + " and is granted " + + "(novox/hq ADR 0363); after max-deliver it gives an event up, and the controller keeps it in " + + "DEAD_LETTERS until a person delivers it again or drops it", + } +} + +// heardGrants are what a principal publishes to read and acknowledge the consumers it hears module streams +// through: asking about each, pulling from it, acknowledging it — its own name on each stream, no other's. +func heardGrants(p Principal, hears []Heard) []string { + var out []string + durable := consumerDurable(p) + for _, h := range hears { + if !safeSubject.MatchString(h.Stream) || len(h.Filters) == 0 { + continue + } + out = append(out, + "$JS.API.CONSUMER.INFO."+h.Stream+"."+durable, + "$JS.API.CONSUMER.MSG.NEXT."+h.Stream+"."+durable, + "$JS.ACK."+h.Stream+"."+durable+".>") + } + return out +} + +// ownStreamGrants are the subjects a module publishes its own streams' events on, and no other's. +func ownStreamGrants(module string, streams []ModuleStream) []string { + var out []string + for _, s := range streams { + if s.Module != module { + continue + } + out = append(out, s.Subjects()...) + } + return out +} + +// StreamIssued is one of a module's own streams as its membership names it (ADR 0363 §2): the events it +// carries, as the bus spells them, and the subject an event of them is published on, `{event}` the event. +type StreamIssued struct { + Stream string `json:"stream"` + Events []string `json:"events"` + Subject string `json:"subject"` +} + +// HeardIssued is one module stream an assignment hears, as its membership names it (ADR 0363 §8): the +// stream and the durable consumer on it the runtime binds. +type HeardIssued struct { + Stream string `json:"stream"` + Consumer string `json:"consumer"` +} + +// streamsIssuedFor is every stream of its own a module's membership names. +func streamsIssuedFor(d Declared) []StreamIssued { + var out []StreamIssued + for _, s := range d.Streams { + if s.Module != d.Module || !safeSubject.MatchString(d.Module) { + continue + } + out = append(out, StreamIssued{Stream: s.Name, Events: append([]string(nil), s.Events...), + Subject: ModuleStreamSubject(d.Module, "{event}")}) + } + return out +} + +// HeardIssuedFor is every module stream an assignment hears, as its membership names it. +func HeardIssuedFor(node string, d Declared) []HeardIssued { + var out []HeardIssued + for _, h := range d.Hears { + if len(h.Filters) == 0 { + continue + } + out = append(out, HeardIssued{Stream: h.Stream, Consumer: node + "_" + d.Module}) + } + sort.Slice(out, func(i, j int) bool { return out[i].Stream < out[j].Stream }) + return out +} + +// Pruner is the part of a JetStream connection that removes a module's consumer on a module's stream it +// may no longer hear (ADR 0363 §6). Optional: what only records what it is asked to assert has none. +type Pruner interface { + ConsumerNames(stream string) ([]string, error) + RemoveConsumer(stream, name string) error +} + +// PruneHeard removes, from each module stream, every durable consumer the composition no longer asks for, +// and answers what it removed as `/`. A consumer that loses its grant loses what it reads +// through, whatever its runtime still holds; a stream that does not exist yet has nothing to remove. +func PruneHeard(p Pruner, streams []ModuleStream, wanted []Consumer) ([]string, error) { + keep := map[string]bool{} + for _, c := range wanted { + keep[c.Stream+"/"+c.Name] = true + } + var removed []string + for _, s := range streams { + names, err := p.ConsumerNames(s.Name) + if err != nil { + return removed, fmt.Errorf("listing the consumers on %s: %w", s.Name, err) + } + for _, n := range names { + if keep[s.Name+"/"+n] { + continue + } + if err := p.RemoveConsumer(s.Name, n); err != nil { + return removed, fmt.Errorf("removing %s's consumer on %s, which is no longer granted: %w", n, s.Name, err) + } + removed = append(removed, s.Name+"/"+n) + } + } + return removed, nil +} diff --git a/internal/broker/modstreams_live_test.go b/internal/broker/modstreams_live_test.go new file mode 100644 index 00000000..1a2366bc --- /dev/null +++ b/internal/broker/modstreams_live_test.go @@ -0,0 +1,62 @@ +package broker + +import ( + "slices" + "testing" + "time" + + "github.com/novox/mesh-controller/internal/testbus" +) + +// On a real bus: a module's stream is made beside EVENTS without a clash, keeps what the module publishes on +// its own subjects, a granted consumer hears only its part, a narrowed grant narrows it, and a withdrawn one +// is removed (novox/hq ADR 0363). +// +// go test ./internal/broker/ -run TestAModuleStreamOnARealBus +func TestAModuleStreamOnARealBus(t *testing.T) { + js, err := Dial(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + defer js.Close() + for _, s := range []Stream{MeshStreams()[3], mailStream.Stream()} { + _ = js.js.DeleteStream(s.Name) + if err := js.EnsureStream(s); err != nil { + t.Fatalf("%s: %v", s.Name, err) + } + defer func(name string) { _ = js.js.DeleteStream(name) }(s.Name) + } + for _, account := range []string{"mesh", "work"} { + if _, err := js.js.Publish(ModuleStreamSubject("mesh-mail", "mail."+account+".message.added"), []byte("{}")); err != nil { + t.Fatalf("publishing %s's mail: %v", account, err) + } + } + if info, _ := js.js.StreamInfo(EventsStream); info.State.Msgs != 0 { + t.Fatalf("EVENTS kept %d of the module stream's events", info.State.Msgs) + } + + reader := Principal{Kind: KindModule, Node: "ace", Module: "reader", Hears: []Heard{mailHeard}} + c := heardConsumer(reader, mailHeard) + if err := js.EnsureConsumer(c); err != nil { + t.Fatal(err) + } + time.Sleep(50 * time.Millisecond) + if info, _ := js.js.ConsumerInfo("MAIL", c.Name); info.NumPending != 1 { + t.Fatalf("the reader granted `mesh` has %d pending, want 1", info.NumPending) + } + narrowed := c + narrowed.Filters = []string{"mesh.stream.mesh-mail.event.mail.mesh.message.changed"} + if err := js.EnsureConsumer(narrowed); err != nil { + t.Fatalf("narrowing the grant: %v", err) + } + removed, err := PruneHeard(js, []ModuleStream{mailStream}, nil) + if err != nil { + t.Fatal(err) + } + if !slices.Equal(removed, []string{"MAIL/ace_reader"}) { + t.Fatalf("removed %v", removed) + } + if names, _ := js.ConsumerNames("MAIL"); len(names) != 0 { + t.Fatalf("still on MAIL: %v", names) + } +} diff --git a/internal/broker/modstreams_test.go b/internal/broker/modstreams_test.go new file mode 100644 index 00000000..2518d989 --- /dev/null +++ b/internal/broker/modstreams_test.go @@ -0,0 +1,176 @@ +package broker + +import ( + "reflect" + "slices" + "testing" +) + +// mailStream is mesh-mail's own stream as the bus holds it (novox/hq ADR 0363). +var mailStream = ModuleStream{Module: "mesh-mail", Name: "MAIL", + Events: []string{"mail.*.message.added", "mail.*.message.changed", "mail.*.message.removed"}, + MaxAgeDays: 7, MaxMiB: 1024, Restricted: true} + +// mailHeard is what a reader granted the account `mesh` hears of it. +var mailHeard = Heard{Stream: "MAIL", Emitter: "mesh-mail", Filters: []string{ + "mesh.stream.mesh-mail.event.mail.mesh.message.added", "mesh.stream.mesh-mail.event.mail.mesh.message.changed"}} + +// A module's stream is its own: subjects EVENTS does not claim, its age and size, kept by limits (ADR 0363 §1, §2). +func TestAModuleStreamTakesSubjectsEventsDoesNot(t *testing.T) { + s := mailStream.Stream() + if s.Name != "MAIL" || s.Retention != RetentionLimits || s.MaxAge != 7*24*60*60 || s.MaxBytes != 1024<<20 || s.DiscardNew { + t.Fatalf("composed as %+v", s) + } + want := []string{"mesh.stream.mesh-mail.event.mail.*.message.added", "mesh.stream.mesh-mail.event.mail.*.message.changed", + "mesh.stream.mesh-mail.event.mail.*.message.removed"} + if !reflect.DeepEqual(s.Subjects, want) { + t.Fatalf("subjects %v", s.Subjects) + } + for _, own := range MeshStreams() { + for _, a := range own.Subjects { + for _, b := range s.Subjects { + if SubjectsOverlap(a, b) { + t.Fatalf("%s claims %s, which overlaps %s of %s: the bus would refuse the stream", s.Name, b, a, own.Name) + } + } + } + } + if WirePattern("mail.**") != "mail.>" { + t.Fatal("`**` is spelled `>` on the wire") + } +} + +// The provider publishes its stream's events on their subjects and nowhere on EVENTS for them; a consumer +// granted a part hears it through its own consumer on the stream, and one not granted has nothing there +// (ADR 0363 §2, §5). +func TestTheProviderPublishesItsStreamAndOnlyAGrantedConsumerHearsIt(t *testing.T) { + provider, err := PermissionsFor(Principal{Kind: KindModule, Node: "novox", Module: "mesh-mail", + Streams: []ModuleStream{mailStream}}) + if err != nil { + t.Fatal(err) + } + if !MayPublish(provider, "mesh.stream.mesh-mail.event.mail.work.message.added") { + t.Fatal("the provider may not publish its own stream's event") + } + if MayPublish(provider, "mesh.mod.mesh-mail.event.mail.work.message.added") { + t.Fatal("the provider may publish its stream's event on EVENTS") + } + if MayPublish(provider, "mesh.stream.other.event.mail.work.message.added") { + t.Fatal("the provider may publish on another module's stream") + } + + reader := Principal{Kind: KindModule, Node: "ace", Module: "reader", Hears: []Heard{mailHeard}} + perms, err := PermissionsFor(reader) + if err != nil { + t.Fatal(err) + } + for _, s := range []string{"$JS.API.CONSUMER.MSG.NEXT.MAIL.ace_reader", "$JS.API.CONSUMER.INFO.MAIL.ace_reader", + "$JS.ACK.MAIL.ace_reader.1"} { + if !MayPublish(perms, s) { + t.Errorf("the granted reader may not publish %s", s) + } + } + if !MaySubscribe(perms, "mesh.stream.mesh-mail.event.mail.mesh.message.added") || + MaySubscribe(perms, "mesh.stream.mesh-mail.event.mail.work.message.added") { + t.Fatal("the reader's subscription is not what it was granted") + } + if MayPublish(perms, "$JS.API.CONSUMER.MSG.NEXT.MAIL.ace_other") { + t.Fatal("the reader may pull another's consumer") + } + + stranger, err := PermissionsFor(Principal{Kind: KindModule, Node: "ace", Module: "stranger", + Consumes: []string{"billing.paid"}}) + if err != nil { + t.Fatal(err) + } + if MayPublish(stranger, "$JS.API.CONSUMER.MSG.NEXT.MAIL.ace_stranger") || + MaySubscribe(stranger, "mesh.stream.mesh-mail.event.mail.mesh.message.added") { + t.Fatal("a module not granted reaches the stream") + } +} + +// A granted consumer has a durable consumer on the stream, with its granted subjects as filters, and an EVENTS +// consumer that hears what it gives up on delivered again (ADR 0363 §5, §7); the runtime carrying it may read both. +func TestAGrantedConsumerHasAConsumerOnTheStream(t *testing.T) { + records := Records{Nodes: []string{"ace", "novox"}, Assigned: map[string][]Declared{ + "novox": {{Module: "mesh-mail", Streams: []ModuleStream{mailStream}}}, + "ace": {{Module: RuntimeModule}, {Module: "reader", Hears: []Heard{mailHeard}, HearsAStream: true}, {Module: "stranger", HearsAStream: true}}, + }} + users, err := Users(records) + if err != nil { + t.Fatal(err) + } + var onMail, onEvents []Consumer + for _, c := range ConsumersOf(users) { + switch c.Consumer.Stream { + case "MAIL": + onMail = append(onMail, c.Consumer) + case EventsStream: + onEvents = append(onEvents, c.Consumer) + } + } + if len(onMail) != 1 || onMail[0].Name != "ace_reader" || !slices.Equal(onMail[0].Filters, mailHeard.Filters) { + t.Fatalf("on MAIL: %+v", onMail) + } + if len(onEvents) != 1 || onEvents[0].Name != "ace_reader" || !slices.Equal(onEvents[0].Filters, []string{AgainFilter("ace_reader")}) { + t.Fatalf("on EVENTS: %+v", onEvents) + } + for _, u := range users { + if u.Kind != KindNodeTools || u.Node != "ace" { + continue + } + perms, err := PermissionsFor(u) + if err != nil { + t.Fatal(err) + } + if !MayPublish(perms, "$JS.API.CONSUMER.MSG.NEXT.MAIL.ace_reader") || !MayPublish(perms, "$JS.ACK.MAIL.ace_reader.1") { + t.Fatal("the runtime carrying the reader may not read its consumer on MAIL") + } + if MayPublish(perms, "$JS.API.CONSUMER.MSG.NEXT.MAIL.ace_stranger") { + t.Fatal("the runtime may read a consumer no module it carries is granted") + } + } +} + +// The membership says where a module's stream events go and which consumers on module streams to bind +// (ADR 0363 §2, §8). +func TestTheMembershipNamesTheStreamAndWhatIsHeard(t *testing.T) { + where := Placements{Nodes: map[string][]string{}} + provider := MembershipFor("novox", Declared{Module: "mesh-mail", Streams: []ModuleStream{mailStream}}, where) + if len(provider.Streams) != 1 || provider.Streams[0].Stream != "MAIL" || + provider.Streams[0].Subject != "mesh.stream.mesh-mail.event.{event}" || + !slices.Equal(provider.Streams[0].Events, mailStream.Events) { + t.Fatalf("the provider's membership: %+v", provider.Streams) + } + reader := MembershipFor("ace", Declared{Module: "reader", Hears: []Heard{mailHeard}}, where) + if !reflect.DeepEqual(reader.Hears, []HeardIssued{{Stream: "MAIL", Consumer: "ace_reader"}}) { + t.Fatalf("the reader's membership: %+v", reader.Hears) + } + if none := MembershipFor("ace", Declared{Module: "stranger", HearsAStream: true}, where); none.Hears != nil { + t.Fatalf("a module not granted is told to bind %+v", none.Hears) + } +} + +type fakePruner struct { + on map[string][]string + removed []string +} + +func (f *fakePruner) ConsumerNames(stream string) ([]string, error) { return f.on[stream], nil } +func (f *fakePruner) RemoveConsumer(stream, name string) error { + f.removed = append(f.removed, stream+"/"+name) + return nil +} + +// A consumer that lost its grant has its consumer on the stream removed by the next assertion (ADR 0363 §6). +func TestAConsumerThatLostItsGrantIsRemoved(t *testing.T) { + f := &fakePruner{on: map[string][]string{"MAIL": {"ace_reader", "ace_former"}}} + wanted := []Consumer{{Name: "ace_reader", Stream: "MAIL"}, {Name: "ace_former", Stream: EventsStream}} + removed, err := PruneHeard(f, []ModuleStream{mailStream}, wanted) + if err != nil { + t.Fatal(err) + } + if !slices.Equal(removed, []string{"MAIL/ace_former"}) || !slices.Equal(f.removed, removed) { + t.Fatalf("removed %v", removed) + } +} diff --git a/internal/broker/nats.go b/internal/broker/nats.go index 3e5a42ca..9323c6fd 100644 --- a/internal/broker/nats.go +++ b/internal/broker/nats.go @@ -141,6 +141,10 @@ type Principal struct { KeyedReads []KeyedRead // Logs is the local names of the logs this principal's module keeps (novox/hq ADR 0297). Logs []string + // Streams are the streams this principal's module owns, and Hears the module streams it hears with the + // subjects it was granted (novox/hq ADR 0363). + Streams []ModuleStream + Hears []Heard // SnapshotsTheBus is the bus's own module, the one holding mesh-broker (novox/hq ADR 0235). Its // whole authority is BusSnapshotGrants: it copies the streams for the night's backup and nothing @@ -637,6 +641,8 @@ func PermissionsFor(p Principal) (Permissions, error) { for _, e := range p.Emits { pub = append(pub, own+".event."+e) } + // And its own streams' events, on their own subjects (novox/hq ADR 0363 §2). + pub = append(pub, ownStreamGrants(p.Module, p.Streams)...) // Every tool under its own name, not a list: the tools a module serves are what its code // answers, and a second copy of that list in the manifest would be a second source of // truth for the mesh to keep in step (2026-09-28: every module that served a tool was @@ -678,6 +684,16 @@ func PermissionsFor(p Principal) (Permissions, error) { sub = append(sub, subject) } + // 2a. What it hears of a module's own stream, as it was granted (novox/hq ADR 0363 §5): the subjects, + // and its own consumer on that stream, asked about, pulled from and acknowledged. + for _, h := range p.Hears { + if len(h.Filters) == 0 { + continue + } + sub = append(sub, h.Filters...) + } + pub = append(pub, heardGrants(p, p.Hears)...) + // 2b. Events of a role it watches, under the seat's own namespace. Subscribe only: watching a // role is hearing what it announced, not taking part in it. for _, w := range p.Watches { @@ -800,6 +816,8 @@ func PermissionsFor(p Principal) (Permissions, error) { for _, e := range d.Emits { pub = append(pub, own+".event."+e) } + // And its own streams' events (novox/hq ADR 0363 §2). + pub = append(pub, ownStreamGrants(d.Module, d.Streams)...) for _, s := range d.Holds { if servedOnlyByTheController(s) { continue @@ -838,7 +856,10 @@ func PermissionsFor(p Principal) (Permissions, error) { // consumer is still the controller's to make, from the module's own principal. for _, d := range p.Carries { own := Principal{Kind: KindModule, Node: p.Node, Module: d.Module, Emits: d.Emits, - Consumes: d.Consumes, Serves: d.Serves, Holds: d.Holds, Uses: d.Uses, Watches: d.Watches} + Consumes: d.Consumes, Serves: d.Serves, Holds: d.Holds, Uses: d.Uses, Watches: d.Watches, + Streams: d.Streams, Hears: d.Hears} + // And the module streams it hears, each through its own consumer there (novox/hq ADR 0363 §8). + pub = append(pub, heardGrants(own, d.Hears)...) if _, consumes := ConsumerFor(own); !consumes { continue } diff --git a/internal/broker/users.go b/internal/broker/users.go index 1768b028..2c0ea4c5 100644 --- a/internal/broker/users.go +++ b/internal/broker/users.go @@ -40,6 +40,13 @@ type Declared struct { State []Bucket // Logs are the logs it keeps, each a stream its instances append to (novox/hq ADR 0297). Logs []Log + // Streams are the streams it owns, each its events' one place (novox/hq ADR 0363). + Streams []ModuleStream + // Hears are the module streams it hears on this machine, each with the subjects it was granted + // (novox/hq ADR 0363 §5); HearsAStream says its `consumes` names a module that owns a stream, granted + // or not, so its membership is issued again when a grant changes. + Hears []Heard + HearsAStream bool // Reads are other modules' state it reads, each `.` (novox/hq ADR 0201). Reads []string // KeyedReads are keys of other modules' state it reads, each one key alone: what a seat's holder is @@ -129,7 +136,7 @@ func Users(r Records) ([]Principal, error) { Holds: d.Holds, Uses: d.Uses, Watches: d.Watches, Invokes: d.Invokes, State: stateNames(d.State), Reads: d.Reads, SnapshotsTheBus: d.SnapshotsTheBus, PerMachine: perMachineNames(d.State), KeyedReads: d.KeyedReads, - Logs: logNames(d.Logs), + Logs: logNames(d.Logs), Streams: d.Streams, Hears: d.Hears, }) } if runtimeHere { diff --git a/internal/catalogue/manifest.go b/internal/catalogue/manifest.go index 9fc35664..a816c6c5 100644 --- a/internal/catalogue/manifest.go +++ b/internal/catalogue/manifest.go @@ -523,6 +523,12 @@ type Manifest struct { // (novox/hq ADR 0297). A log is its owner's alone: no other module reads it. Logs []LogDeclaration `json:"logs,omitempty"` + // Streams are the streams this module owns on the bus (novox/hq ADR 0363): each named locally, the + // events it carries named as `emits` names them but with `*` and `**`, and bounded in age and size; the + // controller creates each. One may be restricted by a trusted setting of this module, so a consumer + // hears a part of it only where that setting's value names it. + Streams []StreamDeclaration `json:"streams,omitempty"` + // Settings are the defaults this module gives its settings (novox/hq ADR 0262): each key a file, // a contribution or a served fact asks for as `${setting:}`, its default, and why that // default. Only a preference has one — a font size, a width, a number of workers — and a value @@ -1792,6 +1798,8 @@ func ParseManifest(raw []byte) (Manifest, error) { problems = append(problems, StateProblems(m)...) // And what it may call its logs, and how large it may ask them to grow (logs.go, novox/hq ADR 0297). problems = append(problems, LogProblems(m)...) + // And the streams it owns, and who may hear them (streams.go, novox/hq ADR 0363). + problems = append(problems, StreamProblems(m)...) // And the defaults it gives its settings (setting_defaults.go, novox/hq ADR 0262). problems = append(problems, SettingProblems(m)...) // And what it requires optionally, and the files it delivers only while bound (novox/hq ADR 0330). diff --git a/internal/catalogue/seats_declared.go b/internal/catalogue/seats_declared.go index cfda1c45..e6e51cb0 100644 --- a/internal/catalogue/seats_declared.go +++ b/internal/catalogue/seats_declared.go @@ -245,6 +245,8 @@ func IsUndeclaredSeat(problem string) bool { return strings.HasSuffix(problem, U func CatalogueProblems(shelf Shelf) []string { var problems []string + // A stream name has one owner (novox/hq ADR 0363). + problems = append(problems, StreamNameProblems(shelf)...) // Who declares what, and who declared it first. declaredBy := map[string]string{} diff --git a/internal/catalogue/streams.go b/internal/catalogue/streams.go new file mode 100644 index 00000000..d1e91afb --- /dev/null +++ b/internal/catalogue/streams.go @@ -0,0 +1,479 @@ +package catalogue + +import ( + "bytes" + "encoding/json" + "fmt" + "slices" + "strings" +) + +// A module's own stream (novox/hq ADR 0363). +// +// A module declares a stream as it declares a log (ADR 0297) or a bucket (ADR 0201): named locally, the +// events it carries named locally as `emits` names them, its age and its size within the mesh's bounds. +// The controller creates it and stays the only writer of stream definitions. A stream may be restricted +// by a trusted config value of its own module: then a consumer hears a part of it only where its own +// `consumes` asks and that value names it. + +// The mesh's bounds on a module's stream: its age in days and its size in MiB, and what it gets when it +// says nothing. +const ( + StreamLeastDays = 1 + StreamMostDays = 30 + StreamDefaultDays = 7 + StreamLeastMiB = 1 + StreamMostMiB = 8192 + StreamDefaultMiB = 1024 +) + +// GrantKey is the name in a grant's pattern that each key of the setting's value fills. +const GrantKey = "{key}" + +// ReservedStreamPrefixes are the starts of the stream names the mesh derives, which no module's stream +// may take: a seat's work queue, a log, a bucket, an object store, and the dead letters. +var ReservedStreamPrefixes = []string{"SEAT_", "LOG_", "KV_", "OBJ_", "DEAD_"} + +// MeshStreamNames are the mesh's own streams by name. Written here as well as in the broker, which this +// package cannot import; a test keeps them agreeing. +var MeshStreamNames = []string{"CONTROL", "NODES", "ASSIGNMENTS", "EVENTS", "DEAD_LETTER_NOTICES", "DEAD_LETTERS"} + +// StreamDeclaration is one stream a module owns. +type StreamDeclaration struct { + Name string `json:"name"` + // Events are the events it carries, named locally; `*` is one name and `**`, last only, the rest. + Events []string `json:"events"` + // MaxAgeDays and MaxMiB bound it; zero is the default. + MaxAgeDays int `json:"max-age-days,omitempty"` + MaxMiB int `json:"max-mib,omitempty"` + // Grant restricts it, when said. + Grant *StreamGrant `json:"grant,omitempty"` +} + +// StreamGrant says which trusted setting of the module grants a consumer a part of the stream, and +// which part each of its keys names. +type StreamGrant struct { + // Config is the trusted config value of the module that grants it (ADR 0348 §8): set only at the + // controller's terminal or on the operator's Approve. + Config string `json:"config"` + // Events is a pattern holding GrantKey exactly once, which each key of the setting's value fills. + Events string `json:"events"` +} + +// UnmarshalJSON reads a stream strictly: an unknown key is refused here, not passed over. +func (s *StreamDeclaration) UnmarshalJSON(raw []byte) error { + type plain StreamDeclaration + var p plain + dec := json.NewDecoder(bytes.NewReader(raw)) + dec.DisallowUnknownFields() + if err := dec.Decode(&p); err != nil { + return fmt.Errorf("a stream is {name, events, max-age-days, max-mib, grant}: %w", typedUnknown(err)) + } + *s = StreamDeclaration(p) + return nil +} + +// UnmarshalJSON reads a grant strictly. +func (g *StreamGrant) UnmarshalJSON(raw []byte) error { + type plain StreamGrant + var p plain + dec := json.NewDecoder(bytes.NewReader(raw)) + dec.DisallowUnknownFields() + if err := dec.Decode(&p); err != nil { + return fmt.Errorf("a stream's grant is {config, events}: %w", typedUnknown(err)) + } + *g = StreamGrant(p) + return nil +} + +// Days is the stream's age in days, the default where none is said. +func (s StreamDeclaration) Days() int { + if s.MaxAgeDays == 0 { + return StreamDefaultDays + } + return s.MaxAgeDays +} + +// Cap is the stream's size in MiB, the default where none is said. +func (s StreamDeclaration) Cap() int { + if s.MaxMiB == 0 { + return StreamDefaultMiB + } + return s.MaxMiB +} + +// BusName is the stream's name on the bus: its local name in capitals, a hyphen an underscore. +func (s StreamDeclaration) BusName() string { return StreamBusName(s.Name) } + +// StreamBusName is a local stream name as the bus holds it. +func StreamBusName(local string) string { + return strings.ToUpper(strings.ReplaceAll(local, "-", "_")) +} + +// Carries says whether an event, or every event a pattern names, is one this stream carries: within one +// of its patterns. +func (s StreamDeclaration) Carries(pattern string) bool { + for _, p := range s.Events { + if within(pattern, p) { + return true + } + } + return false +} + +// StreamProblems is what is wrong with a manifest's streams, judged on the one manifest. +func StreamProblems(m Manifest) []string { + if len(m.Streams) == 0 { + return nil + } + var problems []string + if !stateName.MatchString(m.Module) { + problems = append(problems, fmt.Sprintf( + "%s declares a stream, and a module's name is part of its stream's subjects, which take one plain "+ + "name — no dot (novox/hq ADR 0363)", m.Module)) + } + seen := map[string]bool{} + terminal := TerminalKeys(m) + for i, s := range m.Streams { + switch { + case !stateName.MatchString(s.Name): + problems = append(problems, fmt.Sprintf( + "%s declares stream %q: a stream is named locally — lower-case letters, digits and hyphens "+ + "(novox/hq ADR 0363)", m.Module, s.Name)) + case seen[s.BusName()]: + problems = append(problems, fmt.Sprintf("%s declares stream %q twice", m.Module, s.Name)) + case slices.Contains(MeshStreamNames, s.BusName()): + problems = append(problems, fmt.Sprintf( + "%s declares stream %q, which is the mesh's own stream %s (novox/hq ADR 0363)", m.Module, s.Name, s.BusName())) + default: + for _, prefix := range ReservedStreamPrefixes { + if strings.HasPrefix(s.BusName(), prefix) { + problems = append(problems, fmt.Sprintf( + "%s declares stream %q, whose name on the bus starts with %s, which the mesh derives names "+ + "with (novox/hq ADR 0363)", m.Module, s.Name, prefix)) + } + } + } + seen[s.BusName()] = true + if s.MaxAgeDays != 0 && (s.MaxAgeDays < StreamLeastDays || s.MaxAgeDays > StreamMostDays) { + problems = append(problems, fmt.Sprintf("%s keeps stream %q %d days; a stream keeps between %d and %d "+ + "(novox/hq ADR 0363)", m.Module, s.Name, s.MaxAgeDays, StreamLeastDays, StreamMostDays)) + } + if s.MaxMiB != 0 && (s.MaxMiB < StreamLeastMiB || s.MaxMiB > StreamMostMiB) { + problems = append(problems, fmt.Sprintf("%s caps stream %q at %d MiB; a stream holds between %d and %d "+ + "(novox/hq ADR 0363)", m.Module, s.Name, s.MaxMiB, StreamLeastMiB, StreamMostMiB)) + } + if len(s.Events) == 0 { + problems = append(problems, fmt.Sprintf("%s declares stream %q with no events", m.Module, s.Name)) + } + for _, e := range s.Events { + if err := eventPattern(e, false); err != nil { + problems = append(problems, fmt.Sprintf("%s's stream %q carries %q: %v", m.Module, s.Name, e, err)) + } + } + // **An event goes to one place** (ADR 0363 §2): an emitted event a stream carries, or two streams + // carrying one event, would leave the module's runtime to choose. + for _, e := range m.EmitsAll() { + if s.Carries(e) { + problems = append(problems, fmt.Sprintf( + "%s emits %q, which its stream %q carries: an event goes to EVENTS or to the module's stream, "+ + "so name it in one place (novox/hq ADR 0363)", m.Module, e, s.Name)) + } + } + for _, other := range m.Streams[i+1:] { + for _, a := range s.Events { + for _, b := range other.Events { + if patternsOverlap(a, b) { + problems = append(problems, fmt.Sprintf( + "%s's streams %q and %q both carry %q and %q: an event goes to one stream (novox/hq "+ + "ADR 0363)", m.Module, s.Name, other.Name, a, b)) + } + } + } + } + if g := s.Grant; g != nil { + if !slices.Contains(terminal, g.Config) { + problems = append(problems, fmt.Sprintf( + "%s's stream %q is granted by %q, which is not a trusted config value of the module, set only at "+ + "the controller's terminal or on the operator's Approve: a grant of what a module hears is the "+ + "operator's (novox/hq ADR 0348 §8, ADR 0363)", m.Module, s.Name, g.Config)) + } + if strings.Count(g.Events, GrantKey) != 1 { + problems = append(problems, fmt.Sprintf( + "%s's stream %q grants %q, which must hold %s exactly once, as a name of its own (novox/hq ADR 0363)", + m.Module, s.Name, g.Events, GrantKey)) + } else if err := eventPattern(g.Events, true); err != nil { + problems = append(problems, fmt.Sprintf("%s's stream %q grants %q: %v", m.Module, s.Name, g.Events, err)) + } + } + } + return problems +} + +// eventPattern checks one local event pattern: names, `*` for one, `**` last for the rest; with key, +// GrantKey is a name too. +func eventPattern(pattern string, key bool) error { + if pattern == "" { + return fmt.Errorf("a pattern needs a name") + } + parts := strings.Split(pattern, ".") + for i, part := range parts { + switch { + case part == TheRest: + if i != len(parts)-1 { + return fmt.Errorf("%q stands for the rest of a name, so nothing may follow it", TheRest) + } + case part == OneName: + case key && part == GrantKey: + case !eventName.MatchString(part): + return fmt.Errorf("%q is not a usable name: lower-case letters, digits and dashes", part) + } + } + return nil +} + +// StreamsHearing is every stream of a module that carries some of a consumed pattern's events, and +// whether one of them carries all of them. +func (m Manifest) StreamsHearing(pattern string) (hearing []StreamDeclaration, whole bool) { + for _, s := range m.Streams { + for _, p := range s.Events { + if patternsOverlap(pattern, p) { + hearing = append(hearing, s) + break + } + } + if s.Carries(pattern) { + whole = true + } + } + return hearing, whole +} + +// within says every event pattern a names is one b names too. +func within(a, b string) bool { + x, y := strings.Split(a, "."), strings.Split(b, ".") + for i := 0; ; i++ { + switch { + case i == len(y): + return i == len(x) + case y[i] == TheRest: + return i < len(x) + case i == len(x): + return false + case x[i] == TheRest: + return false + case y[i] == OneName: + continue + case x[i] == OneName: + return false + case x[i] != y[i]: + return false + } + } +} + +// patternsOverlap says some event both patterns name. +func patternsOverlap(a, b string) bool { + x, y := strings.Split(a, "."), strings.Split(b, ".") + for i := 0; ; i++ { + switch { + case i == len(x) && i == len(y): + return true + case i == len(x) || i == len(y): + return false + case x[i] == TheRest || y[i] == TheRest: + return true + case x[i] == OneName || y[i] == OneName || x[i] == y[i]: + continue + default: + return false + } + } +} + +// StreamNameProblems are the rules about stream names no single manifest can be judged against: two +// modules declaring one name. +func StreamNameProblems(shelf Shelf) []string { + var problems []string + declaredBy := map[string]string{} + for _, module := range shelfOrder(shelf) { + for _, s := range shelf[module].Streams { + name := s.BusName() + if first, taken := declaredBy[name]; taken && first != module { + problems = append(problems, fmt.Sprintf( + "%s declares the stream %s, which %s already declares; a stream has one owner (novox/hq ADR 0363)", + module, name, first)) + continue + } + declaredBy[name] = module + } + } + return problems +} + +// GrantedKeys reads a restricting config value (novox/hq ADR 0363 §4): each key whose list names the +// module wherever it runs, or the module on this machine as `.`. A value of another shape +// grants nothing; a key that is not one usable name grants nothing. +func GrantedKeys(value any, module, machine string) []string { + if text, isText := value.(string); isText { + // A layer may hold the object as its JSON text. + var parsed any + if json.Unmarshal([]byte(text), &parsed) != nil { + return nil + } + value = parsed + } + entries, ok := value.(map[string]any) + if !ok { + return nil + } + var out []string + for key, listed := range entries { + if !eventName.MatchString(key) { + continue + } + names, ok := listed.([]any) + if !ok { + continue + } + for _, n := range names { + s, _ := n.(string) + if s == module || (machine != "" && s == machine+"."+module) { + out = append(out, key) + break + } + } + } + slices.Sort(out) + return out +} + +// Granted is a grant's pattern with its key filled. +func (g StreamGrant) Granted(key string) string { return strings.Replace(g.Events, GrantKey, key, 1) } + +// narrow is the events both patterns name, as one pattern; false when they name none in common. +func narrow(a, b string) (string, bool) { + x, y := strings.Split(a, "."), strings.Split(b, ".") + var out []string + for i := 0; ; i++ { + switch { + case i == len(x) && i == len(y): + return strings.Join(out, "."), true + case i == len(x) || i == len(y): + return "", false + case x[i] == TheRest: + return strings.Join(append(out, y[i:]...), "."), true + case y[i] == TheRest: + return strings.Join(append(out, x[i:]...), "."), true + case x[i] == OneName: + out = append(out, y[i]) + case y[i] == OneName || x[i] == y[i]: + out = append(out, x[i]) + default: + return "", false + } + } +} + +// HeardPart is what a consumer hears of one module's stream: the patterns, local, it is granted. +type HeardPart struct { + Emitter string + Stream StreamDeclaration + Patterns []string +} + +// Hearing is what a consumer's `consumes` hears of the module streams the shelf declares, on one machine +// (novox/hq ADR 0363 §5): for each stream, its `consumes` narrowed to the stream's events and, for a +// restricted stream, to the parts the provider's setting grants it. values answers a restricting setting's +// value on each machine running the provider. Answered too: whether any `consumes` names a module owning a +// stream, granted or not, and the `consumes` left for EVENTS — those naming something the provider emits +// there. +// +// Only an entry naming its emitter literally reaches a module's stream: a wildcard emitter, `**` among them, +// is EVENTS' alone, so nothing that hears every event hears a restricted stream by saying so. +func Hearing(consumer Manifest, shelf map[string]Manifest, machine string, + values func(provider, setting string) []any) (parts []HeardPart, touches bool, onEvents []string) { + byStream := map[string]*HeardPart{} + var order []string + for _, c := range consumer.Consumes { + emitter, rest, named := strings.Cut(c, ".") + provider, known := shelf[emitter] + if !named || emitter == OneName || emitter == TheRest || !known || len(provider.Streams) == 0 { + onEvents = append(onEvents, c) + continue + } + hearing, whole := provider.StreamsHearing(rest) + if len(hearing) > 0 { + touches = true + } + // Kept for EVENTS while the provider says something there it names: a stream carrying it whole, or + // a provider that says nothing of it on EVENTS, leaves it nothing to hear there. + if !whole && slices.ContainsFunc(provider.EmitsAll(), func(e string) bool { return patternsOverlap(rest, e) }) { + onEvents = append(onEvents, c) + } + for _, s := range hearing { + var granted []string + if s.Grant != nil { + for _, v := range values(emitter, s.Grant.Config) { + for _, key := range GrantedKeys(v, consumer.Module, machine) { + granted = append(granted, s.Grant.Granted(key)) + } + } + } + var got []string + for _, p := range s.Events { + x, ok := narrow(rest, p) + if !ok { + continue + } + if s.Grant == nil { + got = append(got, x) + continue + } + for _, g := range granted { + if y, ok := narrow(x, g); ok { + got = append(got, y) + } + } + } + if len(got) == 0 { + continue + } + id := emitter + "/" + s.BusName() + if byStream[id] == nil { + byStream[id] = &HeardPart{Emitter: emitter, Stream: s} + order = append(order, id) + } + byStream[id].Patterns = append(byStream[id].Patterns, got...) + } + } + for _, id := range order { + h := byStream[id] + h.Patterns = uncovered(h.Patterns) + parts = append(parts, *h) + } + return parts, touches, onEvents +} + +// uncovered is a list of patterns without those another names whole, and without repeats: the bus refuses +// a consumer whose filters overlap. +func uncovered(patterns []string) []string { + var out []string + for i, p := range patterns { + covered := false + for j, q := range patterns { + if i == j { + continue + } + if within(p, q) && (!within(q, p) || j < i) { + covered = true + break + } + } + if !covered { + out = append(out, p) + } + } + slices.Sort(out) + return out +} diff --git a/internal/catalogue/streams_test.go b/internal/catalogue/streams_test.go new file mode 100644 index 00000000..f0d0aee4 --- /dev/null +++ b/internal/catalogue/streams_test.go @@ -0,0 +1,215 @@ +package catalogue + +import ( + "encoding/json" + "slices" + "strings" + "testing" +) + +// mailManifest is a module owning a restricted stream, as mesh-mail declares it (novox/hq ADR 0363), with +// the trusted mergeable file that takes its grants setting; extra is spliced in at the top. +func mailManifest(t *testing.T, streams, extra string) string { + t.Helper() + return `{"module":"mesh-mail","version":"1",` + extra + `"streams":` + streams + `,` + + `"resources":[{"id":"state","type":"directory","mode":"0700","place":"."},` + + `{"id":"grants","trusted":true,"type":"file","path":"${dir:state}/grants.json","mode":"0600",` + + `"merge":"json","takes-settings":["grants"],"content":"{\"grants\": {}}"}]}` +} + +const mailStream = `[{"name":"mail","events":["mail.*.message.added","mail.*.message.changed",` + + `"mail.*.message.removed"],"max-age-days":7,"grant":{"config":"grants","events":"mail.{key}.**"}}]` + +// A module declares a stream of its own (ADR 0363 §1): its events with a variable name (§3), its age and +// size with defaults, and a grant by a trusted setting of its own (§4). +func TestAModuleDeclaresAStreamOfItsOwn(t *testing.T) { + m, err := ParseManifest([]byte(mailManifest(t, mailStream, ""))) + if err != nil { + t.Fatal(err) + } + if len(m.Streams) != 1 { + t.Fatalf("streams not read: %+v", m.Streams) + } + s := m.Streams[0] + if s.BusName() != "MAIL" || s.Days() != 7 || s.Cap() != StreamDefaultMiB || s.Grant == nil || + s.Grant.Config != "grants" || s.Grant.Granted("mesh") != "mail.mesh.**" { + t.Fatalf("stream read as %+v", s) + } + if !s.Carries("mail.work.message.added") || s.Carries("mail.work.body") || !s.Carries("mail.*.message.removed") { + t.Fatal("the stream's patterns do not say what it carries") + } + plain, err := ParseManifest([]byte(`{"module":"feed","version":"1","streams":[{"name":"feed-items","events":["item.**"]}]}`)) + if err != nil { + t.Fatalf("a stream without a grant was refused: %v", err) + } + if got := plain.Streams[0]; got.BusName() != "FEED_ITEMS" || got.Days() != StreamDefaultDays || got.Grant != nil { + t.Fatalf("defaults read as %+v", got) + } +} + +// Registration refuses a stream the mesh could not hold or would leave ambiguous (ADR 0363 §1, §2, §4). +func TestAStreamIsNamedBoundedAndGrantedByATrustedConfigValue(t *testing.T) { + grants := `"grant":{"config":"grants","events":"mail.{key}.**"}` + for _, c := range []struct{ manifest, says string }{ + {mailManifest(t, `[{"name":"Mail","events":["mail.**"]}]`, ""), `a stream is named locally`}, + {mailManifest(t, `[{"name":"mail.x","events":["mail.**"]}]`, ""), `a stream is named locally`}, + {mailManifest(t, `[{"name":"events","events":["mail.**"]}]`, ""), `the mesh's own stream EVENTS`}, + {mailManifest(t, `[{"name":"log-x","events":["mail.**"]}]`, ""), `starts with LOG_`}, + {mailManifest(t, `[{"name":"mail","events":["mail.**"]},{"name":"mail","events":["x.**"]}]`, ""), `twice`}, + {mailManifest(t, `[{"name":"mail","events":[]}]`, ""), `with no events`}, + {mailManifest(t, `[{"name":"mail","events":["mail.**.x"]}]`, ""), `nothing may follow it`}, + {mailManifest(t, `[{"name":"mail","events":["mail.**"],"max-age-days":31}]`, ""), `between 1 and 30`}, + {mailManifest(t, `[{"name":"mail","events":["mail.**"],"max-mib":9000}]`, ""), `between 1 and 8192`}, + {mailManifest(t, `[{"name":"mail","events":["mail.**"],"stream":"MAIL"}]`, ""), `{name, events`}, + {mailManifest(t, `[{"name":"mail","events":["mail.**"]}]`, `"emits":["mail.mesh.message.added"],`), + `which its stream "mail" carries`}, + {mailManifest(t, `[{"name":"mail","events":["mail.**"]},{"name":"more","events":["mail.x.*"]}]`, ""), + `both carry`}, + {mailManifest(t, `[{"name":"mail","events":["mail.**"],"grant":{"config":"tags","events":"mail.{key}.**"}}]`, ""), + `is not a trusted config value`}, + {mailManifest(t, `[{"name":"mail","events":["mail.**"],"grant":{"config":"grants","events":"mail.**"}}]`, ""), + `exactly once`}, + {mailManifest(t, `[{"name":"mail","events":["mail.**"],"grant":{"config":"grants","events":"{key}.{key}"}}]`, ""), + `exactly once`}, + {mailManifest(t, `[{"name":"mail","events":["mail.**"],`+grants+`,"why":1}]`, ""), `{name, events`}, + {`{"module":"a.b","version":"1","streams":[{"name":"x","events":["x"]}]}`, `no dot`}, + } { + _, err := ParseManifest([]byte(c.manifest)) + if err == nil { + t.Errorf("%s was accepted", c.manifest) + continue + } + if !strings.Contains(err.Error(), c.says) { + t.Errorf("%s refused for the wrong reason: %v", c.manifest, err) + } + } +} + +// A stream name has one owner across the catalogue (ADR 0363 §1). +func TestTwoModulesCannotDeclareOneStream(t *testing.T) { + a := Manifest{Module: "a", Streams: []StreamDeclaration{{Name: "mail", Events: []string{"x"}}}} + b := Manifest{Module: "b", Streams: []StreamDeclaration{{Name: "mail", Events: []string{"y"}}}} + got := strings.Join(CatalogueProblems(Shelf{"a": a, "b": b}), "; ") + if !strings.Contains(got, "b declares the stream MAIL, which a already declares") { + t.Fatalf("a second owner of MAIL was not refused: %s", got) + } + if got := StreamNameProblems(Shelf{"a": a}); len(got) != 0 { + t.Fatalf("one owner refused: %v", got) + } +} + +// The setting's value names who hears each key: a module wherever it runs, or on one machine (ADR 0363 §4). +func TestARestrictingSettingNamesAModuleOrAModuleOnAMachine(t *testing.T) { + var value any + if err := json.Unmarshal([]byte(`{"mesh":["reader","person.jochen"],"work":["ace.reader"],"home":["other"],`+ + `"Bad.Key":["reader"]}`), &value); err != nil { + t.Fatal(err) + } + if got := GrantedKeys(value, "reader", "ace"); !slices.Equal(got, []string{"mesh", "work"}) { + t.Fatalf("on ace reader is granted %v", got) + } + if got := GrantedKeys(value, "reader", "shanks"); !slices.Equal(got, []string{"mesh"}) { + t.Fatalf("on shanks reader is granted %v", got) + } + if got := GrantedKeys(`{"mesh":["reader"]}`, "reader", "ace"); !slices.Equal(got, []string{"mesh"}) { + t.Fatalf("a value held as its JSON text grants %v", got) + } + if got := GrantedKeys([]any{"reader"}, "reader", "ace"); len(got) != 0 { + t.Fatalf("a value of another shape granted %v", got) + } +} + +// A consumer hears a restricted stream only where its consumes asks and the setting grants it, and only +// by naming the emitter (ADR 0363 §5); what a stream carries whole leaves EVENTS. +func TestAConsumerHearsOnlyWhatItAsksAndIsGranted(t *testing.T) { + provider, err := ParseManifest([]byte(mailManifest(t, mailStream, `"emits":["account.added"],`))) + if err != nil { + t.Fatal(err) + } + shelf := map[string]Manifest{"mesh-mail": provider} + var value any + _ = json.Unmarshal([]byte(`{"mesh":["reader"],"work":["shanks.reader"]}`), &value) + values := func(module, setting string) []any { + if module == "mesh-mail" && setting == "grants" { + return []any{value} + } + return nil + } + reader := Manifest{Module: "reader", Consumes: []string{"mesh-mail.mail.**", "mesh-mail.account.added", "billing.paid"}} + + parts, touches, onEvents := Hearing(reader, shelf, "ace", values) + if !touches || len(parts) != 1 || parts[0].Stream.BusName() != "MAIL" { + t.Fatalf("heard %+v", parts) + } + want := []string{"mail.mesh.message.added", "mail.mesh.message.changed", "mail.mesh.message.removed"} + if !slices.Equal(parts[0].Patterns, want) { + t.Fatalf("on ace reader hears %v, want %v", parts[0].Patterns, want) + } + if !slices.Equal(onEvents, []string{"mesh-mail.account.added", "billing.paid"}) { + t.Fatalf("left on EVENTS: %v", onEvents) + } + parts, _, _ = Hearing(reader, shelf, "shanks", values) + if len(parts) != 1 || len(parts[0].Patterns) != 6 { + t.Fatalf("on shanks reader hears %+v", parts) + } + + narrow := Manifest{Module: "reader", Consumes: []string{"mesh-mail.mail.*.message.added"}} + parts, _, _ = Hearing(narrow, shelf, "ace", values) + if len(parts) != 1 || !slices.Equal(parts[0].Patterns, []string{"mail.mesh.message.added"}) { + t.Fatalf("a narrower consumes heard %+v", parts) + } + + stranger := Manifest{Module: "stranger", Consumes: []string{"mesh-mail.mail.**"}} + parts, touches, onEvents = Hearing(stranger, shelf, "ace", values) + if len(parts) != 0 || !touches || len(onEvents) != 0 { + t.Fatalf("a module not granted heard %+v (touches %v, on EVENTS %v)", parts, touches, onEvents) + } + everything := Manifest{Module: "audit", Consumes: []string{"**", "*.mail.**"}} + parts, touches, _ = Hearing(everything, shelf, "ace", values) + if len(parts) != 0 || touches { + t.Fatalf("a wildcard emitter reached a module's stream: %+v", parts) + } + parts, _, _ = Hearing(reader, shelf, "ace", func(string, string) []any { return nil }) + if len(parts) != 0 { + t.Fatalf("with no setting reader heard %+v", parts) + } +} + +// A stream without a grant is heard by whoever asks, within its patterns. +func TestAnUnrestrictedStreamIsHeardByWhoeverAsks(t *testing.T) { + feed := Manifest{Module: "feed", Streams: []StreamDeclaration{{Name: "items", Events: []string{"item.*.added", "item.*.gone"}}}} + reader := Manifest{Module: "reader", Consumes: []string{"feed.item.**", "feed.item.x.added"}} + parts, touches, onEvents := Hearing(reader, map[string]Manifest{"feed": feed}, "ace", nil) + if !touches || len(onEvents) != 0 || len(parts) != 1 || + !slices.Equal(parts[0].Patterns, []string{"item.*.added", "item.*.gone"}) { + t.Fatalf("heard %+v, on EVENTS %v", parts, onEvents) + } +} + +func TestNarrowingTwoPatterns(t *testing.T) { + for _, c := range []struct{ a, b, want string }{ + {"mail.**", "mail.*.message.added", "mail.*.message.added"}, + {"mail.*.message.added", "mail.mesh.**", "mail.mesh.message.added"}, + {"a.*", "*.b", "a.b"}, + {"**", "a.**", "a.**"}, + {"a.b", "a.c", ""}, + {"a", "a.**", ""}, + } { + got, ok := narrow(c.a, c.b) + if (c.want == "") == ok || got != c.want { + t.Errorf("narrow(%s, %s) = %q, %v; want %q", c.a, c.b, got, ok, c.want) + } + } + if got := uncovered([]string{"a.**", "a.b", "a.b", "c"}); !slices.Equal(got, []string{"a.**", "c"}) { + t.Fatalf("uncovered: %v", got) + } +} + +// The mesh's own stream names are written in the broker too; this list must name the same. +func TestStreamNamesTheMeshReservesIncludeEveryOwnStream(t *testing.T) { + for _, n := range []string{"EVENTS", "CONTROL", "NODES", "ASSIGNMENTS"} { + if !slices.Contains(MeshStreamNames, n) { + t.Fatalf("%s is missing", n) + } + } +} diff --git a/internal/inventory/busrecords.go b/internal/inventory/busrecords.go index fa540dff..d5519763 100644 --- a/internal/inventory/busrecords.go +++ b/internal/inventory/busrecords.go @@ -61,12 +61,23 @@ func (i *Inventory) BusRecords(ctx context.Context) (broker.Records, error) { out := broker.Records{Assigned: map[string][]broker.Declared{}, People: map[string][]string{}, Interchangeable: map[string]bool{}} + assigned := map[string][]string{} for _, n := range nodes { - out.Nodes = append(out.Nodes, n.Name) modules, err := i.Assigned(ctx, n.Name) if err != nil { return broker.Records{}, fmt.Errorf("cannot read what %s runs: %w", n.Name, err) } + assigned[n.Name] = modules + } + // What each restricting setting says on each machine running its module (novox/hq ADR 0363 §5): read + // once, for every consumer composed below. + grants, err := i.streamGrantValues(ctx, declared, nodes, assigned) + if err != nil { + return broker.Records{}, err + } + for _, n := range nodes { + out.Nodes = append(out.Nodes, n.Name) + modules := assigned[n.Name] for _, module := range modules { m, known := declared[module] if !known { @@ -89,6 +100,9 @@ func (i *Inventory) BusRecords(ctx context.Context) (broker.Records, error) { d.KeyedReads = append(d.KeyedReads, broker.KeyedRead{Read: sr.Read, Key: sr.Key}) } d.Holds = heldHere(d.Holds, holdings, n.Name, module) + // And the module streams it hears on this machine, as granted (novox/hq ADR 0363 §5): what a + // stream carries whole leaves EVENTS' consumes. + d.Hears, d.HearsAStream, d.Consumes = hearsOf(m, declared, n.Name, grants, d.Consumes) out.Assigned[n.Name] = append(out.Assigned[n.Name], d) if m.Instances == catalogue.InstancesInterchangeable { out.Interchangeable[m.Module] = true @@ -169,6 +183,8 @@ func declaredFor(m catalogue.Manifest, seats map[string]catalogue.SeatDeclaratio Reads: m.Reads, // And the logs it keeps (novox/hq ADR 0297). Logs: logsOf(m), + // And the streams it owns (novox/hq ADR 0363). + Streams: streamsOf(m), // And the tools its health asks (novox/hq ADR 0240): the machine's node-engine is granted them. Checks: catalogue.HealthChecks(m), // And whether it runs as an account of its own (novox/hq ADR 0259 §8). @@ -362,3 +378,101 @@ func (i *Inventory) DeclaredTrafficSeats(ctx context.Context) ([]broker.Seat, er } return out, nil } + +// streamsOf is the streams a module owns, as the bus holds them (novox/hq ADR 0363). +func streamsOf(m catalogue.Manifest) []broker.ModuleStream { + var out []broker.ModuleStream + for _, s := range m.Streams { + events := make([]string, 0, len(s.Events)) + for _, e := range s.Events { + events = append(events, broker.WirePattern(e)) + } + out = append(out, broker.ModuleStream{Module: m.Module, Name: s.BusName(), Events: events, + MaxAgeDays: s.Days(), MaxMiB: s.Cap(), Restricted: s.Grant != nil}) + } + return out +} + +// DeclaredStreams is every module stream the catalogue declares, registered modules assigned or not: a +// stream exists from registration, as a log does (novox/hq ADR 0363 §1). +func (i *Inventory) DeclaredStreams(ctx context.Context) ([]broker.ModuleStream, error) { + declared, err := i.Catalogue(ctx) + if err != nil { + return nil, fmt.Errorf("cannot read the catalogue: %w", err) + } + var out []broker.ModuleStream + for _, m := range declared { + out = append(out, streamsOf(m)...) + } + return out, nil +} + +// grantValues is each restricting setting's value on each machine running its module, by `/`. +type grantValues map[string][]any + +// streamGrantValues reads, for every module owning a restricted stream, the setting that restricts it on every +// machine the module runs on: the mesh's layer under that machine's, the machine's winning (novox/hq ADR 0363 §5). +func (i *Inventory) streamGrantValues(ctx context.Context, declared map[string]catalogue.Manifest, + nodes []Node, assigned map[string][]string) (grantValues, error) { + out := grantValues{} + for _, n := range nodes { + for _, module := range assigned[n.Name] { + m, known := declared[module] + if !known { + continue + } + var settings []string + for _, s := range m.Streams { + if s.Grant != nil && !slices.Contains(settings, s.Grant.Config) { + settings = append(settings, s.Grant.Config) + } + } + if len(settings) == 0 { + continue + } + layers, err := i.SettingsFor(ctx, n.Name, module) + if err != nil { + return nil, fmt.Errorf("cannot read %s's settings on %s, which grant who hears its stream: %w", + module, n.Name, err) + } + for _, setting := range settings { + if v, set := settingValue(layers, setting); set { + out[module+"/"+setting] = append(out[module+"/"+setting], v) + } + } + } + } + return out, nil +} + +// settingValue is one key's value over a machine's layers, the last that sets it winning. Pure. +func settingValue(layers []catalogue.Layer, key string) (any, bool) { + var value any + set := false + for _, l := range layers { + if v, ok := l.Values[key]; ok { + value, set = v, true + } + } + return value, set +} + +// hearsOf is what a module hears of the module streams on one machine, and the consumes left for EVENTS +// (novox/hq ADR 0363 §5). Pure. +func hearsOf(m catalogue.Manifest, declared map[string]catalogue.Manifest, node string, grants grantValues, + consumes []string) ([]broker.Heard, bool, []string) { + reading := m + reading.Consumes = consumes + parts, touches, onEvents := catalogue.Hearing(reading, declared, node, func(provider, setting string) []any { + return grants[provider+"/"+setting] + }) + var out []broker.Heard + for _, p := range parts { + h := broker.Heard{Stream: p.Stream.BusName(), Emitter: p.Emitter} + for _, pattern := range p.Patterns { + h.Filters = append(h.Filters, broker.ModuleStreamSubject(p.Emitter, broker.WirePattern(pattern))) + } + out = append(out, h) + } + return out, touches, onEvents +} diff --git a/internal/inventory/streams_test.go b/internal/inventory/streams_test.go new file mode 100644 index 00000000..9efef62f --- /dev/null +++ b/internal/inventory/streams_test.go @@ -0,0 +1,45 @@ +package inventory + +import ( + "encoding/json" + "slices" + "testing" + + "github.com/novox/mesh-controller/internal/catalogue" +) + +// A machine's layer wins over the mesh's for the setting that restricts a stream (novox/hq ADR 0363 §5). +func TestARestrictingSettingIsReadMachineOverMesh(t *testing.T) { + layers := []catalogue.Layer{ + {From: catalogue.MeshWideLayer, Values: map[string]any{"grants": "mesh", "other": 1}}, + {From: "novox", Values: map[string]any{"grants": "novox"}}, + } + if v, set := settingValue(layers, "grants"); !set || v != "novox" { + t.Fatalf("read %v, %v", v, set) + } + if _, set := settingValue(layers[:1], "absent"); set { + t.Fatal("a key no layer sets read as set") + } +} + +// What a consumer hears is composed into the bus's subjects, and what a stream carries leaves EVENTS. +func TestWhatAModuleHearsIsComposedOntoTheStreamsSubjects(t *testing.T) { + provider := catalogue.Manifest{Module: "mesh-mail", Streams: []catalogue.StreamDeclaration{{Name: "mail", + Events: []string{"mail.*.message.added"}, Grant: &catalogue.StreamGrant{Config: "grants", Events: "mail.{key}.**"}}}} + var value any + _ = json.Unmarshal([]byte(`{"mesh":["reader"]}`), &value) + grants := grantValues{"mesh-mail/grants": {value}} + reader := catalogue.Manifest{Module: "reader"} + hears, touches, onEvents := hearsOf(reader, map[string]catalogue.Manifest{"mesh-mail": provider}, "ace", grants, + []string{"mesh-mail.mail.**", "billing.paid"}) + if !touches || len(hears) != 1 || hears[0].Stream != "MAIL" || + !slices.Equal(hears[0].Filters, []string{"mesh.stream.mesh-mail.event.mail.mesh.message.added"}) { + t.Fatalf("heard %+v", hears) + } + if !slices.Equal(onEvents, []string{"billing.paid"}) { + t.Fatalf("left on EVENTS: %v", onEvents) + } + if got := streamsOf(provider); len(got) != 1 || got[0].Name != "MAIL" || got[0].MaxAgeDays != 7 || !got[0].Restricted { + t.Fatalf("the provider's stream: %+v", got) + } +}