diff --git a/internal/broker/derived.go b/internal/broker/derived.go index c724b439..55503638 100644 --- a/internal/broker/derived.go +++ b/internal/broker/derived.go @@ -190,18 +190,33 @@ func ConsumersOf(users []Principal) []ModuleConsumer { var out []ModuleConsumer seen := map[string]bool{} add := func(p Principal) { - c, needed := ConsumerFor(p) - if !needed || seen[p.Node+"/"+p.Module] { + key := p.Node + "/" + p.Module + if seen[key] { return } - seen[p.Node+"/"+p.Module] = true - out = append(out, ModuleConsumer{Node: p.Node, Module: p.Module, Consumer: c}) - // And one on each module stream it hears (novox/hq ADR 0363 §5). - for _, h := range p.Hears { - if len(h.Filters) == 0 { - continue + c, needed := ConsumerFor(p) + // **A module with a work-queue consumer of its own needs no other reason to be visited here** + // (novox/hq ADR 0436 §1, §2, §6): mesh-mail's sender consumes and watches nothing of its own, so + // ConsumerFor alone would have found nothing needed and skipped it before its one queue consumer + // was ever composed. + if !needed && len(p.Queues) == 0 { + return + } + seen[key] = true + if needed { + out = append(out, ModuleConsumer{Node: p.Node, Module: p.Module, Consumer: c}) + // And one on each module stream it hears (novox/hq ADR 0363 §5). + for _, h := range p.Hears { + if len(h.Filters) == 0 { + continue + } + out = append(out, ModuleConsumer{Node: p.Node, Module: p.Module, Consumer: heardConsumer(p, h)}) } - out = append(out, ModuleConsumer{Node: p.Node, Module: p.Module, Consumer: heardConsumer(p, h)}) + } + // And the one durable consumer of each of its own work-queue streams, for the one process its + // bus.queues names (novox/hq ADR 0436 §1, §2, §6). + for _, q := range p.Queues { + out = append(out, ModuleConsumer{Node: p.Node, Module: p.Module, Consumer: queueConsumer(p, q)}) } } for _, p := range users { diff --git a/internal/broker/modstream_work_queue_test.go b/internal/broker/modstream_work_queue_test.go index ea7107b0..8c515465 100644 --- a/internal/broker/modstream_work_queue_test.go +++ b/internal/broker/modstream_work_queue_test.go @@ -36,3 +36,21 @@ func TestANonWorkQueueStreamStillDropsItsOldest(t *testing.T) { t.Fatalf("composed as %+v; want limits retention, never discarding new, no duplicate window", s) } } + +// 1178/AC1: the process a work-queue stream's bus.queues names gets exactly one durable consumer, +// __, with the stream's declared max-deliver — composed even though mesh-mail's +// sender otherwise consumes and watches nothing of its own, which alone would give ConsumerFor nothing +// needed. Explicit acknowledgement is EnsureConsumer's own rule for every consumer it makes (jetstream.go), +// so it is not re-asserted here. +func TestTheOneProcessAWorkQueuesBusNamesGetsItsOwnConsumer(t *testing.T) { + sender := Principal{Kind: KindModule, Node: "novox", Module: "mesh-mail", PasswordHash: "x", + Queues: []QueueOwner{{Process: "send", Stream: "MAIL_SENDS", MaxDeliver: 5}}} + got := ConsumersOf([]Principal{sender}) + if len(got) != 1 { + t.Fatalf("mesh-mail's consumers: %+v", got) + } + c := got[0].Consumer + if c.Name != "novox_mesh-mail_send" || c.Stream != "MAIL_SENDS" || c.MaxDeliver != 5 { + t.Fatalf("the queue's consumer: %+v", c) + } +} diff --git a/internal/broker/modstreams.go b/internal/broker/modstreams.go index 33e78271..cdc01984 100644 --- a/internal/broker/modstreams.go +++ b/internal/broker/modstreams.go @@ -94,10 +94,44 @@ func (s ModuleStream) Stream() Stream { out.Retention = RetentionWorkQueue out.DiscardNew = true out.DuplicatesSeconds = s.Duplicates * 60 * 60 + // Said on the bus too, reusing the one flag a restricted stream's dead letters already age and hide + // by (novox/hq ADR 0436 §6, ADR 0376 §3-4): a work-queue stream's grant is always empty (ADR 0436 + // §6 refuses one that carries a grant), so this never collides with a trusted-config restriction. + out.Restricted = true } return out } +// QueueOwner is the one process of a module that is the durable consumer of one of its own work-queue +// streams (novox/hq ADR 0436 §1, §2, §6): named in that process's bus.queues, and checked at registration +// to be the only one (catalogue.WorkQueueStreamProblems). +type QueueOwner struct { + // Process is the process resource's id — its bus account's name, with the node and module (novox/hq ADR + // 0436 §2). + Process string + // Stream is the work-queue stream's name on the bus. + Stream string + // MaxDeliver bounds how many times a message is handed over before it is given up as a dead letter + // (novox/hq ADR 0436 §6). + MaxDeliver int +} + +// queueConsumer is the durable consumer the one process a module's bus.queues names is granted on its own +// work-queue stream (novox/hq ADR 0436 §1, §2, §6): no filter, since it is the stream's only consumer — +// unlike heardConsumer, which is one of possibly several hearing a module's ordinary stream. +func queueConsumer(p Principal, q QueueOwner) Consumer { + return Consumer{ + Name: p.Node + "_" + p.Module + "_" + q.Process, + Stream: q.Stream, + AckWaitSeconds: 30, + MaxDeliver: q.MaxDeliver, + Why: fmt.Sprintf("%s's process %s is the one consumer %s's bus.queues names for its work-queue "+ + "stream %s (novox/hq ADR 0436 §6); after max-deliver %d it gives a message up, and the "+ + "controller keeps it in %s, offered only drop", p.Module, q.Process, p.Module, q.Stream, + q.MaxDeliver, DeadLettersStream), + } +} + // ModuleStreams is every declared module stream as the controller asserts it, in name order. func ModuleStreams(declared []ModuleStream) []Stream { sorted := append([]ModuleStream(nil), declared...) diff --git a/internal/broker/modstreams_workqueue_integration_test.go b/internal/broker/modstreams_workqueue_integration_test.go new file mode 100644 index 00000000..fbab47a9 --- /dev/null +++ b/internal/broker/modstreams_workqueue_integration_test.go @@ -0,0 +1,101 @@ +//go:build integration + +package broker + +import ( + "testing" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "github.com/novox/mesh-controller/internal/testbus" +) + +// 1178/AC4, on a real bus: the controller ensuring mesh-mail's work-queue stream and its one consumer gives +// the bus's own guarantees — a duplicate publish within the window is taken once, a message naked past +// max-deliver is not handed over again, and a full stream refuses a new publish (ADR 0436 §6). +// +// go test -tags integration ./internal/broker/ -run TestAWorkQueueStreamOnARealBus +func TestAWorkQueueStreamOnARealBus(t *testing.T) { + js, err := Dial(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + defer js.Close() + + // Small enough that a handful of messages fill it, so "the stream is full" is reachable within a test. + stream := ModuleStream{Module: "mesh-mail", Name: "MAIL_SENDS", Events: []string{"mail.send.approved"}, + MaxAgeDays: 1, MaxMiB: 1, WorkQueue: true, Duplicates: 1} + if err := js.EnsureStream(stream.Stream()); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = js.Context().DeleteStream(stream.Name) }) + + owner := QueueOwner{Process: "send", Stream: stream.Name, MaxDeliver: 5} + sender := Principal{Kind: KindModule, Node: "novox", Module: "mesh-mail", PasswordHash: "x"} + consumer := queueConsumer(sender, owner) + consumer.AckWaitSeconds = 1 + if err := js.EnsureConsumer(consumer); err != nil { + t.Fatal(err) + } + + subject := ModuleStreamSubject("mesh-mail", "mail.send.approved") + + // A duplicate publish within the window is kept once (ADR 0436 §6's duplicate window). + publish := func(id string, size int) (*nats.PubAck, error) { + return js.Context().Publish(subject, make([]byte, size), nats.MsgId(id)) + } + if _, err := publish("dup-1", 16); err != nil { + t.Fatal(err) + } + if _, err := publish("dup-1", 16); err != nil { + t.Fatal(err) + } + if info, err := js.Context().StreamInfo(stream.Name); err != nil || info.State.Msgs != 1 { + t.Fatalf("a message published twice with the same id: %v msgs, err %v; want 1", info, err) + } + + // Naked five times, the consumer is not handed the message a sixth time. + api, err := jetstream.New(js.Conn()) + if err != nil { + t.Fatal(err) + } + reading, err := api.Consumer(t.Context(), stream.Name, consumer.Name) + if err != nil { + t.Fatal(err) + } + for i := 0; i < owner.MaxDeliver; i++ { + batch, err := reading.Fetch(1, jetstream.FetchMaxWait(2*time.Second)) + if err != nil { + t.Fatal(err) + } + got := false + for msg := range batch.Messages() { + got = true + _ = msg.Nak() + } + if !got { + t.Fatalf("delivery %d of %d never arrived", i+1, owner.MaxDeliver) + } + } + batch, err := reading.Fetch(1, jetstream.FetchMaxWait(2*time.Second)) + if err != nil { + t.Fatal(err) + } + for range batch.Messages() { + t.Fatalf("the message was delivered a %dth time, past its max-deliver of %d", owner.MaxDeliver+1, owner.MaxDeliver) + } + + // Full, the stream refuses a new publish rather than dropping what it holds (DiscardNew). + var fullErr error + for i := 0; i < 1<<16; i++ { + if _, err := publish("fill-"+time.Now().String(), 1<<16); err != nil { + fullErr = err + break + } + } + if fullErr == nil { + t.Fatal("a work-queue stream accepted every publish; want it to refuse one once full") + } +} diff --git a/internal/broker/nats.go b/internal/broker/nats.go index ecec2bba..cef0fef8 100644 --- a/internal/broker/nats.go +++ b/internal/broker/nats.go @@ -170,6 +170,9 @@ type Principal struct { // subjects it was granted (novox/hq ADR 0363). Streams []ModuleStream Hears []Heard + // Queues are its own work-queue streams and the one process each is the consumer of (novox/hq ADR 0436 + // §1, §2, §6). + Queues []QueueOwner // SnapshotsTheBus is the bus's own module, the one holding mesh-broker (novox/hq ADR 0235). Its // whole authority is BusSnapshotGrants: it copies the streams for the night's backup and nothing diff --git a/internal/broker/users.go b/internal/broker/users.go index 9ad43933..ac325a4f 100644 --- a/internal/broker/users.go +++ b/internal/broker/users.go @@ -46,6 +46,9 @@ type Declared struct { Logs []Log // Streams are the streams it owns, each its events' one place (novox/hq ADR 0363). Streams []ModuleStream + // Queues are its own work-queue streams and the one process each is the consumer of (novox/hq ADR 0436 + // §1, §2, §6): one durable consumer per entry, named `__`. + Queues []QueueOwner // Hears are the module streams it hears on this machine, each with the subjects it was granted // (novox/hq ADR 0363 §5); HearsAStream says its `consumes` names a module that owns a stream, granted // or not, so its membership is issued again when a grant changes. @@ -155,7 +158,7 @@ func Users(r Records) ([]Principal, error) { Holds: d.Holds, Uses: d.Uses, Watches: d.Watches, Invokes: d.Invokes, State: stateNames(d.State), Reads: d.Reads, SnapshotsTheBus: d.SnapshotsTheBus, PerMachine: perMachineNames(d.State), KeyedReads: d.KeyedReads, - Logs: logNames(d.Logs), Streams: d.Streams, Hears: ComposedHears(d), + Logs: logNames(d.Logs), Streams: d.Streams, Hears: ComposedHears(d), Queues: d.Queues, }) // **The vault's holder asks the desks and the issuing providers itself** (issue 723), under the // account its own runtime runs on (ADR 0259 §8's runs-as): the one module granted it. diff --git a/internal/catalogue/manifest.go b/internal/catalogue/manifest.go index bcbb4c69..85bbb067 100644 --- a/internal/catalogue/manifest.go +++ b/internal/catalogue/manifest.go @@ -1953,6 +1953,7 @@ func parseManifest(raw []byte, predefined PredefinedList) (Manifest, error) { problems = append(problems, LogProblems(m)...) // And the streams it owns, and who may hear them (streams.go, novox/hq ADR 0363). problems = append(problems, StreamProblems(m)...) + problems = append(problems, WorkQueueStreamProblems(m)...) // And the defaults it gives its settings (setting_defaults.go, novox/hq ADR 0262), or, written in the // config form, the whole declaration of each value (config.go, novox/hq ADR 0348). if m.DeclaresConfig() { diff --git a/internal/catalogue/process_bus_test.go b/internal/catalogue/process_bus_test.go index 22ac13dc..5551ec17 100644 --- a/internal/catalogue/process_bus_test.go +++ b/internal/catalogue/process_bus_test.go @@ -18,10 +18,13 @@ func mailSenderProcessManifest(t *testing.T, bus string) string { process += `,"bus":` + bus } process += `}` + // The stream itself is plain here, not retention work-queue: 1179 checks only that a process names a + // stream the module owns, never its retention — that rule is 1178's (WorkQueueStreamProblems), and is + // tested against its own fixture there (work_queue_consumer_test.go) so this one stays accepted whether + // or not a process currently names "mail-sends" in bus.queues. return `{"module":"mesh-mail","version":"1","emits":["mail.send.sent","mail.send.failed"],` + `"state":["sends"],"streams":[{"name":"mail-sends","events":["mail.send.approved"],"max-age-days":1,` + - `"max-mib":64,"retention":"work-queue","max-deliveries":5,"duplicates":24}],` + - `"resources":[{"id":"account","type":"user","name":"mesh-mail-send"},` + process + `]}` + `"max-mib":64}],"resources":[{"id":"account","type":"user","name":"mesh-mail-send"},` + process + `]}` } const mailSenderBus = `{"emits":["mail.send.sent","mail.send.failed"],"queues":["mail-sends"],` + diff --git a/internal/catalogue/seats_declared.go b/internal/catalogue/seats_declared.go index d53d3e04..2a9ad3fe 100644 --- a/internal/catalogue/seats_declared.go +++ b/internal/catalogue/seats_declared.go @@ -280,6 +280,8 @@ func CatalogueProblems(shelf Shelf) []string { problems = append(problems, StreamNameProblems(shelf)...) // A module that hears a restricted stream runs as an account of its own (novox/hq ADR 0376 §2). problems = append(problems, RestrictedHearingProblems(shelf)...) + // A work-queue stream is never asked for by another module's consumes (novox/hq ADR 0436 §6). + problems = append(problems, WorkQueueConsumedElsewhereProblems(shelf)...) // Who declares what, and who declared it first. declaredBy := map[string]string{} diff --git a/internal/catalogue/stream_work_queue_test.go b/internal/catalogue/stream_work_queue_test.go index f37d6bfe..9f8ccf64 100644 --- a/internal/catalogue/stream_work_queue_test.go +++ b/internal/catalogue/stream_work_queue_test.go @@ -13,9 +13,14 @@ import ( const mailSendsStream = `[{"name":"mail-sends","events":["mail.send.approved"],"max-age-days":1,"max-mib":64,` + `"retention":"work-queue","max-deliveries":5,"duplicates":24}]` -// 1177/AC1: a well-formed work-queue stream is accepted and read back whole. +// 1177/AC1: a well-formed work-queue stream is accepted and read back whole. Named by one process in +// bus.queues (novox/hq ADR 0436 §6, checked by 1178's WorkQueueStreamProblems), so this stays accepted once +// that rule exists. func TestAWorkQueueStreamIsDeclaredAndAccepted(t *testing.T) { - m, err := ParseManifest([]byte(mailManifest(t, mailSendsStream, ""))) + m, err := ParseManifest([]byte(`{"module":"mesh-mail","version":"1","streams":` + mailSendsStream + + `,"resources":[{"id":"account","type":"user","name":"mesh-mail-send"},` + + `{"id":"send","type":"process","run":["./mesh-mail","send"],"user":"mesh-mail-send",` + + `"bus":{"queues":["mail-sends"]}}]}`)) if err != nil { t.Fatalf("a well-formed work-queue stream was refused: %v", err) } diff --git a/internal/catalogue/streams.go b/internal/catalogue/streams.go index 62c1eb1a..15c6c2b0 100644 --- a/internal/catalogue/streams.go +++ b/internal/catalogue/streams.go @@ -351,6 +351,64 @@ func patternsOverlap(a, b string) bool { } } +// WorkQueueStreamProblems refuses a work-queue stream (novox/hq ADR 0436 §6) that is not read by exactly +// one process of its own module: named by none, it queues forever unconsumed; named by two, each would take +// half of what the other needed, silently. +func WorkQueueStreamProblems(m Manifest) []string { + var problems []string + named := map[string]int{} + for _, b := range m.ProcessBuses() { + for _, q := range b.Queues { + named[q]++ + } + } + for _, s := range m.Streams { + if s.Retention != StreamRetentionWorkQueue { + continue + } + switch named[s.Name] { + case 0: + problems = append(problems, fmt.Sprintf( + "%s's work-queue stream %q: no process names it in bus.queues; a work-queue stream is read by "+ + "exactly one process (novox/hq ADR 0436 §6)", m.Module, s.Name)) + case 1: + default: + problems = append(problems, fmt.Sprintf( + "%s's work-queue stream %q: %d processes name it in bus.queues; a work-queue stream is read by "+ + "exactly one (novox/hq ADR 0436 §6)", m.Module, s.Name, named[s.Name])) + } + } + return problems +} + +// WorkQueueConsumedElsewhereProblems refuses another module's `consumes` naming a work-queue stream's events +// (novox/hq ADR 0436 §6): read by its module's one process alone, never broadcast to a consumer that only +// subscribes. Modelled on RestrictedHearingProblems. +func WorkQueueConsumedElsewhereProblems(shelf Shelf) []string { + var problems []string + for _, module := range shelfOrder(shelf) { + m := shelf[module] + for _, c := range m.Consumes { + emitter, rest, named := strings.Cut(c, ".") + provider, known := shelf[emitter] + if !named || emitter == OneName || emitter == TheRest || !known || emitter == module { + continue + } + hearing, _ := provider.StreamsHearing(rest) + for _, s := range hearing { + if s.Retention != StreamRetentionWorkQueue { + continue + } + problems = append(problems, fmt.Sprintf( + "%s consumes %s, which %s's work-queue stream %s carries; a work-queue stream is read by "+ + "its module's one process alone, never by another module's consumes (novox/hq ADR 0436 "+ + "§6)", module, c, emitter, s.BusName())) + } + } + } + return problems +} + // StreamNameProblems are the rules about stream names no single manifest can be judged against: two // modules declaring one name. func StreamNameProblems(shelf Shelf) []string { diff --git a/internal/catalogue/work_queue_consumer_test.go b/internal/catalogue/work_queue_consumer_test.go new file mode 100644 index 00000000..4f98b6f1 --- /dev/null +++ b/internal/catalogue/work_queue_consumer_test.go @@ -0,0 +1,75 @@ +package catalogue + +import ( + "strings" + "testing" +) + +// A work-queue stream is read by exactly one process of its own module (novox/hq ADR 0436 §6): named once +// in a process's bus.queues, and never asked for by another module's consumes. + +// mailSenderManifestWithQueue is the mesh-mail fixture (ADR 0436 §7, §8) with its process's bus.queues set +// as given, so each 1178/AC2 case can vary just that. +func mailSenderManifestWithQueue(queues string) Manifest { + return Manifest{ + Module: "mesh-mail", Version: "1", Emits: []string{"mail.send.sent", "mail.send.failed"}, + State: []StateDeclaration{{Name: "sends"}}, + Streams: []StreamDeclaration{{Name: "mail-sends", Events: []string{"mail.send.approved"}, + Retention: StreamRetentionWorkQueue, MaxDeliveries: 5}}, + Resources: []map[string]any{ + {"id": "account", "type": "user", "name": "mesh-mail-send"}, + {"id": "send", "type": "process", "run": []any{"./mesh-mail", "send"}, "user": "mesh-mail-send", + "bus": map[string]any{"emits": []any{"mail.send.sent", "mail.send.failed"}, "queues": queueList(queues)}}, + }, + } +} + +func queueList(q string) []any { + if q == "" { + return nil + } + out := []any{} + for _, s := range strings.Split(q, ",") { + out = append(out, s) + } + return out +} + +// 1178/AC2 (named): a work-queue stream no process names is refused. +func TestAWorkQueueStreamNoProcessNamesIsRefused(t *testing.T) { + m := mailSenderManifestWithQueue("") + got := strings.Join(WorkQueueStreamProblems(m), "; ") + if !strings.Contains(got, "mail-sends") || !strings.Contains(got, "no process names it") { + t.Fatalf("an unnamed work-queue stream: %q", got) + } +} + +// 1178/AC2 (named twice): a work-queue stream two processes name is refused. +func TestAWorkQueueStreamTwoProcessesNameIsRefused(t *testing.T) { + m := mailSenderManifestWithQueue("mail-sends") + m.Resources = append(m.Resources, map[string]any{"id": "resend", "type": "process", + "run": []any{"./mesh-mail", "send"}, "user": "mesh-mail-send", + "bus": map[string]any{"emits": []any{"mail.send.sent"}, "queues": []any{"mail-sends"}}}) + got := strings.Join(WorkQueueStreamProblems(m), "; ") + if !strings.Contains(got, "mail-sends") || !strings.Contains(got, "2 processes name") { + t.Fatalf("a work-queue stream two processes name: %q", got) + } +} + +// 1178/AC2 (named once): a work-queue stream exactly one process names is accepted. +func TestAWorkQueueStreamOneProcessNamesIsAccepted(t *testing.T) { + m := mailSenderManifestWithQueue("mail-sends") + if got := WorkQueueStreamProblems(m); len(got) != 0 { + t.Fatalf("a work-queue stream one process names was refused: %v", got) + } +} + +// 1178/AC2 (another module's consumes): a work-queue stream another module's consumes asks for is refused. +func TestAWorkQueueStreamAnotherModulesConsumesIsRefused(t *testing.T) { + sender := mailSenderManifestWithQueue("mail-sends") + reader := Manifest{Module: "reader", Version: "1", Consumes: []string{"mesh-mail.mail.send.approved"}} + got := strings.Join(WorkQueueConsumedElsewhereProblems(Shelf{"mesh-mail": sender, "reader": reader}), "; ") + if !strings.Contains(got, "reader") || !strings.Contains(got, "MAIL_SENDS") { + t.Fatalf("another module's consumes of a work-queue stream: %q", got) + } +} diff --git a/internal/inventory/busrecords.go b/internal/inventory/busrecords.go index 380fa149..e7a0c4f6 100644 --- a/internal/inventory/busrecords.go +++ b/internal/inventory/busrecords.go @@ -187,6 +187,9 @@ func declaredFor(m catalogue.Manifest, seats map[string]catalogue.SeatDeclaratio Logs: logsOf(m), // And the streams it owns (novox/hq ADR 0363). Streams: streamsOf(m), + // And its own work-queue streams, each with the one process its bus.queues names as the + // consumer (novox/hq ADR 0436 §1, §2, §6). + Queues: queueOwnersOf(m), // And the tools its health asks (novox/hq ADR 0240): the machine's node-engine is granted them. Checks: catalogue.HealthChecks(m), // And whether it runs as an account of its own (novox/hq ADR 0259 §8). @@ -428,6 +431,37 @@ func streamsOf(m catalogue.Manifest) []broker.ModuleStream { return out } +// queueOwnersOf is every work-queue stream of a module's own that one of its processes names in bus.queues, +// with that process's id and the stream's declared max-deliveries (novox/hq ADR 0436 §1, §2, §6). In +// resource order, so composition is repeatable; a queue naming a stream the module does not own, or one two +// processes name, is caught at registration (catalogue.WorkQueueStreamProblems) and never reaches here. +func queueOwnersOf(m catalogue.Manifest) []broker.QueueOwner { + buses := m.ProcessBuses() + if len(buses) == 0 { + return nil + } + byName := map[string]catalogue.StreamDeclaration{} + for _, s := range m.Streams { + byName[s.Name] = s + } + var out []broker.QueueOwner + for _, r := range m.Resources { + id := fmt.Sprint(r["id"]) + b, declared := buses[id] + if !declared { + continue + } + for _, q := range b.Queues { + s, known := byName[q] + if !known || s.Retention != catalogue.StreamRetentionWorkQueue { + continue + } + out = append(out, broker.QueueOwner{Process: id, Stream: s.BusName(), MaxDeliver: s.MaxDeliveries}) + } + } + return out +} + // DeclaredStreams is every module stream the catalogue declares, registered modules assigned or not: a // stream exists from registration, as a log does (novox/hq ADR 0363 §1). func (i *Inventory) DeclaredStreams(ctx context.Context) ([]broker.ModuleStream, error) { diff --git a/internal/inventory/streams_test.go b/internal/inventory/streams_test.go index b79df9bf..4a4cd0fb 100644 --- a/internal/inventory/streams_test.go +++ b/internal/inventory/streams_test.go @@ -65,3 +65,25 @@ func TestANonWorkQueueStreamIsCarriedAsNotAWorkQueue(t *testing.T) { t.Fatalf("the plain stream: %+v", got) } } + +// 1178/AC1: the process a work-queue stream's bus.queues names is read from the manifest, by its resource +// id, with the stream's own max-deliveries — what the broker composes the one durable consumer from. +func TestTheQueueOwnersOfAWorkQueueStreamAreReadFromItsProcess(t *testing.T) { + sender := catalogue.Manifest{Module: "mesh-mail", Emits: []string{"mail.send.sent", "mail.send.failed"}, + Streams: []catalogue.StreamDeclaration{{Name: "mail-sends", Events: []string{"mail.send.approved"}, + Retention: catalogue.StreamRetentionWorkQueue, MaxDeliveries: 5}}, + Resources: []map[string]any{ + {"id": "account", "type": "user", "name": "mesh-mail-send"}, + {"id": "send", "type": "process", "run": []any{"./mesh-mail", "send"}, "user": "mesh-mail-send", + "bus": map[string]any{"emits": []any{"mail.send.sent", "mail.send.failed"}, + "queues": []any{"mail-sends"}}}, + }} + got := queueOwnersOf(sender) + if len(got) != 1 || got[0].Process != "send" || got[0].Stream != "MAIL_SENDS" || got[0].MaxDeliver != 5 { + t.Fatalf("the sender's queue owners: %+v", got) + } + // And a module with no process declaring bus has none. + if got := queueOwnersOf(catalogue.Manifest{Module: "feed"}); got != nil { + t.Fatalf("a module with no process bus has queue owners: %+v", got) + } +} diff --git a/internal/link/deadletters.go b/internal/link/deadletters.go index 94b442cb..b806ad2d 100644 --- a/internal/link/deadletters.go +++ b/internal/link/deadletters.go @@ -45,6 +45,10 @@ const ( // DeadHeaderAgesOut when its stream's maximum age runs out, counted from when it was first published. DeadHeaderRestricted = "Mesh-Dead-Restricted" DeadHeaderAgesOut = "Mesh-Dead-Ages-Out" + // DeadHeaderWorkQueue says the message came from a work-queue stream (novox/hq ADR 0436 §6): aged and + // shown as a restricted stream's are (it sets DeadHeaderRestricted too), but never delivered again — + // only dropped. + DeadHeaderWorkQueue = "Mesh-Dead-Work-Queue" // AgainHeader marks a message delivered again, with the kept copy's id it came from. AgainHeader = "Mesh-Delivered-Again" ) @@ -74,6 +78,10 @@ type DeadLetter struct { Restricted bool `json:"restricted,omitempty"` AgesOut time.Time `json:"ages_out,omitzero"` Withheld string `json:"withheld,omitempty"` + // WorkQueue says it came from a work-queue stream (novox/hq ADR 0436 §6): shown and aged as a restricted + // stream's are (Restricted is also set), but AgainTo refuses it unconditionally — only dropped, never + // delivered again. + WorkQueue bool `json:"work-queue,omitempty"` // Body and Headers are the message itself, given only for one dead letter asked by its id, and never for // a restricted one; a header with several values keeps them all. Body string `json:"body,omitempty"` @@ -91,17 +99,19 @@ const leastLeftToDeliver = time.Minute const withheldWords = "its body and its own headers are not shown: it came from a restricted stream, which only " + "the modules its grant names may read (novox/hq ADR 0376)" -// sourceAge is whether a stream is restricted (novox/hq ADR 0363 §4), and its maximum age: read from the stream -// itself, where the controller says both. A stream that no longer exists is neither. -func sourceAge(js nats.JetStreamContext, stream string) (bool, time.Duration, error) { +// sourceAge is whether a stream is restricted (novox/hq ADR 0363 §4; a work-queue stream is restricted too, +// novox/hq ADR 0436 §6, aged and hidden the same way), whether it is a work queue, and its maximum age: read +// from the stream itself, where the controller says all three. A stream that no longer exists is neither. +func sourceAge(js nats.JetStreamContext, stream string) (restricted, workQueue bool, maxAge time.Duration, err error) { info, err := js.StreamInfo(stream) if errors.Is(err, nats.ErrStreamNotFound) { - return false, 0, nil + return false, false, 0, nil } if err != nil { - return false, 0, fmt.Errorf("whether %s is restricted cannot be read: %w", stream, err) + return false, false, 0, fmt.Errorf("whether %s is restricted cannot be read: %w", stream, err) } - return info.Config.Metadata[broker.RestrictedMetadata] == "true", info.Config.MaxAge, nil + workQueue = info.Config.Retention == nats.WorkQueuePolicy + return info.Config.Metadata[broker.RestrictedMetadata] == "true", workQueue, info.Config.MaxAge, nil } // ttlOf is a time left as a message's time to live, in whole seconds and never more than what is left. @@ -135,13 +145,16 @@ func KeepDeadLetter(js nats.JetStreamContext, notice []byte) (DeadLetter, error) // **A restricted stream's message ages out with its stream** (novox/hq ADR 0376 §3): kept with a time to live // of what is left of the stream's maximum age, counted from when it was first published; one already past it // is kept as a record without its body. Every other dead letter keeps no age, as before (ADR 0264). - restricted, maxAge, err := sourceAge(js, a.Stream) + restricted, workQueue, maxAge, err := sourceAge(js, a.Stream) if err != nil { return DeadLetter{}, err } if restricted { kept.Header.Set(DeadHeaderRestricted, "true") } + if workQueue { + kept.Header.Set(DeadHeaderWorkQueue, "true") + } raw, err := js.GetMsg(a.Stream, a.StreamSeq) agesOut := time.Time{} if err == nil && restricted && maxAge > 0 { @@ -205,6 +218,7 @@ func deadLetterOf(id uint64, subject string, h nats.Header, data []byte, whole b d.GaveUp, _ = time.Parse(time.RFC3339Nano, h.Get(DeadHeaderGaveUp)) d.Published, _ = time.Parse(time.RFC3339Nano, h.Get(DeadHeaderPublished)) d.Restricted = h.Get(DeadHeaderRestricted) == "true" + d.WorkQueue = h.Get(DeadHeaderWorkQueue) == "true" d.AgesOut, _ = time.Parse(time.RFC3339Nano, h.Get(DeadHeaderAgesOut)) if whole { d.Body = string(data) @@ -349,6 +363,13 @@ func AgainTo(js nats.JetStreamContext, d DeadLetter) (string, error) { case d.Subject == "": return "", fmt.Errorf("dead letter %d does not say the subject it was published on, so it cannot be "+ "delivered again. Drop it", d.ID) + case d.WorkQueue: + // A work-queue stream's one consumer takes its messages in order (novox/hq ADR 0436 clause 6): + // delivering a given-up one back in would let it jump ahead of whatever the stream already holds, so + // unlike a module's ordinary stream (ADR 0363 §7) it is never delivered again, however much of its + // age is left. + return "", fmt.Errorf("dead letter %d is from %s, a work queue: it is only dropped, never delivered "+ + "again (novox/hq ADR 0436 clause 6). Drop it", d.ID, d.Stream) case broker.TheControllersAsk(d.Stream, d.Subject): // The seat's worker takes it, and only while it is on the bus: without one the queue would keep // the ask for a holder that may never come, and it would be let go as delivered meanwhile. @@ -593,7 +614,7 @@ func AgingDeadLetters(js nats.JetStreamContext, held map[string]int) (map[string stream, consumer, _ := strings.Cut(key, ".") is, known := restricted[stream] if !known { - r, _, err := sourceAge(js, stream) + r, _, _, err := sourceAge(js, stream) if err != nil { return nil, err } diff --git a/internal/link/deadletters_work_queue_test.go b/internal/link/deadletters_work_queue_test.go new file mode 100644 index 00000000..c0c5e730 --- /dev/null +++ b/internal/link/deadletters_work_queue_test.go @@ -0,0 +1,50 @@ +package link + +import ( + "strings" + "testing" + "time" + + "github.com/nats-io/nats.go" +) + +// 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). + +// 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). +func TestAWorkQueueDeadLettersHeaderIsReadBack(t *testing.T) { + agesOut := time.Now().Add(time.Hour).UTC() + h := nats.Header{} + h.Set(DeadHeaderStream, "MAIL_SENDS") + h.Set(DeadHeaderConsumer, "novox_mesh-mail_send") + h.Set(DeadHeaderSubject, "mesh.stream.mesh-mail.event.mail.send.approved") + h.Set(DeadHeaderRestricted, "true") + h.Set(DeadHeaderWorkQueue, "true") + h.Set(DeadHeaderAgesOut, agesOut.Format(time.RFC3339Nano)) + + d := deadLetterOf(42, "mesh.stream.mesh-mail.event.mail.send.approved", h, []byte("secret body"), false) + if !d.WorkQueue { + t.Fatalf("a work-queue dead letter read as %+v", d) + } + if !d.Restricted { + t.Fatalf("a work-queue dead letter is not read as restricted, so it would be shown with its body: %+v", d) + } + if !d.AgesOut.Equal(agesOut) { + t.Fatalf("ages out at %v, want %v — it should age with its stream", d.AgesOut, agesOut) + } +} + +// 1178/AC3: asking to deliver a work-queue dead letter again is refused, naming ADR 0436 clause 6 — before +// anything is asked of the bus, so it holds whatever max-deliver or age is left. +func TestAWorkQueueDeadLetterIsNeverDeliveredAgain(t *testing.T) { + d := DeadLetter{ID: 7, Stream: "MAIL_SENDS", Consumer: "novox_mesh-mail_send", + Subject: "mesh.stream.mesh-mail.event.mail.send.approved", WorkQueue: true, Restricted: true, + AgesOut: time.Now().Add(time.Hour)} + if _, err := AgainTo(nil, d); err == nil || !strings.Contains(err.Error(), "ADR 0436 clause 6") || + !strings.Contains(err.Error(), "Drop it") { + t.Fatalf("delivering a work-queue dead letter again: %v; want a refusal naming ADR 0436 clause 6", err) + } +}