diff --git a/backend/internal/notification/dispatcher_test.go b/backend/internal/notification/dispatcher_test.go index 9a9b0fc06..96a91e388 100644 --- a/backend/internal/notification/dispatcher_test.go +++ b/backend/internal/notification/dispatcher_test.go @@ -17,8 +17,10 @@ type fakeRuntimeStore struct { mu sync.Mutex unrouted []domain.Notification deliveries []DeliveryRow + byID map[string]DeliveryRow routed []domain.NotificationID releases int + retryNext time.Time failEnqueue map[domain.NotificationID]error } @@ -45,6 +47,13 @@ func (f *fakeRuntimeStore) EnqueueDelivery(_ context.Context, row DeliveryRow) ( return row, true, nil } +func (f *fakeRuntimeStore) GetDelivery(_ context.Context, id string) (DeliveryRow, bool, error) { + f.mu.Lock() + defer f.mu.Unlock() + row, ok := f.byID[id] + return row, ok, nil +} + func (f *fakeRuntimeStore) ClaimDueDeliveries(context.Context, string, string, time.Time, int, time.Duration) ([]DeliveryRow, error) { return nil, nil } @@ -57,7 +66,10 @@ func (f *fakeRuntimeStore) ReleaseExpiredDeliveryLeases(context.Context, time.Ti func (f *fakeRuntimeStore) MarkDeliverySent(context.Context, string, string, time.Time) error { return nil } -func (f *fakeRuntimeStore) MarkDeliveryRetry(context.Context, string, string, string, time.Time) error { +func (f *fakeRuntimeStore) MarkDeliveryRetry(_ context.Context, _ string, _ string, _ string, next time.Time) error { + f.mu.Lock() + defer f.mu.Unlock() + f.retryNext = next return nil } func (f *fakeRuntimeStore) MarkDeliveryFailed(context.Context, string, string, string, time.Time) error { @@ -119,6 +131,29 @@ func TestRoutePendingDeliveryFailureDoesNotBlockOtherNotifications(t *testing.T) } } +func TestMarkDeliveryErrorUsesAttemptAwareBackoff(t *testing.T) { + now := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) + cfg := config.DefaultNotificationConfig() + cfg.Retry.BaseDelay = time.Second + cfg.Retry.MaxDelay = time.Minute + store := &fakeRuntimeStore{byID: map[string]DeliveryRow{ + "del_1": {ID: "del_1", Attempts: 2, MaxAttempts: 5}, + }} + mgr := NewManager(store, StaticSettings(cfg), discardLogger()) + mgr.clock = func() time.Time { return now } + + if err := mgr.MarkDeliveryError(context.Background(), "del_1", "timeout", "timed out"); err != nil { + t.Fatal(err) + } + store.mu.Lock() + next := store.retryNext + store.mu.Unlock() + delay := next.Sub(now) + if delay < 3200*time.Millisecond || delay > 4800*time.Millisecond { + t.Fatalf("retry delay for third attempt = %s, want jittered 4s backoff", delay) + } +} + func sampleDomainNotification(id domain.NotificationID, priority string) domain.Notification { return domain.Notification{ Seq: 1, diff --git a/backend/internal/notification/manager.go b/backend/internal/notification/manager.go index ea505ff8c..80f23f7db 100644 --- a/backend/internal/notification/manager.go +++ b/backend/internal/notification/manager.go @@ -3,6 +3,7 @@ package notification import ( "context" "errors" + "fmt" "log/slog" "time" @@ -136,11 +137,21 @@ func (m *Manager) MarkDeliverySent(ctx context.Context, id, externalID string) e func (m *Manager) MarkDeliveryError(ctx context.Context, id, code, message string) error { settings := m.settings.Settings(ctx) policy := RetryPolicyFromConfig(settings.Retry) + now := m.clock().UTC() // The store is the source of truth for attempts/max-attempt terminal // handling. Permanent classification short-circuits to failed; otherwise we - // provide the next retry timestamp for retry_wait rows. + // fetch the current delivery attempts and provide the attempt-aware next + // retry timestamp for retry_wait rows. if ClassifyError(code) == ErrorPermanent { - return m.store.MarkDeliveryFailed(ctx, id, code, message, m.clock().UTC()) + return m.store.MarkDeliveryFailed(ctx, id, code, message, now) } - return m.store.MarkDeliveryRetry(ctx, id, code, message, policy.NextAttemptAt(m.clock().UTC(), 1)) + row, ok, err := m.store.GetDelivery(ctx, id) + if err != nil { + return err + } + if !ok { + return fmt.Errorf("notification delivery %s not found", id) + } + nextAttemptNo := row.Attempts + 1 + return m.store.MarkDeliveryRetry(ctx, id, code, message, policy.NextAttemptAt(now, nextAttemptNo)) } diff --git a/backend/internal/notification/store.go b/backend/internal/notification/store.go index 9be4aef59..b29d1ebc6 100644 --- a/backend/internal/notification/store.go +++ b/backend/internal/notification/store.go @@ -14,6 +14,7 @@ type Store interface { ListUnroutedNotifications(ctx context.Context, limit int) ([]domain.Notification, error) MarkNotificationRouted(ctx context.Context, id domain.NotificationID, at time.Time) error + GetDelivery(ctx context.Context, id string) (DeliveryRow, bool, error) EnqueueDelivery(ctx context.Context, row DeliveryRow) (DeliveryRow, bool, error) ClaimDueDeliveries(ctx context.Context, sink string, owner string, now time.Time, limit int, lease time.Duration) ([]DeliveryRow, error) ReleaseExpiredDeliveryLeases(ctx context.Context, now time.Time) (int, error) diff --git a/backend/internal/storage/sqlite/migrations/0003_notification_deliveries.sql b/backend/internal/storage/sqlite/migrations/0003_notification_deliveries.sql index 5cf486887..e1e8e85e2 100644 --- a/backend/internal/storage/sqlite/migrations/0003_notification_deliveries.sql +++ b/backend/internal/storage/sqlite/migrations/0003_notification_deliveries.sql @@ -1,6 +1,9 @@ -- +goose Up -- +goose StatementBegin ALTER TABLE notifications ADD COLUMN routed_at TIMESTAMP; +CREATE INDEX idx_notifications_unrouted + ON notifications(seq) + WHERE routed_at IS NULL; CREATE TABLE notification_deliveries ( id TEXT PRIMARY KEY, @@ -115,5 +118,6 @@ DROP TRIGGER IF EXISTS notification_deliveries_cdc_update; DROP TRIGGER IF EXISTS notification_deliveries_cdc_insert; DROP TABLE IF EXISTS notification_delivery_attempts; DROP TABLE IF EXISTS notification_deliveries; +DROP INDEX IF EXISTS idx_notifications_unrouted; ALTER TABLE notifications DROP COLUMN routed_at; -- +goose StatementEnd diff --git a/backend/internal/storage/sqlite/notification_delivery_store.go b/backend/internal/storage/sqlite/notification_delivery_store.go index 18593b240..c2b8a62a9 100644 --- a/backend/internal/storage/sqlite/notification_delivery_store.go +++ b/backend/internal/storage/sqlite/notification_delivery_store.go @@ -59,7 +59,7 @@ func (s *Store) EnqueueDelivery(ctx context.Context, row notification.DeliveryRo if now.IsZero() { now = time.Now().UTC() } - row, err := notification.NormalizeDelivery(row, now, 5) + row, err := notification.NormalizeDelivery(row, now, row.MaxAttempts) if err != nil { return notification.DeliveryRow{}, false, err }