From dd9df8abd59c878abde53bd39057b69a0a5ae9d4 Mon Sep 17 00:00:00 2001 From: Olusegun Ibraheem Date: Fri, 31 Jul 2026 12:38:17 -0600 Subject: [PATCH 1/4] feat(notification): wire NotificationWithBackoff/RetryUntil to real queue retry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Uses Goravel's documented optional queue.Job interface — ShouldRetry(err error, attempt int) (retryable bool, delay time.Duration) confirmed at https://www.goravel.dev/digging-deeper/queues.html#job-retry. Backoff()/RetryUntil() are evaluated once, eagerly, in Manager.dispatchQueued — while the live notification still exists — and carried through the queue boundary as two new dispatchItem fields. DispatchJob wraps a Deliver() failure in a small deliveryError type so ShouldRetry (called by the worker with only (err, attempt), no access to decoded job state) can recover them via errors.As. DispatchJob itself stays fully stateless — no per-execution fields, since it's registered once and potentially shared across concurrent worker goroutines. Without RetryUntil set, NotificationWithBackoff alone would retry indefinitely, since ShouldRetry had no other bound to check. Adds DefaultMaxRetryAttempts (exported var, default 10), applied only when RetryUntil isn't set — RetryUntil already provides its own bound and takes precedence when both are present. Known limitation: supports a single fixed backoff per notification+channel, not a growing per-attempt schedule — there's no live notification left to call a second time for a bigger number by retry time. --- errors/list.go | 14 +++--- notification/job.go | 63 +++++++++++++++++++++-- notification/job_test.go | 83 +++++++++++++++++++++++++++++++ notification/notification.go | 13 +++++ notification/notification_test.go | 48 ++++++++++++++++++ 5 files changed, 210 insertions(+), 11 deletions(-) diff --git a/errors/list.go b/errors/list.go index 926a767a7..4822c99e9 100644 --- a/errors/list.go +++ b/errors/list.go @@ -187,14 +187,14 @@ var ( NotificationChannelNotQueueable = New("notification channel %q does not support queued dispatch (does not implement ResolvableChannel)").SetModule(ModuleNotification) NotificationInvalidQueuePayload = New("notification queue payload is missing or malformed").SetModule(ModuleNotification) NotificationMailEmptyRoute = New("mail channel: %T.RouteNotificationFor(\"mail\") returned empty address").SetModule(ModuleNotification) - NotificationMailMarshalPayloadFailed = New("mail channel: failed to marshal payload for %T: %w").SetModule(ModuleNotification) - NotificationMailUnmarshalPayloadFailed = New("mail channel: failed to unmarshal payload: %w").SetModule(ModuleNotification) - NotificationMailSendFailed = New("mail channel: failed to send: %w").SetModule(ModuleNotification) + NotificationMailMarshalPayloadFailed = New("mail channel: failed to marshal payload for %T: %v").SetModule(ModuleNotification) + NotificationMailUnmarshalPayloadFailed = New("mail channel: failed to unmarshal payload: %v").SetModule(ModuleNotification) + NotificationMailSendFailed = New("mail channel: failed to send: %v").SetModule(ModuleNotification) NotificationDatabaseEmptyRoute = New("database channel: %T.RouteNotificationFor(\"database\") returned empty ID").SetModule(ModuleNotification) - NotificationDatabaseMarshalDataFailed = New("database channel: failed to marshal payload for %T: %w").SetModule(ModuleNotification) - NotificationDatabaseMarshalRecordFailed = New("database channel: failed to marshal record: %w").SetModule(ModuleNotification) - NotificationDatabaseUnmarshalRecordFailed = New("database channel: failed to unmarshal record: %w").SetModule(ModuleNotification) - NotificationDatabaseInsertFailed = New("database channel: failed to insert notification record: %w").SetModule(ModuleNotification) + NotificationDatabaseMarshalDataFailed = New("database channel: failed to marshal payload for %T: %v").SetModule(ModuleNotification) + NotificationDatabaseMarshalRecordFailed = New("database channel: failed to marshal record: %v").SetModule(ModuleNotification) + NotificationDatabaseUnmarshalRecordFailed = New("database channel: failed to unmarshal record: %v").SetModule(ModuleNotification) + NotificationDatabaseInsertFailed = New("database channel: failed to insert notification record: %v").SetModule(ModuleNotification) NotificationTableRequiresBootstrapSetup = New("notifications:table auto-registration requires the bootstrap setup (see env.IsBootstrapSetup); register the migration manually").SetModule(ModuleNotification) OrmDriverNotSupported = New("invalid driver: %s, only support mysql, postgres, sqlite and sqlserver") diff --git a/notification/job.go b/notification/job.go index 68cbeb628..88bb666a8 100644 --- a/notification/job.go +++ b/notification/job.go @@ -2,6 +2,7 @@ package notification import ( "encoding/json" + "time" "github.com/goravel/framework/contracts/notification" "github.com/goravel/framework/errors" @@ -9,9 +10,11 @@ import ( // dispatchItem is the plain, JSON-serializable unit queued per channel. type dispatchItem struct { - Channel string `json:"channel"` - Route string `json:"route"` - Payload []byte `json:"payload"` + Channel string `json:"channel"` + Route string `json:"route"` + Payload []byte `json:"payload"` + BackoffSeconds int `json:"backoff_seconds,omitempty"` + RetryUntilUnix int64 `json:"retry_until_unix,omitempty"` } func encodeDispatchItem(item dispatchItem) (string, error) { @@ -22,6 +25,19 @@ func encodeDispatchItem(item dispatchItem) (string, error) { return string(b), nil } +var DefaultMaxRetryAttempts = 10 + +type deliveryError struct { + err error + hasBackoff bool + backoff time.Duration + hasRetryUntil bool + retryUntil time.Time +} + +func (e *deliveryError) Error() string { return e.err.Error() } +func (e *deliveryError) Unwrap() error { return e.err } + type DispatchJob struct { manager *Manager } @@ -59,5 +75,44 @@ func (j *DispatchJob) Handle(args ...any) error { return errors.NotificationChannelNotQueueable.Args(item.Channel) } - return resolvable.Deliver(item.Route, item.Payload) + err := resolvable.Deliver(item.Route, item.Payload) + if err == nil { + return nil + } + + if item.BackoffSeconds == 0 && item.RetryUntilUnix == 0 { + return err + } + + wrapped := &deliveryError{err: err} + if item.BackoffSeconds > 0 { + wrapped.hasBackoff = true + wrapped.backoff = time.Duration(item.BackoffSeconds) * time.Second + } + if item.RetryUntilUnix > 0 { + wrapped.hasRetryUntil = true + wrapped.retryUntil = time.Unix(item.RetryUntilUnix, 0) + } + return wrapped +} + +func (j *DispatchJob) ShouldRetry(err error, attempt int) (bool, time.Duration) { + var de *deliveryError + if !errors.As(err, &de) { + return false, 0 + } + + if de.hasRetryUntil { + if time.Now().After(de.retryUntil) { + return false, 0 + } + } else if attempt >= DefaultMaxRetryAttempts { + return false, 0 + } + + if de.hasBackoff { + return true, de.backoff + } + + return false, 0 } diff --git a/notification/job_test.go b/notification/job_test.go index e3a3442f3..8cc990cea 100644 --- a/notification/job_test.go +++ b/notification/job_test.go @@ -3,6 +3,7 @@ package notification import ( "errors" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" @@ -104,3 +105,85 @@ func TestDispatchJob_Handle_PropagatesDeliverError(t *testing.T) { assert.Error(t, err) assert.Contains(t, err.Error(), "smtp down") } + +func TestDispatchJob_ShouldRetry_NoRetry_WhenPlainError(t *testing.T) { + logger := mockslog.NewLog(t) + mgr := NewManager(logger, nil) + mgr.Extend(&fakeResolvableChannel{name: "a", deliverErr: errors.New("smtp down")}) + job := NewDispatchJob(mgr) + + encoded, err := encodeDispatchItem(dispatchItem{Channel: "a", Route: "r", Payload: []byte("{}")}) + assert.NoError(t, err) + + handleErr := job.Handle(encoded) + assert.Error(t, handleErr) + + retryable, delay := job.ShouldRetry(handleErr, 1) + assert.False(t, retryable) + assert.Zero(t, delay) +} + +func TestDispatchJob_ShouldRetry_RetriesWithBackoff(t *testing.T) { + logger := mockslog.NewLog(t) + mgr := NewManager(logger, nil) + mgr.Extend(&fakeResolvableChannel{name: "a", deliverErr: errors.New("smtp down")}) + job := NewDispatchJob(mgr) + + encoded, err := encodeDispatchItem(dispatchItem{ + Channel: "a", Route: "r", Payload: []byte("{}"), + BackoffSeconds: 45, + }) + assert.NoError(t, err) + + handleErr := job.Handle(encoded) + assert.Error(t, handleErr) + + retryable, delay := job.ShouldRetry(handleErr, 1) + assert.True(t, retryable) + assert.Equal(t, 45*time.Second, delay) +} + +func TestDispatchJob_ShouldRetry_StopsAfterRetryUntilDeadline(t *testing.T) { + logger := mockslog.NewLog(t) + mgr := NewManager(logger, nil) + mgr.Extend(&fakeResolvableChannel{name: "a", deliverErr: errors.New("smtp down")}) + job := NewDispatchJob(mgr) + + past := time.Now().Add(-1 * time.Hour).Unix() + encoded, err := encodeDispatchItem(dispatchItem{ + Channel: "a", Route: "r", Payload: []byte("{}"), + BackoffSeconds: 30, RetryUntilUnix: past, + }) + assert.NoError(t, err) + + handleErr := job.Handle(encoded) + assert.Error(t, handleErr) + + retryable, delay := job.ShouldRetry(handleErr, 5) + assert.False(t, retryable, "past RetryUntil should stop retries even though Backoff is set") + assert.Zero(t, delay) +} + +func TestDispatchJob_ShouldRetry_CapsAttempts_WhenNoRetryUntilSet(t *testing.T) { + logger := mockslog.NewLog(t) + mgr := NewManager(logger, nil) + mgr.Extend(&fakeResolvableChannel{name: "a", deliverErr: errors.New("smtp down")}) + job := NewDispatchJob(mgr) + + encoded, err := encodeDispatchItem(dispatchItem{ + Channel: "a", Route: "r", Payload: []byte("{}"), + BackoffSeconds: 5, // RetryUntilUnix deliberately left unset + }) + assert.NoError(t, err) + + handleErr := job.Handle(encoded) + assert.Error(t, handleErr) + + retryable, delay := job.ShouldRetry(handleErr, DefaultMaxRetryAttempts-1) + assert.True(t, retryable, "still below the cap, should retry") + assert.Equal(t, 5*time.Second, delay) + + retryable, delay = job.ShouldRetry(handleErr, DefaultMaxRetryAttempts) + assert.False(t, retryable, "at the cap, should stop even though Backoff is set") + assert.Zero(t, delay) +} diff --git a/notification/notification.go b/notification/notification.go index 39217e55f..373da2caa 100644 --- a/notification/notification.go +++ b/notification/notification.go @@ -114,6 +114,8 @@ func (m *Manager) dispatchQueued( sq contractsnotification.ShouldQueue, ) error { shouldSend, _ := n.(contractsnotification.NotificationWithShouldSend) + withBackoff, _ := n.(contractsnotification.NotificationWithBackoff) + withRetryUntil, _ := n.(contractsnotification.NotificationWithRetryUntil) var errs []error for _, name := range n.Via(notifiable) { @@ -143,6 +145,17 @@ func (m *Manager) dispatchQueued( } item := dispatchItem{Channel: name, Route: route, Payload: payload} + // Captured now, while n is still live — DispatchJob.ShouldRetry + // can't call these itself later, see job.go. + if withBackoff != nil { + item.BackoffSeconds = withBackoff.Backoff(name) + } + if withRetryUntil != nil { + if ru := withRetryUntil.RetryUntil(); !ru.IsZero() { + item.RetryUntilUnix = ru.Unix() + } + } + encoded, err := encodeDispatchItem(item) if err != nil { errs = append(errs, err) diff --git a/notification/notification_test.go b/notification/notification_test.go index 3b43922c4..c9a2b3abf 100644 --- a/notification/notification_test.go +++ b/notification/notification_test.go @@ -1,9 +1,11 @@ package notification import ( + "encoding/json" "errors" "strings" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" @@ -535,3 +537,49 @@ func TestManager_Send_QueuedNotification_PassesConnectionAndQueue(t *testing.T) err := mgr.Send(queueTestNotifiable{}, &queueableNotificationWithRouting{}) assert.NoError(t, err) } + +type backoffNotification struct { + channels []string + backoff int + retryUntil time.Time +} + +func (n *backoffNotification) Via(_ contractsnotification.Notifiable) []string { return n.channels } +func (n *backoffNotification) OnQueue() string { return "" } +func (n *backoffNotification) OnConnection() string { return "" } +func (n *backoffNotification) Backoff(_ string) int { return n.backoff } +func (n *backoffNotification) RetryUntil() time.Time { return n.retryUntil } + +func TestManager_Send_QueuedNotification_CapturesBackoffAndRetryUntil(t *testing.T) { + logger := mockslog.NewLog(t) + q := mocksqueue.NewQueue(t) + pending := mocksqueue.NewPendingJob(t) + + var captured []contractsqueue.Arg + q.EXPECT(). + Job(mock.AnythingOfType("*notification.DispatchJob"), mock.Anything). + Run(func(_ contractsqueue.Job, args ...[]contractsqueue.Arg) { + if len(args) > 0 { + captured = args[0] + } + }). + Return(pending).Once() + pending.EXPECT().Dispatch().Return(nil).Once() + + mgr := NewManager(logger, q) + mgr.Extend(&fakeResolvableChannel{name: "a"}) + + retryUntil := time.Now().Add(2 * time.Hour).Truncate(time.Second) + n := &backoffNotification{channels: []string{"a"}, backoff: 30, retryUntil: retryUntil} + + err := mgr.Send(&fakeNotifiable{}, n) + assert.NoError(t, err) + assert.NotEmpty(t, captured) + + var item dispatchItem + raw, ok := captured[0].Value.(string) + assert.True(t, ok) + assert.NoError(t, json.Unmarshal([]byte(raw), &item)) + assert.Equal(t, 30, item.BackoffSeconds) + assert.Equal(t, retryUntil.Unix(), item.RetryUntilUnix) +} From aeadda8fa61f12cc67863b50aa781c659a06e33d Mon Sep 17 00:00:00 2001 From: Olusegun Ibraheem Date: Sun, 9 Aug 2026 12:49:51 -0600 Subject: [PATCH 2/4] wip: align notification backoff/retry with broadcasting pattern --- contracts/notification/notification.go | 41 +++-- notification/job.go | 133 ++++++++++------ notification/job_test.go | 201 +++++++++++++++---------- notification/notification_test.go | 146 ++++++++++++------ 4 files changed, 335 insertions(+), 186 deletions(-) diff --git a/contracts/notification/notification.go b/contracts/notification/notification.go index 06aa75375..728a39801 100644 --- a/contracts/notification/notification.go +++ b/contracts/notification/notification.go @@ -25,21 +25,36 @@ type NotificationWithAfterSending interface { AfterSending(notifiable Notifiable, channel string) error } -// NOT CURRENTLY WIRED — implementing this on a notification has no -// effect today. -type NotificationWithBackoff interface { +// NotificationWithTries is an optional extension for queued +// notifications that want to cap retry attempts on delivery failure. +// Mirrors contracts/broadcasting.ShouldBroadcastWithTries exactly, for +// consistency across the two queued-retry mechanisms in this codebase. +type NotificationWithTries interface { Notification - // Backoff returns the number of seconds to wait before retrying - // after channel is the channel that failed. - Backoff(channel string) int -} - -// NOT CURRENTLY WIRED — same root cause as NotificationWithBackoff: -type NotificationWithRetryUntil interface { + // Tries returns the maximum number of attempts for the given + // channel. 0 / not implementing this interface means the + // notification is single-shot on that channel — no retry at all, + // not even once, matching ShouldBroadcastWithTries's semantics. + Tries(channel string) int +} + +// NotificationWithBackoff is an optional extension for queued +// notifications that want to control the delay before each retry +// attempt. Mirrors contracts/broadcasting.ShouldBroadcastWithBackoff +// exactly: Backoff(channel) is called once, while the notification is +// still live, and returns the FULL per-attempt schedule up front — not +// re-invoked per retry. ShouldRetry indexes into the captured slice by +// attempt number, and the last value repeats for any attempt beyond the +// slice's length (min(attempt-1, len(backoff)-1)), matching Laravel's +// Worker::calculateBackoff and BroadcastJob.ShouldRetry precisely. +// +// Only takes effect together with NotificationWithTries; without Tries +// the notification is single-shot regardless of Backoff. +type NotificationWithBackoff interface { Notification - // RetryUntil returns the time after which delivery attempts should - // stop being retried. - RetryUntil() time.Time + // Backoff returns the delay before each retry attempt on channel, + // in order; the last value repeats for subsequent attempts. + Backoff(channel string) []time.Duration } type Notifiable interface { diff --git a/notification/job.go b/notification/job.go index 88bb666a8..0e08d3dcf 100644 --- a/notification/job.go +++ b/notification/job.go @@ -2,6 +2,7 @@ package notification import ( "encoding/json" + "sync" "time" "github.com/goravel/framework/contracts/notification" @@ -9,12 +10,15 @@ import ( ) // dispatchItem is the plain, JSON-serializable unit queued per channel. +// Tries/Backoff mirror broadcasting's broadcastItem exactly (same field +// names, same wire format — Backoff in milliseconds, not seconds) for +// consistency between the two queued-retry mechanisms in this codebase. type dispatchItem struct { - Channel string `json:"channel"` - Route string `json:"route"` - Payload []byte `json:"payload"` - BackoffSeconds int `json:"backoff_seconds,omitempty"` - RetryUntilUnix int64 `json:"retry_until_unix,omitempty"` + Channel string `json:"channel"` + Route string `json:"route"` + Payload []byte `json:"payload"` + Tries int `json:"tries,omitempty"` + Backoff []int64 `json:"backoff,omitempty"` // per-attempt delay in ms } func encodeDispatchItem(item dispatchItem) (string, error) { @@ -25,21 +29,38 @@ func encodeDispatchItem(item dispatchItem) (string, error) { return string(b), nil } -var DefaultMaxRetryAttempts = 10 - -type deliveryError struct { - err error - hasBackoff bool - backoff time.Duration - hasRetryUntil bool - retryUntil time.Time -} - -func (e *deliveryError) Error() string { return e.err.Error() } -func (e *deliveryError) Unwrap() error { return e.err } - +// DispatchJob delivers one resolved channel item. It's registered once +// with the queue at Boot() (see service_provider.go) rather than +// constructed per-dispatch, since persisting queue drivers (database, +// Redis) look up a registered Job by Signature() and call Handle() on a +// freshly constructed instance with the dispatch-time []queue.Arg. That's +// why Manager.dispatchQueued resolves each channel's payload eagerly via +// ResolvableChannel.Resolve — while notifiable/notification are still +// live — and queues only the resulting plain (channel, route, payload, +// tries, backoff). +// +// Retry state (item) is a shared, mutex-guarded field rather than +// carried in the error, matching broadcasting.BroadcastJob's exact +// pattern rather than an independently-derived design — see that type's +// own doc comment for the full reasoning, copied here: +// +// item is the payload of the task being processed, set by Handle and +// read by ShouldRetry. DispatchJob is a shared singleton, so access is +// guarded by mu. +// +// Limitation: the mutex guarantees memory safety, not logical +// isolation. ShouldRetry has no access to task args by contract, so a +// concurrent failed task can still overwrite this payload between +// another task's Handle returning an error and its ShouldRetry call. +// Clearing item on Handle's success path converts the common +// interleaving case into the safe single-shot fallback; a full fix +// requires per-task state passed by the queue worker, which is out of +// scope. type DispatchJob struct { manager *Manager + + mu sync.Mutex + item *dispatchItem } func NewDispatchJob(manager *Manager) *DispatchJob { @@ -52,19 +73,32 @@ func (j *DispatchJob) Signature() string { func (j *DispatchJob) Handle(args ...any) error { if len(args) != 1 { + j.mu.Lock() + j.item = nil + j.mu.Unlock() return errors.NotificationInvalidQueuePayload } raw, ok := args[0].(string) if !ok { + j.mu.Lock() + j.item = nil + j.mu.Unlock() return errors.NotificationInvalidQueuePayload } var item dispatchItem if err := json.Unmarshal([]byte(raw), &item); err != nil { + j.mu.Lock() + j.item = nil + j.mu.Unlock() return errors.NotificationInvalidQueuePayload } + j.mu.Lock() + j.item = &item + j.mu.Unlock() + ch := j.manager.Channel(item.Channel) if ch == nil { return errors.NotificationChannelNotFound.Args(item.Channel) @@ -75,44 +109,49 @@ func (j *DispatchJob) Handle(args ...any) error { return errors.NotificationChannelNotQueueable.Args(item.Channel) } - err := resolvable.Deliver(item.Route, item.Payload) - if err == nil { - return nil - } - - if item.BackoffSeconds == 0 && item.RetryUntilUnix == 0 { + if err := resolvable.Deliver(item.Route, item.Payload); err != nil { return err } - wrapped := &deliveryError{err: err} - if item.BackoffSeconds > 0 { - wrapped.hasBackoff = true - wrapped.backoff = time.Duration(item.BackoffSeconds) * time.Second - } - if item.RetryUntilUnix > 0 { - wrapped.hasRetryUntil = true - wrapped.retryUntil = time.Unix(item.RetryUntilUnix, 0) - } - return wrapped + // A successful task is never consulted by ShouldRetry, so release + // the payload: an interleaving concurrent failed task then reads + // the safe single-shot fallback (item == nil) instead of a wrong + // retry policy. + j.mu.Lock() + j.item = nil + j.mu.Unlock() + + return nil } -func (j *DispatchJob) ShouldRetry(err error, attempt int) (bool, time.Duration) { - var de *deliveryError - if !errors.As(err, &de) { +// ShouldRetry implements the optional queue.Job retry-control interface +// documented at https://www.goravel.dev/digging-deeper/queues.html#job-retry. +// Logic mirrors broadcasting.BroadcastJob.ShouldRetry exactly: without +// Tries the notification is single-shot regardless of the worker's own +// tries config; with Tries it retries while attempt < Tries using the +// configured per-attempt Backoff (last value repeats). +func (j *DispatchJob) ShouldRetry(err error, attempt int) (retryable bool, delay time.Duration) { + j.mu.Lock() + item := j.item + j.mu.Unlock() + + if item == nil || item.Tries <= 0 { return false, 0 } - - if de.hasRetryUntil { - if time.Now().After(de.retryUntil) { - return false, 0 - } - } else if attempt >= DefaultMaxRetryAttempts { + if attempt < 1 { + // Defensive: attempts come from the pop-incremented reservation + // (or a chain counter starting at 1), so this is unreachable in + // practice. Returning false avoids an accidental infinite + // retry loop. return false, 0 } - - if de.hasBackoff { - return true, de.backoff + if attempt >= item.Tries { + return false, 0 + } + if len(item.Backoff) == 0 { + return true, 0 } - return false, 0 + idx := min(attempt-1, len(item.Backoff)-1) + return true, time.Duration(item.Backoff[idx]) * time.Millisecond } diff --git a/notification/job_test.go b/notification/job_test.go index 8cc990cea..fc51f04a5 100644 --- a/notification/job_test.go +++ b/notification/job_test.go @@ -1,6 +1,7 @@ package notification import ( + "encoding/json" "errors" "testing" "time" @@ -106,84 +107,124 @@ func TestDispatchJob_Handle_PropagatesDeliverError(t *testing.T) { assert.Contains(t, err.Error(), "smtp down") } -func TestDispatchJob_ShouldRetry_NoRetry_WhenPlainError(t *testing.T) { - logger := mockslog.NewLog(t) - mgr := NewManager(logger, nil) - mgr.Extend(&fakeResolvableChannel{name: "a", deliverErr: errors.New("smtp down")}) - job := NewDispatchJob(mgr) - - encoded, err := encodeDispatchItem(dispatchItem{Channel: "a", Route: "r", Payload: []byte("{}")}) - assert.NoError(t, err) - - handleErr := job.Handle(encoded) - assert.Error(t, handleErr) - - retryable, delay := job.ShouldRetry(handleErr, 1) - assert.False(t, retryable) - assert.Zero(t, delay) -} - -func TestDispatchJob_ShouldRetry_RetriesWithBackoff(t *testing.T) { - logger := mockslog.NewLog(t) - mgr := NewManager(logger, nil) - mgr.Extend(&fakeResolvableChannel{name: "a", deliverErr: errors.New("smtp down")}) - job := NewDispatchJob(mgr) - - encoded, err := encodeDispatchItem(dispatchItem{ - Channel: "a", Route: "r", Payload: []byte("{}"), - BackoffSeconds: 45, - }) - assert.NoError(t, err) - - handleErr := job.Handle(encoded) - assert.Error(t, handleErr) - - retryable, delay := job.ShouldRetry(handleErr, 1) - assert.True(t, retryable) - assert.Equal(t, 45*time.Second, delay) -} - -func TestDispatchJob_ShouldRetry_StopsAfterRetryUntilDeadline(t *testing.T) { - logger := mockslog.NewLog(t) - mgr := NewManager(logger, nil) - mgr.Extend(&fakeResolvableChannel{name: "a", deliverErr: errors.New("smtp down")}) - job := NewDispatchJob(mgr) - - past := time.Now().Add(-1 * time.Hour).Unix() - encoded, err := encodeDispatchItem(dispatchItem{ - Channel: "a", Route: "r", Payload: []byte("{}"), - BackoffSeconds: 30, RetryUntilUnix: past, - }) - assert.NoError(t, err) - - handleErr := job.Handle(encoded) - assert.Error(t, handleErr) - - retryable, delay := job.ShouldRetry(handleErr, 5) - assert.False(t, retryable, "past RetryUntil should stop retries even though Backoff is set") - assert.Zero(t, delay) -} - -func TestDispatchJob_ShouldRetry_CapsAttempts_WhenNoRetryUntilSet(t *testing.T) { - logger := mockslog.NewLog(t) - mgr := NewManager(logger, nil) - mgr.Extend(&fakeResolvableChannel{name: "a", deliverErr: errors.New("smtp down")}) - job := NewDispatchJob(mgr) - - encoded, err := encodeDispatchItem(dispatchItem{ - Channel: "a", Route: "r", Payload: []byte("{}"), - BackoffSeconds: 5, // RetryUntilUnix deliberately left unset - }) - assert.NoError(t, err) - - handleErr := job.Handle(encoded) - assert.Error(t, handleErr) - - retryable, delay := job.ShouldRetry(handleErr, DefaultMaxRetryAttempts-1) - assert.True(t, retryable, "still below the cap, should retry") - assert.Equal(t, 5*time.Second, delay) - - retryable, delay = job.ShouldRetry(handleErr, DefaultMaxRetryAttempts) - assert.False(t, retryable, "at the cap, should stop even though Backoff is set") - assert.Zero(t, delay) +// TestDispatchJob_ShouldRetry mirrors broadcasting/job_test.go's +// TestBroadcastJob_ShouldRetry exactly — same case names, same edge +// cases — since DispatchJob.ShouldRetry now uses the identical +// Tries/Backoff/mutex-guarded-item design as BroadcastJob, adopted for +// consistency across the two queued-retry mechanisms in this codebase. +func TestDispatchJob_ShouldRetry(t *testing.T) { + marshalItem := func(tries int, backoff []int64) string { + item := dispatchItem{ + Channel: "a", + Route: "r", + Payload: []byte("{}"), + Tries: tries, + Backoff: backoff, + } + data, _ := json.Marshal(item) + return string(data) + } + + newJob := func(t *testing.T) *DispatchJob { + logger := mockslog.NewLog(t) + mgr := NewManager(logger, nil) + mgr.Extend(&fakeResolvableChannel{name: "a", deliverErr: errors.New("smtp down")}) + return NewDispatchJob(mgr) + } + + tests := []struct { + name string + payload string + attempt int + err error + want bool + wantD time.Duration + }{ + { + name: "tries zero is single-shot", + payload: marshalItem(0, nil), + attempt: 1, + want: false, + wantD: 0, + }, + { + name: "tries 3 first attempt retries", + payload: marshalItem(3, nil), + attempt: 1, + want: true, + wantD: 0, + }, + { + name: "tries 3 second attempt retries", + payload: marshalItem(3, nil), + attempt: 2, + want: true, + wantD: 0, + }, + { + name: "tries 3 third attempt stops", + payload: marshalItem(3, nil), + attempt: 3, + want: false, + wantD: 0, + }, + { + name: "backoff first attempt", + payload: marshalItem(4, []int64{1000, 2000}), + attempt: 1, + err: errors.New("test error"), // err is ignored by ShouldRetry + want: true, + wantD: 1 * time.Second, + }, + { + name: "backoff second attempt", + payload: marshalItem(4, []int64{1000, 2000}), + attempt: 2, + want: true, + wantD: 2 * time.Second, + }, + { + name: "backoff last value repeats", + payload: marshalItem(4, []int64{1000, 2000}), + attempt: 3, + want: true, + wantD: 2 * time.Second, + }, + { + name: "backoff with attempt 0 is single-shot fallback", + payload: marshalItem(4, []int64{1000, 2000}), + attempt: 0, + want: false, + wantD: 0, + }, + { + name: "backoff stop at final attempt before index", + payload: marshalItem(2, []int64{1000, 2000}), + attempt: 2, + want: false, + wantD: 0, + }, + { + name: "backoff last attempt stops", + payload: marshalItem(4, []int64{1000, 2000}), + attempt: 4, + want: false, + wantD: 0, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + job := newJob(t) + // Handle fails via fakeResolvableChannel's deliverErr, which + // retains the parsed item for ShouldRetry to read — the + // realistic pre-ShouldRetry state, mirroring + // broadcasting/job_test.go's own setup. + assert.Error(t, job.Handle(tt.payload)) + + retryable, delay := job.ShouldRetry(tt.err, tt.attempt) + assert.Equal(t, tt.want, retryable) + assert.Equal(t, tt.wantD, delay) + }) + } } diff --git a/notification/notification_test.go b/notification/notification_test.go index c9a2b3abf..8bb74cbe3 100644 --- a/notification/notification_test.go +++ b/notification/notification_test.go @@ -164,6 +164,23 @@ type queueableNotificationWithRouting struct{ queueableNotification } func (n *queueableNotificationWithRouting) OnQueue() string { return "notifications" } func (n *queueableNotificationWithRouting) OnConnection() string { return "redis" } +// triesBackoffNotification implements NotificationWithTries + +// NotificationWithBackoff, for exercising dispatchQueued's eager +// Tries/Backoff capture. +type triesBackoffNotification struct { + channels []string + tries int + backoff []time.Duration +} + +func (n *triesBackoffNotification) Via(_ contractsnotification.Notifiable) []string { + return n.channels +} +func (n *triesBackoffNotification) OnQueue() string { return "" } +func (n *triesBackoffNotification) OnConnection() string { return "" } +func (n *triesBackoffNotification) Tries(_ string) int { return n.tries } +func (n *triesBackoffNotification) Backoff(_ string) []time.Duration { return n.backoff } + // ---- Manager: SendNow / dispatchSync ---- func TestManager_SendNow_CallsCorrectChannels(t *testing.T) { @@ -405,6 +422,89 @@ func TestManager_Send_QueuedNotification_SkipsChannel_WhenShouldSendReturnsFalse assert.NoError(t, err) } +// TestManager_Send_QueuedNotification_CapturesTriesAndBackoff mirrors +// broadcasting/application.go's Dispatch capture logic exactly: Tries +// and Backoff (converted to milliseconds on the wire) are captured +// eagerly, while n is still live, since DispatchJob.ShouldRetry can't +// call these itself later — see job.go. +func TestManager_Send_QueuedNotification_CapturesTriesAndBackoff(t *testing.T) { + logger := mockslog.NewLog(t) + q := mocksqueue.NewQueue(t) + pending := mocksqueue.NewPendingJob(t) + + var captured []contractsqueue.Arg + q.EXPECT(). + Job(mock.AnythingOfType("*notification.DispatchJob"), mock.Anything). + Run(func(_ contractsqueue.Job, args ...[]contractsqueue.Arg) { + if len(args) > 0 { + captured = args[0] + } + }). + Return(pending).Once() + pending.EXPECT().Dispatch().Return(nil).Once() + + mgr := NewManager(logger, q) + mgr.Extend(&fakeResolvableChannel{name: "a"}) + + n := &triesBackoffNotification{ + channels: []string{"a"}, + tries: 4, + backoff: []time.Duration{1 * time.Second, 2 * time.Second}, + } + + err := mgr.Send(&fakeNotifiable{}, n) + assert.NoError(t, err) + assert.NotEmpty(t, captured) + + var item dispatchItem + raw, ok := captured[0].Value.(string) + assert.True(t, ok) + assert.NoError(t, json.Unmarshal([]byte(raw), &item)) + assert.Equal(t, 4, item.Tries) + assert.Equal(t, []int64{1000, 2000}, item.Backoff) // milliseconds on the wire +} + +// TestManager_Send_QueuedNotification_OmitsBackoff_WhenTriesNotSet +// confirms Backoff only takes effect alongside a positive Tries, +// matching broadcasting's exact rule (Backoff has no effect without +// Tries, so there's no reason to carry it across the queue boundary +// otherwise). +func TestManager_Send_QueuedNotification_OmitsBackoff_WhenTriesNotSet(t *testing.T) { + logger := mockslog.NewLog(t) + q := mocksqueue.NewQueue(t) + pending := mocksqueue.NewPendingJob(t) + + var captured []contractsqueue.Arg + q.EXPECT(). + Job(mock.AnythingOfType("*notification.DispatchJob"), mock.Anything). + Run(func(_ contractsqueue.Job, args ...[]contractsqueue.Arg) { + if len(args) > 0 { + captured = args[0] + } + }). + Return(pending).Once() + pending.EXPECT().Dispatch().Return(nil).Once() + + mgr := NewManager(logger, q) + mgr.Extend(&fakeResolvableChannel{name: "a"}) + + n := &triesBackoffNotification{ + channels: []string{"a"}, + tries: 0, + backoff: []time.Duration{5 * time.Second}, + } + + err := mgr.Send(&fakeNotifiable{}, n) + assert.NoError(t, err) + + var item dispatchItem + raw, ok := captured[0].Value.(string) + assert.True(t, ok) + assert.NoError(t, json.Unmarshal([]byte(raw), &item)) + assert.Zero(t, item.Tries) + assert.Empty(t, item.Backoff) +} + // ---- Route (on-demand notifications) ---- func TestManager_Route_RouteNotificationForReturnsConfiguredAddress(t *testing.T) { @@ -537,49 +637,3 @@ func TestManager_Send_QueuedNotification_PassesConnectionAndQueue(t *testing.T) err := mgr.Send(queueTestNotifiable{}, &queueableNotificationWithRouting{}) assert.NoError(t, err) } - -type backoffNotification struct { - channels []string - backoff int - retryUntil time.Time -} - -func (n *backoffNotification) Via(_ contractsnotification.Notifiable) []string { return n.channels } -func (n *backoffNotification) OnQueue() string { return "" } -func (n *backoffNotification) OnConnection() string { return "" } -func (n *backoffNotification) Backoff(_ string) int { return n.backoff } -func (n *backoffNotification) RetryUntil() time.Time { return n.retryUntil } - -func TestManager_Send_QueuedNotification_CapturesBackoffAndRetryUntil(t *testing.T) { - logger := mockslog.NewLog(t) - q := mocksqueue.NewQueue(t) - pending := mocksqueue.NewPendingJob(t) - - var captured []contractsqueue.Arg - q.EXPECT(). - Job(mock.AnythingOfType("*notification.DispatchJob"), mock.Anything). - Run(func(_ contractsqueue.Job, args ...[]contractsqueue.Arg) { - if len(args) > 0 { - captured = args[0] - } - }). - Return(pending).Once() - pending.EXPECT().Dispatch().Return(nil).Once() - - mgr := NewManager(logger, q) - mgr.Extend(&fakeResolvableChannel{name: "a"}) - - retryUntil := time.Now().Add(2 * time.Hour).Truncate(time.Second) - n := &backoffNotification{channels: []string{"a"}, backoff: 30, retryUntil: retryUntil} - - err := mgr.Send(&fakeNotifiable{}, n) - assert.NoError(t, err) - assert.NotEmpty(t, captured) - - var item dispatchItem - raw, ok := captured[0].Value.(string) - assert.True(t, ok) - assert.NoError(t, json.Unmarshal([]byte(raw), &item)) - assert.Equal(t, 30, item.BackoffSeconds) - assert.Equal(t, retryUntil.Unix(), item.RetryUntilUnix) -} From a883da592486b01df38a49244bb774e27a96710a Mon Sep 17 00:00:00 2001 From: Olusegun Ibraheem Date: Sun, 9 Aug 2026 12:53:58 -0600 Subject: [PATCH 3/4] fix(notification): apply Tries/Backoff alignment with broadcasting --- notification/notification.go | 24 ++++++++++++++++-------- 1 file changed, 16 insertions(+), 8 deletions(-) diff --git a/notification/notification.go b/notification/notification.go index 373da2caa..3dbaea4f5 100644 --- a/notification/notification.go +++ b/notification/notification.go @@ -114,8 +114,6 @@ func (m *Manager) dispatchQueued( sq contractsnotification.ShouldQueue, ) error { shouldSend, _ := n.(contractsnotification.NotificationWithShouldSend) - withBackoff, _ := n.(contractsnotification.NotificationWithBackoff) - withRetryUntil, _ := n.(contractsnotification.NotificationWithRetryUntil) var errs []error for _, name := range n.Via(notifiable) { @@ -146,13 +144,23 @@ func (m *Manager) dispatchQueued( item := dispatchItem{Channel: name, Route: route, Payload: payload} // Captured now, while n is still live — DispatchJob.ShouldRetry - // can't call these itself later, see job.go. - if withBackoff != nil { - item.BackoffSeconds = withBackoff.Backoff(name) + // can't call these itself later, see job.go. Mirrors + // broadcasting/application.go's Dispatch capture logic exactly, + // including only serializing Backoff alongside a positive Tries + // (Backoff has no effect without Tries, so there's no reason to + // carry it across the queue boundary otherwise). + if withTries, ok := n.(contractsnotification.NotificationWithTries); ok && withTries.Tries(name) > 0 { + item.Tries = withTries.Tries(name) } - if withRetryUntil != nil { - if ru := withRetryUntil.RetryUntil(); !ru.IsZero() { - item.RetryUntilUnix = ru.Unix() + if item.Tries > 0 { + if withBackoff, ok := n.(contractsnotification.NotificationWithBackoff); ok { + backoff := withBackoff.Backoff(name) + if len(backoff) > 0 { + item.Backoff = make([]int64, len(backoff)) + for i, d := range backoff { + item.Backoff[i] = d.Milliseconds() + } + } } } From dd66aeddca054f306727b513c4847cdf7c64a421 Mon Sep 17 00:00:00 2001 From: Olusegun Ibraheem Date: Sun, 9 Aug 2026 13:03:37 -0600 Subject: [PATCH 4/4] build(mocks): update generated mocks for notification service --- mocks/notification/NotificationWithBackoff.go | 16 ++- mocks/notification/NotificationWithTries.go | 129 ++++++++++++++++++ 2 files changed, 139 insertions(+), 6 deletions(-) create mode 100644 mocks/notification/NotificationWithTries.go diff --git a/mocks/notification/NotificationWithBackoff.go b/mocks/notification/NotificationWithBackoff.go index 4598e6371..69964b51e 100644 --- a/mocks/notification/NotificationWithBackoff.go +++ b/mocks/notification/NotificationWithBackoff.go @@ -3,6 +3,8 @@ package notification import ( + time "time" + notification "github.com/goravel/framework/contracts/notification" mock "github.com/stretchr/testify/mock" ) @@ -21,18 +23,20 @@ func (_m *NotificationWithBackoff) EXPECT() *NotificationWithBackoff_Expecter { } // Backoff provides a mock function with given fields: channel -func (_m *NotificationWithBackoff) Backoff(channel string) int { +func (_m *NotificationWithBackoff) Backoff(channel string) []time.Duration { ret := _m.Called(channel) if len(ret) == 0 { panic("no return value specified for Backoff") } - var r0 int - if rf, ok := ret.Get(0).(func(string) int); ok { + var r0 []time.Duration + if rf, ok := ret.Get(0).(func(string) []time.Duration); ok { r0 = rf(channel) } else { - r0 = ret.Get(0).(int) + if ret.Get(0) != nil { + r0 = ret.Get(0).([]time.Duration) + } } return r0 @@ -56,12 +60,12 @@ func (_c *NotificationWithBackoff_Backoff_Call) Run(run func(channel string)) *N return _c } -func (_c *NotificationWithBackoff_Backoff_Call) Return(_a0 int) *NotificationWithBackoff_Backoff_Call { +func (_c *NotificationWithBackoff_Backoff_Call) Return(_a0 []time.Duration) *NotificationWithBackoff_Backoff_Call { _c.Call.Return(_a0) return _c } -func (_c *NotificationWithBackoff_Backoff_Call) RunAndReturn(run func(string) int) *NotificationWithBackoff_Backoff_Call { +func (_c *NotificationWithBackoff_Backoff_Call) RunAndReturn(run func(string) []time.Duration) *NotificationWithBackoff_Backoff_Call { _c.Call.Return(run) return _c } diff --git a/mocks/notification/NotificationWithTries.go b/mocks/notification/NotificationWithTries.go new file mode 100644 index 000000000..fc5d4f1c9 --- /dev/null +++ b/mocks/notification/NotificationWithTries.go @@ -0,0 +1,129 @@ +// Code generated by mockery. DO NOT EDIT. + +package notification + +import ( + notification "github.com/goravel/framework/contracts/notification" + mock "github.com/stretchr/testify/mock" +) + +// NotificationWithTries is an autogenerated mock type for the NotificationWithTries type +type NotificationWithTries struct { + mock.Mock +} + +type NotificationWithTries_Expecter struct { + mock *mock.Mock +} + +func (_m *NotificationWithTries) EXPECT() *NotificationWithTries_Expecter { + return &NotificationWithTries_Expecter{mock: &_m.Mock} +} + +// Tries provides a mock function with given fields: channel +func (_m *NotificationWithTries) Tries(channel string) int { + ret := _m.Called(channel) + + if len(ret) == 0 { + panic("no return value specified for Tries") + } + + var r0 int + if rf, ok := ret.Get(0).(func(string) int); ok { + r0 = rf(channel) + } else { + r0 = ret.Get(0).(int) + } + + return r0 +} + +// NotificationWithTries_Tries_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Tries' +type NotificationWithTries_Tries_Call struct { + *mock.Call +} + +// Tries is a helper method to define mock.On call +// - channel string +func (_e *NotificationWithTries_Expecter) Tries(channel interface{}) *NotificationWithTries_Tries_Call { + return &NotificationWithTries_Tries_Call{Call: _e.mock.On("Tries", channel)} +} + +func (_c *NotificationWithTries_Tries_Call) Run(run func(channel string)) *NotificationWithTries_Tries_Call { + _c.Call.Run(func(args mock.Arguments) { + run(args[0].(string)) + }) + return _c +} + +func (_c *NotificationWithTries_Tries_Call) Return(_a0 int) *NotificationWithTries_Tries_Call { + _c.Call.Return(_a0) + return _c +} + +func (_c *NotificationWithTries_Tries_Call) RunAndReturn(run func(string) int) *NotificationWithTries_Tries_Call { + _c.Call.Return(run) + return _c +} + +// Via provides a mock function with given fields: notifiable +func (_m *NotificationWithTries) Via(notifiable notification.Notifiable) []string { + ret := _m.Called(notifiable) + + if len(ret) == 0 { + panic("no return value specified for Via") + } + + var r0 []string + if rf, ok := ret.Get(0).(func(notification.Notifiable) []string); ok { + r0 = rf(notifiable) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).([]string) + } + } + + return r0 +} + +// NotificationWithTries_Via_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Via' +type NotificationWithTries_Via_Call struct { + *mock.Call +} + +// Via is a helper method to define mock.On call +// - notifiable notification.Notifiable +func (_e *NotificationWithTries_Expecter) Via(notifiable interface{}) *NotificationWithTries_Via_Call { + return &NotificationWithTries_Via_Call{Call: _e.mock.On("Via", notifiable)} +} + +func (_c *NotificationWithTries_Via_Call) Run(run func(notifiable notification.Notifiable)) *NotificationWithTries_Via_Call { + _c.Call.Run(func(args mock.Arguments) { + run(args[0].(notification.Notifiable)) + }) + return _c +} + +func (_c *NotificationWithTries_Via_Call) Return(_a0 []string) *NotificationWithTries_Via_Call { + _c.Call.Return(_a0) + return _c +} + +func (_c *NotificationWithTries_Via_Call) RunAndReturn(run func(notification.Notifiable) []string) *NotificationWithTries_Via_Call { + _c.Call.Return(run) + return _c +} + +// NewNotificationWithTries creates a new instance of NotificationWithTries. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations. +// The first argument is typically a *testing.T value. +func NewNotificationWithTries(t interface { + mock.TestingT + Cleanup(func()) +}) *NotificationWithTries { + mock := &NotificationWithTries{} + mock.Mock.Test(t) + + t.Cleanup(func() { mock.AssertExpectations(t) }) + + return mock +}