Never wait NextMsg's bound for a dead letter held already overcounts (issue 728)
This commit is contained in:
@@ -607,7 +607,16 @@ func AgingDeadLetters(js nats.JetStreamContext, held map[string]int) (map[string
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("the dead letters of %s cannot be read: %w", key, err)
|
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)
|
m, err := sub.NextMsg(2 * time.Second)
|
||||||
if errors.Is(err, nats.ErrTimeout) {
|
if errors.Is(err, nats.ErrTimeout) {
|
||||||
break // delivered again, dropped or aged out since they were counted
|
break // delivered again, dropped or aged out since they were counted
|
||||||
|
|||||||
@@ -218,3 +218,34 @@ func TestTheAgingOfRestrictedDeadLettersIsRead(t *testing.T) {
|
|||||||
t.Fatalf("aging %+v, kept %+v", aging, d)
|
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)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user