Remove a restricted stream's consumer when the bus refuses to narrow it, so its wider filters deliver nothing more (ADR 0363 §6, review of PR 338)

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))
 		}
 	}

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)
+}

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)

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)
+	}
+}

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)
+	}
+}

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)
 			}

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.

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)
-		}
-	}
-}

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 `<module>/<setting>`.
+// grantValues is each restricting config value on each machine running its module, by `<module>/<key>`.
 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) {