Deliver a module stream's given-up message again only while its consumer is still granted it (ADR 0363 §7)

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, ".")

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 "+

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)
+	}
+}