[go] Fix notifier hanging after a listen or unlisten times out

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
 

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()
 	}

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()