diff --git a/cmd/mesh-controller/deadletters_aging_test.go b/cmd/mesh-controller/deadletters_aging_test.go index 867f97b0..2e4af61d 100644 --- a/cmd/mesh-controller/deadletters_aging_test.go +++ b/cmd/mesh-controller/deadletters_aging_test.go @@ -2,6 +2,7 @@ package main import ( "context" + "strings" "testing" "time" @@ -67,3 +68,31 @@ func TestADeadLetterThatAgesOutClearsItsConditionAndSaysSo(t *testing.T) { t.Fatalf("not plain: %s", why) } } + +// **A message past its restricted stream's age is not kept, only counted** (novox/hq ADR 0376, as the feature's lead +// decided): its advisory is said as a condition whose words give how many aged out undelivered and from which stream. +func TestMessagesNotKeptPastTheirAgeAreCounted(t *testing.T) { + now := time.Date(2026, 10, 15, 12, 0, 0, 0, time.UTC) + f := calm(now) + f.advisories = []link.Advisory{{Kind: link.AdvisoryMaxDeliveries, ID: "MAIL.ace_mailer", Stream: "MAIL", + Consumer: "ace_mailer", Token: link.AdvisoryAgedOut, Said: "the consumer gave up after the stream's age", + First: now.Add(-10 * time.Minute), Last: now.Add(-time.Minute), Count: 2}} + var found []conditions.Observation + for _, o := range watchAdvisories(f) { + if o.Token == link.AdvisoryAgedOut { + found = append(found, o) + } + } + if len(found) != 1 { + t.Fatalf("said %+v", found) + } + o := found[0] + if o.Headline != "2 messages of the stream MAIL aged out undelivered" || o.Times != 2 || + !strings.Contains(o.Summary, "2 messages of the stream MAIL aged out undelivered") { + t.Fatalf("%+v", o) + } + if why, ok := conditions.PlainWords(conditions.Words{Headline: o.Headline, Explanation: o.Explanation, + Resolved: o.Resolved, Needs: o.Needs}); !ok { + t.Fatalf("not plain: %s", why) + } +} diff --git a/cmd/mesh-controller/signals.go b/cmd/mesh-controller/signals.go index 51b7d598..98bcc8d5 100644 --- a/cmd/mesh-controller/signals.go +++ b/cmd/mesh-controller/signals.go @@ -584,6 +584,18 @@ func watchAdvisories(f *signalFacts) []conditions.Observation { "for you yet. It tries again every minute." o.Resolved = "Resolved: the message is kept" } + if a.Kind == link.AdvisoryMaxDeliveries && a.Token == link.AdvisoryAgedOut { + // Given up on after its restricted stream's age, so not kept at all, only counted (novox/hq ADR 0376). + counted := link.AgedOutWords(a.Count, a.Stream) + o.Machine = consumerMachine(a.Stream, a.Consumer) + o.Summary = fmt.Sprintf("%s in the last hour: %s", counted, a.Said) + o.Headline = clip(conditions.Capital(counted), 60) + o.Explanation = conditions.Capital(fmt.Sprintf("%s could not handle %s, and by the time it gave up the "+ + "stream's time limit had passed, so the mesh did not keep them. Nothing to do; this clears an hour "+ + "after the last one.", consumerWho(a.Stream, a.Consumer), map[bool]string{true: "some messages", + false: "a message"}[a.Count > 1])) + o.Resolved = "Resolved: no more messages of the stream " + a.Stream + " aged out undelivered" + } out = append(out, o) } if f.busFaultsErr != nil { diff --git a/internal/link/advisories.go b/internal/link/advisories.go index fc5d0e1d..683603c8 100644 --- a/internal/link/advisories.go +++ b/internal/link/advisories.go @@ -69,17 +69,18 @@ type AdvisoryLog struct { // Advisories is this process's log. var Advisories = &AdvisoryLog{seen: map[string]*Advisory{}} -// Heard records one advisory. -func (l *AdvisoryLog) Heard(a Advisory, at time.Time) { +// Heard records one advisory, and answers how often it has been said while remembered. +func (l *AdvisoryLog) Heard(a Advisory, at time.Time) int { l.mu.Lock() defer l.mu.Unlock() key := a.Kind + "/" + a.ID + "/" + a.Token if had, ok := l.seen[key]; ok { had.Last, had.Said, had.Count = at, a.Said, had.Count+1 - return + return had.Count } a.First, a.Last, a.Count = at, at, 1 l.seen[key] = &a + return 1 } // Since is every advisory said at or after a moment, and forgets the older ones. diff --git a/internal/link/deadletters.go b/internal/link/deadletters.go index 63f7d1c3..4376a9de 100644 --- a/internal/link/deadletters.go +++ b/internal/link/deadletters.go @@ -71,9 +71,12 @@ type DeadLetter struct { Original string `json:"original,omitempty"` // 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"` + Restricted bool `json:"restricted,omitempty"` + // AgedOut says it was not kept at all: from a restricted stream and already past its stream's maximum age + // when its notice was taken, so DEAD_LETTERS holds nothing of it, and it is only counted (novox/hq ADR 0376). + AgedOut bool `json:"-"` + 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"` @@ -134,7 +137,7 @@ func KeepDeadLetter(js nats.JetStreamContext, notice []byte) (DeadLetter, error) 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). + // is not kept at all, only counted. Every other dead letter keeps no age, as before (ADR 0264). restricted, maxAge, err := sourceAge(js, a.Stream) if err != nil { return DeadLetter{}, err @@ -148,12 +151,13 @@ func KeepDeadLetter(js nats.JetStreamContext, notice []byte) (DeadLetter, error) agesOut = raw.Time.Add(maxAge) kept.Header.Set(DeadHeaderAgesOut, agesOut.UTC().Format(time.RFC3339Nano)) } + if err == nil && !agesOut.IsZero() && agesOut.Sub(deadLetterNow()) < time.Second { + // **Past its stream's age, it is not kept at all** (novox/hq ADR 0376 §3, as the feature's lead decided): + // no copy, no subject, no headers — its subject alone names whose mail it was. Only counted, by the caller. + return DeadLetter{Stream: a.Stream, Consumer: a.Consumer, Who: ConsumerInWords(a.Stream, a.Consumer), + Sequence: a.StreamSeq, Deliveries: a.Deliveries, GaveUp: gaveUp, Restricted: true, AgedOut: true}, nil + } 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-") { @@ -552,30 +556,64 @@ func keepingGivenUp(ctx context.Context, bus *broker.JetStream, logger *log.Logg // again after noticeRetry, and said: in the log, and as the consumer's max-deliveries condition. func keepGivenUp(js nats.JetStreamContext, logger *log.Logger) (*nats.Subscription, error) { return js.Subscribe("", func(m *nats.Msg) { - d, err := KeepDeadLetter(js, m.Data) - if err != nil { - logger.Printf("a message a consumer gave up on could NOT be kept: %v", err) - var a maxDeliveries - if json.Unmarshal(m.Data, &a) == nil && a.Stream != "" && a.Consumer != "" { - Advisories.Heard(Advisory{Kind: AdvisoryMaxDeliveries, ID: a.Stream + "." + a.Consumer, - Stream: a.Stream, Consumer: a.Consumer, Token: AdvisoryNotKept, - Said: fmt.Sprintf("%s gave up on message %d, and it could not be kept: %v", - ConsumerInWords(a.Stream, a.Consumer), a.StreamSeq, err)}, time.Now()) - } - _ = m.NakWithDelay(noticeRetry) - return - } - if d.Lost != "" { - logger.Printf("%s gave up on message %d of %s, which its stream no longer held; kept as dead letter %d "+ - "without it", d.Who, d.Sequence, d.Stream, d.ID) + if takeNotice(js, m.Data, logger) { + _ = m.Ack() } else { - logger.Printf("%s gave up on message %d of %s (%s) after %d deliveries; kept as dead letter %d", - d.Who, d.Sequence, d.Stream, d.Subject, d.Deliveries, d.ID) + _ = m.NakWithDelay(noticeRetry) } - _ = m.Ack() }, nats.Bind(broker.DeadLetterNoticesStream, broker.ControllerName), nats.ManualAck()) } +// AdvisoryAgedOut is the token of a max-deliveries advisory about messages of a restricted stream given up on after +// their stream's maximum age: not kept, only counted (novox/hq ADR 0376). +const AdvisoryAgedOut = "aged-out" + +// AgedOutWords is how many messages of a stream aged out undelivered, as the log and the condition say it. +func AgedOutWords(n int, stream string) string { + messages := "a message" + if n > 1 { + messages = fmt.Sprintf("%d messages", n) + } + return fmt.Sprintf("%s of the stream %s aged out undelivered", messages, stream) +} + +// takeNotice keeps the message one notice is about, says what came of it, and answers whether the notice is done +// with: false to take it again after noticeRetry. +func takeNotice(js nats.JetStreamContext, notice []byte, logger *log.Logger) bool { + d, err := KeepDeadLetter(js, notice) + if err != nil { + logger.Printf("a message a consumer gave up on could NOT be kept: %v", err) + var a maxDeliveries + if json.Unmarshal(notice, &a) == nil && a.Stream != "" && a.Consumer != "" { + Advisories.Heard(Advisory{Kind: AdvisoryMaxDeliveries, ID: a.Stream + "." + a.Consumer, + Stream: a.Stream, Consumer: a.Consumer, Token: AdvisoryNotKept, + Said: fmt.Sprintf("%s gave up on message %d, and it could not be kept: %v", + ConsumerInWords(a.Stream, a.Consumer), a.StreamSeq, err)}, time.Now()) + } + return false + } + if d.AgedOut { + // Counted, never kept: said in the log, and as the consumer's aged-out condition, whose count is how many + // (novox/hq ADR 0376). + a := Advisory{Kind: AdvisoryMaxDeliveries, ID: d.Stream + "." + d.Consumer, Stream: d.Stream, + Consumer: d.Consumer, Token: AdvisoryAgedOut, + Said: fmt.Sprintf("%s gave up on messages of %s after that stream's maximum age, so they were not kept: "+ + "%s keeps a message only for its maximum age", d.Who, d.Stream, d.Stream)} + n := Advisories.Heard(a, time.Now()) + logger.Printf("%s gave up on message %d of %s after its stream's maximum age, so it is not kept: %s "+ + "in the last hour (novox/hq ADR 0376)", d.Who, d.Sequence, d.Stream, AgedOutWords(n, d.Stream)) + return true + } + if d.Lost != "" { + logger.Printf("%s gave up on message %d of %s, which its stream no longer held; kept as dead letter %d "+ + "without it", d.Who, d.Sequence, d.Stream, d.ID) + } else { + logger.Printf("%s gave up on message %d of %s (%s) after %d deliveries; kept as dead letter %d", + d.Who, d.Sequence, d.Stream, d.Subject, d.Deliveries, d.ID) + } + return true +} + // 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 { diff --git a/internal/link/deadletters_restricted_test.go b/internal/link/deadletters_restricted_test.go index 47a09ac1..169d8414 100644 --- a/internal/link/deadletters_restricted_test.go +++ b/internal/link/deadletters_restricted_test.go @@ -2,6 +2,7 @@ package link import ( "fmt" + "log" "strings" "testing" "time" @@ -105,8 +106,8 @@ func TestARestrictedDeadLetterIsListedWithoutItsBody(t *testing.T) { } // **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. +// of MAIL's week left, DEAD_LETTERS no longer holds it once they pass; with none left, +// it is not kept at all and only counted. func TestARestrictedDeadLetterAgesOutWithItsStream(t *testing.T) { js := aBus(t) subject := aRestrictedStream(t, js) @@ -127,21 +128,34 @@ func TestARestrictedDeadLetterAgesOutWithItsStream(t *testing.T) { 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) + // Past its age when the notice is taken, it is not kept at all — no copy, no subject, no headers — only + // counted, and the count said in the log and as the consumer's aged-out advisory. + var said strings.Builder + logger := log.New(&said, "", 0) + for i := 0; i < 2; i++ { + seq, published = publishMail(t, js, subject) + withClock(t, func() time.Time { return published.Add(7*24*time.Hour + time.Minute) }) + if !takeNotice(js.Context(), notice(seq), logger) { + t.Fatal("a notice of a message past its age is taken again") + } } - raw, err := js.Context().GetMsg(broker.DeadLettersStream, past.ID) - if err != nil { - t.Fatal(err) + if held, _ := HeldDeadLetters(js.Context()); held["MAIL.ace_mailer"] != 0 { + t.Fatalf("a message past its age was kept: %v", held) } - 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 info, _ := js.Context().StreamInfo(broker.DeadLettersStream); info.State.Msgs != 0 { + t.Fatalf("DEAD_LETTERS holds %d messages, want none", info.State.Msgs) } - 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) + if !strings.Contains(said.String(), "2 messages of the stream MAIL aged out undelivered") { + t.Fatalf("the log said %q", said.String()) + } + counted := 0 + for _, a := range Advisories.Since(time.Now().Add(-time.Minute)) { + if a.Kind == AdvisoryMaxDeliveries && a.ID == "MAIL.ace_mailer" && a.Token == AdvisoryAgedOut { + counted = a.Count + } + } + if counted != 2 { + t.Fatalf("counted %d aged out, want 2", counted) } }