Reconcile active jobs against cancel_attempted_at after notifier reconnects

CHANGELOG.md

@@ -6,6 +6,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
+- A `JobCancel()` sent while a client's notifier is disconnected/reconnecting is no longer lost. Previously, the cancellation's `NOTIFY` could commit while nothing was `LISTEN`ing (Postgres `NOTIFY` is fire-and-forget), leaving the running job unaware of the cancellation until `JobRescuer`'s much longer stuck-job sweep. Each client now reconciles its currently running jobs against `cancel_attempted_at` by primary key every time its notifier (re)establishes healthy listening, closing the gap with no added steady-state query load. [PR #1361](https://github.com/riverqueue/river/pull/1361), as discussed in [#1358](https://github.com/riverqueue/river/issues/1358).
+- `JobCancel()` against a job running in the same process is now delivered immediately for clients with no notifier at all, including an explicit `Config.PollOnly` on an otherwise listener-capable driver. Previously, the local delivery path only activated for drivers that can't support `LISTEN`/`NOTIFY` at all, so `PollOnly: true` on the Postgres or SQLite drivers fell all the way back to `JobRescuer`'s stuck-job sweep for a job it could have cancelled synchronously in-process. [PR #1361](https://github.com/riverqueue/river/pull/1361).
 
 ## [0.45.0] - 2026-08-25
 

client.go

@@ -1456,11 +1456,22 @@ func (c *Client[TTx]) JobCancel(ctx context.Context, jobID int64) (*rivertype.Jo
 		return nil, err
 	}
 
-	c.notifyProducerWithoutListenerQueueControlEvent(job.Queue, &controlEventPayload{
-		Action: controlActionCancel,
-		JobID:  job.ID,
-		Queue:  job.Queue,
-	})
+	// Only a running job has an in-process executor to interrupt (an
+	// available/scheduled/retryable job was cancelled directly above, and a
+	// finalized job needed no cancellation at all), so there's nothing for a
+	// non-running job to dispatch. This also matters for correctness, not
+	// just efficiency: it keeps a burst of cancellations against jobs that
+	// were never running (e.g. cancelling many freshly-inserted jobs before
+	// the client has even been started) from filling the producer's
+	// control-event channel, which TriggerQueueControlEvent below sends to
+	// unconditionally and would otherwise block this call indefinitely.
+	if job.State == rivertype.JobStateRunning {
+		c.notifyProducerOfCancelWithoutNotifier(job.Queue, &controlEventPayload{
+			Action: controlActionCancel,
+			JobID:  job.ID,
+			Queue:  job.Queue,
+		})
+	}
 
 	return job, nil
 }
@@ -2876,6 +2887,58 @@ func (c *Client[TTx]) notifyProducerWithoutListenerQueueControlEvent(queue strin
 	}
 }
 
+// Notifies an internal producer of a job cancellation so a job running in
+// this same process observes it immediately, for clients that have no
+// notifier to deliver it via listen/notify.
+//
+// This deliberately does NOT share notifyProducerWithoutListenerQueueControlEvent
+// above despite the near-identical dispatch: the two helpers guard on
+// different conditions for a reason, and unifying them would either reopen
+// this gap or introduce new bugs in the other one.
+//
+//   - This helper guards on c.notifier == nil (this client has no notifier,
+//     for any reason, including an explicit Config.PollOnly on an otherwise
+//     listener-capable driver). Queue control events guard on
+//     driver.SupportsListener() instead, because pause/resume/metadata
+//     changes already have an independent, documented convergence path in
+//     poll-only mode (producer.pollForSettingChanges, on QueuePollInterval) —
+//     using this same client.notifier == nil guard there would double-deliver
+//     metadata changes not deduplicated the way pause/resume are, and would
+//     panic on QueueUpdate's nil control event when metadata is empty.
+//   - Job cancellation has no such fallback: nothing else ever re-checks a
+//     running job's cancel_attempted_at in poll-only mode, so a lost signal
+//     here is only caught by JobRescuer's much longer stuck-job sweep.
+//
+// The dispatch below (via TriggerQueueControlEvent) sends unconditionally and
+// blocks if the target producer's control-event channel is full — the same
+// characteristic notifyProducerWithoutListenerQueueControlEvent above already
+// has for genuinely listener-incapable drivers. The caller (JobCancel) is
+// responsible for only invoking this for jobs it knows are actually running,
+// both because there's nothing to dispatch for a job with no in-process
+// executor and to avoid a burst of cancellations against non-running jobs
+// (e.g. many freshly-inserted jobs cancelled before the client is started)
+// filling that channel and blocking indefinitely.
+//
+// Should only ever be invoked *outside* a transaction. If invoked within a
+// transaction, the producer wouldn't yet be able to access the state that
+// triggered the notification because it's not committed yet.
+func (c *Client[TTx]) notifyProducerOfCancelWithoutNotifier(queue string, controlEvent *controlEventPayload) {
+	if c.notifier != nil {
+		return
+	}
+
+	c.producersMu.RLock()
+	defer c.producersMu.RUnlock()
+
+	if len(c.producersByQueueName) < 1 {
+		return
+	}
+
+	if producer, ok := c.producersByQueueName[queue]; ok {
+		producer.TriggerQueueControlEvent(controlEvent)
+	}
+}
+
 func (c *Client[TTx]) queueUpdate(ctx context.Context, executorTx riverdriver.ExecutorTx, name string, params *QueueUpdateParams) (*rivertype.Queue, *controlEventPayload, error) {
 	updateMetadata := len(params.Metadata) > 0
 

client_test.go

@@ -1368,6 +1368,112 @@ func Test_Client_Common(t *testing.T) {
 		require.WithinDuration(t, time.Now(), *event.Job.FinalizedAt, 2*time.Second)
 	})
 
+	// Unlike CancelRunningJobPollOnly above (which uses a fake driver that
+	// reports SupportsListener() == false to simulate a driver that can't
+	// listen/notify at all), this uses the real pgx driver with an explicit
+	// Config.PollOnly, which is the actually-common "I don't want a notifier"
+	// case. driver.SupportsListener() is true here, so a same-process
+	// cancellation delivered via JobCancel() has nothing to do with the
+	// listen/notify path at all: it must go through the client's local,
+	// no-notifier dispatch instead.
+	t.Run("CancelRunningJobExplicitPollOnly", func(t *testing.T) {
+		t.Parallel()
+
+		config, bundle := setupConfig(t)
+		config.PollOnly = true
+
+		client, err := NewClient(bundle.driver, 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)
+	})
+
+	// A non-running job (available, scheduled, or retryable) is cancelled
+	// directly in the database and has no in-process executor to notify, so
+	// JobCancel must not dispatch a local control event for it. This matters
+	// beyond efficiency: notifyProducerOfCancelWithoutNotifier's dispatch
+	// blocks if the target producer's control-event channel (buffered to
+	// 100) fills up, which an unstarted (or otherwise not-yet-draining)
+	// producer never does. Cancelling more than that many never-started jobs
+	// against an explicit-PollOnly client (which has no notifier and so
+	// would otherwise take the local dispatch path) must not block.
+	t.Run("CancelManyNonRunningJobsExplicitPollOnlyDoesNotBlock", func(t *testing.T) {
+		t.Parallel()
+
+		config, bundle := setupConfig(t)
+		config.PollOnly = true
+
+		client, err := NewClient(bundle.driver, config)
+		require.NoError(t, err)
+
+		type JobArgs struct {
+			testutil.JobArgsReflectKind[JobArgs]
+		}
+
+		AddWorker(client.config.Workers, WorkFunc(func(ctx context.Context, job *Job[JobArgs]) error {
+			return nil
+		}))
+
+		const numJobs = 101 // more than the producer's 100-slot control-event buffer
+
+		jobIDs := make([]int64, numJobs)
+		for i := range jobIDs {
+			insertRes, err := client.Insert(ctx, &JobArgs{}, nil)
+			require.NoError(t, err)
+			jobIDs[i] = insertRes.Job.ID
+		}
+
+		// Deliberately cancelling before the client is started: nothing is
+		// running yet, and nothing is consuming the (not-yet-existent)
+		// producer's control-event channel either.
+		cancelDone := make(chan error, 1)
+		go func() {
+			for _, jobID := range jobIDs {
+				if _, err := client.JobCancel(ctx, jobID); err != nil {
+					cancelDone <- err
+					return
+				}
+			}
+			cancelDone <- nil
+		}()
+
+		select {
+		case err := <-cancelDone:
+			require.NoError(t, err)
+		case <-time.After(5 * time.Second):
+			t.Fatal("cancelling non-running jobs blocked, presumably on the producer's control-event channel")
+		}
+	})
+
 	t.Run("CancelScheduledJob", func(t *testing.T) {
 		t.Parallel()
 
@@ -1433,15 +1539,27 @@ func Test_Client_Common(t *testing.T) {
 			testutil.JobArgsReflectKind[JobArgs]
 		}
 
+		jobStartedChan := make(chan int64)
+		continueChan := make(chan struct{})
+
 		AddWorker(client.config.Workers, WorkFunc(func(ctx context.Context, job *Job[JobArgs]) error {
+			jobStartedChan <- job.ID
+			<-continueChan
 			return nil
 		}))
 
 		startClient(ctx, t, client)
+		t.Cleanup(func() { close(continueChan) })
 
 		insertRes, err := client.Insert(ctx, &JobArgs{}, nil)
 		require.NoError(t, err)
 
+		// Only a running job dispatches a local control event (see
+		// notifyProducerOfCancelWithoutNotifier), so make sure the job has
+		// actually started before cancelling it.
+		startedJobID := riversharedtest.WaitOrTimeout(t, jobStartedChan)
+		require.Equal(t, insertRes.Job.ID, startedJobID)
+
 		_, err = client.JobCancel(ctx, insertRes.Job.ID)
 		require.NoError(t, err)
 

internal/notifier/notifier.go

@@ -88,12 +88,46 @@ type Notifier struct {
 	testSignals       notifierTestSignals
 	waitInterruptChan chan func()
 
-	mu            sync.RWMutex
-	isConnected   bool
-	isStarted     bool
-	isWaiting     bool
-	subscriptions map[NotificationTopic][]*Subscription
-	waitCancel    context.CancelFunc
+	mu                 sync.RWMutex
+	isConnected        bool
+	isStarted          bool
+	isWaiting          bool
+	listenerReadyFuncs map[*ListenerReadySubscription]ListenerReadyFunc
+	subscriptions      map[NotificationTopic][]*Subscription
+	waitCancel         context.CancelFunc
+}
+
+// ListenerReadyFunc is invoked each time the notifier establishes healthy
+// listening on all of its currently subscribed topics. This includes the
+// very first connect as well as every subsequent reconnect that follows a
+// connection loss, so callers whose work is only meaningful after a genuine
+// reconnect (as opposed to the initial connect) should make that check
+// themselves.
+type ListenerReadyFunc func()
+
+// ListenerReadySubscription is a handle to a func registered with
+// RegisterListenerReadyFunc. Call Unregister to stop receiving invocations.
+type ListenerReadySubscription struct {
+	notifier *Notifier
+}
+
+func (s *ListenerReadySubscription) Unregister() {
+	s.notifier.mu.Lock()
+	defer s.notifier.mu.Unlock()
+	delete(s.notifier.listenerReadyFuncs, s)
+}
+
+// RegisterListenerReadyFunc registers a function to be invoked every time the
+// notifier (re)establishes healthy listening on all of its subscribed topics.
+// It's used to let callers reconcile any state that could have missed a
+// notification while the notifier was disconnected/reconnecting.
+func (n *Notifier) RegisterListenerReadyFunc(listenerReadyFunc ListenerReadyFunc) *ListenerReadySubscription {
+	n.mu.Lock()
+	defer n.mu.Unlock()
+
+	sub := &ListenerReadySubscription{notifier: n}
+	n.listenerReadyFuncs[sub] = listenerReadyFunc
+	return sub
 }
 
 func New(archetype *baseservice.Archetype, listener riverdriver.Listener) *Notifier {
@@ -102,7 +136,8 @@ func New(archetype *baseservice.Archetype, listener riverdriver.Listener) *Notif
 		notificationBuf:   make(chan *riverdriver.Notification, 1000),
 		waitInterruptChan: make(chan func(), 10),
 
-		subscriptions: make(map[NotificationTopic][]*Subscription),
+		listenerReadyFuncs: make(map[*ListenerReadySubscription]ListenerReadyFunc),
+		subscriptions:      make(map[NotificationTopic][]*Subscription),
 	})
 	return notifier
 }
@@ -212,6 +247,16 @@ func (n *Notifier) listenAndWait(ctx context.Context) error {
 	n.testSignals.ListeningBegin.Signal(struct{}{})
 	defer n.testSignals.ListeningEnd.Signal(struct{}{})
 
+	listenerReadyFuncs := func() []ListenerReadyFunc {
+		n.mu.RLock()
+		defer n.mu.RUnlock()
+
+		return maputil.Values(n.listenerReadyFuncs)
+	}()
+	for _, listenerReadyFunc := range listenerReadyFuncs {
+		listenerReadyFunc()
+	}
+
 	drainInterrupts := func() {
 		for {
 			select {

internal/notifier/notifier_test.go

@@ -622,6 +622,54 @@ func TestNotifier(t *testing.T) {
 		// Subscription should still work.
 		require.Equal(t, TopicAndPayload{testTopic1, "msg1"}, riversharedtest.WaitOrTimeout(t, notifyChan))
 	})
+
+	t.Run("RegisterListenerReadyFunc", func(t *testing.T) {
+		t.Parallel()
+
+		notifier, _ := setup(t, nil)
+		notifier.testDisableSleep = true
+
+		var errorNum int
+
+		listenerMock := NewListenerMock(notifier.listener)
+		listenerMock.waitForNotificationFunc = func(ctx context.Context) (*riverdriver.Notification, error) {
+			// Returns an error the first time, but then works after.
+			errorNum++
+			switch errorNum {
+			case 1:
+				return nil, errors.New("error during wait")
+			default:
+				return listenerMock.Listener.WaitForNotification(ctx)
+			}
+		}
+		notifier.listener = listenerMock
+
+		readyChan := make(chan struct{}, 10)
+		sub := notifier.RegisterListenerReadyFunc(func() { readyChan <- struct{}{} })
+		t.Cleanup(sub.Unregister)
+
+		start(t, notifier)
+
+		// Fires once on the initial connect.
+		riversharedtest.WaitOrTimeout(t, readyChan)
+
+		// And again once the notifier reconnects after the injected failure.
+		require.EqualError(t, notifier.testSignals.BackoffError.WaitOrTimeout(), "error during wait")
+		riversharedtest.WaitOrTimeout(t, readyChan)
+
+		// Unregister stops further invocations. There's no direct signal for
+		// "would have fired again", so drain what's pending, unregister, and
+		// confirm a subsequent restart doesn't add anything new.
+		for len(readyChan) > 0 {
+			<-readyChan
+		}
+		sub.Unregister()
+
+		notifier.Stop()
+		require.NoError(t, notifier.Start(ctx))
+		notifier.testSignals.ListeningBegin.WaitOrTimeout()
+		require.Empty(t, readyChan)
+	})
 }
 
 type ListenerMock struct {