From 4f5b44a2990503139c5b9ef60b8689940d16c1be Mon Sep 17 00:00:00 2001 From: Yash Bitla <53489772+yash-bitla@users.noreply.github.com> Date: Wed, 4 Nov 2026 19:44:23 +0530 Subject: [PATCH 1/2] Poll queue settings as a fallback when a notifier is configured A queue pause or resume reaches running clients only by NOTIFY. If the notification is lost, like while the notifier is reconnecting, the client keeps working the queue until it restarts, because pollForSettingChanges only ran in poll-only mode. Run it with a notifier too, once a minute, as #1564 did for job cancellation. Also give the notifier's health check ping a timeout. A connection that goes silent without an error blocked the ping, so the notifier never reconnected. Fixes #1579. --- CHANGELOG.md | 5 ++++ internal/notifier/notifier.go | 6 ++++ internal/notifier/notifier_test.go | 31 +++++++++++++++++++++ producer.go | 43 ++++++++++++++++++++++------- producer_test.go | 44 ++++++++++++++++++++++++++++++ 5 files changed, 119 insertions(+), 10 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3add8bfc0..9395e6197 100644 --- a/CHANGELOG.md +++ b/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 `Client.QueuePause` or `Client.QueueResume` 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 diff --git a/internal/notifier/notifier.go b/internal/notifier/notifier.go index d453eddc3..515aad345 100644 --- a/internal/notifier/notifier.go +++ b/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 } diff --git a/internal/notifier/notifier_test.go b/internal/notifier/notifier_test.go index 8d5c21cc3..00323af3d 100644 --- a/internal/notifier/notifier_test.go +++ b/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() diff --git a/producer.go b/producer.go index a660f7a78..9c06cb2fe 100644 --- a/producer.go +++ b/producer.go @@ -37,11 +37,12 @@ import ( ) const ( - jobCancelPollBatchSize = 1000 - jobCancelPollIntervalDefault = time.Minute - producerReportIntervalDefault = 30 * time.Second - queuePollIntervalDefault = 2 * time.Second - queueReportIntervalDefault = 10 * time.Minute + jobCancelPollBatchSize = 1000 + jobCancelPollIntervalDefault = time.Minute + producerReportIntervalDefault = 30 * time.Second + queuePollIntervalDefault = 2 * time.Second + queueReportIntervalDefault = 10 * time.Minute + queueSettingsPollIntervalDefault = time.Minute ) // Test-only properties. @@ -123,6 +124,13 @@ type producerConfig struct { // queue setting changes and job cancellation requests in poll-only mode // (when no notifier is provided). QueuePollInterval time.Duration + + // QueueSettingsPollInterval is the amount of time between checks for + // queue setting changes (pause, resume, metadata) when a notifier is + // configured. Notifications normally deliver changes right away, so this is + // only a fallback for ones that were lost, like while the notifier was + // reconnecting. In poll-only mode, QueuePollInterval is used instead. + QueueSettingsPollInterval time.Duration // QueueReportInterval is the amount of time between periodic reports // of the queue status. QueueReportInterval time.Duration @@ -152,6 +160,12 @@ func (c *producerConfig) mustValidate() *producerConfig { if c.JobCancelPollInterval <= 0 { panic("producerConfig.JobCancelPollInterval must be greater than zero") } + if c.QueueSettingsPollInterval == 0 { + c.QueueSettingsPollInterval = queueSettingsPollIntervalDefault + } + if c.QueueSettingsPollInterval <= 0 { + panic("producerConfig.QueueSettingsPollInterval must be greater than zero") + } if c.JobStuckCount == nil { c.JobStuckCount = &atomic.Int32{} } @@ -457,15 +471,15 @@ func (p *producer) StartWorkContext(fetchCtx, workCtx context.Context) error { subroutineWG.Add(1) go p.reportProducerStatusLoop(subroutineCtx, &subroutineWG) - // Runs with a notifier too, in case a cancel notification is lost. + // Both run with a notifier too, in case a notification is lost. subroutineWG.Add(1) go p.pollForJobCancellations(subroutineCtx, &subroutineWG) + subroutineWG.Add(1) + go p.pollForSettingChanges(subroutineCtx, &subroutineWG, initiallyPaused, initialMetadata) + if p.config.Notifier == nil { p.Logger.DebugContext(subroutineCtx, p.Name+": No notifier configured; starting in poll mode", "client_id", p.config.ClientID) - - subroutineWG.Add(1) - go p.pollForSettingChanges(subroutineCtx, &subroutineWG, initiallyPaused, initialMetadata) } p.fetchAndRunLoop(fetchCtx, workCtx) @@ -1107,10 +1121,19 @@ func (p *producer) pollForJobCancellationsOnce(ctx context.Context) error { return nil } +// Without a notifier, polling is the only way to learn about queue setting +// changes. With one, it's a slower fallback because a notification is lost if +// it's sent while the notifier isn't listening, like while it's reconnecting. +// Pause and resume are idempotent, so a change seen by both is applied once. func (p *producer) pollForSettingChanges(ctx context.Context, wg *sync.WaitGroup, lastPaused bool, lastMetadata []byte) { defer wg.Done() - ticker := time.NewTicker(p.config.QueuePollInterval) + interval := p.config.QueuePollInterval + if p.config.Notifier != nil { + interval = p.config.QueueSettingsPollInterval + } + + ticker := time.NewTicker(interval) defer ticker.Stop() for { select { diff --git a/producer_test.go b/producer_test.go index 368a7080a..8cdff310a 100644 --- a/producer_test.go +++ b/producer_test.go @@ -635,6 +635,50 @@ func testProducer(t *testing.T, makeProducer func(ctx context.Context, t *testin } }) + // A pause or resume whose notification was lost, like while the notifier + // was reconnecting, is still found by polling. + t.Run("QueueSettingsPolling", func(t *testing.T) { + t.Parallel() + + producer, bundle := setup(t) + AddWorker(bundle.workers, &noOpWorker{}) + + // Each mode polls on its own interval. The other one is too long for the + // test to pass if it's used instead. + if producer.config.Notifier == nil { + producer.config.QueuePollInterval = 20 * time.Millisecond + producer.config.QueueSettingsPollInterval = time.Hour + } else { + producer.config.QueuePollInterval = time.Hour + producer.config.QueueSettingsPollInterval = 20 * time.Millisecond + } + + startProducer(t, ctx, ctx, producer) + + // Pause and resume without a notification. Only polling can find them. + require.NoError(t, bundle.exec.QueuePause(ctx, &riverdriver.QueuePauseParams{ + Name: producer.config.Queue, + Schema: producer.config.Schema, + })) + producer.testSignals.Paused.WaitOrTimeout() + + mustInsert(ctx, t, producer, bundle, &noOpArgs{}) + select { + case update := <-bundle.jobUpdates: + t.Fatalf("Unexpected job update: %+v", update) + case <-time.After(200 * time.Millisecond): + } + + require.NoError(t, bundle.exec.QueueResume(ctx, &riverdriver.QueueResumeParams{ + Name: producer.config.Queue, + Schema: producer.config.Schema, + })) + producer.testSignals.Resumed.WaitOrTimeout() + + update := riversharedtest.WaitOrTimeout(t, bundle.jobUpdates) + require.Equal(t, rivertype.JobStateCompleted, update.Job.State) + }) + // A cancellation request found by polling cancels the attempt that was // checked, not a newer one running under the same ID by the time the // check returns, like one started after the job was retried. From 320f98ef27a749e764a6b635afbf11969e1ef262 Mon Sep 17 00:00:00 2001 From: Brandur Date: Thu, 5 Nov 2026 13:03:12 -0600 Subject: [PATCH 2/2] A few more fixes for queue pause/resume/metadata changes on notifier reconnect Apply a few more fixes on top of the original changes in #1582: * More reliable Go state reconciliation: Polling compares database settings against producer's applied state. Fixes a received pause followed by a lost resume leaving the producer paused indefinitely. Metadata changes followed by reversions are also fixed. * Metadata dedup: Polling skips metadata already applied through a notification, avoiding duplicate callbacks. * Shared fallback interval: Queue settings and cancellation polling use `NotificationPollInterval`, defaulting to one minute. Rust uses the same. * Cleaner shutdown: Queue settings polling stops when fetching stops. Cancellation polling continues while running jobs drain. * Rust recover: Adds periodic queue settings refresh when notifier exists, supplementing existing refresh on reconnect Reuses existing timer. * Expanded tests: Extend Yash's polling test to cover metadata. Add regression for notification /poll interactions and duplicate callbacks. Add shared Go driver and Rust Postgres/SQLite coverage. * Docs and small fixes: Documents fallback polling, updates changeslogs, and fixes the Go notification test helper to encode metadata as JSON rather than base64. --- CHANGELOG.md | 2 +- client.go | 20 +- .../java/com/riverqueue/RunHandleTest.java | 41 ++++ js/src/runtime.test.ts | 99 +++++---- producer.go | 117 ++++------ producer_test.go | 200 +++++++++++++----- .../riverdrivertest/driver_client_test.go | 36 ++++ .../notification_receiver_shared_examples.rb | 26 +++ rust/CHANGELOG.md | 1 + rust/riverqueue/src/__private.rs | 21 +- rust/riverqueue/src/client/builder.rs | 4 +- rust/riverqueue/src/client/mod.rs | 14 +- rust/riverqueue/src/client/producer.rs | 23 +- .../tests/poll_only_cancellation.rs | 157 +++++++++++++- 14 files changed, 549 insertions(+), 212 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9395e6197..c7766e359 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,7 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed -- Fixed a client ignoring `Client.QueuePause` or `Client.QueueResume` 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 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 diff --git a/client.go b/client.go index d78acdd16..0b64c561f 100644 --- a/client.go +++ b/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 diff --git a/java/river/src/test/java/com/riverqueue/RunHandleTest.java b/java/river/src/test/java/com/riverqueue/RunHandleTest.java index 48e8836b6..8bd7dc7a3 100644 --- a/java/river/src/test/java/com/riverqueue/RunHandleTest.java +++ b/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>(); + java.util.function.Consumer 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 { diff --git a/js/src/runtime.test.ts b/js/src/runtime.test.ts index c44c05346..010ef628f 100644 --- a/js/src/runtime.test.ts +++ b/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(); diff --git a/producer.go b/producer.go index 9c06cb2fe..0db1e229c 100644 --- a/producer.go +++ b/producer.go @@ -37,12 +37,11 @@ import ( ) const ( - jobCancelPollBatchSize = 1000 - jobCancelPollIntervalDefault = time.Minute - producerReportIntervalDefault = 30 * time.Second - queuePollIntervalDefault = 2 * time.Second - queueReportIntervalDefault = 10 * time.Minute - queueSettingsPollIntervalDefault = time.Minute + jobCancelPollBatchSize = 1000 + notificationPollIntervalDefault = time.Minute + producerReportIntervalDefault = 30 * time.Second + queuePollIntervalDefault = 2 * time.Second + queueReportIntervalDefault = 10 * time.Minute ) // Test-only properties. @@ -93,13 +92,6 @@ type producerConfig struct { // LISTEN/NOTIFY, but this provides a fallback. FetchPollInterval time.Duration - // JobCancelPollInterval is the amount of time between checks for - // cancellation of running jobs when a notifier is configured. Notifications - // normally deliver cancellations right away, so this is only a fallback for - // ones that were lost, like while the notifier was reconnecting. In - // poll-only mode, QueuePollInterval is used instead. - JobCancelPollInterval time.Duration - PluginLookupByJob *pluginlookup.JobPluginLookup PluginLookupGlobal *pluginlookup.PluginLookup JobStuckHandler JobStuckHandler @@ -108,6 +100,12 @@ type producerConfig struct { JobTimeout time.Duration MaxWorkers int + // NotificationPollInterval is the fallback interval for job cancellations + // and queue settings when a notifier is configured. Notifications normally + // deliver these changes immediately, but may be lost while disconnected. + // In poll-only mode, QueuePollInterval is used instead. + NotificationPollInterval time.Duration + // Notifier is a notifier for subscribing to new job inserts and job // control. If nil, the producer will operate in poll-only mode. Notifier *notifier.Notifier @@ -124,13 +122,6 @@ type producerConfig struct { // queue setting changes and job cancellation requests in poll-only mode // (when no notifier is provided). QueuePollInterval time.Duration - - // QueueSettingsPollInterval is the amount of time between checks for - // queue setting changes (pause, resume, metadata) when a notifier is - // configured. Notifications normally deliver changes right away, so this is - // only a fallback for ones that were lost, like while the notifier was - // reconnecting. In poll-only mode, QueuePollInterval is used instead. - QueueSettingsPollInterval time.Duration // QueueReportInterval is the amount of time between periodic reports // of the queue status. QueueReportInterval time.Duration @@ -154,18 +145,6 @@ func (c *producerConfig) mustValidate() *producerConfig { if c.FetchPollInterval <= 0 { panic("producerConfig.FetchPollInterval must be greater than zero") } - if c.JobCancelPollInterval == 0 { - c.JobCancelPollInterval = jobCancelPollIntervalDefault - } - if c.JobCancelPollInterval <= 0 { - panic("producerConfig.JobCancelPollInterval must be greater than zero") - } - if c.QueueSettingsPollInterval == 0 { - c.QueueSettingsPollInterval = queueSettingsPollIntervalDefault - } - if c.QueueSettingsPollInterval <= 0 { - panic("producerConfig.QueueSettingsPollInterval must be greater than zero") - } if c.JobStuckCount == nil { c.JobStuckCount = &atomic.Int32{} } @@ -181,6 +160,12 @@ func (c *producerConfig) mustValidate() *producerConfig { if c.MaxWorkers == 0 { panic("producerConfig.MaxWorkers is required") } + if c.NotificationPollInterval == 0 { + c.NotificationPollInterval = notificationPollIntervalDefault + } + if c.NotificationPollInterval <= 0 { + panic("producerConfig.NotificationPollInterval must be greater than zero") + } if c.ProducerReportInterval == 0 { c.ProducerReportInterval = producerReportIntervalDefault } @@ -191,7 +176,7 @@ func (c *producerConfig) mustValidate() *producerConfig { c.QueuePollInterval = queuePollIntervalDefault } if c.QueuePollInterval <= 0 { - panic("producerConfig.QueueSettingsPollInterval must be greater than zero") + panic("producerConfig.QueuePollInterval must be greater than zero") } if c.QueueReportInterval == 0 { c.QueueReportInterval = queueReportIntervalDefault @@ -269,6 +254,7 @@ type producer struct { // Receives control messages from the notifier goroutine. Written by notifier // goroutine, only read from main goroutine. queueControlCh chan *controlEventPayload + queueMetadata []byte // last applied metadata; owned by the main goroutine retryPolicy ClientRetryPolicy testSignals producerTestSignals } @@ -381,6 +367,7 @@ func (p *producer) StartWorkContext(fetchCtx, workCtx context.Context) error { } } p.paused = initiallyPaused + p.queueMetadata = initialMetadata id := p.id.Load() id, p.state, err = p.pilot.ProducerInit(fetchCtx, p.exec, &riverpilot.ProducerInitParams{ @@ -476,7 +463,7 @@ func (p *producer) StartWorkContext(fetchCtx, workCtx context.Context) error { go p.pollForJobCancellations(subroutineCtx, &subroutineWG) subroutineWG.Add(1) - go p.pollForSettingChanges(subroutineCtx, &subroutineWG, initiallyPaused, initialMetadata) + go p.pollForSettingChanges(fetchCtx, &subroutineWG) if p.config.Notifier == nil { p.Logger.DebugContext(subroutineCtx, p.Name+": No notifier configured; starting in poll mode", "client_id", p.config.ClientID) @@ -533,6 +520,9 @@ type controlEventPayload struct { JobID int64 `json:"job_id,omitempty"` Metadata json.RawMessage `json:"metadata,omitempty"` Queue string `json:"queue"` + + // Poll snapshots may repeat metadata already applied from a notification. + fromPoll bool } type insertPayload struct { @@ -602,6 +592,10 @@ func (p *producer) fetchAndRunLoop(fetchCtx, workCtx context.Context) { // will have to wait for the next poll event for a cancel to take effect. p.maybeCancelJob(workCtx, msg.JobID) case controlActionMetadataChanged: + if msg.fromPoll && metadataEqual(p.queueMetadata, msg.Metadata) { + continue + } + p.queueMetadata = msg.Metadata p.Logger.DebugContext(workCtx, p.Name+": Queue metadata changed", slog.String("queue", p.config.Queue), slog.String("queue_in_message", msg.Queue)) p.testSignals.MetadataChanged.Signal(struct{}{}) if err := p.pilot.QueueMetadataChanged(workCtx, p.exec, &riverpilot.QueueMetadataChangedParams{ @@ -1069,7 +1063,7 @@ func (p *producer) pollForJobCancellations(ctx context.Context, wg *sync.WaitGro interval := p.config.QueuePollInterval if p.config.Notifier != nil { - interval = p.config.JobCancelPollInterval + interval = p.config.NotificationPollInterval } ticker := time.NewTicker(interval) @@ -1125,12 +1119,12 @@ func (p *producer) pollForJobCancellationsOnce(ctx context.Context) error { // changes. With one, it's a slower fallback because a notification is lost if // it's sent while the notifier isn't listening, like while it's reconnecting. // Pause and resume are idempotent, so a change seen by both is applied once. -func (p *producer) pollForSettingChanges(ctx context.Context, wg *sync.WaitGroup, lastPaused bool, lastMetadata []byte) { +func (p *producer) pollForSettingChanges(ctx context.Context, wg *sync.WaitGroup) { defer wg.Done() interval := p.config.QueuePollInterval if p.config.Notifier != nil { - interval = p.config.QueueSettingsPollInterval + interval = p.config.NotificationPollInterval } ticker := time.NewTicker(interval) @@ -1159,47 +1153,22 @@ func (p *producer) pollForSettingChanges(ctx context.Context, wg *sync.WaitGroup continue } - // Look for a change in the paused state: - shouldBePaused := (updatedQueue.PausedAt != nil) - if lastPaused != shouldBePaused { - action := controlActionPause - if !shouldBePaused { - action = controlActionResume - } - payload := &controlEventPayload{ - Action: action, - Queue: p.config.Queue, - } - p.Logger.DebugContext(ctx, p.Name+": Queue control state changed from polling", - slog.String("queue", p.config.Queue), - slog.String("action", string(action)), - slog.Bool("paused", shouldBePaused), - ) - - select { - case p.queueControlCh <- payload: - lastPaused = shouldBePaused - default: - p.Logger.WarnContext(ctx, p.Name+": Queue control notification dropped due to full buffer", slog.String("action", string(action))) - } + // Reconcile every snapshot with the state the main goroutine actually + // applied, including notifications received between polls. Comparing + // only with the last poll would miss a notified pause followed by a + // lost resume that returns the database to its previous state. + action := controlActionPause + if updatedQueue.PausedAt == nil { + action = controlActionResume } - - // Look for a change in the queue's metadata: - if !metadataEqual(lastMetadata, updatedQueue.Metadata) { - payload := &controlEventPayload{ - Action: controlActionMetadataChanged, - Queue: p.config.Queue, - Metadata: updatedQueue.Metadata, - } - p.Logger.DebugContext(ctx, p.Name+": Queue metadata changed from polling", - slog.String("queue", p.config.Queue), - ) - + for _, payload := range []*controlEventPayload{ + {Action: action, Queue: p.config.Queue}, + {Action: controlActionMetadataChanged, Metadata: updatedQueue.Metadata, Queue: p.config.Queue, fromPoll: true}, + } { select { case p.queueControlCh <- payload: - lastMetadata = updatedQueue.Metadata default: - p.Logger.WarnContext(ctx, p.Name+": Queue control notification dropped due to full buffer", slog.String("action", string(controlActionMetadataChanged))) + p.Logger.WarnContext(ctx, p.Name+": Queue control notification dropped due to full buffer", slog.String("action", string(payload.Action))) } } diff --git a/producer_test.go b/producer_test.go index 8cdff310a..500bbe3d0 100644 --- a/producer_test.go +++ b/producer_test.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "slices" + "sync" "sync/atomic" "testing" "time" @@ -89,6 +90,17 @@ func (e *jobGetCancelRequestedFuncExecutor) JobGetCancelRequested(ctx context.Co return e.jobGetCancelRequestedFunc(params) } +// queueGetFuncExecutor replaces QueueGet on the wrapped executor. +type queueGetFuncExecutor struct { + riverdriver.Executor + + queueGetFunc func(context.Context, *riverdriver.QueueGetParams) (*rivertype.Queue, error) +} + +func (e *queueGetFuncExecutor) QueueGet(ctx context.Context, params *riverdriver.QueueGetParams) (*rivertype.Queue, error) { + return e.queueGetFunc(ctx, params) +} + // undecodableKindPilot reports locked jobs of one kind as undecodable. Postgres' // column types don't allow a job row that can't be decoded, so this simulates // one. @@ -569,10 +581,10 @@ func testProducer(t *testing.T, makeProducer func(ctx context.Context, t *testin // Each mode polls on its own interval. The other one is too long // for the test to pass if it's used instead. if producer.config.Notifier == nil { - producer.config.JobCancelPollInterval = time.Hour + producer.config.NotificationPollInterval = time.Hour producer.config.QueuePollInterval = 20 * time.Millisecond } else { - producer.config.JobCancelPollInterval = 20 * time.Millisecond + producer.config.NotificationPollInterval = 20 * time.Millisecond producer.config.QueuePollInterval = time.Hour } @@ -635,50 +647,6 @@ func testProducer(t *testing.T, makeProducer func(ctx context.Context, t *testin } }) - // A pause or resume whose notification was lost, like while the notifier - // was reconnecting, is still found by polling. - t.Run("QueueSettingsPolling", func(t *testing.T) { - t.Parallel() - - producer, bundle := setup(t) - AddWorker(bundle.workers, &noOpWorker{}) - - // Each mode polls on its own interval. The other one is too long for the - // test to pass if it's used instead. - if producer.config.Notifier == nil { - producer.config.QueuePollInterval = 20 * time.Millisecond - producer.config.QueueSettingsPollInterval = time.Hour - } else { - producer.config.QueuePollInterval = time.Hour - producer.config.QueueSettingsPollInterval = 20 * time.Millisecond - } - - startProducer(t, ctx, ctx, producer) - - // Pause and resume without a notification. Only polling can find them. - require.NoError(t, bundle.exec.QueuePause(ctx, &riverdriver.QueuePauseParams{ - Name: producer.config.Queue, - Schema: producer.config.Schema, - })) - producer.testSignals.Paused.WaitOrTimeout() - - mustInsert(ctx, t, producer, bundle, &noOpArgs{}) - select { - case update := <-bundle.jobUpdates: - t.Fatalf("Unexpected job update: %+v", update) - case <-time.After(200 * time.Millisecond): - } - - require.NoError(t, bundle.exec.QueueResume(ctx, &riverdriver.QueueResumeParams{ - Name: producer.config.Queue, - Schema: producer.config.Schema, - })) - producer.testSignals.Resumed.WaitOrTimeout() - - update := riversharedtest.WaitOrTimeout(t, bundle.jobUpdates) - require.Equal(t, rivertype.JobStateCompleted, update.Job.State) - }) - // A cancellation request found by polling cancels the attempt that was // checked, not a newer one running under the same ID by the time the // check returns, like one started after the job was retried. @@ -1334,8 +1302,7 @@ func testProducer(t *testing.T, makeProducer func(ctx context.Context, t *testin // Update with equivalent metadata but different field ordering: reorderedMetadata := []byte(`{"baz":123,"foo":"bar"}`) updateMetadata(reorderedMetadata) - // do not emit a notification here because this isn't a "real" update and - // notifier mode doesn't check for metadata equivalence. + // Omit the notification so polling must recognize equivalent metadata. // Should not receive a metadata changed signal since the JSON is equivalent: select { @@ -1356,6 +1323,141 @@ func testProducer(t *testing.T, makeProducer func(ctx context.Context, t *testin // Should receive a metadata changed signal since the JSON is different: producer.testSignals.MetadataChanged.WaitOrTimeout() }) + + // A pause, resume, or metadata change whose notification was lost, like + // while the notifier was reconnecting, is still found by polling. + t.Run("QueueSettingsPolling", func(t *testing.T) { + t.Parallel() + + producer, bundle := setup(t) + AddWorker(bundle.workers, &noOpWorker{}) + + // Each mode polls on its own interval. The other one is too long for the + // test to pass if it's used instead. + if producer.config.Notifier == nil { + producer.config.QueuePollInterval = 20 * time.Millisecond + producer.config.NotificationPollInterval = time.Hour + } else { + producer.config.QueuePollInterval = time.Hour + producer.config.NotificationPollInterval = 20 * time.Millisecond + } + + startProducer(t, ctx, ctx, producer) + + // Pause and resume without a notification. Only polling can find them. + require.NoError(t, bundle.exec.QueuePause(ctx, &riverdriver.QueuePauseParams{ + Name: producer.config.Queue, + Schema: producer.config.Schema, + })) + producer.testSignals.Paused.WaitOrTimeout() + + mustInsert(ctx, t, producer, bundle, &noOpArgs{}) + select { + case update := <-bundle.jobUpdates: + t.Fatalf("Unexpected job update: %+v", update) + case <-time.After(200 * time.Millisecond): + } + + require.NoError(t, bundle.exec.QueueResume(ctx, &riverdriver.QueueResumeParams{ + Name: producer.config.Queue, + Schema: producer.config.Schema, + })) + producer.testSignals.Resumed.WaitOrTimeout() + + update := riversharedtest.WaitOrTimeout(t, bundle.jobUpdates) + require.Equal(t, rivertype.JobStateCompleted, update.Job.State) + + // Metadata is also recovered when its notification is lost. + _, err := bundle.exec.QueueUpdate(ctx, &riverdriver.QueueUpdateParams{ + Metadata: []byte(`{"version":1}`), + MetadataDoUpdate: true, + Name: producer.config.Queue, + Schema: producer.config.Schema, + }) + require.NoError(t, err) + producer.testSignals.MetadataChanged.WaitOrTimeout() + }) + + t.Run("QueueSettingsPollingDeduplicatesMetadata", func(t *testing.T) { + t.Parallel() + + producer, bundle := setup(t) + + producer.config.NotificationPollInterval = 20 * time.Millisecond + producer.config.QueuePollInterval = 20 * time.Millisecond + + // Hold the first poll until the notification has been applied, so + // the poll must compare against metadata learned from a notification. + var allowPoll testsignal.TestSignal[struct{}] + allowPoll.Init(t) + var firstPoll sync.Once + producer.exec = &queueGetFuncExecutor{ + Executor: producer.exec, + queueGetFunc: func(ctx context.Context, params *riverdriver.QueueGetParams) (*rivertype.Queue, error) { + firstPoll.Do(func() { + select { + case <-allowPoll.WaitC(): + case <-ctx.Done(): + } + }) + return bundle.exec.QueueGet(ctx, params) + }, + } + startProducer(t, ctx, ctx, producer) + + metadata := []byte(`{"version":1}`) + _, err := bundle.exec.QueueUpdate(ctx, &riverdriver.QueueUpdateParams{ + Metadata: metadata, + MetadataDoUpdate: true, + Name: producer.config.Queue, + Schema: producer.config.Schema, + }) + require.NoError(t, err) + producer.TriggerQueueControlEvent(&controlEventPayload{ + Action: controlActionMetadataChanged, + Metadata: metadata, + Queue: producer.config.Queue, + }) + producer.testSignals.MetadataChanged.WaitOrTimeout() + + allowPoll.Signal(struct{}{}) + producer.testSignals.PolledQueueConfig.WaitOrTimeout() + select { + case <-producer.testSignals.MetadataChanged.WaitC(): + t.Fatal("Polling repeated metadata already delivered by notification") + case <-time.After(100 * time.Millisecond): + } + }) + + t.Run("QueueSettingsPollingReconcilesNotifications", func(t *testing.T) { + t.Parallel() + + producer, _ := setup(t) + producer.config.NotificationPollInterval = 50 * time.Millisecond + producer.config.QueuePollInterval = 50 * time.Millisecond + startProducer(t, ctx, ctx, producer) + + // A pause and metadata update were notified, then both reverted in + // the database without another notification. The database still has + // its initial state, so comparing only successive polls misses this. + producer.TriggerQueueControlEvent(&controlEventPayload{Action: controlActionPause, Queue: producer.config.Queue}) + producer.testSignals.Paused.WaitOrTimeout() + producer.TriggerQueueControlEvent(&controlEventPayload{ + Action: controlActionMetadataChanged, Metadata: []byte(`{"version":1}`), Queue: producer.config.Queue, + }) + producer.testSignals.MetadataChanged.WaitOrTimeout() + + producer.testSignals.Resumed.WaitOrTimeout() + producer.testSignals.MetadataChanged.WaitOrTimeout() + // Further polls of unchanged settings must not emit duplicate events. + select { + case <-producer.testSignals.MetadataChanged.WaitC(): + t.Fatal("Unchanged metadata was applied again") + case <-time.After(100 * time.Millisecond): + } + producer.testSignals.Paused.RequireEmpty() + producer.testSignals.Resumed.RequireEmpty() + }) } func TestProducer_jitteredFetchPollInterval(t *testing.T) { @@ -1384,7 +1486,7 @@ func emitQueueNotification(t *testing.T, ctx context.Context, exec riverdriver.E "action": action, } if metadata != nil { - payload["metadata"] = metadata + payload["metadata"] = json.RawMessage(metadata) } payloadBytes, err := json.Marshal(payload) diff --git a/riverdriver/riverdrivertest/driver_client_test.go b/riverdriver/riverdrivertest/driver_client_test.go index a044c870b..1afb65832 100644 --- a/riverdriver/riverdrivertest/driver_client_test.go +++ b/riverdriver/riverdrivertest/driver_client_test.go @@ -1778,6 +1778,42 @@ func ExerciseClient[TTx any](ctx context.Context, t *testing.T, require.Equal(t, insertRes2.Job.ID, event.Job.ID) }) + t.Run("QueuePauseAndResumeWithoutNotifications", func(t *testing.T) { + t.Parallel() + + config, bundle := setupConfig(t) + config.PollOnly = true + client, err := river.NewClient(bundle.driver, config) + require.NoError(t, err) + events := subscribe(t, client) + startClient(ctx, t, client) + + // Driver operations persist controls without sending notifications. + // Exercise the polling path on every supported database and driver. + require.NoError(t, bundle.exec.QueuePause(ctx, &riverdriver.QueuePauseParams{ + Name: river.QueueDefault, Schema: bundle.schema, + })) + require.Equal(t, river.EventKindQueuePaused, riversharedtest.WaitOrTimeout(t, events).Kind) + inserted, err := client.Insert(ctx, &noOpArgs{}, nil) + require.NoError(t, err) + select { + case event := <-events: + t.Fatalf("Unexpected event while queue was paused: %+v", event) + case <-time.After(100 * time.Millisecond): + } + job, err := client.JobGet(ctx, inserted.Job.ID) + require.NoError(t, err) + require.Equal(t, rivertype.JobStateAvailable, job.State) + + require.NoError(t, bundle.exec.QueueResume(ctx, &riverdriver.QueueResumeParams{ + Name: river.QueueDefault, Schema: bundle.schema, + })) + require.Equal(t, river.EventKindQueueResumed, riversharedtest.WaitOrTimeout(t, events).Kind) + completed := riversharedtest.WaitOrTimeout(t, events) + require.Equal(t, river.EventKindJobCompleted, completed.Kind) + require.Equal(t, inserted.Job.ID, completed.Job.ID) + }) + t.Run("QueueUpdate", func(t *testing.T) { t.Parallel() diff --git a/ruby/spec/notification_receiver_shared_examples.rb b/ruby/spec/notification_receiver_shared_examples.rb index eabb6c7d6..6be80aed4 100644 --- a/ruby/spec/notification_receiver_shared_examples.rb +++ b/ruby/spec/notification_receiver_shared_examples.rb @@ -255,6 +255,32 @@ def receiver_client(**options, &work) client&.stop_and_cancel end + [false, true].each do |poll_only| + it "recovers lost queue controls with poll_only=#{poll_only}" do + # Suppress both native notifications and SQLite outbox entries. The + # notifier remains configured, but only persisted state can recover. + @driver.define_singleton_method(:runtime_notify) { |*_args| } + worked = Queue.new + client = receiver_client(fetch_poll_interval: 0.01, poll_only: poll_only) { |job| worked << job.id } + events = client.subscribe(:queue_paused, :queue_resumed) + client.start + other = River::Client.new(@driver) + + other.queue_pause(:notifications) + expect(receive_signal(events).kind).to eq(:queue_paused) + row = other.insert(River::JobArgsHash.new(:notification, {}), queue: :notifications).job + expect { Timeout.timeout(0.1) { worked.pop } }.to raise_error(Timeout::Error) + expect(other.job_get(row.id).state).to eq("available") + + other.queue_resume(:notifications) + expect(receive_signal(events).kind).to eq(:queue_resumed) + expect(receive_signal(worked)).to eq(row.id) + ensure + client&.stop_and_cancel + events&.close + end + end + it "observes remote queue pause and wildcard resume without waiting for a poll" do fetches = observe_fetches worked = Queue.new diff --git a/rust/CHANGELOG.md b/rust/CHANGELOG.md index 47175ed11..a1247d628 100644 --- a/rust/CHANGELOG.md +++ b/rust/CHANGELOG.md @@ -13,6 +13,7 @@ Changes to River for Go are recorded in the [repository changelog](../CHANGELOG. ### Fixed - Align migration CLI output with Go: `migrate-list` shows every bundled version and name, marking the latest applied version with `*`; migration progress uses the Go format and reports when nothing remains to apply; successful validation is silent. SQL output now requires `--show-sql`, including with `--dry-run`. [PR #1580](https://github.com/riverqueue/river/pull/1580). +- Recover queue pause, resume, and metadata changes whose notifications were lost by polling persisted settings once a minute, including while the listener remains disconnected. [Issue #1579](https://github.com/riverqueue/river/issues/1579). ## [0.5.0] - 2026-11-02 diff --git a/rust/riverqueue/src/__private.rs b/rust/riverqueue/src/__private.rs index 4828369c6..8f329b348 100644 --- a/rust/riverqueue/src/__private.rs +++ b/rust/riverqueue/src/__private.rs @@ -109,13 +109,12 @@ pub trait ClientBuilderExt: Sized { /// jobs, should reject that configuration the same way. fn leader_election_disabled(&self) -> bool; - /// Sets how often a client with a notifier checks its running jobs for - /// cancellation requests in case a notification was lost, a minute by - /// default. Checks happen on the two-second poll for queue settings, so - /// shorter intervals check that often. It's a control for tests, not a - /// tuning option. + /// Sets how often a client with a notifier checks for lost job cancellation + /// and queue control notifications, a minute by default. Checks happen on + /// the two-second poll for queue settings, so shorter intervals check that + /// often. It's a control for tests, not a tuning option. #[must_use] - fn job_cancel_poll_interval(self, interval: Duration) -> Self; + fn notification_poll_interval(self, interval: Duration) -> Self; /// Installs a pilot from a companion crate. #[must_use] @@ -129,15 +128,15 @@ pub trait ClientBuilderExt: Sized { } impl ClientBuilderExt for crate::ClientBuilder { - fn job_cancel_poll_interval(mut self, interval: Duration) -> Self { - self.job_cancel_poll_interval = interval; - self - } - fn leader_election_disabled(&self) -> bool { self.leader_election_disabled } + fn notification_poll_interval(mut self, interval: Duration) -> Self { + self.notification_poll_interval = interval; + self + } + fn pilot(self, pilot: P) -> Self { self.with_pilot(pilot) } diff --git a/rust/riverqueue/src/client/builder.rs b/rust/riverqueue/src/client/builder.rs index 8fe8f3a5b..3ac95a09f 100644 --- a/rust/riverqueue/src/client/builder.rs +++ b/rust/riverqueue/src/client/builder.rs @@ -350,11 +350,11 @@ pub struct ClientBuilder { pub(super) hooks: Vec>, pub(super) id: String, pub(super) insert_middleware: Vec>, - pub(crate) job_cancel_poll_interval: Duration, pub(super) job_stuck_threshold: Duration, pub(super) job_timeout: Option, pub(crate) leader_election_disabled: bool, pub(super) maintenance: MaintenanceConfig, + pub(crate) notification_poll_interval: Duration, pub(super) periodic_jobs: Vec, pub(super) pilot: Arc, pub(super) poll_only: bool, @@ -815,13 +815,13 @@ impl ClientBuilder { id: self.id, insert_middleware: self.insert_middleware, insert_notify_limiter: InsertNotifyLimiter::new(self.fetch_cooldown), - job_cancel_poll_interval: self.job_cancel_poll_interval, job_stuck_threshold: self.job_stuck_threshold, job_timeout: self.job_timeout, leader_election_disabled: self.leader_election_disabled, leadership_wakeups, live_queues: watch::channel(std::collections::HashSet::new()).0, maintenance, + notification_poll_interval: self.notification_poll_interval, #[cfg(test)] notifier_start_panics: AtomicU64::new(0), peer_owners: Mutex::new(HashMap::new()), diff --git a/rust/riverqueue/src/client/mod.rs b/rust/riverqueue/src/client/mod.rs index 2a5f2fbd6..8bbcbfe9c 100644 --- a/rust/riverqueue/src/client/mod.rs +++ b/rust/riverqueue/src/client/mod.rs @@ -79,10 +79,9 @@ use crate::{ pub(crate) const ATTEMPTED_BY_MAX: i32 = 100; const EVENT_BUFFER_CAPACITY: usize = 10_000; -/// How often a client with a notifier checks its running jobs for -/// cancellation requests whose notification was lost, like River Go's -/// producers. -const JOB_CANCEL_POLL_INTERVAL_DEFAULT: Duration = Duration::from_mins(1); +/// How often a client with a notifier checks for lost job cancellation +/// and queue control notifications, like River Go's producers. +const NOTIFICATION_POLL_INTERVAL_DEFAULT: Duration = Duration::from_mins(1); const PENDING_CANCELLATION_LIMIT: usize = 10_000; const PENDING_CANCELLATION_RETENTION: Duration = Duration::from_mins(1); // Large queues otherwise become limited by a single Postgres claim round trip. @@ -126,9 +125,6 @@ pub(crate) struct ClientInner { /// Skips a queue's insert notification sent within the fetch cooldown /// of the previous one. pub(crate) insert_notify_limiter: InsertNotifyLimiter, - /// How often a client with a notifier checks its running jobs for - /// cancellation requests in case a notification was lost. - job_cancel_poll_interval: Duration, job_stuck_threshold: Duration, pub(crate) job_timeout: Option, leader_election_disabled: bool, @@ -140,6 +136,8 @@ pub(crate) struct ClientInner { /// names reserved until they stop. live_queues: watch::Sender>, pub(crate) maintenance: MaintenanceConfig, + /// Fallback for lost job cancellation and queue control notifications. + notification_poll_interval: Duration, /// Notification listener starts that panic before doing anything, so /// tests can exercise the supervisor's restart path. #[cfg(test)] @@ -317,11 +315,11 @@ impl Client { hooks: Vec::new(), id: default_client_id(), insert_middleware: Vec::new(), - job_cancel_poll_interval: JOB_CANCEL_POLL_INTERVAL_DEFAULT, job_stuck_threshold: JOB_STUCK_THRESHOLD_DEFAULT, job_timeout: Some(JOB_TIMEOUT_DEFAULT), leader_election_disabled: false, maintenance: MaintenanceConfig::default(), + notification_poll_interval: NOTIFICATION_POLL_INTERVAL_DEFAULT, periodic_jobs: Vec::new(), pilot: Arc::new(NoopPilot), poll_only: false, diff --git a/rust/riverqueue/src/client/producer.rs b/rust/riverqueue/src/client/producer.rs index aebf8d65a..7cdc52029 100644 --- a/rust/riverqueue/src/client/producer.rs +++ b/rust/riverqueue/src/client/producer.rs @@ -145,7 +145,7 @@ fn jittered_fetch_poll_interval(interval: Duration) -> Duration { /// request, checking every [`QUEUE_CONFIG_POLL_INTERVAL`] while the client /// has no notifier, like River Go's producers without one. With a notifier, /// it's a fallback for a notification that was lost, like while the notifier -/// was reconnecting, so it checks only every `job_cancel_poll_interval`. +/// was reconnecting, so it checks only every `notification_poll_interval`. /// Whether the client has a notifier is checked each time, since a Postgres /// server's lack of `LISTEN`/`NOTIFY` is detected only once the client runs. /// It keeps checking after `stopping` fires, so jobs can still be cancelled @@ -158,7 +158,7 @@ async fn poll_job_cancellations(inner: Arc, stopping: CancellationT let mut last_checked = tokio::time::Instant::now(); loop { poll.tick().await; - if inner.has_notifier() && last_checked.elapsed() < inner.job_cancel_poll_interval { + if inner.has_notifier() && last_checked.elapsed() < inner.notification_poll_interval { continue; } last_checked = tokio::time::Instant::now(); @@ -731,6 +731,7 @@ pub(super) async fn run_queue( tokio::pin!(poll); let mut queue_config_poll = tokio::time::interval(QUEUE_CONFIG_POLL_INTERVAL); queue_config_poll.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + let mut last_queue_poll = tokio::time::Instant::now(); let outcome = loop { if let Some(failure) = attempts.failure.take() { @@ -761,13 +762,17 @@ pub(super) async fn run_queue( // More capacity may allow a claim now. (true, false) }, - // Like Go's `pollForSettingChanges`, only a client without a - // notifier polls queue state; one with a notifier learns of - // changes from control notifications, and a listener reconnect - // broadcasts a refresh of every queue. Whether a Postgres - // server supports notifications is detected only once the client - // runs, so this is checked on each tick. - _ = queue_config_poll.tick() => (false, !inner.has_notifier()), + // Notifications and reconnects normally refresh settings immediately. + // Keep a slower fallback too, including while the listener stays + // disconnected. Server capabilities are detected after startup. + _ = queue_config_poll.tick() => { + let refresh = !inner.has_notifier() + || last_queue_poll.elapsed() >= inner.notification_poll_interval; + if refresh { + last_queue_poll = tokio::time::Instant::now(); + } + (false, refresh) + }, () = &mut poll => { poll.as_mut().reset( tokio::time::Instant::now() diff --git a/rust/riverqueue/tests/poll_only_cancellation.rs b/rust/riverqueue/tests/poll_only_cancellation.rs index 5fb002de9..117cfbf17 100644 --- a/rust/riverqueue/tests/poll_only_cancellation.rs +++ b/rust/riverqueue/tests/poll_only_cancellation.rs @@ -1,4 +1,4 @@ -//! Cancellation of running jobs found by polling, in clients without +//! Running job cancellations and queue controls found by polling, in clients without //! notifications and in clients whose notification was lost, on every //! backend. //! @@ -11,12 +11,16 @@ mod support; use std::{convert::Infallible, sync::Arc, time::Duration}; +use async_trait::async_trait; use riverqueue::{ - __private::{ClientBuilderExt, Database}, + __private::{ + ClientBuilderExt, Database, Pilot, PilotError, PilotProducer, ProducerConfiguration, + ProducerStartContext, + }, Client, EventKind, Job, JobArgs, JobState, QueueConfig, WorkContext, WorkOutcome, Workers, }; use serde::{Deserialize, Serialize}; -use tokio::sync::Semaphore; +use tokio::sync::{Semaphore, mpsc}; /// Every wait in these tests is bounded by this timeout. It covers a few of /// the two-second polls for cancellation requests. @@ -88,6 +92,60 @@ impl Backend { } } } + /// Changes queue state without notifying, as if the listener were down. + async fn set_paused_without_notification(&self, paused: bool) { + match self { + #[cfg(all(feature = "postgres", river_postgres_tests))] + Self::Postgres(schema) => { + sqlx::query(sqlx::AssertSqlSafe(format!( + "UPDATE {} SET paused_at = CASE WHEN $1 THEN CURRENT_TIMESTAMP ELSE NULL END, \ + metadata = jsonb_build_object('paused', $1::boolean) WHERE name = 'default'", + schema.table("river_queue") + ))) + .bind(paused) + .execute(&schema.pool) + .await + .unwrap(); + } + #[cfg(feature = "sqlite")] + Self::Sqlite(pool, _) => { + sqlx::query( + "UPDATE river_queue SET paused_at = CASE WHEN ? THEN \ + strftime('%Y-%m-%dT%H:%M:%fZ', 'now') ELSE NULL END, \ + metadata = jsonb_object('paused', json(CASE WHEN ? THEN 'true' ELSE 'false' END)) \ + WHERE name = 'default'", + ) + .bind(paused) + .bind(paused) + .execute(pool) + .await + .unwrap(); + } + } + } +} + +/// Records metadata offered to a producer session by queue state recovery. +#[derive(Clone)] +struct QueueSettingsPilot { + metadata: mpsc::UnboundedSender>, +} + +#[async_trait] +impl Pilot for QueueSettingsPilot { + async fn start_producer( + &self, + _context: ProducerStartContext, + ) -> Result>, PilotError> { + Ok(Some(Box::new(self.clone()))) + } +} + +#[async_trait] +impl PilotProducer for QueueSettingsPilot { + fn configuration_changed(&self, configuration: &ProducerConfiguration) { + let _ = self.metadata.send(configuration.queue.metadata.clone()); + } } /// Workers that block until their attempt is cancelled, adding a permit to @@ -163,7 +221,7 @@ async fn checks_for_lost_cancellations_only_after_interval(backend: Backend) { async fn polls_for_lost_cancellation_notification(backend: Backend) { let started = Arc::new(Semaphore::new(0)); let client = Client::builder(backend.database()) - .job_cancel_poll_interval(Duration::ZERO) + .notification_poll_interval(Duration::ZERO) .queue(riverqueue::QUEUE_DEFAULT, QueueConfig::new(1)) .without_leader_election() .workers(waiting_workers(&started)) @@ -195,6 +253,78 @@ async fn polls_for_lost_cancellation_notification(backend: Backend) { backend.cleanup().await; } +/// A notifier must not disable polling of queue controls: a lost pause +/// stops claims and a lost resume starts them again without reconnecting. +async fn polls_for_lost_queue_controls(backend: Backend, poll_only: bool) { + let started = Arc::new(Semaphore::new(0)); + let (metadata, mut metadata_updates) = mpsc::unbounded_channel(); + let builder = Client::builder(backend.database()) + .pilot(QueueSettingsPilot { metadata }) + .notification_poll_interval(Duration::ZERO) + .queue(riverqueue::QUEUE_DEFAULT, QueueConfig::new(1)) + .without_leader_election() + .workers(waiting_workers(&started)); + let builder = if poll_only { + builder.without_notifications() + } else { + builder + }; + let client = builder.build().unwrap(); + let mut events = client + .subscribe(&[EventKind::QueuePaused, EventKind::QueueResumed]) + .unwrap(); + let mut run = client.start().unwrap(); + run.wait_ready().await.unwrap(); + + backend.set_paused_without_notification(true).await; + let paused = tokio::time::timeout(TIMEOUT, events.recv()) + .await + .unwrap() + .unwrap(); + assert_eq!(paused.kind(), EventKind::QueuePaused); + assert_eq!( + serde_json::Value::Object( + tokio::time::timeout(TIMEOUT, metadata_updates.recv()) + .await + .unwrap() + .unwrap() + ), + serde_json::json!({"paused": true}) + ); + client.insert(WaitArgs {}).await.unwrap(); + assert!( + tokio::time::timeout(Duration::from_millis(100), started.acquire()) + .await + .is_err() + ); + + backend.set_paused_without_notification(false).await; + let resumed = tokio::time::timeout(TIMEOUT, events.recv()) + .await + .unwrap() + .unwrap(); + assert_eq!(resumed.kind(), EventKind::QueueResumed); + assert_eq!( + serde_json::Value::Object( + tokio::time::timeout(TIMEOUT, metadata_updates.recv()) + .await + .unwrap() + .unwrap() + ), + serde_json::json!({"paused": false}) + ); + tokio::time::timeout(TIMEOUT, started.acquire()) + .await + .unwrap() + .unwrap() + .forget(); + tokio::time::timeout(TIMEOUT, run.stop_and_cancel()) + .await + .unwrap() + .unwrap(); + backend.cleanup().await; +} + /// A client without notifications cancels its running job once it polls the /// cancellation another client requested, including while it's stopping and /// waiting for that job. @@ -259,6 +389,17 @@ mod postgres { .await; } + #[tokio::test] + async fn polls_for_lost_queue_controls() { + for poll_only in [false, true] { + super::polls_for_lost_queue_controls( + Backend::Postgres(support::PostgresSchema::new("river_poll_queue_controls").await), + poll_only, + ) + .await; + } + } + #[tokio::test] async fn polls_for_remote_cancellation() { super::polls_for_remote_cancellation(Backend::Postgres( @@ -284,6 +425,14 @@ mod sqlite { super::polls_for_lost_cancellation_notification(Backend::Sqlite(pool, path)).await; } + #[tokio::test] + async fn polls_for_lost_queue_controls() { + for poll_only in [false, true] { + let (pool, path) = support::sqlite_file_pool(4).await; + super::polls_for_lost_queue_controls(Backend::Sqlite(pool, path), poll_only).await; + } + } + #[tokio::test] async fn polls_for_remote_cancellation() { let (pool, path) = support::sqlite_file_pool(4).await;