diff --git a/internal/broker/derived.go b/internal/broker/derived.go index b1aa7848..65c97048 100644 --- a/internal/broker/derived.go +++ b/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 diff --git a/internal/broker/jetstream.go b/internal/broker/jetstream.go index 71fcb474..836675d4 100644 --- a/internal/broker/jetstream.go +++ b/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. -// -// **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 { +// 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 +// by (novox/hq ADR 0436 §6, ADR 0376 §3) — is testable without a server. +func StreamConfigFor(s Stream) *nats.StreamConfig { 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: diff --git a/internal/broker/modstream_work_queue_test.go b/internal/broker/modstream_work_queue_test.go index 8c515465..115bb973 100644 --- a/internal/broker/modstream_work_queue_test.go +++ b/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) + } +} diff --git a/internal/link/deadletters_work_queue_test.go b/internal/link/deadletters_work_queue_test.go index c0c5e730..8aa5b74b 100644 --- a/internal/link/deadletters_work_queue_test.go +++ b/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).