diff --git a/CHANGELOG.md b/CHANGELOG.md index a60905e59..2a8c85b59 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed - Fixed `AddWorkerSafely` leaving a worker's primary kind and earlier aliases registered when a later kind alias conflicts with an existing registration. Failed registrations now leave the worker registry unchanged. [PR #1440](https://github.com/riverqueue/river/pull/1440). +- Fixed the notifier retrying without backoff and flooding logs when the context passed to `Client.Start` reaches its deadline. [PR #1495](https://github.com/riverqueue/river/pull/1495). ## [0.49.0] - 2026-10-05 diff --git a/internal/notifier/notifier.go b/internal/notifier/notifier.go index 8b48fbdd8..e860cb17b 100644 --- a/internal/notifier/notifier.go +++ b/internal/notifier/notifier.go @@ -141,9 +141,11 @@ func (n *Notifier) Start(ctx context.Context) error { n.deliverNotifications(ctx) }) - for attempt := 0; ; attempt++ { + for attempt := 0; ctx.Err() == nil; attempt++ { if err := n.listenAndWait(ctx); err != nil { - if errors.Is(err, context.Canceled) { + // Only the service context ending should stop retries. A listener + // operation may time out while the service context is still live. + if ctx.Err() != nil { break } diff --git a/internal/notifier/notifier_test.go b/internal/notifier/notifier_test.go index a6dc16ab6..76ad0b820 100644 --- a/internal/notifier/notifier_test.go +++ b/internal/notifier/notifier_test.go @@ -8,6 +8,7 @@ import ( "strconv" "sync" "testing" + "testing/synctest" "time" "github.com/jackc/pgx/v5/pgxpool" @@ -624,9 +625,132 @@ func TestNotifier(t *testing.T) { }) } +func TestNotifier_Start(t *testing.T) { + t.Parallel() + + type testBundle struct { + listener *ListenerMock + notifier *Notifier + } + + setup := func(t *testing.T) *testBundle { + t.Helper() + + listener := &ListenerMock{ + closeFunc: func(ctx context.Context) error { return nil }, + connectFunc: func(ctx context.Context) error { return ctx.Err() }, + waitForNotificationFunc: func(ctx context.Context) (*riverdriver.Notification, error) { + <-ctx.Done() + return nil, fmt.Errorf("notification wait: %w", ctx.Err()) + }, + } + notifier := New(riversharedtest.BaseServiceArchetype(t), listener) + notifier.testSignals.Init(t) + t.Cleanup(notifier.Stop) + + return &testBundle{listener: listener, notifier: notifier} + } + + t.Run("BackoffDeadlineExceeded", func(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + bundle := setup(t) + + waitErr := errors.New("error during wait") + bundle.listener.waitForNotificationFunc = func(ctx context.Context) (*riverdriver.Notification, error) { + return nil, waitErr + } + + var numConnects int + bundle.listener.connectFunc = func(ctx context.Context) error { + numConnects++ + return ctx.Err() + } + + // Expire during the first backoff, which lasts about a second. + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + require.NoError(t, bundle.notifier.Start(ctx)) + require.ErrorIs(t, bundle.notifier.testSignals.BackoffError.WaitOrTimeout(), waitErr) + + riversharedtest.WaitOrTimeout(t, bundle.notifier.Stopped()) + require.ErrorIs(t, ctx.Err(), context.DeadlineExceeded) + require.Equal(t, 1, numConnects) + bundle.notifier.testSignals.BackoffError.RequireEmpty() + }) + }) + + t.Run("ContextCanceled", func(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + bundle := setup(t) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + require.NoError(t, bundle.notifier.Start(ctx)) + bundle.notifier.testSignals.ListeningBegin.WaitOrTimeout() + + synctest.Wait() + cancel() + riversharedtest.WaitOrTimeout(t, bundle.notifier.Stopped()) + bundle.notifier.testSignals.BackoffError.RequireEmpty() + }) + }) + + t.Run("ContextDeadlineExceeded", func(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + bundle := setup(t) + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + require.NoError(t, bundle.notifier.Start(ctx)) + bundle.notifier.testSignals.ListeningBegin.WaitOrTimeout() + + riversharedtest.WaitOrTimeout(t, bundle.notifier.Stopped()) + require.ErrorIs(t, ctx.Err(), context.DeadlineExceeded) + require.False(t, bundle.notifier.isConnected) + require.False(t, bundle.notifier.isStarted) + bundle.notifier.testSignals.BackoffError.RequireEmpty() + }) + }) + + t.Run("ListenerDeadlineExceeded", func(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + bundle := setup(t) + + var numWaits int + waitFunc := bundle.listener.waitForNotificationFunc + bundle.listener.waitForNotificationFunc = func(ctx context.Context) (*riverdriver.Notification, error) { + numWaits++ + if numWaits == 1 { + return nil, fmt.Errorf("listener timeout: %w", context.DeadlineExceeded) + } + return waitFunc(ctx) + } + + require.NoError(t, bundle.notifier.Start(context.Background())) + bundle.notifier.testSignals.ListeningBegin.WaitOrTimeout() + require.ErrorIs(t, bundle.notifier.testSignals.BackoffError.WaitOrTimeout(), context.DeadlineExceeded) + bundle.notifier.testSignals.ListeningBegin.WaitOrTimeout() + + synctest.Wait() + bundle.notifier.Stop() + require.Equal(t, 2, numWaits) + bundle.notifier.testSignals.BackoffError.RequireEmpty() + }) + }) +} + type ListenerMock struct { riverdriver.Listener + closeFunc func(ctx context.Context) error connectFunc func(ctx context.Context) error listenFunc func(ctx context.Context, topic string) error pingFunc func(ctx context.Context) error @@ -637,6 +761,7 @@ func NewListenerMock(listener riverdriver.Listener) *ListenerMock { return &ListenerMock{ Listener: listener, + closeFunc: listener.Close, connectFunc: listener.Connect, listenFunc: listener.Listen, pingFunc: listener.Ping, @@ -644,6 +769,10 @@ func NewListenerMock(listener riverdriver.Listener) *ListenerMock { } } +func (l *ListenerMock) Close(ctx context.Context) error { + return l.closeFunc(ctx) +} + func (l *ListenerMock) Connect(ctx context.Context) error { return l.connectFunc(ctx) }