Poll queue settings as a fallback when a notifier is configured

CHANGELOG.md

@@ -7,6 +7,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- Fixed a client ignoring queue pause, resume, or metadata changes until it restarted when the queue control notification was lost, as happens if it's sent while the client's notifier is reconnecting. Clients with a notifier now also check their queues' settings once a minute. [Issue #1579](https://github.com/riverqueue/river/issues/1579).
+- Fixed the notifier never reconnecting when its connection went silent without an error, like after a dropped network path, because its health check ping had no timeout. The ping now times out after 10 seconds. [Issue #1579](https://github.com/riverqueue/river/issues/1579).
+
 ## [0.51.0] - 2026-10-31
 
 ### Added

client.go

@@ -2816,8 +2816,9 @@ func (c *Client[TTx]) QueueListTx(ctx context.Context, tx TTx, params *QueueList
 // queues at once, use the special queue name "*".
 //
 // Clients with a configured notifier should receive a notification about the
-// paused queue(s) within a few milliseconds of the transaction commit. Clients
-// in poll-only mode will pause after their next poll for queue configuration.
+// paused queue(s) within a few milliseconds of the transaction commit. Missed
+// notifications are recovered by polling once a minute. Clients in poll-only
+// mode will pause after their next poll for queue configuration.
 //
 // The provided context is used for the underlying Postgres update and can be
 // used to cancel the operation or apply a timeout. The opts are reserved for
@@ -2856,8 +2857,9 @@ func (c *Client[TTx]) QueuePause(ctx context.Context, name string, opts *QueuePa
 // queues at once, use the special queue name "*".
 //
 // Clients with a configured notifier should receive a notification about the
-// paused queue(s) within a few milliseconds of the transaction commit. Clients
-// in poll-only mode will pause after their next poll for queue configuration.
+// paused queue(s) within a few milliseconds of the transaction commit. Missed
+// notifications are recovered by polling once a minute. Clients in poll-only
+// mode will pause after their next poll for queue configuration.
 //
 // The provided context is used for the underlying Postgres update and can be
 // used to cancel the operation or apply a timeout. The opts are reserved for
@@ -2886,8 +2888,9 @@ func (c *Client[TTx]) QueuePauseTx(ctx context.Context, tx TTx, name string, opt
 // name "*".
 //
 // Clients with a configured notifier should receive a notification about the
-// resumed queue(s) within a few milliseconds of the transaction commit. Clients
-// in poll-only mode will resume after their next poll for queue configuration.
+// resumed queue(s) within a few milliseconds of the transaction commit. Missed
+// notifications are recovered by polling once a minute. Clients in poll-only
+// mode will resume after their next poll for queue configuration.
 //
 // The provided context is used for the underlying Postgres update and can be
 // used to cancel the operation or apply a timeout. The opts are reserved for
@@ -2927,8 +2930,9 @@ func (c *Client[TTx]) QueueResume(ctx context.Context, name string, opts *QueueP
 // name "*".
 //
 // Clients with a configured notifier should receive a notification about the
-// resumed queue(s) within a few milliseconds of the transaction commit. Clients
-// in poll-only mode will resume after their next poll for queue configuration.
+// resumed queue(s) within a few milliseconds of the transaction commit. Missed
+// notifications are recovered by polling once a minute. Clients in poll-only
+// mode will resume after their next poll for queue configuration.
 //
 // The provided context is used for the underlying Postgres update and can be
 // used to cancel the operation or apply a timeout. The opts are reserved for

internal/notifier/notifier.go

@@ -441,6 +441,12 @@ func (n *Notifier) waitOnce(ctx context.Context) error {
 		// the ping a no-op that always returned context.Canceled. With the fix,
 		// dead or flaky connections are now actively detected, which may trigger
 		// reconnections that were previously silently swallowed.
+		//
+		// The ping also needs a timeout. A connection that goes silent without
+		// an error, like after a dropped network path, would otherwise block
+		// here and the notifier would never reconnect.
+		pingCtx, pingCancel := context.WithTimeout(pingCtx, listenerTimeout)
+		defer pingCancel()
 		if err := n.listener.Ping(pingCtx); err != nil {
 			return err
 		}

internal/notifier/notifier_test.go

@@ -776,6 +776,37 @@ func TestNotifier_Start(t *testing.T) {
 		})
 	})
 
+	t.Run("PingTimeoutReconnects", func(t *testing.T) {
+		t.Parallel()
+
+		synctest.Test(t, func(t *testing.T) {
+			bundle := setup(t)
+
+			// The first ping hangs like one on a connection that went silent.
+			var numPings int
+			bundle.listener.pingFunc = func(ctx context.Context) error {
+				numPings++
+				if numPings == 1 {
+					<-ctx.Done()
+					return ctx.Err()
+				}
+				return nil
+			}
+
+			require.NoError(t, bundle.notifier.Start(context.Background()))
+			bundle.notifier.testSignals.ListeningBegin.WaitOrTimeout()
+
+			// Fake time: wait out the ping interval and the ping timeout.
+			time.Sleep(5*time.Second + listenerTimeout)
+			require.ErrorIs(t, bundle.notifier.testSignals.BackoffError.WaitOrTimeout(), context.DeadlineExceeded)
+			bundle.notifier.testSignals.ListeningBegin.WaitOrTimeout()
+
+			synctest.Wait()
+			bundle.notifier.Stop()
+			bundle.notifier.testSignals.BackoffError.RequireEmpty()
+		})
+	})
+
 	t.Run("ListenerDeadlineExceeded", func(t *testing.T) {
 		t.Parallel()
 

java/river/src/test/java/com/riverqueue/RunHandleTest.java

@@ -556,6 +556,47 @@ public boolean rescue(Client client, Rescue request) {
     }
   }
 
+  @ParameterizedTest
+  @ValueSource(booleans = {false, true})
+  @SuppressWarnings("try") // Resources intentionally scope background activity.
+  void fetchesRespectLostQueueControlNotifications(boolean pollOnly) throws Exception {
+    var client = client();
+    var worked = new LinkedBlockingQueue<Job<String>>();
+    java.util.function.Consumer<Boolean> setPaused =
+        paused ->
+            client.transaction(
+                connection -> {
+                  // Persist the control without NOTIFY or a SQLite outbox entry,
+                  // as if the listener were disconnected for both operations.
+                  try (var statement =
+                      Sql.prepare(
+                          connection,
+                          Sql.query(client.database(), paused ? "queue_pause" : "queue_resume"),
+                          client.database().timestamp(Instant.now()),
+                          "default",
+                          "default")) {
+                    assertEquals(1, statement.executeUpdate());
+                  }
+                  return null;
+                });
+    try (var run =
+        workers(client)
+            .pollOnly(pollOnly)
+            .pollInterval(Duration.ofMillis(10))
+            .worker(TYPE, context -> worked.add(context.job()))
+            .build()
+            .start()) {
+      setPaused.accept(true);
+      var inserted = client.insert(TYPE, "lost controls").job();
+      assertNull(worked.poll(100, TimeUnit.MILLISECONDS));
+      assertEquals(Job.State.AVAILABLE, client.jobs().get(inserted.id()).state());
+      setPaused.accept(false);
+      var completed = worked.poll(5, TimeUnit.SECONDS);
+      assertNotNull(completed);
+      assertEquals(inserted.id(), completed.id());
+    }
+  }
+
   @Test
   @SuppressWarnings("try") // Resources intentionally scope background activity.
   void goQueueNotificationsRespectTopicsAndSurviveMalformedPayloads() throws Exception {

js/src/runtime.test.ts

@@ -3567,56 +3567,63 @@ describe("Client runtime", () => {
     expect(errors).toContain("River onEvent hook failed");
   });
 
-  it("reports persisted queue pause transitions recovered by polling", async () => {
-    const driver = new FakeRuntimeDriver();
-    const definition = defineJob({ kind: "test" });
-    const observed: string[] = [];
-    const client = new Client(driver, {
-      hooks: {
-        onEvent: ({ kind }) => {
-          observed.push(kind);
+  it.each([false, true])(
+    "recovers lost queue controls with pollOnly=%s",
+    async (pollOnly) => {
+      const driver = new FakeRuntimeDriver();
+      const definition = defineJob({ kind: "test" });
+      const observed: string[] = [];
+      const client = new Client(driver, {
+        hooks: {
+          onEvent: ({ kind }) => {
+            observed.push(kind);
+          },
         },
-      },
-      leaderElectionDisabled: true,
-      pollOnly: true,
-      queueControlPollInterval: { milliseconds: 1 },
-      queueHeartbeatInterval: { milliseconds: 60_000 },
-      queues: {
-        default: { maxWorkers: 1, pollInterval: { milliseconds: 10_000 } },
-      },
-      workers: new Workers().add(definition, () => undefined),
-    });
-    const events = client.subscribe({
-      kinds: ["queue_paused", "queue_resumed"],
-    });
-    const run = await client.start();
-    const queue = driver.queues.get("default")!;
-    const pausedAt = Temporal.Now.instant();
+        leaderElectionDisabled: true,
+        pollOnly,
+        queueControlPollInterval: { milliseconds: 1 },
+        queueHeartbeatInterval: { milliseconds: 60_000 },
+        queues: {
+          default: { maxWorkers: 1, pollInterval: { milliseconds: 10_000 } },
+        },
+        workers: new Workers().add(definition, () => undefined),
+      });
+      const events = client.subscribe({
+        kinds: ["queue_paused", "queue_resumed"],
+      });
+      const run = await client.start();
+      const queue = driver.queues.get("default")!;
+      const pausedAt = Temporal.Now.instant();
 
-    driver.queues.set("default", { ...queue, pausedAt, updatedAt: pausedAt });
-    await waitUntil(() => run.diagnostics.queues.default?.paused === true);
-    expect((await events.next()).value).toMatchObject({
-      kind: "queue_paused",
-      queue: { name: "default", pausedAt },
-    });
+      driver.queues.set("default", { ...queue, pausedAt, updatedAt: pausedAt });
+      await waitUntil(() => run.diagnostics.queues.default?.paused === true);
+      expect((await events.next()).value).toMatchObject({
+        kind: "queue_paused",
+        queue: { name: "default", pausedAt },
+      });
 
-    const resumedAt = Temporal.Now.instant();
-    driver.queues.set("default", {
-      ...queue,
-      pausedAt: null,
-      updatedAt: resumedAt,
-    });
-    await waitUntil(() => run.diagnostics.queues.default?.paused === false);
-    expect((await events.next()).value).toMatchObject({
-      kind: "queue_resumed",
-      queue: { name: "default", pausedAt: null },
-    });
-    expect(observed.filter((kind) => kind === "queue_paused")).toHaveLength(1);
-    expect(observed.filter((kind) => kind === "queue_resumed")).toHaveLength(1);
+      const resumedAt = Temporal.Now.instant();
+      driver.queues.set("default", {
+        ...queue,
+        pausedAt: null,
+        updatedAt: resumedAt,
+      });
+      await waitUntil(() => run.diagnostics.queues.default?.paused === false);
+      expect((await events.next()).value).toMatchObject({
+        kind: "queue_resumed",
+        queue: { name: "default", pausedAt: null },
+      });
+      expect(observed.filter((kind) => kind === "queue_paused")).toHaveLength(
+        1
+      );
+      expect(observed.filter((kind) => kind === "queue_resumed")).toHaveLength(
+        1
+      );
 
-    events.close();
-    await run.stop();
-  });
+      events.close();
+      await run.stop();
+    }
+  );
 
   it("publishes same-client queue control exactly once", async () => {
     const driver = new FakeRuntimeDriver();