diff --git a/CHANGELOG.md b/CHANGELOG.md index 1b66a6430..3a6b73c75 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Fixed `Client.InsertManyFast` panicking after its transaction committed on a client configured with queues when the driver doesn't support a listener, such as a `riverdatabasesql` driver created with `New` instead of `NewWithPgxListener`. It now wakes producers for the inserted jobs' queues right away, like `Client.InsertMany`. [PR #1558](https://github.com/riverqueue/river/pull/1558). - Fixed `Client.QueueUpdate` without metadata crashing the program when the client works that queue and the driver doesn't support a listener. [PR #1558](https://github.com/riverqueue/river/pull/1558). +- Fixed the notifier's run loop blocking forever when a topic `LISTEN` or `UNLISTEN` handed to it finished after its caller had stopped waiting (after 5 seconds, with `timed out waiting for interrupt operation`), as can happen on a slow or unresponsive connection. `Client.Stop` then never completed, and neither did a `Client.Start` that failed this way. [PR #1559](https://github.com/riverqueue/river/pull/1559). ## [0.50.0] - 2026-10-26 diff --git a/internal/notifier/notifier.go b/internal/notifier/notifier.go index e860cb17b..d453eddc3 100644 --- a/internal/notifier/notifier.go +++ b/internal/notifier/notifier.go @@ -463,7 +463,9 @@ func (n *Notifier) waitOnce(ctx context.Context) error { // // MUST be called with the `n.mu` mutex already locked. func (n *Notifier) sendInterruptAndReceiveResult(operation func() error) error { - errChan := make(chan error) + // Buffered so the main loop can always hand off a result and move on, even + // if the timeout below has fired and nothing is waiting for it anymore. + errChan := make(chan error, 1) n.waitInterruptChan <- func() { errChan <- operation() } diff --git a/internal/notifier/notifier_test.go b/internal/notifier/notifier_test.go index 76ad0b820..8d5c21cc3 100644 --- a/internal/notifier/notifier_test.go +++ b/internal/notifier/notifier_test.go @@ -718,6 +718,64 @@ func TestNotifier_Start(t *testing.T) { }) }) + t.Run("InterruptTimeout", func(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + bundle := setup(t) + + const ( + fastTopic = "fast_topic" + slowTopic = "slow_topic" + ) + + bundle.listener.listenFunc = func(ctx context.Context, topic string) error { + if topic != slowTopic { + return nil + } + + // Longer than the 5 second interrupt timeout so that the listen + // finishes only after Listen has given up on it, but shorter than + // listenerTimeout. + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(6 * time.Second): + return nil + } + } + + var numWaits int + waitFunc := bundle.listener.waitForNotificationFunc + bundle.listener.waitForNotificationFunc = func(ctx context.Context) (*riverdriver.Notification, error) { + numWaits++ + return waitFunc(ctx) + } + + require.NoError(t, bundle.notifier.Start(context.Background())) + bundle.notifier.testSignals.ListeningBegin.WaitOrTimeout() + synctest.Wait() + + _, err := bundle.notifier.Listen(context.Background(), slowTopic, nil) + require.EqualError(t, err, "timed out waiting for interrupt operation") + + // The listen finishes a second after Listen gave up on it. With + // nothing left to receive its result, the main loop should still go + // back to waiting for notifications and handling interrupts. + time.Sleep(time.Second) + synctest.Wait() + require.Equal(t, 2, numWaits) + + _, err = bundle.notifier.Listen(context.Background(), fastTopic, nil) + require.NoError(t, err) + synctest.Wait() + require.Equal(t, 3, numWaits) + + bundle.notifier.Stop() + bundle.notifier.testSignals.BackoffError.RequireEmpty() + }) + }) + t.Run("ListenerDeadlineExceeded", func(t *testing.T) { t.Parallel()