From 9aeb5f44935184b1854ba0c23eefb4a3d3876aa1 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 15 Oct 2026 14:56:14 +0200 Subject: [PATCH] =?UTF-8?q?Keep=20a=20restricted=20message=20past=20its=20?= =?UTF-8?q?stream's=20age=20as=20a=20record=20without=20its=20body,=20as?= =?UTF-8?q?=20ADR=200376=20=C2=A73=20says=20(issue=20728)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reverts 45b6c347, which kept nothing of such a message and raised an aged-out condition: that departed from an accepted decision on a lead's word alone. The concern behind it (the record keeps the subject, with no age) is its own issue. --- cmd/mesh-controller/deadletters_aging_test.go | 29 ------ cmd/mesh-controller/signals.go | 12 --- internal/link/advisories.go | 7 +- internal/link/deadletters.go | 94 ++++++------------- internal/link/deadletters_restricted_test.go | 42 +++------ 5 files changed, 45 insertions(+), 139 deletions(-) diff --git a/cmd/mesh-controller/deadletters_aging_test.go b/cmd/mesh-controller/deadletters_aging_test.go index 2e4af61d..867f97b0 100644 --- a/cmd/mesh-controller/deadletters_aging_test.go +++ b/cmd/mesh-controller/deadletters_aging_test.go @@ -2,7 +2,6 @@ package main import ( "context" - "strings" "testing" "time" @@ -68,31 +67,3 @@ 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 98bcc8d5..51b7d598 100644 --- a/cmd/mesh-controller/signals.go +++ b/cmd/mesh-controller/signals.go @@ -584,18 +584,6 @@ 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 683603c8..fc5d0e1d 100644 --- a/internal/link/advisories.go +++ b/internal/link/advisories.go @@ -69,18 +69,17 @@ type AdvisoryLog struct { // Advisories is this process's log. var Advisories = &AdvisoryLog{seen: map[string]*Advisory{}} -// Heard records one advisory, and answers how often it has been said while remembered. -func (l *AdvisoryLog) Heard(a Advisory, at time.Time) int { +// Heard records one advisory. +func (l *AdvisoryLog) Heard(a Advisory, at time.Time) { 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 had.Count + return } 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 4376a9de..63f7d1c3 100644 --- a/internal/link/deadletters.go +++ b/internal/link/deadletters.go @@ -71,12 +71,9 @@ 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"` - // 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"` + 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"` @@ -137,7 +134,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 not kept at all, only counted. Every other dead letter keeps no age, as before (ADR 0264). + // 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 @@ -151,13 +148,12 @@ 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-") { @@ -556,64 +552,30 @@ 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) { - if takeNotice(js, m.Data, logger) { - _ = m.Ack() - } else { + 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) + } 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.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 169d8414..47a09ac1 100644 --- a/internal/link/deadletters_restricted_test.go +++ b/internal/link/deadletters_restricted_test.go @@ -2,7 +2,6 @@ package link import ( "fmt" - "log" "strings" "testing" "time" @@ -106,8 +105,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 not kept at all and only counted. +// 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) @@ -128,34 +127,21 @@ func TestARestrictedDeadLetterAgesOutWithItsStream(t *testing.T) { t.Fatalf("still held after its age: %v", held) } - // 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") - } + 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) } - if held, _ := HeldDeadLetters(js.Context()); held["MAIL.ace_mailer"] != 0 { - t.Fatalf("a message past its age was kept: %v", held) + raw, err := js.Context().GetMsg(broker.DeadLettersStream, past.ID) + if err != nil { + t.Fatal(err) } - if info, _ := js.Context().StreamInfo(broker.DeadLettersStream); info.State.Msgs != 0 { - t.Fatalf("DEAD_LETTERS holds %d messages, want none", info.State.Msgs) + 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 !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) + 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) } }