Review round 1: a carried module's queue consumer survives PruneHeard, a work-queue stream's restricted metadata is proven on the server config (issue 1178)

ConsumersOf's carried-module path (a module the tool runner carries, no
account of its own — mesh-mail's real shape) was not passing Queues
through, so its work-queue consumer was missing from the composed list
and PruneHeard would have removed what issueProcess made for it. Fixed,
with a test through Users -> ConsumersOf.

Extracted EnsureStream's config building into StreamConfigFor, a pure
function, so the dead-letters test can assert the restricted metadata
EnsureStream would actually send for a work-queue stream instead of
setting the dead letter's header by hand, which passed whether or not
modstreams.go's out.Restricted = true was there at all.

internal/broker/derived.go

@@ -229,9 +229,13 @@ func ConsumersOf(users []Principal) []ModuleConsumer {
 			continue
 		}
 		for _, d := range p.Carries {
+			// **A carried module's work-queue consumer is still the controller's to make** (novox/hq ADR
+			// 0436 §1, §2, §6), the same reason its ordinary consumer is (ADR 0198): mesh-mail has no
+			// account of its own (NoAccount) and is carried here, but its process does, and the consumer
+			// it reads through must exist and survive PruneHeard whether or not the module itself speaks.
 			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,
-				Streams: d.Streams, Hears: ComposedHears(d)})
+				Streams: d.Streams, Hears: ComposedHears(d), Queues: d.Queues})
 		}
 	}
 	return out

internal/broker/jetstream.go

@@ -154,16 +154,10 @@ func Borrow(j *JetStream) *JetStream {
 	return &lent
 }
 
-// EnsureStream creates the stream if it is absent and brings it to match if it is present.
+// StreamConfigFor is the server's own config for one of the mesh's streams, pure so the shape it builds —
-//
+// not least the restricted metadata a work-queue or a grant-restricted stream's dead letters age and hide
-// **Idempotent, because the controller asserts on every start** rather than creating once at
+// by (novox/hq ADR 0436 §6, ADR 0376 §3) — is testable without a server.
-// genesis: a stream somebody deleted, or a mesh raised from a restored backup, has to converge
+func StreamConfigFor(s Stream) *nats.StreamConfig {
-// rather than run without the guarantee its messages assume.
-//
-// An update, not a delete and recreate. Recreating would discard every message the stream holds
-// and every consumer's position in it — which for CONTROL means the pushes being held through a
-// store restart, exactly the guarantee the stream exists for.
-func (j *JetStream) EnsureStream(s Stream) error {
 	want := &nats.StreamConfig{
 		Name:              s.Name,
 		Subjects:          s.Subjects,
@@ -193,6 +187,20 @@ func (j *JetStream) EnsureStream(s Stream) error {
 		want.MaxMsgsPerSubject = 1
 		want.MaxAge = 0
 	}
+	return want
+}
+
+// EnsureStream creates the stream if it is absent and brings it to match if it is present.
+//
+// **Idempotent, because the controller asserts on every start** rather than creating once at
+// genesis: a stream somebody deleted, or a mesh raised from a restored backup, has to converge
+// rather than run without the guarantee its messages assume.
+//
+// An update, not a delete and recreate. Recreating would discard every message the stream holds
+// and every consumer's position in it — which for CONTROL means the pushes being held through a
+// store restart, exactly the guarantee the stream exists for.
+func (j *JetStream) EnsureStream(s Stream) error {
+	want := StreamConfigFor(s)
 
 	switch _, err := j.js.StreamInfo(s.Name); {
 	case err == nil:

internal/broker/modstream_work_queue_test.go

@@ -54,3 +54,28 @@ func TestTheOneProcessAWorkQueuesBusNamesGetsItsOwnConsumer(t *testing.T) {
 		t.Fatalf("the queue's consumer: %+v", c)
 	}
 }
+
+// review round 1, finding 2 (1178): mesh-mail's real shape has no account of its own (NoAccount) and is
+// carried by the machine's tool runner — so its queue consumer has to come from the carried-module path in
+// ConsumersOf, not only from a module with its own principal, or PruneHeard would remove what issueProcess
+// made for it (novox/hq ADR 0436 §1, §2, §6).
+func TestAWorkQueueConsumerSurvivesForAModuleTheRuntimeCarries(t *testing.T) {
+	users, err := Users(Records{Nodes: []string{"novox"}, Assigned: map[string][]Declared{"novox": {
+		{Module: RuntimeModule},
+		{Module: "mesh-mail", NoAccount: true,
+			Queues: []QueueOwner{{Process: "send", Stream: "MAIL_SENDS", MaxDeliver: 5}}},
+	}}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	got := ConsumersOf(users)
+	var found bool
+	for _, c := range got {
+		if c.Module == "mesh-mail" && c.Consumer.Name == "novox_mesh-mail_send" && c.Consumer.Stream == "MAIL_SENDS" {
+			found = true
+		}
+	}
+	if !found {
+		t.Fatalf("mesh-mail's queue consumer is missing from the carried module's composition: %+v", got)
+	}
+}

internal/link/deadletters_work_queue_test.go

@@ -6,12 +6,31 @@ import (
 	"time"
 
 	"github.com/nats-io/nats.go"
+
+	"github.com/novox/mesh-controller/internal/broker"
 )
 
 // A work-queue stream's dead letters are kept and shown as a restricted stream's are — aged with the
 // stream, never with their body or headers — and can only be dropped, never delivered again (novox/hq ADR
 // 0436 §6, changing ADR 0363 §7 for such a stream).
 
+// review round 1, finding 6 (1178/AC3): a work-queue stream really is restricted on the bus — its own
+// Stream() says so, and that is what EnsureStream's config sets the metadata from (jetstream.go's
+// streamConfigFor) — not merely a header this test set by hand, which would still pass with
+// modstreams.go's `out.Restricted = true` removed.
+func TestAWorkQueueStreamIsRestrictedOnTheBus(t *testing.T) {
+	mailSends := broker.ModuleStream{Module: "mesh-mail", Name: "MAIL_SENDS", Events: []string{"mail.send.approved"},
+		MaxAgeDays: 1, MaxMiB: 8, WorkQueue: true, Duplicates: 1}
+	stream := mailSends.Stream()
+	if !stream.Restricted {
+		t.Fatalf("a work-queue stream's own Stream() is not restricted: %+v", stream)
+	}
+	cfg := broker.StreamConfigFor(stream)
+	if cfg.Metadata[broker.RestrictedMetadata] != "true" {
+		t.Fatalf("the server config EnsureStream would send does not carry the restricted metadata: %+v", cfg.Metadata)
+	}
+}
+
 // 1178/AC3: a dead letter's headers say it came from a work-queue stream, and reading it back carries that
 // (and the restricted flag its aging and hiding already key off, deadLetterOf being the one place both are
 // read: see DeadLetter.Restricted's doc).