Fail overdue notifications before sending them (N-01)
The worker checked the 24-hour give-up deadline only after Send failed, so an overdue row whose delivery succeeded was marked sent and a stale notification reached the household. Check the deadline before Send: a row claimed 24 hours or more after its creation is marked failed, with the reason and any previous error in last_error, and the sender is never called. A delivery failing before the deadline still retries with the same backoff. Tests cover a succeeding and a failing sender at exactly 24 hours and after, a row just under the deadline still being sent, and the retry run now ending with a failure at the deadline instead of a 30th attempt. Task: addeb256-a4d4-4acf-8acd-6b6778d967ee Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -458,15 +458,20 @@ func TestGiveUpAfter24Hours(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Attempts at 0, 1, 3, 7, 15, 31 and 63 minutes, then hourly to 1383
|
// Attempts at 0, 1, 3, 7, 15, 31 and 63 minutes, then hourly to 1383
|
||||||
// minutes (22 more), then the last at the deadline: 30 in all.
|
// minutes (22 more): 29 in all. The row is due once more at the
|
||||||
const wantAttempts = 30
|
// deadline, where it is failed without being sent.
|
||||||
|
const wantAttempts = 29
|
||||||
r := onlyRow(t, st, key, ChannelPush)
|
r := onlyRow(t, st, key, ChannelPush)
|
||||||
if r.Status != "failed" || r.Attempts != wantAttempts || r.NextAttemptAt.Valid ||
|
if r.Status != "failed" || r.Attempts != wantAttempts || r.NextAttemptAt.Valid ||
|
||||||
r.LastError.String != "push service unavailable" || r.UpdatedAt != timestamp(deadline) {
|
r.LastError.String != "not delivered within 24 hours of creation; last error: push service unavailable" ||
|
||||||
|
r.UpdatedAt != timestamp(deadline) {
|
||||||
t.Fatalf("row = %+v, want failed at %s after %d attempts", r, timestamp(deadline), wantAttempts)
|
t.Fatalf("row = %+v, want failed at %s after %d attempts", r, timestamp(deadline), wantAttempts)
|
||||||
}
|
}
|
||||||
if len(times) != wantAttempts || !times[len(times)-1].Equal(deadline) || !times[len(times)-2].Equal(start.Add(1383*time.Minute)) {
|
if len(times) != wantAttempts+1 || !times[len(times)-1].Equal(deadline) || !times[len(times)-2].Equal(start.Add(1383*time.Minute)) {
|
||||||
t.Fatalf("attempted at %v, want %d attempts ending 1383 min and 24 h after creation", times, wantAttempts)
|
t.Fatalf("due at %v, want %d attempts ending 1383 min after creation, then the deadline", times, wantAttempts)
|
||||||
|
}
|
||||||
|
if n := len(sender.messages()); n != wantAttempts {
|
||||||
|
t.Fatalf("%d deliveries, want %d", n, wantAttempts)
|
||||||
}
|
}
|
||||||
|
|
||||||
// A failed row is never delivered again.
|
// A failed row is never delivered again.
|
||||||
@@ -479,24 +484,68 @@ func TestGiveUpAfter24Hours(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestOverdueRowFailsOnFirstFailure(t *testing.T) {
|
// TestOverdueRowIsNeverDelivered covers a row first claimed 24 hours or
|
||||||
|
// more after its creation, as when the service was down: it is failed
|
||||||
|
// without calling the sender, whether the sender would succeed or fail.
|
||||||
|
func TestOverdueRowIsNeverDelivered(t *testing.T) {
|
||||||
|
for _, tc := range []struct {
|
||||||
|
name string
|
||||||
|
at time.Duration
|
||||||
|
fail error
|
||||||
|
}{
|
||||||
|
{"succeeding sender at 24 h", 24 * time.Hour, nil},
|
||||||
|
{"succeeding sender after 24 h", 25 * time.Hour, nil},
|
||||||
|
{"failing sender at 24 h", 24 * time.Hour, errors.New("gone")},
|
||||||
|
{"failing sender after 24 h", 25 * time.Hour, errors.New("gone")},
|
||||||
|
} {
|
||||||
|
t.Run(tc.name, func(t *testing.T) {
|
||||||
|
ctx := context.Background()
|
||||||
|
svc, st, c := newTestService(t)
|
||||||
|
setPref(t, st, PlanReady, true, false)
|
||||||
|
sender := &fakeSender{fail: tc.fail}
|
||||||
|
if err := svc.Register(ChannelPush, sender); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
const key = "plan_ready:2026-10-12"
|
||||||
|
if err := svc.Emit(ctx, planReady(key)); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
c.set(start.Add(tc.at))
|
||||||
|
if _, err := svc.deliverDue(ctx); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if n := len(sender.messages()); n != 0 {
|
||||||
|
t.Fatalf("%d deliveries of an overdue row, want 0", n)
|
||||||
|
}
|
||||||
|
r := onlyRow(t, st, key, ChannelPush)
|
||||||
|
if r.Status != "failed" || r.Attempts != 0 || r.NextAttemptAt.Valid ||
|
||||||
|
r.LastError.String != "not delivered within 24 hours of creation" || r.UpdatedAt != timestamp(start.Add(tc.at)) {
|
||||||
|
t.Fatalf("row = %+v, want failed without an attempt", r)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestRowJustBeforeDeadlineIsDelivered pins the boundary from the other
|
||||||
|
// side: a row claimed just under 24 hours after its creation is sent.
|
||||||
|
func TestRowJustBeforeDeadlineIsDelivered(t *testing.T) {
|
||||||
ctx := context.Background()
|
ctx := context.Background()
|
||||||
svc, st, c := newTestService(t)
|
svc, st, c := newTestService(t)
|
||||||
setPref(t, st, PlanReady, true, false)
|
setPref(t, st, PlanReady, true, false)
|
||||||
if err := svc.Register(ChannelPush, &fakeSender{fail: errors.New("gone")}); err != nil {
|
sender := &fakeSender{}
|
||||||
|
if err := svc.Register(ChannelPush, sender); err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
const key = "plan_ready:2026-10-12"
|
const key = "plan_ready:2026-10-12"
|
||||||
if err := svc.Emit(ctx, planReady(key)); err != nil {
|
if err := svc.Emit(ctx, planReady(key)); err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
// The service was down for more than a day.
|
c.set(start.Add(24*time.Hour - time.Millisecond))
|
||||||
c.set(start.Add(25 * time.Hour))
|
|
||||||
if _, err := svc.deliverDue(ctx); err != nil {
|
if _, err := svc.deliverDue(ctx); err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
if r := onlyRow(t, st, key, ChannelPush); r.Status != "failed" || r.Attempts != 1 || r.LastError.String != "gone" {
|
if r := onlyRow(t, st, key, ChannelPush); r.Status != "sent" || r.Attempts != 1 || len(sender.messages()) != 1 {
|
||||||
t.Fatalf("row = %+v, want failed after 1 attempt", r)
|
t.Fatalf("row = %+v after %d deliveries, want sent once", r, len(sender.messages()))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+25
-12
@@ -19,9 +19,10 @@ const (
|
|||||||
// further failure doubles it, up to retryMax.
|
// further failure doubles it, up to retryMax.
|
||||||
retryFirst = time.Minute
|
retryFirst = time.Minute
|
||||||
retryMax = time.Hour
|
retryMax = time.Hour
|
||||||
// giveUpAfter is how long after its creation a row is retried. The
|
// giveUpAfter is how long after its creation a row may be delivered.
|
||||||
// last retry is due at that moment, and a delivery failing then or
|
// The last retry is due at that moment; a row claimed then or later is
|
||||||
// later marks the row failed.
|
// marked failed without being sent, and a delivery failing at or
|
||||||
|
// after it marks the row failed.
|
||||||
giveUpAfter = 24 * time.Hour
|
giveUpAfter = 24 * time.Hour
|
||||||
// sendTimeout bounds one Send.
|
// sendTimeout bounds one Send.
|
||||||
sendTimeout = 30 * time.Second
|
sendTimeout = 30 * time.Second
|
||||||
@@ -175,7 +176,8 @@ func (s *Service) deliverDue(ctx context.Context) (time.Duration, error) {
|
|||||||
// deliver sends a claimed row through its channel's sender and records the
|
// deliver sends a claimed row through its channel's sender and records the
|
||||||
// outcome. A row that cannot be sent at all, because its payload does not
|
// outcome. A row that cannot be sent at all, because its payload does not
|
||||||
// decode or its channel has no sender, is failed at once: retrying cannot
|
// decode or its channel has no sender, is failed at once: retrying cannot
|
||||||
// help.
|
// help. A row 24 hours or more past its creation is failed without being
|
||||||
|
// sent: it is too stale to deliver.
|
||||||
func (s *Service) deliver(ctx context.Context, row db.NotificationOutbox) {
|
func (s *Service) deliver(ctx context.Context, row db.NotificationOutbox) {
|
||||||
logger := s.logger.With("notification_id", row.ID, "event_key", row.EventKey, "channel", row.Channel)
|
logger := s.logger.With("notification_id", row.ID, "event_key", row.EventKey, "channel", row.Channel)
|
||||||
var payload Payload
|
var payload Payload
|
||||||
@@ -194,9 +196,27 @@ func (s *Service) deliver(ctx context.Context, row db.NotificationOutbox) {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
created, err := parseTimestamp(row.CreatedAt)
|
||||||
|
if err != nil {
|
||||||
|
reason := fmt.Sprintf("created_at does not parse: %v", err)
|
||||||
|
logger.ErrorContext(ctx, "notification failed", "reason", reason)
|
||||||
|
s.giveUp(ctx, logger, row, 0, reason)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
deadline := created.Add(giveUpAfter)
|
||||||
|
if !s.now().Before(deadline) {
|
||||||
|
reason := "not delivered within 24 hours of creation"
|
||||||
|
if row.LastError.Valid {
|
||||||
|
reason += "; last error: " + row.LastError.String
|
||||||
|
}
|
||||||
|
logger.ErrorContext(ctx, "notification failed; giving up", "reason", reason)
|
||||||
|
s.giveUp(ctx, logger, row, 0, reason)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
attempt := row.Attempts + 1
|
attempt := row.Attempts + 1
|
||||||
sendCtx, cancel := context.WithTimeout(ctx, sendTimeout)
|
sendCtx, cancel := context.WithTimeout(ctx, sendTimeout)
|
||||||
err := sender.Send(sendCtx, Message{
|
err = sender.Send(sendCtx, Message{
|
||||||
ID: row.ID, EventKey: row.EventKey, Channel: Channel(row.Channel), Payload: payload, Attempt: attempt,
|
ID: row.ID, EventKey: row.EventKey, Channel: Channel(row.Channel), Payload: payload, Attempt: attempt,
|
||||||
})
|
})
|
||||||
cancel()
|
cancel()
|
||||||
@@ -215,13 +235,6 @@ func (s *Service) deliver(ctx context.Context, row db.NotificationOutbox) {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
created, perr := parseTimestamp(row.CreatedAt)
|
|
||||||
if perr != nil {
|
|
||||||
logger.ErrorContext(ctx, "notification failed", "reason", "created_at does not parse", "error", perr)
|
|
||||||
s.giveUp(ctx, logger, row, 1, err.Error())
|
|
||||||
return
|
|
||||||
}
|
|
||||||
deadline := created.Add(giveUpAfter)
|
|
||||||
if !now.Before(deadline) {
|
if !now.Before(deadline) {
|
||||||
logger.ErrorContext(ctx, "notification failed; giving up", "attempt", attempt, "error", err)
|
logger.ErrorContext(ctx, "notification failed; giving up", "attempt", attempt, "error", err)
|
||||||
s.giveUp(ctx, logger, row, 1, err.Error())
|
s.giveUp(ctx, logger, row, 1, err.Error())
|
||||||
|
|||||||
Reference in New Issue
Block a user