Age a restricted stream's dead letters out with it and never show their body (issue 728)

DEAD_LETTERS kept a message given up on MAIL forever and showed it whole to any caller of dead-letters, outside the grant and past MAIL's week. Per ADR 0376 §3-§6: the kept copy carries a per-message time to live of what is left of its stream's age (a copy already past it is kept without its body), DEAD_LETTERS and EVENTS allow per-message TTL, the list and show give no body or own headers for it, delivering it again needs a minute left and the copy keeps the time left, and its max-deliveries condition clears saying how many aged out undelivered and from which stream.

internal/broker/jetstream.go

@@ -173,6 +173,10 @@ func (j *JetStream) EnsureStream(s Stream) error {
 		Description:       s.Why,
 	}
 	want.AllowDirect = s.Direct
+	want.AllowMsgTTL = s.AllowMsgTTL
+	if s.Restricted {
+		want.Metadata = map[string]string{RestrictedMetadata: "true"}
+	}
 	if s.MaxBytes > 0 {
 		want.MaxBytes = s.MaxBytes
 	}

internal/broker/modstreams.go

@@ -77,6 +77,8 @@ func (s ModuleStream) Stream() Stream {
 		MaxAge:    s.MaxAgeDays * 24 * 60 * 60,
 		MaxBytes:  int64(s.MaxMiB) * 1024 * 1024,
 		Why:       why,
+		// Said on the bus, so a dead letter kept from it ages out with it (novox/hq ADR 0376 §3).
+		Restricted: s.Restricted,
 	}
 }
 

internal/broker/msgttl_live_test.go

@@ -0,0 +1,86 @@
+package broker
+
+import (
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go"
+
+	"github.com/novox/mesh-controller/internal/testbus"
+)
+
+// **DEAD_LETTERS and EVENTS take a message's own time to live once updated in place** (novox/hq ADR 0376 §3, §5),
+// on the pinned server: each made as it was before, without one, is brought to its definition by the controller's
+// assertion, keeps what it held, and removes a message once its time to live passes. A restricted module stream says
+// so in its metadata; one without a grant does not.
+//
+//	go test ./internal/broker/ -run TestTheStreamsTakeAMessagesTimeToLive
+func TestTheStreamsTakeAMessagesTimeToLive(t *testing.T) {
+	js, err := Dial(testbus.URL(t))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer js.Close()
+	for _, s := range MeshStreams() {
+		if s.Name != EventsStream && s.Name != DeadLettersStream {
+			if s.AllowMsgTTL {
+				t.Errorf("%s takes a time to live of a message's own, which nothing asks of it", s.Name)
+			}
+			continue
+		}
+		if !s.AllowMsgTTL {
+			t.Fatalf("%s is not defined to take a message's time to live", s.Name)
+		}
+		_ = js.js.DeleteStream(s.Name)
+		before := s
+		before.AllowMsgTTL = false
+		if err := js.EnsureStream(before); err != nil {
+			t.Fatalf("%s as it was: %v", s.Name, err)
+		}
+		subject := DeadLetterSubject("EVENTS", "ace_reader")
+		if s.Name == EventsStream {
+			subject = "mesh.mod.gitea.event.pull.merged"
+		}
+		if _, err := js.js.Publish(subject, []byte("held")); err != nil {
+			t.Fatal(err)
+		}
+		if err := js.EnsureStream(s); err != nil {
+			t.Fatalf("%s updated in place: %v", s.Name, err)
+		}
+		info, err := js.js.StreamInfo(s.Name)
+		if err != nil || !info.Config.AllowMsgTTL || info.State.Msgs != 1 {
+			t.Fatalf("%s after the update: %+v (%v)", s.Name, info, err)
+		}
+		msg := &nats.Msg{Subject: subject, Data: []byte("brief"), Header: nats.Header{}}
+		msg.Header.Set(nats.MsgTTLHdr, "1s")
+		ack, err := js.js.PublishMsg(msg)
+		if err != nil {
+			t.Fatalf("%s refused a message's time to live: %v", s.Name, err)
+		}
+		deadline := time.Now().Add(8 * time.Second)
+		for {
+			if _, err := js.js.GetMsg(s.Name, ack.Sequence); err != nil {
+				break
+			}
+			if time.Now().After(deadline) {
+				t.Fatalf("%s still holds a message past its time to live", s.Name)
+			}
+			time.Sleep(200 * time.Millisecond)
+		}
+		if info, _ := js.js.StreamInfo(s.Name); info.State.Msgs != 1 {
+			t.Fatalf("%s holds %d, want only the message without an age", s.Name, info.State.Msgs)
+		}
+		_ = js.js.PurgeStream(s.Name)
+	}
+	if !mailStream.Stream().Restricted || newsStream.Stream().Restricted {
+		t.Fatal("a module stream does not say whether it is restricted")
+	}
+	_ = js.js.DeleteStream("MAIL")
+	if err := js.EnsureStream(mailStream.Stream()); err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = js.js.DeleteStream("MAIL") }()
+	if info, _ := js.js.StreamInfo("MAIL"); info.Config.Metadata[RestrictedMetadata] != "true" {
+		t.Fatalf("MAIL's metadata: %v", info.Config.Metadata)
+	}
+}

internal/broker/streams.go

@@ -59,8 +59,17 @@ type Stream struct {
 	// DuplicatesSeconds is the window in which a message id published twice is kept once; zero for the
 	// server's default (two minutes).
 	DuplicatesSeconds int
+	// AllowMsgTTL lets a message carry a time to live of its own (`Nats-TTL`), for what must not outlive the
+	// stream it came from (novox/hq ADR 0376 §3, §5).
+	AllowMsgTTL bool
+	// Restricted marks a module's stream restricted by a grant (novox/hq ADR 0363 §4), in the stream's own
+	// metadata, so whoever keeps one of its messages reads it from the bus (ADR 0376 §3).
+	Restricted bool
 }
 
+// RestrictedMetadata is the key a restricted stream's metadata says it under.
+const RestrictedMetadata = "mesh-restricted"
+
 // AssignmentsStream holds every assignment's membership, the newest per subject.
 const AssignmentsStream = "ASSIGNMENTS"
 
@@ -174,6 +183,8 @@ func MeshStreams() []Stream {
 			Retention:         RetentionLimits,
 			MaxAge:            7 * 24 * 60 * 60,
 			MaxMsgsPerSubject: 10000,
+			// A restricted dead letter delivered again keeps what is left of its stream's age (ADR 0376 §5).
+			AllowMsgTTL: true,
 			Why: "a subscriber that was down catches up; tool traffic under the same prefix is " +
 				"excluded by the event token; per-subject caps keep a noisy emitter from " +
 				"evicting a quiet one without splitting the stream",
@@ -194,9 +205,11 @@ func MeshStreams() []Stream {
 			MaxBytes:          DeadLettersBytes,
 			DiscardNew:        true,
 			DuplicatesSeconds: 24 * 60 * 60,
+			AllowMsgTTL:       true,
 			Why: "every message a consumer gave up on, with its consumer, subject, deliveries and when, kept " +
-				"until a person delivers it again or drops it with why (novox/hq issue 330); no age, and full " +
-				"it refuses the next copy rather than drop one it holds",
+				"until a person delivers it again or drops it with why (novox/hq issue 330); no age but a restricted " +
+				"stream's message, which ages out with its stream (ADR 0376), and full it refuses the next copy " +
+				"rather than drop one it holds",
 		},
 	}
 }