fix: forward controlActionCancel to cancelCh in poll-mode fetchAndRunLoop (#1245)

When using drivers that don't support LISTEN/NOTIFY (e.g. riverdatabasesql),
job cancel events are routed in-process via queueControlCh. The
controlActionCancel case was missing from fetchAndRunLoop's queueControlCh
handler, causing cancel events to be silently dropped and ctx.Done() to never
fire inside a running Work() call.

Forward the job ID to cancelCh so the existing maybeCancelJob call handles it,
matching the behaviour of the LISTEN/NOTIFY path in handleControlNotification.

Adds a test that verifies ctx.Done() fires in a running job after JobCancel is
called when using a poll-only driver (SupportsListener() == false).

Author: ata-the-legend

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- Fix `JobCancel` having no effect on running jobs when using a poll-only driver (e.g. `riverdatabasesql`). The `controlActionCancel` event was silently dropped in `fetchAndRunLoop`'s `queueControlCh` handler instead of being forwarded to `maybeCancelJob`. Note: this fix only works within a single process; cross-process cancels in poll-only setups must wait for the next poll cycle. [PR #1245](https://github.com/riverqueue/river/pull/1245).
+
 ## [0.39.0] - 2026-06-03
 
 ⚠️ **Breaking API change:** `rivermigrate.Migrator.Validate` and `rivermigrate.Migrator.ValidateTx` now take a `*rivermigrate.ValidateOpts` parameter. Pass `nil` to preserve previous behavior. We normally endeavor not to make any breaking API changes, but this one will keep the API in a much nicer state, and is on an ancillary function that most installations won't be using. [PR #1259](https://github.com/riverqueue/river/pull/1259)

client_test.go

@@ -1043,6 +1043,46 @@ func Test_Client_Common(t *testing.T) {
 		})
 	})
 
+	t.Run("CancelRunningJobPollOnly", func(t *testing.T) {
+		t.Parallel()
+
+		config, bundle := setupConfig(t)
+
+		client, err := NewClient(NewDriverPollOnly(bundle.dbPool), config)
+		require.NoError(t, err)
+
+		jobStartedChan := make(chan int64)
+
+		type JobArgs struct {
+			testutil.JobArgsReflectKind[JobArgs]
+		}
+
+		AddWorker(client.config.Workers, WorkFunc(func(ctx context.Context, job *Job[JobArgs]) error {
+			jobStartedChan <- job.ID
+			<-ctx.Done()
+			return ctx.Err()
+		}))
+
+		subscribeChan := subscribe(t, client)
+		startClient(ctx, t, client)
+		riversharedtest.WaitOrTimeout(t, client.baseStartStop.Started())
+
+		insertRes, err := client.Insert(ctx, &JobArgs{}, nil)
+		require.NoError(t, err)
+
+		startedJobID := riversharedtest.WaitOrTimeout(t, jobStartedChan)
+		require.Equal(t, insertRes.Job.ID, startedJobID)
+
+		updatedJob, err := client.JobCancel(ctx, insertRes.Job.ID)
+		require.NoError(t, err)
+		require.Equal(t, rivertype.JobStateRunning, updatedJob.State)
+
+		event := riversharedtest.WaitOrTimeout(t, subscribeChan)
+		require.Equal(t, EventKindJobCancelled, event.Kind)
+		require.Equal(t, rivertype.JobStateCancelled, event.Job.State)
+		require.WithinDuration(t, time.Now(), *event.Job.FinalizedAt, 2*time.Second)
+	})
+
 	t.Run("CancelScheduledJob", func(t *testing.T) {
 		t.Parallel()
 

producer.go

@@ -528,8 +528,10 @@ func (p *producer) fetchAndRunLoop(fetchCtx, workCtx context.Context) {
 		case msg := <-p.queueControlCh:
 			switch msg.Action {
 			case controlActionCancel:
-				// Separate this case to make linter happy:
-				p.Logger.DebugContext(workCtx, p.Name+": Unhandled queue control action", "action", msg.Action)
+				// This path is only expected to take effect in poll-only mode, and
+				// only works for the case of a single process. Multi-process setups
+				// will have to wait for the next poll event for a cancel to take effect.
+				p.maybeCancelJob(workCtx, msg.JobID)
 			case controlActionMetadataChanged:
 				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{}{})