Merge pull request 'The grants part keeps the consumers a module has on another module's stream (issue 1921)' (#158) from fix/1921-heard-consumers into main

modules/mesh-grants/cmd/mesh-grants/main.go

@@ -132,7 +132,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		}, jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) }))
 		return err
 	}
-	logger.Printf("started on the account %s: publishing memberships, writing grants and the user list sealed to the bus node, which it follows, and creating the account consumers and the declared buckets, logs and streams the bus lacks; a change to a held stream is only said", account.User)
+	logger.Printf("started on the account %s: publishing memberships, writing grants and the user list sealed to the bus node, which it follows, and creating the account consumers (the heard-stream ones too) and the declared buckets, logs and streams the bus lacks; a held stream and a held consumer only ever gain the subjects and filters a manifest adds", account.User)
 	f.Run(ctx)
 	return nil
 }

modules/mesh-grants/grants_test.go

@@ -816,11 +816,24 @@ func TestGrants(t *testing.T) {
 		if heard := p.Compose(); heard.Withheld == nil || !slices.Equal(heard.Withheld.Deleted, []string{key}) || b.own[grants.StateWithheld][key] != nil {
 			t.Errorf("once reader runs as its own account the pass did %+v (errors %v), want %s deleted", heard.Withheld, heard.Errors, key)
 		}
-		for _, add := range b.filterAdds { // and once granted, still only what the records compose for the account's consumer
-			if strings.Contains(add, "mesh-mail") || !strings.Contains(add, "+mesh.") {
-				t.Errorf("a filter the records do not compose was added: %v", b.filterAdds)
+		// Once reader runs as an account of its own, its held one-filter consumer on the restricted stream gains exactly the
+		// subject its grant allows (mail.box...), and the filter it held stays.
+		var mailAdds []string
+		for _, add := range b.filterAdds {
+			if strings.HasPrefix(add, "MAIL.a_reader +") {
+				mailAdds = append(mailAdds, add)
 			}
 		}
+		if len(mailAdds) != 1 || !strings.Contains(mailAdds[0], "mesh-mail.event.mail.box.message.added") || strings.Contains(mailAdds[0], "mail.*") {
+			t.Errorf("the granted hearing's filter was not added exactly once: %v", b.filterAdds)
+		}
+		// And a heard-stream consumer the bus lacks is created.
+		b.mu.Lock()
+		b.consumers["MAIL"] = nil
+		b.mu.Unlock()
+		if heard := p.Compose(); heard.Consumers == nil || !slices.Contains(heard.Consumers.Created, "MAIL.a_reader") {
+			t.Errorf("a missing heard-stream consumer was not created: %+v", heard.Consumers)
+		}
 	})
 
 	// ADR 0601 clause 3: a held module's seats, hearings and reads are composed from the manifest its running build was judged

modules/mesh-grants/internal/buscompose/derived.go

@@ -248,11 +248,19 @@ func ConsumersOf(users []Principal) []ModuleConsumer {
 }
 
 // AccountConsumers are the durable consumers an account needs (novox/hq ADR 0561 §6): of ConsumersOf, each module's on
-// EVENTS and each process's on its own work queue, never one on a module stream it hears (those the controller makes with
-// the stream's grant). The grants part creates them from the lease write, with these configurations.
+// EVENTS, each process's on its own work queue, and each module's on another module's stream it hears (the controller's
+// last duty, the grants part's since the controller ended, ADR 0547, ADR 0640): a restricted stream's only for a module
+// that runs as an account of its own, with the subjects its grant allows (ComposedHears). The grants part creates them,
+// and adds the filters a held one lacks, with these configurations.
 func AccountConsumers(users []Principal) []ModuleConsumer {
 	queues := map[string]bool{}
+	heard := map[string]bool{} // node/module/stream a module hears of another module's stream
 	for _, p := range users {
+		for _, h := range p.Hears {
+			if len(h.Filters) > 0 {
+				heard[p.Node+"/"+p.Module+"/"+h.Stream] = true
+			}
+		}
 		for _, q := range p.Queues {
 			queues[q.Stream] = true
 		}
@@ -264,7 +272,8 @@ func AccountConsumers(users []Principal) []ModuleConsumer {
 	}
 	var out []ModuleConsumer
 	for _, mc := range ConsumersOf(users) {
-		if mc.Consumer.Stream == consumerStream(Principal{Kind: KindModule}) || queues[mc.Consumer.Stream] {
+		if mc.Consumer.Stream == consumerStream(Principal{Kind: KindModule}) || queues[mc.Consumer.Stream] ||
+			heard[mc.Node+"/"+mc.Module+"/"+mc.Consumer.Stream] {
 			out = append(out, mc)
 		}
 	}

modules/mesh-grants/internal/grants/part.go

@@ -139,7 +139,7 @@ type Result struct {
 }
 
 // NothingTakesEffect is what every composition says it did on the bus beyond its own state and the memberships.
-const NothingTakesEffect = "memberships (in its own state memberships, unless one would leave a seat a placed module claims with no holder, or drop a capability its holder still asks for, which stays a shadow difference), each bucket, log or stream a manifest declares that the bus lacks, created once with the manifest's limits (the operator, 2026-10-26 08:54), the user list the bus node follows, the grants and the account consumers; a change to a stream the bus holds is only said"
+const NothingTakesEffect = "memberships (in its own state memberships, unless one would leave a seat a placed module claims with no holder, or drop a capability its holder still asks for, which stays a shadow difference), each bucket, log or stream a manifest declares that the bus lacks, created once with the manifest's limits (the operator, 2026-10-26 08:54), the user list the bus node follows, the grants and the account consumers (a module's on another module's stream too); a stream or consumer the bus holds only ever gains the subjects and filters a manifest adds"
 
 // Part is the grants part: it reads, calls and writes through these, and nothing else.
 type Part struct {

modules/mesh-grants/internal/grants/tools.go

@@ -285,7 +285,7 @@ func (p *Part) AssertOne(module string) (any, error) {
 		}
 	}
 	if missing == 0 {
-		out["says"] = "the bus holds every bucket, log and stream the manifest declares: nothing to create; a difference is only said"
+		out["says"] = "the bus holds every bucket, log and stream the manifest declares: nothing to create; a held stream only ever gains the subjects a manifest adds, any other difference is only said"
 		return out, nil
 	}
 	ask, _ := json.Marshal(Ask{By: "grants_assert", At: p.now(), Assert: module})