diff --git a/CHANGELOG.md b/CHANGELOG.md index 3a6b73c75..e5176eb2d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Fixed a running job not being cancelled after `Client.JobCancel` when the cancellation notification was lost, as happens if it's sent while the client's notifier is reconnecting. Clients with a notifier now also check their running jobs for cancellation requests once a minute. `Client.JobRetry` now clears an earlier attempt's cancellation request so that the retried job isn't cancelled again. [PR #1564](https://github.com/riverqueue/river/pull/1564). - 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). diff --git a/client.go b/client.go index e0abd8956..0cf0b4e3c 100644 --- a/client.go +++ b/client.go @@ -1504,7 +1504,9 @@ func (c *Client[TTx]) Driver() riverdriver.Driver[TTx] { // If the job is currently running, it is not immediately cancelled, but is // instead marked for cancellation. The client running the job will also be // notified to cancel the running job's context. When running without a notifier, -// clients poll for cancellation requests every two seconds. Although the job's +// clients poll for cancellation requests every two seconds. Clients with a +// notifier also poll once a minute in case a notification was missed, like +// while the notifier was reconnecting. Although the job's // context will be cancelled, since Go does not provide a mechanism to // interrupt a running goroutine the job will continue running until it returns. // As always, it is important for workers to respect context cancellation and @@ -1561,7 +1563,9 @@ func (c *Client[TTx]) JobCancel(ctx context.Context, jobID int64) (*rivertype.Jo // If the job is currently running, it is not immediately cancelled, but is // instead marked for cancellation. The client running the job will also be // notified to cancel the running job's context. When running without a notifier, -// clients poll for cancellation requests every two seconds. Although the job's +// clients poll for cancellation requests every two seconds. Clients with a +// notifier also poll once a minute in case a notification was missed, like +// while the notifier was reconnecting. Although the job's // context will be cancelled, since Go does not provide a mechanism to // interrupt a running goroutine the job will continue running until it returns. // As always, it is important for workers to respect context cancellation and diff --git a/java/CHANGELOG.md b/java/CHANGELOG.md index 4d45e1841..963e4c7c4 100644 --- a/java/CHANGELOG.md +++ b/java/CHANGELOG.md @@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed + +- Fixed a cancelled job that was retried being cancelled again as soon as its new attempt started. Retrying a job now clears its earlier attempt's cancellation request, like River for Go. [PR #1564](https://github.com/riverqueue/river/pull/1564). + ## [0.2.0] - 2026-10-26 ### Changed diff --git a/java/river/src/main/resources/com/riverqueue/postgres.sql b/java/river/src/main/resources/com/riverqueue/postgres.sql index d5a92e3ca..517578057 100644 --- a/java/river/src/main/resources/com/riverqueue/postgres.sql +++ b/java/river/src/main/resources/com/riverqueue/postgres.sql @@ -56,7 +56,7 @@ DELETE FROM {schema}river_job WHERE id=? AND state != 'running' RETURNING {colum -- name: delete_filtered DELETE FROM {schema}river_job WHERE {where} AND id=? AND state != 'running' RETURNING {columns} -- name: retry -UPDATE {schema}river_job SET state='available', finalized_at=NULL, +UPDATE {schema}river_job SET state='available', finalized_at=NULL, metadata=metadata - 'cancel_attempted_at', max_attempts=CASE WHEN attempt=max_attempts THEN max_attempts+1 ELSE max_attempts END, scheduled_at=? WHERE id=? AND state != 'running' AND NOT (state='available' AND scheduled_at < ?) RETURNING {columns} -- name: reset_jobs diff --git a/java/river/src/main/resources/com/riverqueue/sqlite.sql b/java/river/src/main/resources/com/riverqueue/sqlite.sql index 75e2d53b0..dc140d684 100644 --- a/java/river/src/main/resources/com/riverqueue/sqlite.sql +++ b/java/river/src/main/resources/com/riverqueue/sqlite.sql @@ -55,7 +55,7 @@ DELETE FROM {schema}river_job WHERE id=? AND state != 'running' RETURNING {colum -- name: delete_filtered DELETE FROM {schema}river_job WHERE {where} AND id=? AND state != 'running' RETURNING {columns} -- name: retry -UPDATE {schema}river_job SET state='available', finalized_at=NULL, +UPDATE {schema}river_job SET state='available', finalized_at=NULL, metadata=jsonb_remove(metadata,'$.cancel_attempted_at'), max_attempts=CASE WHEN attempt=max_attempts THEN max_attempts+1 ELSE max_attempts END, scheduled_at=? WHERE id=? AND state != 'running' AND NOT (state='available' AND scheduled_at < ?) RETURNING {columns} -- name: reset_jobs diff --git a/java/river/src/test/java/com/riverqueue/RetryTest.java b/java/river/src/test/java/com/riverqueue/RetryTest.java index 683b2a0eb..fb76a4d2c 100644 --- a/java/river/src/test/java/com/riverqueue/RetryTest.java +++ b/java/river/src/test/java/com/riverqueue/RetryTest.java @@ -9,6 +9,7 @@ import java.time.Instant; import java.time.ZoneOffset; import java.util.List; +import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; @@ -152,6 +153,24 @@ public List claim( assertTrue(errors.isEmpty(), () -> "Unexpected worker errors: " + errors); } + // Like Go, a retry clears a cancellation requested during an earlier attempt + // so that it can't cancel the retried one. Other metadata is preserved. + @Test + void retryClearsAnEarlierCancellationRequest() { + var client = new Client(databases.open(directory.resolve("river.db"))); + var inserted = + client + .insert( + TYPE, "retry", InsertOptions.builder().metadata(Map.of("preserved", true)).build()) + .job(); + assertTrue(client.cancel(inserted.id()).metadata().has("cancel_attempted_at")); + + var retried = client.retry(inserted.id()); + assertEquals(Job.State.AVAILABLE, retried.state()); + assertFalse(retried.metadata().has("cancel_attempted_at")); + assertTrue(retried.metadata().path("preserved").asBoolean()); + } + @ParameterizedTest @ValueSource(booleans = {false, true}) void rolledBackRetriesPreserveEarlierCommittedNotifications(boolean earlierInsert) { diff --git a/js/CHANGELOG.md b/js/CHANGELOG.md index 1a8f980cb..124185cc4 100644 --- a/js/CHANGELOG.md +++ b/js/CHANGELOG.md @@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed + +- Fixed a cancelled job that was retried being cancelled again when the client checked for cancellation requests, which it does while polling and after its notification stream reconnects. Retrying a job now clears its earlier attempt's cancellation request, like River for Go. [PR #1564](https://github.com/riverqueue/river/pull/1564). + ## [0.3.0] - 2026-10-26 ### Added diff --git a/js/driver/pg/src/driver.integration.test.ts b/js/driver/pg/src/driver.integration.test.ts index 99588eebe..de12c27e9 100644 --- a/js/driver/pg/src/driver.integration.test.ts +++ b/js/driver/pg/src/driver.integration.test.ts @@ -564,6 +564,19 @@ describe("PgDriver integration", () => { expect(unchanged!.state).toBe("running"); }); + it("clears an earlier attempt's cancellation request on retry", async () => { + const inserted = await driver.jobInsert( + insertParams(`${filePrefix}_retry_cancelled`) + ); + const cancelled = await driver.jobCancel(inserted.job.id); + expect(cancelled!.metadata.cancel_attempted_at).toBeDefined(); + + const retried = await driver.jobRetry(inserted.job.id); + + expect(retried!.state).toBe("available"); + expect(retried!.metadata.cancel_attempted_at).toBeUndefined(); + }); + it("returns the committed row to the loser of a cancel or retry race", async () => { const inserted = await driver.jobInsert( insertParams(`${filePrefix}_race`, { diff --git a/js/driver/pg/src/sql/jobs.ts b/js/driver/pg/src/sql/jobs.ts index 37b91ecb6..928c14fc4 100644 --- a/js/driver/pg/src/sql/jobs.ts +++ b/js/driver/pg/src/sql/jobs.ts @@ -827,6 +827,9 @@ export async function jobRetryWithOptions( ELSE river_job.max_attempts END, finalized_at = NULL, + -- A cancellation requested during an earlier attempt must not cancel + -- the next one. + metadata = river_job.metadata - 'cancel_attempted_at', scheduled_at = coalesce($2::timestamptz, now()) FROM job_to_update WHERE river_job.id = job_to_update.id diff --git a/js/driver/sqlite/src/compatibility.test.ts b/js/driver/sqlite/src/compatibility.test.ts index e73c52129..de5a208a3 100644 --- a/js/driver/sqlite/src/compatibility.test.ts +++ b/js/driver/sqlite/src/compatibility.test.ts @@ -629,11 +629,12 @@ describe("SqliteDriver compatibility with Go's riversqlite", () => { const retried = await insert("golden_retry", "available"); await driver.jobCancelDetailed(retried, { now }); await driver.jobRetryDetailed(retried, { now: now.add({ hours: 1 }) }); + // Like Go, a retry clears the earlier cancellation request. expect(stored("golden_retry")).toEqual({ errors: null, errors_null: 1n, finalized_at: null, - metadata: cancelled, + metadata: { m: "x" }, scheduled_at: "2026-01-02 06:06:07.123", state: "available", }); diff --git a/js/driver/sqlite/src/driver.test.ts b/js/driver/sqlite/src/driver.test.ts index a9a41e008..f817367b4 100644 --- a/js/driver/sqlite/src/driver.test.ts +++ b/js/driver/sqlite/src/driver.test.ts @@ -326,8 +326,11 @@ describe("SqliteDriver", () => { const retried = await driver.jobRetryDetailed(inserted.id, { now }); expect(retried.status).toBe("retried"); - if (retried.status === "retried") + if (retried.status === "retried") { expect(retried.job.state).toBe("available"); + // An earlier attempt's cancellation must not cancel the retried one. + expect(retried.job.metadata.cancel_attempted_at).toBeUndefined(); + } expect((await driver.jobRetryDetailed(inserted.id, { now })).status).toBe( "unchanged" ); diff --git a/js/driver/sqlite/src/driver.ts b/js/driver/sqlite/src/driver.ts index 98a0891c5..bb5fc30cb 100644 --- a/js/driver/sqlite/src/driver.ts +++ b/js/driver/sqlite/src/driver.ts @@ -1446,6 +1446,9 @@ export class SqliteRuntime implements RuntimeDriver< ELSE max_attempts END, finalized_at = NULL, + -- A cancellation requested during an earlier attempt must not + -- cancel the next one. + metadata = jsonb_remove(metadata, '$.cancel_attempted_at'), scheduled_at = ? WHERE id = ? AND state != 'running' diff --git a/producer.go b/producer.go index fd5a4e9f5..a660f7a78 100644 --- a/producer.go +++ b/producer.go @@ -38,6 +38,7 @@ import ( const ( jobCancelPollBatchSize = 1000 + jobCancelPollIntervalDefault = time.Minute producerReportIntervalDefault = 30 * time.Second queuePollIntervalDefault = 2 * time.Second queueReportIntervalDefault = 10 * time.Minute @@ -91,6 +92,13 @@ 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 @@ -112,8 +120,8 @@ type producerConfig struct { QueueEventCallback func(event *Event) // QueuePollInterval is the amount of time between periodic checks for - // queue setting changes and job cancellation requests. This is only used in - // poll-only mode (when no notifier is provided). + // queue setting changes and job cancellation requests in poll-only mode + // (when no notifier is provided). QueuePollInterval time.Duration // QueueReportInterval is the amount of time between periodic reports // of the queue status. @@ -138,6 +146,12 @@ 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.JobStuckCount == nil { c.JobStuckCount = &atomic.Int32{} } @@ -199,7 +213,8 @@ type producer struct { startstop.BaseStartStop // Jobs which are currently being worked. The polling goroutine snapshots - // their IDs; only the main goroutine accesses the executors themselves. + // them and cancels the snapshotted executors it finds cancellation + // requests for; otherwise only the main goroutine accesses executors. activeJobsMu sync.Mutex activeJobs map[int64]*jobexecutor.JobExecutor @@ -215,8 +230,8 @@ type producer struct { pilot riverpilot.Pilot workers *Workers - // Receives job IDs to cancel. Written by notifier and polling goroutines, - // only read from the main goroutine. + // Receives job IDs to cancel. Written by the notifier goroutine, only read + // from the main goroutine. cancelCh chan int64 // Set to true when the producer thinks it should trigger another fetch as @@ -442,14 +457,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. + subroutineWG.Add(1) + go p.pollForJobCancellations(subroutineCtx, &subroutineWG) + 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) - - subroutineWG.Add(1) - go p.pollForJobCancellations(subroutineCtx, &subroutineWG) } p.fetchAndRunLoop(fetchCtx, workCtx) @@ -1030,10 +1046,19 @@ func (p *producer) handleWorkerDone(job *rivertype.JobRow) { // Cancellation polling is independent of fetching: saturated and paused producers // must still cancel jobs, including while draining during graceful shutdown. +// +// Without a notifier, polling is the only way to learn about cancellations. +// 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. func (p *producer) pollForJobCancellations(ctx context.Context, wg *sync.WaitGroup) { defer wg.Done() - ticker := time.NewTicker(p.config.QueuePollInterval) + interval := p.config.QueuePollInterval + if p.config.Notifier != nil { + interval = p.config.JobCancelPollInterval + } + + ticker := time.NewTicker(interval) defer ticker.Stop() for { select { @@ -1051,9 +1076,14 @@ func (p *producer) pollForJobCancellations(ctx context.Context, wg *sync.WaitGro } func (p *producer) pollForJobCancellationsOnce(ctx context.Context) error { + // Snapshot executors rather than IDs. By the time a query returns, a job + // could have finished and been retried, with a new attempt running under + // the same ID that must not be cancelled. Cancelling an executor that's + // already finished only logs a warning. p.activeJobsMu.Lock() - jobIDs := slices.Collect(maps.Keys(p.activeJobs)) + executors := maps.Clone(p.activeJobs) p.activeJobsMu.Unlock() + jobIDs := slices.Collect(maps.Keys(executors)) // Bound query size (including SQLite's parameter count) and skip database // access entirely when there are no active jobs. @@ -1069,10 +1099,8 @@ func (p *producer) pollForJobCancellationsOnce(ctx context.Context) error { } for _, id := range cancelIDs { - select { - case <-ctx.Done(): - return ctx.Err() - case p.cancelCh <- id: + if executor := executors[id]; executor != nil { + executor.Cancel(ctx) } } } diff --git a/producer_test.go b/producer_test.go index 8d8545b7a..368a7080a 100644 --- a/producer_test.go +++ b/producer_test.go @@ -13,6 +13,7 @@ import ( "github.com/stretchr/testify/require" "github.com/riverqueue/river/internal/jobcompleter" + "github.com/riverqueue/river/internal/jobexecutor" "github.com/riverqueue/river/internal/notifier" "github.com/riverqueue/river/internal/pluginlookup" "github.com/riverqueue/river/internal/rivercommon" @@ -76,6 +77,18 @@ func (p *beforeJobGetAvailablePilot) JobGetAvailable( return p.Pilot.JobGetAvailable(ctx, exec, state, params) } +// jobGetCancelRequestedFuncExecutor replaces JobGetCancelRequested on the +// wrapped executor. +type jobGetCancelRequestedFuncExecutor struct { + riverdriver.Executor + + jobGetCancelRequestedFunc func(params *riverdriver.JobGetCancelRequestedParams) ([]int64, error) +} + +func (e *jobGetCancelRequestedFuncExecutor) JobGetCancelRequested(ctx context.Context, params *riverdriver.JobGetCancelRequestedParams) ([]int64, error) { + return e.jobGetCancelRequestedFunc(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. @@ -550,12 +563,18 @@ func testProducer(t *testing.T, makeProducer func(ctx context.Context, t *testin t.Parallel() producer, bundle := setup(t) - if producer.config.Notifier != nil { - t.Skip("requires polling without a notifier") - } producer.config.FetchPollInterval = time.Hour producer.config.MaxWorkers = 1 - producer.config.QueuePollInterval = 20 * time.Millisecond + + // 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.QueuePollInterval = 20 * time.Millisecond + } else { + producer.config.JobCancelPollInterval = 20 * time.Millisecond + producer.config.QueuePollInterval = time.Hour + } var jobStarted testsignal.TestSignal[int64] var workerErr testsignal.TestSignal[error] @@ -588,19 +607,25 @@ func testProducer(t *testing.T, makeProducer func(ctx context.Context, t *testin Name: producer.config.Queue, Schema: producer.config.Schema, })) + if producer.config.Notifier != nil { + // Only Client.QueuePause sends a notification. + producer.TriggerQueueControlEvent(&controlEventPayload{Action: controlActionPause, Queue: producer.config.Queue}) + } producer.testSignals.Paused.WaitOrTimeout() case "Stopping": fetchCancel() producer.testSignals.ExecutorShutdownStarted.WaitOrTimeout() } - // Use the executor directly so no local client shortcut can cancel - // the worker. No more jobs can be fetched in any of these states. - _, err := bundle.exec.JobCancel(ctx, &riverdriver.JobCancelParams{ - ID: jobID, - CancelAttemptedAt: time.Now().UTC(), - ControlTopic: string(notifier.NotificationTopicControl), - Schema: producer.config.Schema, + // Request cancellation the way JobCancel does, but without a + // notification, as if it was lost while the notifier was + // reconnecting. Only polling can find it. No more jobs can be + // fetched in any of these states. + _, err := bundle.exec.JobUpdate(ctx, &riverdriver.JobUpdateParams{ + ID: jobID, + Metadata: fmt.Appendf(nil, `{"cancel_attempted_at":%q}`, time.Now().UTC().Format(time.RFC3339Nano)), + MetadataDoMerge: true, + Schema: producer.config.Schema, }) require.NoError(t, err) require.ErrorIs(t, workerErr.WaitOrTimeout(), rivertype.ErrJobCancelledRemotely) @@ -610,6 +635,48 @@ func testProducer(t *testing.T, makeProducer func(ctx context.Context, t *testin } }) + // 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. + t.Run("CancellationPollingCancelsCheckedAttempt", func(t *testing.T) { + t.Parallel() + + producer, bundle := setup(t) + + var ( + job = &rivertype.JobRow{ID: 123} + checkedCause error + retriedCause error + ) + checkedExecutor := baseservice.Init(bundle.archetype, &jobexecutor.JobExecutor{ + CancelFunc: func(cause error) { checkedCause = cause }, + JobRow: job, + }) + retriedExecutor := baseservice.Init(bundle.archetype, &jobexecutor.JobExecutor{ + CancelFunc: func(cause error) { retriedCause = cause }, + JobRow: job, + }) + producer.addActiveJob(job.ID, checkedExecutor) + + producer.exec = &jobGetCancelRequestedFuncExecutor{ + Executor: producer.exec, + jobGetCancelRequestedFunc: func(params *riverdriver.JobGetCancelRequestedParams) ([]int64, error) { + require.Equal(t, []int64{job.ID}, params.ID) + + // The checked attempt finishes and a retried one starts. + producer.activeJobsMu.Lock() + producer.activeJobs[job.ID] = retriedExecutor + producer.activeJobsMu.Unlock() + + return []int64{job.ID}, nil + }, + } + + require.NoError(t, producer.pollForJobCancellationsOnce(ctx)) + require.ErrorIs(t, checkedCause, rivertype.ErrJobCancelledRemotely) + require.NoError(t, retriedCause) + }) + // A locked job whose row can't be decoded isn't worked. Its attempt fails // with the decode error, and the other jobs locked with it are worked // normally. diff --git a/riverdriver/riverdatabasesql/internal/dbsqlc/river_job.sql.go b/riverdriver/riverdatabasesql/internal/dbsqlc/river_job.sql.go index 4bd6cbfab..0f7f8142f 100644 --- a/riverdriver/riverdatabasesql/internal/dbsqlc/river_job.sql.go +++ b/riverdriver/riverdatabasesql/internal/dbsqlc/river_job.sql.go @@ -1320,6 +1320,9 @@ updated_job AS ( state = 'available', max_attempts = CASE WHEN attempt = max_attempts THEN max_attempts + 1 ELSE max_attempts END, finalized_at = NULL, + -- A cancellation requested during an earlier attempt must not cancel + -- the next one. + metadata = metadata - 'cancel_attempted_at', scheduled_at = coalesce($2::timestamptz, now()) FROM job_to_update WHERE river_job.id = job_to_update.id diff --git a/riverdriver/riverdrivertest/job_update.go b/riverdriver/riverdrivertest/job_update.go index 150445d63..e537229a5 100644 --- a/riverdriver/riverdrivertest/job_update.go +++ b/riverdriver/riverdrivertest/job_update.go @@ -607,6 +607,30 @@ func exerciseJobUpdate[TTx any](ctx context.Context, t *testing.T, executorWithT require.WithinDuration(t, now, jobAfter.ScheduledAt, 5*time.Second) }) + // A cancellation requested during an earlier attempt must not apply to + // the retried one. Other metadata is preserved. + t.Run("ClearsCancelAttemptedAt", func(t *testing.T) { + t.Parallel() + + exec, _ := setup(ctx, t) + + now := time.Now().UTC() + + job := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{ + FinalizedAt: &now, + Metadata: []byte(`{"cancel_attempted_at":"2026-01-01T00:00:00Z","foo":"bar"}`), + State: new(rivertype.JobStateCancelled), + }) + + jobAfter, err := exec.JobRetry(ctx, &riverdriver.JobRetryParams{ + ID: job.ID, + Now: &now, + }) + require.NoError(t, err) + require.Equal(t, rivertype.JobStateAvailable, jobAfter.State) + require.JSONEq(t, `{"foo":"bar"}`, string(jobAfter.Metadata)) + }) + t.Run("DoesNotAlterScheduledAtIfInThePastAndJobAlreadyAvailable", func(t *testing.T) { // We don't want to update ScheduledAt if the job was already available // because doing so can make it lose its place in line. diff --git a/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql b/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql index d87361114..cc921c261 100644 --- a/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql +++ b/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql @@ -533,6 +533,9 @@ updated_job AS ( state = 'available', max_attempts = CASE WHEN attempt = max_attempts THEN max_attempts + 1 ELSE max_attempts END, finalized_at = NULL, + -- A cancellation requested during an earlier attempt must not cancel + -- the next one. + metadata = metadata - 'cancel_attempted_at', scheduled_at = coalesce(sqlc.narg('now')::timestamptz, now()) FROM job_to_update WHERE river_job.id = job_to_update.id diff --git a/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql.go b/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql.go index a58f06ff0..165b85af0 100644 --- a/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql.go +++ b/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql.go @@ -1284,6 +1284,9 @@ updated_job AS ( state = 'available', max_attempts = CASE WHEN attempt = max_attempts THEN max_attempts + 1 ELSE max_attempts END, finalized_at = NULL, + -- A cancellation requested during an earlier attempt must not cancel + -- the next one. + metadata = metadata - 'cancel_attempted_at', scheduled_at = coalesce($2::timestamptz, now()) FROM job_to_update WHERE river_job.id = job_to_update.id diff --git a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql index 8ef471786..fd239638e 100644 --- a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql +++ b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql @@ -588,6 +588,9 @@ SET state = 'available', max_attempts = CASE WHEN attempt = max_attempts THEN max_attempts + 1 ELSE max_attempts END, finalized_at = NULL, + -- A cancellation requested during an earlier attempt must not cancel the + -- next one. + metadata = jsonb_remove(metadata, '$.cancel_attempted_at'), scheduled_at = coalesce(cast(sqlc.narg('now') AS text), datetime('now', 'subsec')) WHERE id = @id -- Do not touch running jobs: diff --git a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go index 8fc19993d..2e4718390 100644 --- a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go +++ b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go @@ -1452,6 +1452,9 @@ SET state = 'available', max_attempts = CASE WHEN attempt = max_attempts THEN max_attempts + 1 ELSE max_attempts END, finalized_at = NULL, + -- A cancellation requested during an earlier attempt must not cancel the + -- next one. + metadata = jsonb_remove(metadata, '$.cancel_attempted_at'), scheduled_at = coalesce(cast(?1 AS text), datetime('now', 'subsec')) WHERE id = ?2 -- Do not touch running jobs: diff --git a/rust/CHANGELOG.md b/rust/CHANGELOG.md index 45edf8a4d..5aa55487f 100644 --- a/rust/CHANGELOG.md +++ b/rust/CHANGELOG.md @@ -10,6 +10,10 @@ Changes to River for Go are recorded in the [repository changelog](../CHANGELOG. ## [Unreleased] +### Fixed + +- Fixed a running job not being cancelled when its cancellation notification was lost, as happens if it's sent while the client's listener is reconnecting. Clients with notifications now also check their running jobs for cancellation requests once a minute, like River for Go. Retrying a job now clears its earlier attempt's cancellation request so that the retried job isn't cancelled again. [PR #1564](https://github.com/riverqueue/river/pull/1564). + ## [0.4.0] - 2026-10-26 ### Added diff --git a/rust/riverqueue/src/__private.rs b/rust/riverqueue/src/__private.rs index 5c3f36930..9008d3c1e 100644 --- a/rust/riverqueue/src/__private.rs +++ b/rust/riverqueue/src/__private.rs @@ -109,6 +109,14 @@ 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. + #[must_use] + fn job_cancel_poll_interval(self, interval: Duration) -> Self; + /// Installs a pilot from a companion crate. #[must_use] fn pilot(self, pilot: P) -> Self; @@ -121,6 +129,11 @@ 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 } diff --git a/rust/riverqueue/src/client/builder.rs b/rust/riverqueue/src/client/builder.rs index 0c378c36d..8fe8f3a5b 100644 --- a/rust/riverqueue/src/client/builder.rs +++ b/rust/riverqueue/src/client/builder.rs @@ -350,6 +350,7 @@ 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, @@ -814,6 +815,7 @@ 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, diff --git a/rust/riverqueue/src/client/mod.rs b/rust/riverqueue/src/client/mod.rs index 68b517853..81da170df 100644 --- a/rust/riverqueue/src/client/mod.rs +++ b/rust/riverqueue/src/client/mod.rs @@ -79,6 +79,10 @@ 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); 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. @@ -122,6 +126,9 @@ 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, @@ -310,6 +317,7 @@ 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, diff --git a/rust/riverqueue/src/client/producer.rs b/rust/riverqueue/src/client/producer.rs index fa68924f9..4125785f1 100644 --- a/rust/riverqueue/src/client/producer.rs +++ b/rust/riverqueue/src/client/producer.rs @@ -16,9 +16,7 @@ use tokio_util::sync::CancellationToken; use tracing::{debug, error, warn}; use crate::__private::DatabaseConnection as PilotDatabaseConnection; -use crate::client::attempts::{ - FetchRegistrationGuard, register_running_attempt, signal_running_attempt, -}; +use crate::client::attempts::{FetchRegistrationGuard, register_running_attempt}; use crate::client::backoff::{RestartBackoff, exponential_backoff}; use crate::client::completer::CompletionUpdate; use crate::client::executor::{AbortOnDrop, execute_job}; @@ -56,8 +54,9 @@ pub(super) async fn run_dynamic_queues( mut changes: watch::Receiver, queues_ready: oneshot::Sender<()>, ) -> Result<(), Error> { - // Without a notifier, running jobs learn of cancellation requests by - // polling until every producer has drained. Whether the client has one is + // Running jobs learn of cancellation requests by polling until every + // producer has drained: regularly without a notifier, and occasionally + // with one in case a notification was lost. Whether the client has one is // known only once a Postgres server's capabilities are detected, so the // poll checks each time. let _cancellation_poll = AbortOnDrop(tokio::spawn(poll_job_cancellations( @@ -144,9 +143,11 @@ fn jittered_fetch_poll_interval(interval: Duration) -> Duration { /// Cancels this client's running attempts whose jobs have a cancellation /// request, checking every [`QUEUE_CONFIG_POLL_INTERVAL`] while the client -/// has no notifier, like River Go's producers without one. That's checked -/// each time, since a Postgres server's lack of `LISTEN`/`NOTIFY` is -/// detected only once the client runs. +/// 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`. +/// 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 /// while producers drain; the caller aborts it once they have. async fn poll_job_cancellations(inner: Arc, stopping: CancellationToken) { @@ -154,36 +155,39 @@ async fn poll_job_cancellations(inner: Arc, stopping: CancellationT poll.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); // The first tick completes at once; running jobs start later. poll.tick().await; + let mut last_checked = tokio::time::Instant::now(); loop { poll.tick().await; - if inner.has_notifier() { + if inner.has_notifier() && last_checked.elapsed() < inner.job_cancel_poll_interval { continue; } - let mut ids = inner + last_checked = tokio::time::Instant::now(); + // Snapshot attempts rather than IDs. By the time a query returns, a + // job could have finished and been retried, with a new attempt + // running under the same ID that must not be cancelled. + let mut attempts = inner .running .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) - .keys() - .copied() + .iter() + .map(|(id, cancellation)| (*id, cancellation.clone())) .collect::>(); - ids.sort_unstable(); - for batch in ids.chunks(JOB_CANCEL_POLL_BATCH_SIZE) { + attempts.sort_unstable_by_key(|(id, _)| *id); + for batch in attempts.chunks(JOB_CANCEL_POLL_BATCH_SIZE) { + let ids = batch.iter().map(|(id, _)| *id).collect::>(); let requested = tokio::time::timeout( JOB_CANCEL_POLL_TIMEOUT, - crate::storage::job_cancel_requested(&inner, batch), + crate::storage::job_cancel_requested(&inner, &ids), ) .await .map_err(|_| Error::runtime_context("job cancellation poll", "timed out")) .and_then(|requested| requested); match requested { Ok(requested) => { - for id in requested { - signal_running_attempt( - &inner.running, - &inner.pending_cancellations, - &inner.fetch_registration_windows, - id, - ); + for (id, cancellation) in batch { + if requested.contains(id) { + cancellation.cancel(); + } } } Err(poll_error) => { diff --git a/rust/riverqueue/src/database/sqlite.rs b/rust/riverqueue/src/database/sqlite.rs index 2152e1341..b9b9d5c51 100644 --- a/rust/riverqueue/src/database/sqlite.rs +++ b/rust/riverqueue/src/database/sqlite.rs @@ -957,6 +957,7 @@ pub(crate) async fn retry( ELSE max_attempts END, finalized_at = NULL, + metadata = jsonb_remove(metadata, '$.cancel_attempted_at'), scheduled_at = ? WHERE id = ? AND state != 'running' diff --git a/rust/riverqueue/src/storage/postgres.rs b/rust/riverqueue/src/storage/postgres.rs index 316eb3587..e5b3bf151 100644 --- a/rust/riverqueue/src/storage/postgres.rs +++ b/rust/riverqueue/src/storage/postgres.rs @@ -239,7 +239,8 @@ impl Backend for PostgresBackend<'_> { "WITH locked AS (SELECT id FROM {table} WHERE id = $1 FOR UPDATE), \ updated AS (UPDATE {table} AS job SET state = 'available', \ max_attempts = CASE WHEN attempt = max_attempts THEN max_attempts + 1 ELSE max_attempts END, \ - finalized_at = NULL, scheduled_at = now() \ + finalized_at = NULL, metadata = metadata - 'cancel_attempted_at', \ + scheduled_at = now() \ FROM locked WHERE job.id = locked.id AND job.state != 'running' \ AND NOT (job.state = 'available' AND job.scheduled_at < now()) RETURNING job.*) \ SELECT {}, false AS unique_skipped_as_duplicate FROM updated AS job \ diff --git a/rust/riverqueue/tests/client_handles.rs b/rust/riverqueue/tests/client_handles.rs index 8f00521a1..c58e35cc5 100644 --- a/rust/riverqueue/tests/client_handles.rs +++ b/rust/riverqueue/tests/client_handles.rs @@ -775,6 +775,30 @@ macro_rules! scenarios { fixture.cleanup().await; } + // Like Go's `JobRetry`, a retry clears the cancellation requested + // during an earlier attempt so that it can't cancel the retried one. + #[tokio::test(flavor = "multi_thread")] + async fn retry_clears_a_cancellation_request() { + let fixture = Fixture::new().await; + let jobs = fixture.client.jobs(); + let metadata: riverqueue::JobMetadata = r#"{"preserved":true}"#.parse().unwrap(); + let id = fixture + .client + .insert(args("cancel then retry")) + .opts(InsertOpts::default().with_metadata(metadata)) + .await + .unwrap() + .id(); + let cancelled = jobs.cancel(id).await.unwrap(); + assert!(cancelled.metadata.contains_key("cancel_attempted_at")); + + let retried = jobs.retry(id).await.unwrap(); + assert_eq!(retried.state, JobState::Available); + assert!(!retried.metadata.contains_key("cancel_attempted_at")); + assert_eq!(retried.metadata.get::("preserved").unwrap(), Some(true)); + fixture.cleanup().await; + } + #[tokio::test(flavor = "multi_thread")] async fn local_queues_change_configuration_without_lock_errors() { let fixture = Fixture::new().await; diff --git a/rust/riverqueue/tests/poll_only_cancellation.rs b/rust/riverqueue/tests/poll_only_cancellation.rs index 76b6928d1..4d999a0c7 100644 --- a/rust/riverqueue/tests/poll_only_cancellation.rs +++ b/rust/riverqueue/tests/poll_only_cancellation.rs @@ -1,4 +1,5 @@ -//! Cancellation of running jobs in clients without notifications, on every +//! Cancellation of running jobs found by polling, in clients without +//! notifications and in clients whose notification was lost, on every //! backend. //! //! Postgres tests run in a unique schema and fail rather than skip when @@ -11,8 +12,8 @@ mod support; use std::{convert::Infallible, sync::Arc, time::Duration}; use riverqueue::{ - __private::Database, Client, EventKind, Job, JobArgs, JobState, QueueConfig, WorkContext, - WorkOutcome, Workers, + __private::{ClientBuilderExt, Database}, + Client, EventKind, Job, JobArgs, JobState, QueueConfig, WorkContext, WorkOutcome, Workers, }; use serde::{Deserialize, Serialize}; use tokio::sync::Semaphore; @@ -54,6 +55,144 @@ impl Backend { Self::Sqlite(pool, path) => support::sqlite_cleanup(pool, path).await, } } + + /// Requests a job's cancellation the way `JobCancel` does, but without a + /// notification, as if it was lost. + async fn request_cancellation_without_notification(&self, id: i64) { + let cancel_attempted_at = chrono::Utc::now().to_rfc3339(); + match self { + #[cfg(all(feature = "postgres", river_postgres_tests))] + Self::Postgres(schema) => { + sqlx::query(sqlx::AssertSqlSafe(format!( + "UPDATE {} SET metadata = jsonb_set(metadata, '{{cancel_attempted_at}}', to_jsonb($2::text)) \ + WHERE id = $1", + schema.table("river_job") + ))) + .bind(id) + .bind(cancel_attempted_at) + .execute(&schema.pool) + .await + .unwrap(); + } + #[cfg(feature = "sqlite")] + Self::Sqlite(pool, _) => { + sqlx::query( + "UPDATE river_job SET metadata = jsonb_set(metadata, '$.cancel_attempted_at', ?) \ + WHERE id = ?", + ) + .bind(cancel_attempted_at) + .bind(id) + .execute(pool) + .await + .unwrap(); + } + } + } +} + +/// Workers that block until their attempt is cancelled, adding a permit to +/// `started` when one starts. +fn waiting_workers(started: &Arc) -> Workers { + let mut workers = Workers::new(); + let started = Arc::clone(started); + workers + .add_fn(move |context: WorkContext, _job: Job| { + let started = Arc::clone(&started); + async move { + started.add_permits(1); + context.cancellation_token().cancelled().await; + // Any outcome but completion becomes the cancellation. + Ok::<_, Infallible>(WorkOutcome::Snooze(Duration::from_hours(1))) + } + }) + .unwrap(); + workers +} + +/// A client with notifications checks for cancellation requests only once +/// its interval (a minute by default) has passed, not on every two-second +/// tick like a client without them. A notified cancellation still arrives +/// right away. +async fn checks_for_lost_cancellations_only_after_interval(backend: Backend) { + let started = Arc::new(Semaphore::new(0)); + let client = Client::builder(backend.database()) + .queue(riverqueue::QUEUE_DEFAULT, QueueConfig::new(1)) + .without_leader_election() + .workers(waiting_workers(&started)) + .build() + .unwrap(); + let mut events = client.subscribe(&[EventKind::JobCancelled]).unwrap(); + let mut run = client.start().unwrap(); + run.wait_ready().await.unwrap(); + + let id = client.insert(WaitArgs {}).await.unwrap().job.row.id; + tokio::time::timeout(TIMEOUT, started.acquire()) + .await + .expect("the job should start") + .unwrap() + .forget(); + backend.request_cancellation_without_notification(id).await; + // This spans at least one tick, on which a client without notifications + // would have found the request. + assert!( + tokio::time::timeout(Duration::from_secs(3), events.recv()) + .await + .is_err(), + "the job shouldn't be cancelled before the interval passes" + ); + + client.jobs().cancel(id).await.unwrap(); + let event = tokio::time::timeout(TIMEOUT, events.recv()) + .await + .expect("the job should be cancelled") + .unwrap(); + let job = &event.as_job().expect("a job event").job; + assert_eq!(job.id, id); + assert_eq!(job.state, JobState::Cancelled, "{job:?}"); + tokio::time::timeout(TIMEOUT, run.stop()) + .await + .expect("the client should stop") + .unwrap(); + backend.cleanup().await; +} + +/// A client with notifications still cancels its running job when the +/// cancellation's notification was lost, like if it was sent while the +/// client's notifier was reconnecting, because it also checks for +/// cancellation requests occasionally. +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) + .queue(riverqueue::QUEUE_DEFAULT, QueueConfig::new(1)) + .without_leader_election() + .workers(waiting_workers(&started)) + .build() + .unwrap(); + let mut events = client.subscribe(&[EventKind::JobCancelled]).unwrap(); + let mut run = client.start().unwrap(); + run.wait_ready().await.unwrap(); + + let id = client.insert(WaitArgs {}).await.unwrap().job.row.id; + tokio::time::timeout(TIMEOUT, started.acquire()) + .await + .expect("the job should start") + .unwrap() + .forget(); + backend.request_cancellation_without_notification(id).await; + + let event = tokio::time::timeout(TIMEOUT, events.recv()) + .await + .expect("the job should be cancelled") + .unwrap(); + let job = &event.as_job().expect("a job event").job; + assert_eq!(job.id, id); + assert_eq!(job.state, JobState::Cancelled, "{job:?}"); + tokio::time::timeout(TIMEOUT, run.stop()) + .await + .expect("the client should stop") + .unwrap(); + backend.cleanup().await; } /// A client without notifications cancels its running job once it polls the @@ -62,24 +201,11 @@ impl Backend { async fn polls_for_remote_cancellation(backend: Backend) { for while_stopping in [false, true] { let started = Arc::new(Semaphore::new(0)); - let mut workers = Workers::new(); - let worker_started = Arc::clone(&started); - workers - .add_fn(move |context: WorkContext, _job: Job| { - let started = Arc::clone(&worker_started); - async move { - started.add_permits(1); - context.cancellation_token().cancelled().await; - // Any outcome but completion becomes the cancellation. - Ok::<_, Infallible>(WorkOutcome::Snooze(Duration::from_hours(1))) - } - }) - .unwrap(); let client = Client::builder(backend.database()) .queue(riverqueue::QUEUE_DEFAULT, QueueConfig::new(1)) .without_leader_election() .without_notifications() - .workers(workers) + .workers(waiting_workers(&started)) .build() .unwrap(); let other = Client::builder(backend.database()).build().unwrap(); @@ -117,6 +243,22 @@ async fn polls_for_remote_cancellation(backend: Backend) { mod postgres { use super::*; + #[tokio::test] + async fn checks_for_lost_cancellations_only_after_interval() { + super::checks_for_lost_cancellations_only_after_interval(Backend::Postgres( + support::PostgresSchema::new("river_poll_cancel_interval").await, + )) + .await; + } + + #[tokio::test] + async fn polls_for_lost_cancellation_notification() { + super::polls_for_lost_cancellation_notification(Backend::Postgres( + support::PostgresSchema::new("river_poll_cancel_lost").await, + )) + .await; + } + #[tokio::test] async fn polls_for_remote_cancellation() { super::polls_for_remote_cancellation(Backend::Postgres( @@ -130,6 +272,18 @@ mod postgres { mod sqlite { use super::*; + #[tokio::test] + async fn checks_for_lost_cancellations_only_after_interval() { + let (pool, path) = support::sqlite_file_pool(4).await; + super::checks_for_lost_cancellations_only_after_interval(Backend::Sqlite(pool, path)).await; + } + + #[tokio::test] + async fn polls_for_lost_cancellation_notification() { + let (pool, path) = support::sqlite_file_pool(4).await; + super::polls_for_lost_cancellation_notification(Backend::Sqlite(pool, path)).await; + } + #[tokio::test] async fn polls_for_remote_cancellation() { let (pool, path) = support::sqlite_file_pool(4).await;