From 101498185e15d39faa8b358d88e79dd6c073224a Mon Sep 17 00:00:00 2001 From: dev_agent Date: Fri, 9 Oct 2026 12:27:44 -0400 Subject: [PATCH] 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 --- internal/notify/notify_test.go | 71 ++++++++++++++++++++++++++++------ internal/notify/worker.go | 37 ++++++++++++------ 2 files changed, 85 insertions(+), 23 deletions(-) diff --git a/internal/notify/notify_test.go b/internal/notify/notify_test.go index a453516..16b3470 100644 --- a/internal/notify/notify_test.go +++ b/internal/notify/notify_test.go @@ -458,15 +458,20 @@ func TestGiveUpAfter24Hours(t *testing.T) { } // 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. - const wantAttempts = 30 + // minutes (22 more): 29 in all. The row is due once more at the + // deadline, where it is failed without being sent. + const wantAttempts = 29 r := onlyRow(t, st, key, ChannelPush) 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) } - if len(times) != wantAttempts || !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) + if len(times) != wantAttempts+1 || !times[len(times)-1].Equal(deadline) || !times[len(times)-2].Equal(start.Add(1383*time.Minute)) { + 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. @@ -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() svc, st, c := newTestService(t) 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) } const key = "plan_ready:2026-10-12" if err := svc.Emit(ctx, planReady(key)); err != nil { t.Fatal(err) } - // The service was down for more than a day. - c.set(start.Add(25 * time.Hour)) + c.set(start.Add(24*time.Hour - time.Millisecond)) if _, err := svc.deliverDue(ctx); err != nil { t.Fatal(err) } - if r := onlyRow(t, st, key, ChannelPush); r.Status != "failed" || r.Attempts != 1 || r.LastError.String != "gone" { - t.Fatalf("row = %+v, want failed after 1 attempt", r) + if r := onlyRow(t, st, key, ChannelPush); r.Status != "sent" || r.Attempts != 1 || len(sender.messages()) != 1 { + t.Fatalf("row = %+v after %d deliveries, want sent once", r, len(sender.messages())) } } diff --git a/internal/notify/worker.go b/internal/notify/worker.go index 532d4f8..796e7c2 100644 --- a/internal/notify/worker.go +++ b/internal/notify/worker.go @@ -19,9 +19,10 @@ const ( // further failure doubles it, up to retryMax. retryFirst = time.Minute retryMax = time.Hour - // giveUpAfter is how long after its creation a row is retried. The - // last retry is due at that moment, and a delivery failing then or - // later marks the row failed. + // giveUpAfter is how long after its creation a row may be delivered. + // The last retry is due at that moment; a row claimed then or later is + // marked failed without being sent, and a delivery failing at or + // after it marks the row failed. giveUpAfter = 24 * time.Hour // sendTimeout bounds one Send. 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 // 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 -// 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) { logger := s.logger.With("notification_id", row.ID, "event_key", row.EventKey, "channel", row.Channel) var payload Payload @@ -194,9 +196,27 @@ func (s *Service) deliver(ctx context.Context, row db.NotificationOutbox) { 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 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, }) cancel() @@ -215,13 +235,6 @@ func (s *Service) deliver(ctx context.Context, row db.NotificationOutbox) { 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) { logger.ErrorContext(ctx, "notification failed; giving up", "attempt", attempt, "error", err) s.giveUp(ctx, logger, row, 1, err.Error())