From d55e148a1da233a7a5bb76cdc3ee8ff7efb7bb5b Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 15 Oct 2026 15:28:56 +0200 Subject: [PATCH] Never wait NextMsg's bound for a dead letter held already overcounts (issue 728) --- internal/link/deadletters.go | 11 ++++++- internal/link/deadletters_restricted_test.go | 31 ++++++++++++++++++++ 2 files changed, 41 insertions(+), 1 deletion(-) diff --git a/internal/link/deadletters.go b/internal/link/deadletters.go index 63f7d1c3..94b442cb 100644 --- a/internal/link/deadletters.go +++ b/internal/link/deadletters.go @@ -607,7 +607,16 @@ func AgingDeadLetters(js nats.JetStreamContext, held map[string]int) (map[string if err != nil { return nil, fmt.Errorf("the dead letters of %s cannot be read: %w", key, err) } - for i := 0; i < n; i++ { + // held can count more than DEAD_LETTERS still holds for this consumer — one aged out, + // delivered again or dropped since it was counted (novox/hq issue 728 review round 1): read + // the ordered consumer's pending count once, from right when it was subscribed, and look for + // at most that many messages, so a look never waits out NextMsg's bound for one that is not + // there. That bound stays only as a guard against a pending count that is itself stale. + limit := n + if pending, err := sub.InitialConsumerPending(); err == nil && int(pending) < limit { + limit = int(pending) + } + for i := 0; i < limit; 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 diff --git a/internal/link/deadletters_restricted_test.go b/internal/link/deadletters_restricted_test.go index 47a09ac1..c621bbc2 100644 --- a/internal/link/deadletters_restricted_test.go +++ b/internal/link/deadletters_restricted_test.go @@ -218,3 +218,34 @@ func TestTheAgingOfRestrictedDeadLettersIsRead(t *testing.T) { t.Fatalf("aging %+v, kept %+v", aging, d) } } + +// **A look never waits out NextMsg's bound for a dead letter that is not there** (novox/hq issue 728 review round +// 1): held can count more than DEAD_LETTERS still holds for a consumer, one aged out, delivered again or dropped +// since it was counted. AgingDeadLetters reads the ordered consumer's pending count once and looks for at most +// that many, so it returns in well under the 2-second NextMsg bound rather than waiting it out for messages that +// are gone. +func TestTheAgingOfRestrictedDeadLettersDoesNotWaitForWhatHeldOvercounts(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) + } + // held says 3 for this consumer; DEAD_LETTERS holds only the one just kept. + held := map[string]int{"MAIL.ace_mailer": 3} + + started := time.Now() + aging, err := AgingDeadLetters(js.Context(), held) + elapsed := time.Since(started) + if err != nil { + t.Fatal(err) + } + if elapsed >= time.Second { + t.Fatalf("took %s, well over the 2-second bound it should never wait out", elapsed) + } + if len(aging["MAIL.ace_mailer"]) != 1 || aging["MAIL.ace_mailer"][0].ID != d.ID { + t.Fatalf("aging %+v, kept %+v", aging, d) + } +}