[go, java, js, rust] Alternative implementation for reconciling jobs against notifier reconnect

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

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

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

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

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

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<Client.Decoded> 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) {

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

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`, {

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

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",
     });

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"
     );

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'

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