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

Original file line number	Diff line number	Diff line change
Expand Up		@@ -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)
Expand Down

40 changes: 40 additions & 0 deletions client_test.go

Show comments View file Open in desktop

Original file line number	Diff line number	Diff line change
Expand Up		@@ -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, 2time.Second)
	})

		t.Run("CancelScheduledJob", func(t *testing.T) {
		t.Parallel()

Expand Down

6 changes: 4 additions & 2 deletions producer.go

Show comments View file Open in desktop

Original file line number	Diff line number	Diff line change
Expand Up		@@ -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{}{})
Expand Down

Navigation Menu