diff --git a/internal/broker/modstreams.go b/internal/broker/modstreams.go index fe9174a1..8345a46b 100644 --- a/internal/broker/modstreams.go +++ b/internal/broker/modstreams.go @@ -24,6 +24,9 @@ func ModuleStreamSubject(module, event string) string { return moduleStreamPrefix + module + ".event." + event } +// OnAModuleStream says a subject is one a module's own stream takes. +func OnAModuleStream(subject string) bool { return strings.HasPrefix(subject, moduleStreamPrefix) } + // WirePattern is a local event pattern as the bus spells it: `**` is `>`. func WirePattern(local string) string { parts := strings.Split(local, ".") diff --git a/internal/link/deadletters.go b/internal/link/deadletters.go index 0b4f41fa..2b6002dd 100644 --- a/internal/link/deadletters.go +++ b/internal/link/deadletters.go @@ -259,7 +259,8 @@ func DeadLetterNamed(js nats.JetStreamContext, id uint64) (DeadLetter, error) { // work queue has one worker per subject, so only the worker that gave it up takes it, and DeliverAgain // removes the original from the queue first, so there are never two (novox/hq issue 334); only while the // worker that gave it up is on the bus. Any other stream's -// message is refused, with why: an ask to a seat the controller does not ask would need a publish it is +// message is refused, with why — but a module's own stream's, which goes back as an event while its consumer is +// still granted it (novox/hq ADR 0363 §7): an ask to a seat the controller does not ask would need a publish it is // not granted, in its asker's name (ADR 0264's consequences, ADR 0259 §3), and a stream that is neither // would reach every consumer of its subject. func AgainTo(js nats.JetStreamContext, d DeadLetter) (string, error) { @@ -280,6 +281,29 @@ func AgainTo(js nats.JetStreamContext, d DeadLetter) (string, error) { return "", fmt.Errorf("whether %s can receive dead letter %d cannot be read: %w", d.Who, d.ID, err) } return d.Subject, nil + case d.Stream != broker.EventsStream && broker.OnAModuleStream(d.Subject): + // **A module stream's event goes back to its consumer only while it is still granted** (novox/hq ADR + // 0363 §7): its consumer on that stream must still be there and still filter the subject, or delivering + // it again would hand a module what its grant no longer names. Then it is delivered as an event is, on + // EVENTS under the consumer's own again-subject, which the module's consumer there filters. + info, err := js.ConsumerInfo(d.Stream, d.Consumer) + if errors.Is(err, nats.ErrConsumerNotFound) { + return "", fmt.Errorf("dead letter %d was given up by %s, which no longer hears %s, so it would reach a "+ + "module no longer granted it. Drop it", d.ID, d.Who, d.Stream) + } + if err != nil { + return "", fmt.Errorf("whether %s still hears %s cannot be read: %w", d.Who, d.Stream, err) + } + granted := false + for _, f := range append([]string{info.Config.FilterSubject}, info.Config.FilterSubjects...) { + if f != "" && broker.SubjectsOverlap(f, d.Subject) { + granted = true + } + } + if !granted { + return "", fmt.Errorf("dead letter %d is on %s, which %s is no longer granted. Drop it", d.ID, d.Subject, d.Who) + } + d.Stream = broker.EventsStream case d.Stream != broker.EventsStream: return "", fmt.Errorf("dead letter %d is from %s, and only an event or an ask the controller made is "+ "delivered again: publishing it again would reach every consumer of its subject, or need a grant the "+ diff --git a/internal/link/deadletters_module_stream_test.go b/internal/link/deadletters_module_stream_test.go new file mode 100644 index 00000000..ce08a68c --- /dev/null +++ b/internal/link/deadletters_module_stream_test.go @@ -0,0 +1,48 @@ +package link + +import ( + "strings" + "testing" + + "github.com/novox/mesh-controller/internal/broker" +) + +// A message given up on a module's own stream is delivered again on EVENTS to the consumer that gave it up, +// and only while that consumer is still granted it (novox/hq ADR 0363 §7). +func TestAModuleStreamsDeadLetterGoesBackOnlyWhileGranted(t *testing.T) { + js := aBus(t) + mail := broker.ModuleStream{Module: "mesh-mail", Name: "MAIL", Events: []string{"mail.*.message.added"}, + MaxAgeDays: 7, MaxMiB: 16, Restricted: true} + if err := js.EnsureStream(mail.Stream()); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = js.Context().DeleteStream("MAIL") }) + subject := broker.ModuleStreamSubject("mesh-mail", "mail.mesh.message.added") + heard := broker.Heard{Stream: "MAIL", Emitter: "mesh-mail", Filters: []string{subject}} + reader := broker.Principal{Kind: broker.KindModule, Node: "ace", Module: "reader", Hears: []broker.Heard{heard}, + PasswordHash: "x"} + users := broker.ConsumersOf([]broker.Principal{reader}) + if len(users) != 2 { + t.Fatalf("the reader's consumers: %+v", users) + } + for _, c := range users { + if err := js.EnsureConsumer(c.Consumer); err != nil { + t.Fatal(err) + } + } + d := DeadLetter{ID: 9, Stream: "MAIL", Consumer: "ace_reader", Subject: subject} + got, err := AgainTo(js.Context(), d) + if err != nil { + t.Fatal(err) + } + if got != broker.AgainSubject("ace_reader", subject) { + t.Fatalf("delivered again on %s", got) + } + + if err := js.RemoveConsumer("MAIL", "ace_reader"); err != nil { + t.Fatal(err) + } + if _, err := AgainTo(js.Context(), d); err == nil || !strings.Contains(err.Error(), "no longer hears MAIL") { + t.Fatalf("delivered again to a consumer no longer granted: %v", err) + } +}