From d03e24f42396603f3383f1e4a00c1e04d1156a30 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 15 Oct 2026 14:11:37 +0200 Subject: [PATCH] Age a restricted stream's dead letters out with it and never show their body (issue 728) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- cmd/mesh-controller/deadletters_aging.go | 117 ++++++++++ cmd/mesh-controller/deadletters_aging_test.go | 69 ++++++ cmd/mesh-controller/probes.go | 6 + cmd/mesh-controller/watchdogs.go | 13 +- internal/broker/jetstream.go | 4 + internal/broker/modstreams.go | 2 + internal/broker/msgttl_live_test.go | 86 +++++++ internal/broker/streams.go | 17 +- internal/link/deadletters.go | 163 ++++++++++++- internal/link/deadletters_restricted_test.go | 220 ++++++++++++++++++ 10 files changed, 690 insertions(+), 7 deletions(-) create mode 100644 cmd/mesh-controller/deadletters_aging.go create mode 100644 cmd/mesh-controller/deadletters_aging_test.go create mode 100644 internal/broker/msgttl_live_test.go create mode 100644 internal/link/deadletters_restricted_test.go diff --git a/cmd/mesh-controller/deadletters_aging.go b/cmd/mesh-controller/deadletters_aging.go new file mode 100644 index 00000000..e7b2b466 --- /dev/null +++ b/cmd/mesh-controller/deadletters_aging.go @@ -0,0 +1,117 @@ +package main + +import ( + "context" + "fmt" + "sort" + "strings" + "time" + + "github.com/novox/mesh-controller/internal/conditions" + "github.com/novox/mesh-controller/internal/link" +) + +// A dead letter from a restricted stream ages out with its stream (novox/hq ADR 0376 §3, §6). +// +// DEAD_LETTERS lets such a message go when its stream's maximum age runs out, so a consumer's `max-deliveries` +// condition can clear with nobody having delivered it again or dropped it. That is said: the watchdogs remember, +// from one look to the next, when each restricted dead letter ages out, count those gone once their time passed, +// say each in the controller's log as it goes, and clear the condition with words saying how many aged out +// undelivered and from which stream. + +// agingTally is what the watchdogs remember of restricted dead letters between looks. +type agingTally struct { + // seen is each consumer's restricted dead letters at the last look, by id, with when each ages out. + seen map[string]map[uint64]time.Time + // agedOut is how many of each consumer's aged out undelivered while its condition was open. + agedOut map[string]int +} + +// look compares this look's restricted dead letters with the last: newly is how many of each consumer's aged out +// since, and cleared how many aged out in all for each consumer DEAD_LETTERS no longer holds anything of, when any +// did. held is every consumer's count this look; aging its restricted dead letters. +func (t *agingTally) look(now time.Time, held map[string]int, aging map[string][]link.Aging) (newly, cleared map[string]int) { + if t.seen == nil { + t.seen, t.agedOut = map[string]map[uint64]time.Time{}, map[string]int{} + } + newly, cleared = map[string]int{}, map[string]int{} + for key, before := range t.seen { + still := map[uint64]bool{} + for _, a := range aging[key] { + still[a.ID] = true + } + for id, agesOut := range before { + if !still[id] && !agesOut.After(now) { + newly[key]++ + } + } + if newly[key] == 0 { + delete(newly, key) + } + t.agedOut[key] += newly[key] + } + for key, n := range t.agedOut { + if held[key] > 0 { + continue + } + if n > 0 { + cleared[key] = n + } + delete(t.agedOut, key) + } + t.seen = map[string]map[uint64]time.Time{} + for key, letters := range aging { + t.seen[key] = map[uint64]time.Time{} + for _, a := range letters { + t.seen[key][a.ID] = a.AgesOut + } + } + return newly, cleared +} + +// agedOutResolved is the line a consumer's condition clears with when messages it gave up on aged out undelivered. +func agedOutResolved(key string, n int) string { + stream, _, _ := strings.Cut(key, ".") + messages := "a message" + if n > 1 { + messages = fmt.Sprintf("%d messages", n) + } + return fmt.Sprintf("Resolved: %s of the stream %s aged out undelivered", messages, stream) +} + +// sayAgedOut says, in the log, what aged out since the last look, and clears the condition of every consumer whose +// dead letters are all gone with the words saying how many aged out. Before the rows are reconciled, so the +// condition is cleared with these words and not its own. +func (w *watchdogs) sayAgedOut(ctx context.Context, f *signalFacts) []string { + if f.advisoriesErr != nil || f.deadLettersAging == nil { + return nil + } + newly, cleared := w.aging.look(f.now, f.deadLetters, f.deadLettersAging) + var problems []string + for _, key := range countedKeys(newly) { + stream, consumer, _ := strings.Cut(key, ".") + fmt.Printf("%d message(s) %s gave up on aged out of %s undelivered: %s keeps them only for its maximum age "+ + "(novox/hq ADR 0376)\n", newly[key], link.ConsumerInWords(stream, consumer), "DEAD_LETTERS", stream) + } + for _, key := range countedKeys(cleared) { + stream, consumer, _ := strings.Cut(key, ".") + fmt.Printf("%s no longer has a message kept: %d aged out of %s undelivered, so its condition clears "+ + "(novox/hq ADR 0376)\n", link.ConsumerInWords(stream, consumer), cleared[key], stream) + if _, err := w.keeper.ClearSaying(ctx, conditions.Key(conditions.ScopeBus, key, link.AdvisoryMaxDeliveries), + fmt.Sprintf("its %d dead letter(s) of %s aged out undelivered", cleared[key], stream), + agedOutResolved(key, cleared[key])); err != nil { + problems = append(problems, err.Error()) + } + } + return problems +} + +// countedKeys is a count's keys in order. +func countedKeys(m map[string]int) []string { + out := make([]string, 0, len(m)) + for k := range m { + out = append(out, k) + } + sort.Strings(out) + return out +} diff --git a/cmd/mesh-controller/deadletters_aging_test.go b/cmd/mesh-controller/deadletters_aging_test.go new file mode 100644 index 00000000..867f97b0 --- /dev/null +++ b/cmd/mesh-controller/deadletters_aging_test.go @@ -0,0 +1,69 @@ +package main + +import ( + "context" + "testing" + "time" + + "github.com/novox/mesh-controller/internal/conditions" + "github.com/novox/mesh-controller/internal/link" +) + +// **A dead letter that ages out clears its condition and says so** (novox/hq ADR 0376 §6), on a fake clock: while +// DEAD_LETTERS holds a restricted stream's messages for a consumer its max-deliveries condition is open; once they +// have aged out it clears, with words saying how many aged out undelivered and from which stream. One dropped before +// its age is not counted as aged out, and a consumer of EVENTS clears as before. +func TestADeadLetterThatAgesOutClearsItsConditionAndSaysSo(t *testing.T) { + now := time.Date(2026, 10, 15, 12, 0, 0, 0, time.UTC) + store := conditions.NewInMemory() + told := &conditions.Told{} + k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store, Teller: told, + Now: func() time.Time { return now }}) + w := &watchdogs{keeper: k, started: now.Add(-time.Hour)} + + look := func(at time.Time, held map[string]int, aging map[string][]link.Aging) { + f := calm(at) + f.deadLetters, f.deadLettersAging = held, aging + w.see(t.Context(), f) + } + mail := "MAIL.ace_mailer" + look(now, map[string]int{mail: 3, "EVENTS.media_sonarr": 1}, map[string][]link.Aging{mail: { + {ID: 1, AgesOut: now.Add(time.Hour)}, {ID: 2, AgesOut: now.Add(2 * time.Hour)}, {ID: 3, AgesOut: now.Add(3 * time.Hour)}}}) + if open, _ := k.Open(t.Context()); len(open) != 2 { + t.Fatalf("open: %+v", open) + } + // One dropped by a person before its age: not aged out. + look(now.Add(30*time.Minute), map[string]int{mail: 2, "EVENTS.media_sonarr": 1}, map[string][]link.Aging{mail: { + {ID: 2, AgesOut: now.Add(2 * time.Hour)}, {ID: 3, AgesOut: now.Add(3 * time.Hour)}}}) + look(now.Add(150*time.Minute), map[string]int{mail: 1}, map[string][]link.Aging{mail: {{ID: 3, AgesOut: now.Add(3 * time.Hour)}}}) + open, _ := k.Open(t.Context()) + if len(open) != 1 || open[0].Key != conditions.Key(conditions.ScopeBus, mail, link.AdvisoryMaxDeliveries) { + t.Fatalf("after one aged out: %+v", open) + } + look(now.Add(4*time.Hour), map[string]int{}, map[string][]link.Aging{}) + if open, _ := k.Open(t.Context()); len(open) != 0 { + t.Fatalf("after all aged out: %+v", open) + } + k.Close(context.Background()) // everything said is said once it closes + var mailCleared, sonarrCleared string + for _, e := range told.Said() { + if e.Change != conditions.ChangeCleared { + continue + } + switch e.Condition.Key { + case conditions.Key(conditions.ScopeBus, mail, link.AdvisoryMaxDeliveries): + mailCleared = e.Condition.Resolved + case conditions.Key(conditions.ScopeBus, "EVENTS.media_sonarr", link.AdvisoryMaxDeliveries): + sonarrCleared = e.Condition.Resolved + } + } + if mailCleared != "Resolved: 2 messages of the stream MAIL aged out undelivered" { + t.Fatalf("MAIL's condition cleared with %q", mailCleared) + } + if sonarrCleared != "Resolved: the messages it gave up on were delivered again or dropped" { + t.Fatalf("an EVENTS consumer's condition cleared with %q", sonarrCleared) + } + if why, ok := conditions.Plain(mailCleared); !ok { + t.Fatalf("not plain: %s", why) + } +} diff --git a/cmd/mesh-controller/probes.go b/cmd/mesh-controller/probes.go index 481571f7..deafc4e9 100644 --- a/cmd/mesh-controller/probes.go +++ b/cmd/mesh-controller/probes.go @@ -929,6 +929,12 @@ func streamDiffers(want broker.Stream, have nats.StreamConfig) string { differs = append(differs, fmt.Sprintf("keeps one of a message id for %s, defined %s", have.Duplicates, time.Duration(want.DuplicatesSeconds)*time.Second)) } + if want.AllowMsgTTL && !have.AllowMsgTTL { + differs = append(differs, "takes no time to live of a message's own, defined to (novox/hq ADR 0376)") + } + if want.Restricted && have.Metadata[broker.RestrictedMetadata] != "true" { + differs = append(differs, "does not say it is restricted, so a message kept from it would not age out with it") + } if want.DiscardNew && have.Discard != nats.DiscardNew { differs = append(differs, "drops what it holds when full, defined to refuse what comes next") } diff --git a/cmd/mesh-controller/watchdogs.go b/cmd/mesh-controller/watchdogs.go index a8d2f793..4b5fde61 100644 --- a/cmd/mesh-controller/watchdogs.go +++ b/cmd/mesh-controller/watchdogs.go @@ -91,7 +91,10 @@ type signalFacts struct { // deadLettersNewest is when DEAD_LETTERS kept the newest message it holds for each of those // consumers: the condition counts the messages given up on, not the looks (novox/hq issue 440). deadLettersNewest map[string]time.Time - advisoriesErr error + // deadLettersAging is, for each consumer whose dead letters came from a restricted stream, when each ages out + // (novox/hq ADR 0376 §6); nil when it was not read. + deadLettersAging map[string][]link.Aging + advisoriesErr error // busFaultsErr is why the bus module's events could not be read this tick (novox/hq ADR 0357). busFaultsErr error @@ -211,6 +214,9 @@ type watchdogs struct { // confirm holds back a row blind for one tick: a read that timed out once on a loaded store. confirm confirming + // aging is what is remembered of restricted dead letters between looks (novox/hq ADR 0376 §6). + aging agingTally + mu sync.Mutex ticked time.Time last *signalFacts @@ -264,6 +270,8 @@ func (w *watchdogs) tick(ctx context.Context) { func (w *watchdogs) see(running context.Context, f *signalFacts) { var problems []string var blind []conditions.Observation + // What aged out of DEAD_LETTERS, said and cleared in its own words before the rows clear it in theirs. + problems = append(problems, w.sayAgedOut(running, f)...) for _, row := range watchedRows() { if err := row.needs(f); err != nil { blind = append(blind, blindRow(row, err)) @@ -382,6 +390,9 @@ func (w *watchdogs) gather(ctx context.Context) *signalFacts { } } } + if f.advisoriesErr == nil && w.js != nil { + f.deadLettersAging, f.advisoriesErr = link.AgingDeadLetters(w.js.Context(), f.deadLetters) + } f.handActs, f.handActsErr = w.gatherHandActs(ctx, now) f.refusedSends, f.refusedSendsErr = inv.RefusedSendsSince(ctx, now.Add(-generationRefusalsSaid)) f.doing, f.doingErr = inv.NotDoingWhatTheyWereTold(ctx) diff --git a/internal/broker/jetstream.go b/internal/broker/jetstream.go index 9f9dca93..495c5290 100644 --- a/internal/broker/jetstream.go +++ b/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 } diff --git a/internal/broker/modstreams.go b/internal/broker/modstreams.go index f909c671..fc584b67 100644 --- a/internal/broker/modstreams.go +++ b/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, } } diff --git a/internal/broker/msgttl_live_test.go b/internal/broker/msgttl_live_test.go new file mode 100644 index 00000000..02589420 --- /dev/null +++ b/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) + } +} diff --git a/internal/broker/streams.go b/internal/broker/streams.go index 124dfa86..54d8c4c1 100644 --- a/internal/broker/streams.go +++ b/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", }, } } diff --git a/internal/link/deadletters.go b/internal/link/deadletters.go index 2b6002dd..63f7d1c3 100644 --- a/internal/link/deadletters.go +++ b/internal/link/deadletters.go @@ -41,6 +41,10 @@ const ( DeadHeaderLost = "Mesh-Dead-Lost" // DeadHeaderMsgID is the message's own de-duplication id, kept, since the copy carries its own. DeadHeaderMsgID = "Mesh-Dead-Msg-Id" + // DeadHeaderRestricted says the message came from a restricted stream (novox/hq ADR 0376), and + // 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" // AgainHeader marks a message delivered again, with the kept copy's id it came from. AgainHeader = "Mesh-Delivered-Again" ) @@ -65,12 +69,46 @@ type DeadLetter struct { // Original says what became of the ask it was kept from, in its seat's work queue, when it was // delivered again (novox/hq issue 334). Original string `json:"original,omitempty"` - // Body and Headers are the message itself, given only for one dead letter asked by its id; a header - // with several values keeps them all. + // Restricted says it came from a restricted stream (novox/hq ADR 0376): its body and its own headers are + // never shown, said in Withheld, and it ages out at AgesOut, when its stream's maximum age runs out. + Restricted bool `json:"restricted,omitempty"` + AgesOut time.Time `json:"ages_out,omitzero"` + Withheld string `json:"withheld,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"` Headers map[string][]string `json:"headers,omitempty"` } +// deadLetterNow is the clock dead letters age by: a variable, so a test can move it. +var deadLetterNow = time.Now + +// leastLeftToDeliver is the least of its stream's age a restricted dead letter must have left to be delivered +// again (novox/hq ADR 0376 §5). +const leastLeftToDeliver = time.Minute + +// withheldWords is what the dead-letters verb says in place of a restricted dead letter's body. +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) { + info, err := js.StreamInfo(stream) + if errors.Is(err, nats.ErrStreamNotFound) { + return false, 0, nil + } + if err != nil { + return 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 +} + +// ttlOf is a time left as a message's time to live, in whole seconds and never more than what is left. +func ttlOf(left time.Duration) string { + return strconv.FormatInt(int64(left/time.Second), 10) + "s" +} + // maxDeliveries is the part of the server's notice a kept message is made from. type maxDeliveries struct { Stream string `json:"stream"` @@ -94,8 +132,28 @@ func KeepDeadLetter(js nats.JetStreamContext, notice []byte) (DeadLetter, error) gaveUp = time.Now() } kept := &nats.Msg{Subject: broker.DeadLetterSubject(a.Stream, a.Consumer), Header: nats.Header{}} + // **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) + if err != nil { + return DeadLetter{}, err + } + if restricted { + kept.Header.Set(DeadHeaderRestricted, "true") + } raw, err := js.GetMsg(a.Stream, a.StreamSeq) + agesOut := time.Time{} + if err == nil && restricted && maxAge > 0 { + agesOut = raw.Time.Add(maxAge) + kept.Header.Set(DeadHeaderAgesOut, agesOut.UTC().Format(time.RFC3339Nano)) + } switch { + case err == nil && !agesOut.IsZero() && agesOut.Sub(deadLetterNow()) < time.Second: + kept.Header.Set(DeadHeaderSubject, raw.Subject) + kept.Header.Set(DeadHeaderPublished, raw.Time.UTC().Format(time.RFC3339Nano)) + kept.Header.Set(DeadHeaderLost, fmt.Sprintf("message %d of %s had passed its stream's maximum age of %s when "+ + "the notice was taken, so it is kept without its body (novox/hq ADR 0376)", a.StreamSeq, a.Stream, maxAge)) case err == nil: for k, v := range raw.Header { if strings.HasPrefix(k, "Nats-") { @@ -109,6 +167,9 @@ func KeepDeadLetter(js nats.JetStreamContext, notice []byte) (DeadLetter, error) kept.Header.Set(DeadHeaderSubject, raw.Subject) kept.Header.Set(DeadHeaderPublished, raw.Time.UTC().Format(time.RFC3339Nano)) kept.Data = raw.Data + if !agesOut.IsZero() { + kept.Header.Set(nats.MsgTTLHdr, ttlOf(agesOut.Sub(deadLetterNow()))) + } case errors.Is(err, nats.ErrMsgNotFound) || errors.Is(err, nats.ErrStreamNotFound): kept.Header.Set(DeadHeaderLost, fmt.Sprintf("%s no longer held message %d when the notice was taken: %v", a.Stream, a.StreamSeq, err)) @@ -143,6 +204,8 @@ func deadLetterOf(id uint64, subject string, h nats.Header, data []byte, whole b d.Deliveries, _ = strconv.ParseUint(h.Get(DeadHeaderDeliveries), 10, 64) 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.AgesOut, _ = time.Parse(time.RFC3339Nano, h.Get(DeadHeaderAgesOut)) if whole { d.Body = string(data) d.Headers = map[string][]string{} @@ -230,6 +293,9 @@ func DeadLetters(js nats.JetStreamContext, consumer string, most int) ([]DeadLet if consumer != "" && consumer != d.Consumer && consumer != d.Stream+"."+d.Consumer { continue } + if d.Restricted { + d.Withheld = withheldWords + } out = append(out, d) } return out, total, nil @@ -238,8 +304,21 @@ func DeadLetters(js nats.JetStreamContext, consumer string, most int) ([]DeadLet // ErrNoDeadLetter is a dead letter's id DEAD_LETTERS does not hold. var ErrNoDeadLetter = errors.New("no such dead letter") -// DeadLetterNamed is one kept message whole: what it said, and the headers it said it with. +// DeadLetterNamed is one kept message whole: what it said, and the headers it said it with — **but for one from +// a restricted stream, whose body and own headers are never shown** (novox/hq ADR 0376 §4). A grant names modules, +// and this is asked by people and by tool runners' bus accounts, which a grant never names: withholding the body +// from every caller is that grant's check, said plainly. func DeadLetterNamed(js nats.JetStreamContext, id uint64) (DeadLetter, error) { + d, err := deadLetterWhole(js, id) + if err != nil || !d.Restricted { + return d, err + } + d.Body, d.Headers, d.Withheld = "", nil, withheldWords + return d, nil +} + +// deadLetterWhole is one kept message whole, a restricted one's body too: for delivering it again, never to show. +func deadLetterWhole(js nats.JetStreamContext, id uint64) (DeadLetter, error) { raw, err := js.GetMsg(broker.DeadLettersStream, id) if errors.Is(err, nats.ErrMsgNotFound) { return DeadLetter{}, fmt.Errorf("%w: %s holds no message %d — delivered again or dropped already, "+ @@ -286,6 +365,16 @@ func AgainTo(js nats.JetStreamContext, d DeadLetter) (string, error) { // 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. + // + // **A restricted one only while at least a minute of its stream's age is left** (novox/hq ADR 0376 §5): + // past that it can only be dropped, and the copy keeps what is left (DeliverAgain). + if d.Restricted { + if left := d.AgesOut.Sub(deadLetterNow()); d.AgesOut.IsZero() || left < leastLeftToDeliver { + return "", fmt.Errorf("dead letter %d has aged out: %s keeps a message only for its maximum age, and "+ + "less than a minute of it is left (it ages out at %s), so it is not delivered again. Drop it "+ + "(novox/hq ADR 0376)", d.ID, d.Stream, d.AgesOut.UTC().Format(time.RFC3339)) + } + } 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 "+ @@ -332,7 +421,16 @@ func AgainTo(js nats.JetStreamContext, d DeadLetter) (string, error) { // from DEAD_LETTERS; an ask's original is removed from its work queue before. The message carries its own headers and AgainHeader; its de-duplication id is the // kept copy's, so asking twice delivers it once. func DeliverAgain(js nats.JetStreamContext, id uint64) (DeadLetter, string, error) { - d, err := DeadLetterNamed(js, id) + d, to, err := deliverAgain(js, id) + if d.Restricted { + // What it said stays unsaid to whoever asked it delivered (novox/hq ADR 0376 §4). + d.Body, d.Headers, d.Withheld = "", nil, withheldWords + } + return d, to, err +} + +func deliverAgain(js nats.JetStreamContext, id uint64) (DeadLetter, string, error) { + d, err := deadLetterWhole(js, id) if err != nil { return d, "", err } @@ -349,6 +447,10 @@ func DeliverAgain(js nats.JetStreamContext, id uint64) (DeadLetter, string, erro } again.Header.Set(AgainHeader, strconv.FormatUint(id, 10)) again.Header.Set(nats.MsgIdHdr, "again."+broker.DeadLettersStream+"."+strconv.FormatUint(id, 10)) + if d.Restricted { + // The copy keeps what is left of its stream's age (novox/hq ADR 0376 §5). + again.Header.Set(nats.MsgTTLHdr, ttlOf(d.AgesOut.Sub(deadLetterNow()))) + } if _, err := js.PublishMsg(again); err != nil { if d.Original != "" { return d, to, fmt.Errorf("dead letter %d could not be delivered again on %s: %w; it is still kept, so "+ @@ -473,3 +575,56 @@ func keepGivenUp(js nats.JetStreamContext, logger *log.Logger) (*nats.Subscripti _ = m.Ack() }, nats.Bind(broker.DeadLetterNoticesStream, broker.ControllerName), nats.ManualAck()) } + +// Aging is one restricted dead letter as it ages (novox/hq ADR 0376 §6): its id, and when its stream's maximum age +// runs out and DEAD_LETTERS lets it go. +type Aging struct { + ID uint64 + AgesOut time.Time +} + +// AgingDeadLetters is, for each consumer of held whose dead letters came from a restricted stream, each one that +// ages out and when: read by their headers alone, never their bodies. A consumer of any other stream is left out, +// and so is a record that keeps no age. +func AgingDeadLetters(js nats.JetStreamContext, held map[string]int) (map[string][]Aging, error) { + out := map[string][]Aging{} + restricted := map[string]bool{} + for key, n := range held { + stream, consumer, _ := strings.Cut(key, ".") + is, known := restricted[stream] + if !known { + r, _, err := sourceAge(js, stream) + if err != nil { + return nil, err + } + restricted[stream], is = r, r + } + if !is || n <= 0 { + continue + } + sub, err := js.SubscribeSync(broker.DeadLetterSubject(stream, consumer), nats.BindStream(broker.DeadLettersStream), + nats.OrderedConsumer(), nats.HeadersOnly(), nats.DeliverAll()) + if err != nil { + return nil, fmt.Errorf("the dead letters of %s cannot be read: %w", key, err) + } + for i := 0; i < n; i++ { + m, err := sub.NextMsg(2 * time.Second) + if errors.Is(err, nats.ErrTimeout) { + break // delivered again, dropped or aged out since they were counted + } + if err != nil { + _ = sub.Unsubscribe() + return nil, fmt.Errorf("the dead letters of %s cannot be read: %w", key, err) + } + meta, err := m.Metadata() + if err != nil { + continue + } + if at, err := time.Parse(time.RFC3339Nano, m.Header.Get(DeadHeaderAgesOut)); err == nil { + out[key] = append(out[key], Aging{ID: meta.Sequence.Stream, AgesOut: at}) + } + } + _ = sub.Unsubscribe() + } + return out, nil +} diff --git a/internal/link/deadletters_restricted_test.go b/internal/link/deadletters_restricted_test.go new file mode 100644 index 00000000..47a09ac1 --- /dev/null +++ b/internal/link/deadletters_restricted_test.go @@ -0,0 +1,220 @@ +package link + +import ( + "fmt" + "strings" + "testing" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "github.com/novox/mesh-controller/internal/broker" +) + +// A restricted stream of a module, MAIL, kept a week, on the bus (novox/hq ADR 0363), and the subject an event of +// the account `mesh` is published on. +func aRestrictedStream(t *testing.T, js *broker.JetStream) string { + t.Helper() + mail := broker.ModuleStream{Module: "mesh-mail", Name: "MAIL", Events: []string{"mail.*.message.added"}, + MaxAgeDays: 7, MaxMiB: 16, Restricted: true} + _ = js.Context().DeleteStream("MAIL") + if err := js.EnsureStream(mail.Stream()); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = js.Context().DeleteStream("MAIL") }) + return broker.ModuleStreamSubject("mesh-mail", "mail.mesh.message.added") +} + +// withClock sets the clock dead letters are kept and delivered again by, until the test ends. +func withClock(t *testing.T, now func() time.Time) { + t.Helper() + before := deadLetterNow + deadLetterNow = now + t.Cleanup(func() { deadLetterNow = before }) +} + +// publishMail puts one mail event on MAIL with a header of its own, and answers its sequence and when it was stored. +func publishMail(t *testing.T, js *broker.JetStream, subject string) (uint64, time.Time) { + t.Helper() + msg := &nats.Msg{Subject: subject, Data: []byte(`{"from":"someone@example.org"}`), Header: nats.Header{}} + msg.Header.Set("x-account", "mesh") + ack, err := js.Context().PublishMsg(msg) + if err != nil { + t.Fatal(err) + } + raw, err := js.Context().GetMsg("MAIL", ack.Sequence) + if err != nil { + t.Fatal(err) + } + return ack.Sequence, raw.Time +} + +// notice is the server's maximum-deliveries notice for one message of MAIL. +func notice(seq uint64) []byte { + return []byte(fmt.Sprintf(`{"stream":"MAIL","consumer":"ace_mailer","stream_seq":%d,"deliveries":5}`, seq)) +} + +// **A dead letter from a restricted stream is listed without its body, and ages out with its stream** (ADR 0376 +// §3, §4): given up on by its consumer and kept, the list and the show by id give what it was and when it ages out, +// never its body or its own headers, and the kept copy carries a time to live of what is left of MAIL's week. +func TestARestrictedDeadLetterIsListedWithoutItsBody(t *testing.T) { + js := aBus(t) + keeping(t, js) + subject := aRestrictedStream(t, js) + heard := broker.Heard{Stream: "MAIL", Emitter: "mesh-mail", Restricted: true, Filters: []string{subject}} + for _, c := range broker.ConsumersOf([]broker.Principal{{Kind: broker.KindModule, Node: "ace", Module: "mailer", + Hears: []broker.Heard{heard}, PasswordHash: "x"}}) { + if err := js.EnsureConsumer(c.Consumer); err != nil { + t.Fatal(err) + } + } + _, published := publishMail(t, js, subject) + api, err := jetstream.New(js.Conn()) + if err != nil { + t.Fatal(err) + } + mailer, err := api.Consumer(t.Context(), "MAIL", "ace_mailer") + if err != nil { + t.Fatal(err) + } + d := givenUp(t, js, mailer) + weekLeft := published.Add(7 * 24 * time.Hour) + if !d.Restricted || d.Body != "" || d.Headers != nil || d.Subject != subject || d.Sequence != 1 || d.Deliveries != 5 || + d.Published.IsZero() || d.GaveUp.IsZero() || d.Size == 0 || d.AgesOut.Sub(weekLeft).Abs() > time.Second { + t.Fatalf("listed as %+v", d) + } + shown, err := DeadLetterNamed(js.Context(), d.ID) + if err != nil { + t.Fatal(err) + } + if shown.Body != "" || shown.Headers != nil || !shown.Restricted || shown.Withheld == "" || shown.AgesOut.IsZero() { + t.Fatalf("shown by its id as %+v", shown) + } + raw, err := js.Context().GetMsg(broker.DeadLettersStream, d.ID) + if err != nil { + t.Fatal(err) + } + ttl, err := time.ParseDuration(raw.Header.Get(nats.MsgTTLHdr)) + if err != nil || ttl > 7*24*time.Hour || ttl < 7*24*time.Hour-time.Minute { + t.Fatalf("kept with a time to live of %q (%v), want what is left of the week", raw.Header.Get(nats.MsgTTLHdr), err) + } + if string(raw.Data) != `{"from":"someone@example.org"}` { + t.Fatal("the kept copy does not hold the message, so it could not be delivered again") + } +} + +// **Past its stream's age, a restricted dead letter is gone** (ADR 0376 §3), on a fake clock: kept with two seconds +// of MAIL's week left, DEAD_LETTERS no longer holds it once they pass; with none left, it is kept as a record +// without its body, which cannot be delivered again. +func TestARestrictedDeadLetterAgesOutWithItsStream(t *testing.T) { + js := aBus(t) + subject := aRestrictedStream(t, js) + seq, published := publishMail(t, js, subject) + withClock(t, func() time.Time { return published.Add(7*24*time.Hour - 2*time.Second) }) + d, err := KeepDeadLetter(js.Context(), notice(seq)) + if err != nil { + t.Fatal(err) + } + if !d.Restricted || d.Lost != "" { + t.Fatalf("kept as %+v", d) + } + eventually(t, "the dead letter aged out", func() bool { + _, err := js.Context().GetMsg(broker.DeadLettersStream, d.ID) + return err != nil + }) + if held, _ := HeldDeadLetters(js.Context()); held["MAIL.ace_mailer"] != 0 { + t.Fatalf("still held after its age: %v", held) + } + + seq, published = publishMail(t, js, subject) + withClock(t, func() time.Time { return published.Add(7*24*time.Hour + time.Minute) }) + past, err := KeepDeadLetter(js.Context(), notice(seq)) + if err != nil { + t.Fatal(err) + } + raw, err := js.Context().GetMsg(broker.DeadLettersStream, past.ID) + if err != nil { + t.Fatal(err) + } + if len(raw.Data) != 0 || raw.Header.Get("x-account") != "" || !strings.Contains(past.Lost, "maximum age") { + t.Fatalf("kept past its age as %+v with %q", past, raw.Data) + } + if _, _, err := DeliverAgain(js.Context(), past.ID); err == nil || !strings.Contains(err.Error(), "Drop it") { + t.Fatalf("a record past its age was delivered again: %v", err) + } +} + +// **A restricted dead letter is delivered again only within its age, and the copy keeps that age** (ADR 0376 §5): +// with more than a minute left the copy goes on EVENTS under the consumer's again subject with a time to live of +// what is left; with less, it is refused as aged out. +func TestARestrictedDeadLetterGoesBackOnlyWithinItsAge(t *testing.T) { + js := aBus(t) + subject := aRestrictedStream(t, js) + heard := broker.Heard{Stream: "MAIL", Emitter: "mesh-mail", Restricted: true, Filters: []string{subject}} + for _, c := range broker.ConsumersOf([]broker.Principal{{Kind: broker.KindModule, Node: "ace", Module: "mailer", + Hears: []broker.Heard{heard}, PasswordHash: "x"}}) { + if err := js.EnsureConsumer(c.Consumer); err != nil { + t.Fatal(err) + } + } + seq, published := publishMail(t, js, subject) + at := published.Add(7*24*time.Hour - 10*time.Minute) + withClock(t, func() time.Time { return at }) + d, err := KeepDeadLetter(js.Context(), notice(seq)) + if err != nil { + t.Fatal(err) + } + at = published.Add(7*24*time.Hour - 30*time.Second) + if _, _, err := DeliverAgain(js.Context(), d.ID); err == nil || !strings.Contains(err.Error(), "aged out") { + t.Fatalf("delivered again with under a minute left: %v", err) + } + at = published.Add(7*24*time.Hour - 5*time.Minute) + _, to, err := DeliverAgain(js.Context(), d.ID) + if err != nil { + t.Fatal(err) + } + if to != broker.AgainSubject("ace_mailer", subject) { + t.Fatalf("delivered again on %s", to) + } + raw, err := js.Context().GetLastMsg(broker.EventsStream, to) + if err != nil { + t.Fatal(err) + } + ttl, err := time.ParseDuration(raw.Header.Get(nats.MsgTTLHdr)) + if err != nil || ttl > 5*time.Minute || ttl < 4*time.Minute { + t.Fatalf("the copy carries a time to live of %q (%v), want what is left", raw.Header.Get(nats.MsgTTLHdr), err) + } + if string(raw.Data) != `{"from":"someone@example.org"}` || raw.Header.Get("x-account") != "mesh" { + t.Fatalf("the copy is %q %v", raw.Data, raw.Header) + } +} + +// What of a consumer's dead letters ages out, and when, is read without their bodies, and only for a restricted +// stream's (ADR 0376 §6). +func TestTheAgingOfRestrictedDeadLettersIsRead(t *testing.T) { + js := aBus(t) + subject := aRestrictedStream(t, js) + seq, published := publishMail(t, js, subject) + withClock(t, func() time.Time { return published.Add(time.Hour) }) + d, err := KeepDeadLetter(js.Context(), notice(seq)) + if err != nil { + t.Fatal(err) + } + if _, err := KeepDeadLetter(js.Context(), []byte(`{"stream":"EVENTS","consumer":"media_sonarr","stream_seq":42,`+ + `"deliveries":5}`)); err != nil { + t.Fatal(err) + } + held, err := HeldDeadLetters(js.Context()) + if err != nil { + t.Fatal(err) + } + aging, err := AgingDeadLetters(js.Context(), held) + if err != nil { + t.Fatal(err) + } + if len(aging) != 1 || len(aging["MAIL.ace_mailer"]) != 1 || aging["MAIL.ace_mailer"][0].ID != d.ID || + !aging["MAIL.ace_mailer"][0].AgesOut.Equal(d.AgesOut) { + t.Fatalf("aging %+v, kept %+v", aging, d) + } +}