-
Notifications
You must be signed in to change notification settings - Fork 178
fix: forward controlActionCancel to cancelCh in poll-mode fetchAndRunLoop - #1245
Conversation
bgentry
commented
May 10, 2026
I see two issues here:
- I don't think this is actually going to do what you think it will, because there's no way for the producer to get these job cancel messages when in poll-only mode. You would need to periodically poll the database for all running job IDs to see if they've been cancelled.
- Repeatedly polling the database to ask if running jobs are still actually running could have substantial overhead, or would at least result in a lot of query noise for something that happens rarely.
Rather than going down this path, I think it's more likely we'd want to think about a non-Postgres notifier option of some kind (i.e. Redis). This would lose transactionality, but would be far more scalable and compatible with non-NOTIFY Postgres environments, dbsql drivers, or other databases.
Another option which we've talked about before is letting you use a pgx listener/notifer even if you're using database/sql for the rest of your driver: #352 This potentially covers some Postgres use cases that leverage database/sql, albeit only if you have access to a conn or pool that does have LISTEN/NOTIFY support (like by bypassing pgbouncer).
ata-the-legend
commented
May 10, 2026
Thanks for the detailed explanation. One small clarification: the fix doesn't rely on Postgres NOTIFY at all. JobCancel calls notifyProducerWithoutListenerQueueControlEvent, which — when SupportsListener() == false — does an in-process channel send via TriggerQueueControlEvent directly to the producer. My fix wires that into cancelCh so maybeCancelJob fires. I confirmed this in my setup with logs showing job cancelled remotely immediately after JobCancel.
You're right though that this only works single-process — in a multi-process deployment that channel send never reaches the other server's producer. So it's not a general solution and I understand why you wouldn't want to merge it as-is.
For my use case I'm working around it by polling the contract status from inside Work(). Happy to close this PR if it's not the direction you want to go.
Hey @ata-the-legend, sorry for the delay here! I think your last point in that this is a fairly strict improvement for the in-process poll-only path is a good one. We don't handle the multi-process case, but this'll be useful in some situations.
A couple minor things:
- Did you consider just calling
p.maybeCancelJob(workCtx, msg.JobID)instead of sending tocancelCh? This might have the benefit of just making the code a little more readable/direct without the additional channel hop. - Could you add a comment around those lines saying what we've talked about here: 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.
- Do you want to rebase and add a changelog entry?
Thanks! This code is not too easy to understand, so good stuff.
...Loop 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).
33439c2 to
317f573
Compare
ata-the-legend
commented
Jun 7, 2026
Thanks for the feedback! Addressed all three points:
- Replaced the
cancelChhop with a directp.maybeCancelJobcall - Added your suggested comment about poll-only / single-process limitation
- Rebased onto latest master and added a changelog entry
All tests pass.
@brandur
brandur
left a comment
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Thanks!
Problem
When using River with a driver that does not support LISTEN/NOTIFY (e.g.
riverdatabasesql), callingJobCancelon a running job has no effect: the job's context is never cancelled andctx.Done()never fires insideWork().Root Cause
JobCancel(non-tx) callsnotifyProducerWithoutListenerQueueControlEvent, which sends acontrolEventPayload{Action: "cancel"}toqueueControlCh. However, infetchAndRunLoop, thequeueControlChreceive block only handledcontrolActionPause,controlActionResume, andcontrolActionMetadataChanged. ThecontrolActionCancelcase was absent and fell through todefault, silently discarding the event.The
controlActionCancelcase exists correctly inhandleControlNotification(the LISTEN path), but was never added to the parallel poll-mode path.Fix
Add
controlActionCanceltofetchAndRunLoop'squeueControlChswitch, forwardingmsg.JobIDtocancelCh. This is consumed by the existingmaybeCancelJobcall, which cancels the running job's context viacontext.WithCancelCause.Testing
Added a test that verifies
ctx.Done()fires insideWork()afterJobCancelis called when the driver is in poll mode.