diff --git a/cmd/mesh-controller/busobjects.go b/cmd/mesh-controller/busobjects.go index daed0d65..0700d79b 100644 --- a/cmd/mesh-controller/busobjects.go +++ b/cmd/mesh-controller/busobjects.go @@ -83,8 +83,12 @@ func assertBusObjects(ctx context.Context, inv *inventory.Inventory, r broker.Ra failed = append(failed, fmt.Errorf("the worker %s on %s: %w", c.Name, c.Stream, err)) } } + restricted := map[string]bool{} + for _, s := range modStreams { + restricted[s.Name] = s.Restricted + } for _, c := range consumers { - if err := r.EnsureConsumer(c.Consumer); err != nil { + if err := broker.EnsureHeard(r, c.Consumer, restricted[c.Consumer.Stream]); err != nil { failed = append(failed, fmt.Errorf("how %s on %s hears what it consumes: %w", c.Module, c.Node, err)) } } diff --git a/internal/broker/modstreams.go b/internal/broker/modstreams.go index 8345a46b..e6e3a315 100644 --- a/internal/broker/modstreams.go +++ b/internal/broker/modstreams.go @@ -47,7 +47,7 @@ type ModuleStream struct { Events []string MaxAgeDays int MaxMiB int - // Restricted says a trusted setting of the module grants who hears it (ADR 0363 §4). + // Restricted says a trusted config value of the module grants who hears it (ADR 0363 §4). Restricted bool } @@ -68,7 +68,7 @@ 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" + why += "; heard only by the consumers its trusted config value grants" } return Stream{ Name: s.Name, @@ -92,7 +92,7 @@ func ModuleStreams(declared []ModuleStream) []Stream { } // 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 +// within the stream and, for a restricted stream, within what the provider's trusted config value grants it // (ADR 0363 §5). Composed only with at least one subject. type Heard struct { Stream string @@ -220,3 +220,24 @@ func PruneHeard(p Pruner, streams []ModuleStream, wanted []Consumer) ([]string, } return removed, nil } + +// EnsureHeard asserts a consumer on a module's stream (ADR 0363 §5, §6). **On a restricted stream, a consumer +// the bus will not bring to match is removed**: a pull consumer delivers to whoever pulls it, so one left with +// its wider filters after a narrowing was refused would keep handing over what the grant no longer names, +// whatever the user list now says. The next assertion makes it again with the granted filters. The refusal is +// returned either way, saying what was done, so the send says it and raises it. +func EnsureHeard(r Raiser, c Consumer, restricted bool) error { + err := r.EnsureConsumer(c) + if err == nil || !restricted { + return err + } + p, prunes := r.(Pruner) + if !prunes { + return fmt.Errorf("%w; it could not be removed here, and may still deliver more than it is granted", err) + } + if rerr := p.RemoveConsumer(c.Stream, c.Name); rerr != nil { + return fmt.Errorf("%w; REMOVING IT FAILED TOO, so it may still deliver more than it is granted: %v", err, rerr) + } + return fmt.Errorf("%w; REMOVED, so it delivers nothing it is no longer granted: the next assertion makes it "+ + "again with its granted filters (novox/hq ADR 0363 §6)", err) +} diff --git a/internal/broker/modstreams_live_test.go b/internal/broker/modstreams_live_test.go index 1a2366bc..68f42df6 100644 --- a/internal/broker/modstreams_live_test.go +++ b/internal/broker/modstreams_live_test.go @@ -46,9 +46,15 @@ func TestAModuleStreamOnARealBus(t *testing.T) { } narrowed := c narrowed.Filters = []string{"mesh.stream.mesh-mail.event.mail.mesh.message.changed"} - if err := js.EnsureConsumer(narrowed); err != nil { + if err := EnsureHeard(js, narrowed, true); err != nil { t.Fatalf("narrowing the grant: %v", err) } + if info, _ := js.js.ConsumerInfo("MAIL", c.Name); info.Config.FilterSubject != narrowed.Filters[0] || len(info.Config.FilterSubjects) != 0 { + t.Fatalf("narrowed to %q %v, want only %s", info.Config.FilterSubject, info.Config.FilterSubjects, narrowed.Filters[0]) + } + if info, _ := js.js.ConsumerInfo("MAIL", c.Name); info.NumPending != 0 { + t.Fatalf("after narrowing, %d of the reader's former grant are still pending", info.NumPending) + } removed, err := PruneHeard(js, []ModuleStream{mailStream}, nil) if err != nil { t.Fatal(err) diff --git a/internal/broker/modstreams_test.go b/internal/broker/modstreams_test.go index 2518d989..ffa04bd2 100644 --- a/internal/broker/modstreams_test.go +++ b/internal/broker/modstreams_test.go @@ -1,8 +1,10 @@ package broker import ( + "errors" "reflect" "slices" + "strings" "testing" ) @@ -174,3 +176,35 @@ func TestAConsumerThatLostItsGrantIsRemoved(t *testing.T) { t.Fatalf("removed %v", removed) } } + +type refusingRaiser struct { + fakePruner + refuse bool +} + +func (r *refusingRaiser) EnsureStream(Stream) error { return nil } +func (r *refusingRaiser) EnsureConsumer(Consumer) error { + if r.refuse { + return errors.New("consumer update refused") + } + return nil +} + +// A narrowing the bus refuses on a restricted stream removes the consumer, so its wider filters deliver nothing +// more; on an unrestricted stream the refusal is only said (review of PR 338, ADR 0363 §6). +func TestARefusedNarrowingOnARestrictedStreamRemovesTheConsumer(t *testing.T) { + c := Consumer{Name: "ace_reader", Stream: "MAIL", Filters: mailHeard.Filters[:1]} + r := &refusingRaiser{refuse: true} + err := EnsureHeard(r, c, true) + if err == nil || !strings.Contains(err.Error(), "REMOVED") || !slices.Equal(r.removed, []string{"MAIL/ace_reader"}) { + t.Fatalf("restricted: %v, removed %v", err, r.removed) + } + r = &refusingRaiser{refuse: true} + if err := EnsureHeard(r, c, false); err == nil || len(r.removed) != 0 { + t.Fatalf("unrestricted: %v, removed %v", err, r.removed) + } + r = &refusingRaiser{} + if err := EnsureHeard(r, c, true); err != nil || len(r.removed) != 0 { + t.Fatalf("accepted: %v, removed %v", err, r.removed) + } +} diff --git a/internal/broker/stream_names_agreement_test.go b/internal/broker/stream_names_agreement_test.go new file mode 100644 index 00000000..29c4cf6c --- /dev/null +++ b/internal/broker/stream_names_agreement_test.go @@ -0,0 +1,23 @@ +package broker + +import ( + "slices" + "testing" + + "github.com/novox/mesh-controller/internal/catalogue" +) + +// The catalogue's list of the mesh's own stream names, which a module's stream may not take, is the broker's +// own list (novox/hq ADR 0363 §1): read from MeshStreams, so a stream added there cannot be forgotten there. +func TestTheCataloguesReservedStreamNamesAreTheMeshsOwn(t *testing.T) { + var own []string + for _, s := range MeshStreams() { + own = append(own, s.Name) + } + reserved := append([]string(nil), catalogue.MeshStreamNames...) + slices.Sort(own) + slices.Sort(reserved) + if !slices.Equal(own, reserved) { + t.Fatalf("the broker's own streams are %v, the catalogue reserves %v", own, reserved) + } +} diff --git a/internal/catalogue/manifest.go b/internal/catalogue/manifest.go index a816c6c5..fa9d65e6 100644 --- a/internal/catalogue/manifest.go +++ b/internal/catalogue/manifest.go @@ -1654,7 +1654,8 @@ func ParseManifest(raw []byte) (Manifest, error) { name string n int }{{"consumes", len(m.Consumes)}, {"uses", len(m.Uses)}, {"invokes", len(m.Invokes)}, - {"state", len(m.State)}, {"logs", len(m.Logs)}, {"reads", len(m.Reads)}} { + {"state", len(m.State)}, {"logs", len(m.Logs)}, {"reads", len(m.Reads)}, + {"streams", len(m.Streams)}} { if f.n > 0 { said = append(said, f.name) } diff --git a/internal/catalogue/streams.go b/internal/catalogue/streams.go index d1e91afb..0fd19126 100644 --- a/internal/catalogue/streams.go +++ b/internal/catalogue/streams.go @@ -27,7 +27,7 @@ const ( StreamDefaultMiB = 1024 ) -// GrantKey is the name in a grant's pattern that each key of the setting's value fills. +// GrantKey is the name in a grant's pattern that each key of the config value fills. const GrantKey = "{key}" // ReservedStreamPrefixes are the starts of the stream names the mesh derives, which no module's stream @@ -50,13 +50,13 @@ type StreamDeclaration struct { Grant *StreamGrant `json:"grant,omitempty"` } -// StreamGrant says which trusted setting of the module grants a consumer a part of the stream, and +// StreamGrant says which trusted config value 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 is a pattern holding GrantKey exactly once, which each key of the config value fills. Events string `json:"events"` } @@ -384,7 +384,7 @@ type HeardPart struct { // 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 +// restricted stream, to the parts the provider's trusted config value grants it. values answers a restricting config // 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. diff --git a/internal/catalogue/streams_test.go b/internal/catalogue/streams_test.go index f0d0aee4..5ea8c4e7 100644 --- a/internal/catalogue/streams_test.go +++ b/internal/catalogue/streams_test.go @@ -204,12 +204,3 @@ func TestNarrowingTwoPatterns(t *testing.T) { 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 d5519763..8ce9c330 100644 --- a/internal/inventory/busrecords.go +++ b/internal/inventory/busrecords.go @@ -69,7 +69,7 @@ func (i *Inventory) BusRecords(ctx context.Context) (broker.Records, error) { } assigned[n.Name] = modules } - // What each restricting setting says on each machine running its module (novox/hq ADR 0363 §5): read + // What each restricting config value 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 { @@ -407,10 +407,10 @@ func (i *Inventory) DeclaredStreams(ctx context.Context) ([]broker.ModuleStream, return out, nil } -// grantValues is each restricting setting's value on each machine running its module, by `/`. +// grantValues is each restricting config 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 +// streamGrantValues reads, for every module owning a restricted stream, the config value 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) {