Reconcile active jobs against cancel_attempted_at after notifier reconnects

CHANGELOG.md

@@ -12,6 +12,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - 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).
+- 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).
 
 ## [0.50.0] - 2026-10-26
 

client.go

@@ -1231,6 +1231,16 @@ func (c *Client[TTx]) Start(ctx context.Context) error {
 		fetchCtx := withClient(fetchCtx, c)
 		workCtx = withClient(workCtx, c)
 
+		// Keep cancellation notifications and reconnects live until producers
+		// finish draining. The notifier is stopped with the other services below.
+		if c.notifier != nil {
+			if err := c.notifier.StartWithConnectContext(context.WithoutCancel(fetchCtx), fetchCtx); err != nil {
+				workCancel(err)
+				stopServicesOnError()
+				return err
+			}
+		}
+
 		if err := startstop.StartAll(fetchCtx, c.services...); err != nil {
 			workCancel(err)
 			stopServicesOnError()
@@ -1503,8 +1513,8 @@ 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
+// notified to cancel the running job's context. Clients reconcile missed requests
+// after reconnecting, or poll every two seconds without a notifier. 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
@@ -1533,11 +1543,9 @@ 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,
-	})
+	if job.State == rivertype.JobStateRunning {
+		c.notifyProducerOfCancelWithoutNotifier(ctx, job)
+	}
 
 	return job, nil
 }
@@ -1560,8 +1568,8 @@ 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
+// notified to cancel the running job's context. Clients reconcile missed requests
+// after reconnecting, or poll every two seconds without a notifier. 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
@@ -2426,14 +2434,15 @@ func (c *Client[TTx]) producerAdd(queueName string, queueConfig QueueConfig) (*p
 }
 
 func (c *Client[TTx]) producerRemove(ctx context.Context, queueName string) error {
-	c.producersMu.Lock()
-	defer c.producersMu.Unlock()
-
+	c.producersMu.RLock()
 	producer, ok := c.producersByQueueName[queueName]
+	c.producersMu.RUnlock()
 	if !ok {
 		return &QueueNotFoundError{Name: queueName}
 	}
 
+	// QueueBundle serializes additions and removals. Keep the registry readable
+	// while draining so workers can still receive local cancellation requests.
 	shouldStop, stopped, finalizeStop := producer.StopInit()
 	if shouldStop {
 		select {
@@ -2445,7 +2454,9 @@ func (c *Client[TTx]) producerRemove(ctx context.Context, queueName string) erro
 		}
 	}
 
+	c.producersMu.Lock()
 	delete(c.producersByQueueName, queueName)
+	c.producersMu.Unlock()
 
 	return nil
 }
@@ -2833,7 +2844,7 @@ func (c *Client[TTx]) QueuePause(ctx context.Context, name string, opts *QueuePa
 		return err
 	}
 
-	c.notifyProducerWithoutListenerQueueControlEvent(name, controlEvent)
+	c.notifyProducerWithoutListenerQueueControlEvent(ctx, name, controlEvent)
 
 	return nil
 }
@@ -2903,7 +2914,7 @@ func (c *Client[TTx]) QueueResume(ctx context.Context, name string, opts *QueueP
 		return err
 	}
 
-	c.notifyProducerWithoutListenerQueueControlEvent(name, controlEvent)
+	c.notifyProducerWithoutListenerQueueControlEvent(ctx, name, controlEvent)
 
 	return nil
 }
@@ -2966,7 +2977,7 @@ func (c *Client[TTx]) QueueUpdate(ctx context.Context, name string, params *Queu
 	// No control event is returned when metadata isn't being changed, in which
 	// case there's nothing for the producer to act on.
 	if controlEvent != nil {
-		c.notifyProducerWithoutListenerQueueControlEvent(name, controlEvent)
+		c.notifyProducerWithoutListenerQueueControlEvent(ctx, name, controlEvent)
 	}
 
 	return queue, nil
@@ -2989,21 +3000,65 @@ func (c *Client[TTx]) QueueUpdateTx(ctx context.Context, tx TTx, name string, pa
 // 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]) notifyProducerWithoutListenerQueueControlEvent(queue string, controlEvent *controlEventPayload) {
+func (c *Client[TTx]) notifyProducerWithoutListenerQueueControlEvent(ctx context.Context, queue string, controlEvent *controlEventPayload) {
 	if c.driver.SupportsListener() {
 		return
 	}
 
 	c.producersMu.RLock()
-	defer c.producersMu.RUnlock()
+	producer := c.producersByQueueName[queue]
+	c.producersMu.RUnlock()
 
-	if len(c.producersByQueueName) < 1 {
+	if producer != nil {
+		producer.TriggerQueueControlEvent(ctx, controlEvent)
+	}
+}
+
+// Deliver cancellations locally when no notifier is configured, including PollOnly.
+// Only call after commit so the producer can observe the cancellation metadata.
+func (c *Client[TTx]) notifyProducerOfCancelWithoutNotifier(ctx context.Context, job *rivertype.JobRow) {
+	if c.notifier != nil {
+		return
+	}
+
+	c.producersMu.RLock()
+	producer := c.producersByQueueName[job.Queue]
+	c.producersMu.RUnlock()
+	if producer == nil {
+		return
+	}
+
+	state := producer.runState.Load()
+	if state == nil {
+		return
+	}
+	select {
+	case <-state.done:
 		return
+	default:
+	}
+
+	producer.activeJobsMu.Lock()
+	executor := producer.activeJobs[job.ID]
+	if executor != nil {
+		sameAttemptedAt := executor.JobRow.AttemptedAt == job.AttemptedAt ||
+			(executor.JobRow.AttemptedAt != nil && job.AttemptedAt != nil && executor.JobRow.AttemptedAt.Equal(*job.AttemptedAt))
+		if executor.JobRow.Attempt != job.Attempt || !sameAttemptedAt {
+			executor = nil
+		}
 	}
+	producer.activeJobsMu.Unlock()
 
-	if producer, ok := c.producersByQueueName[queue]; ok {
-		producer.TriggerQueueControlEvent(controlEvent)
+	if executor != nil {
+		executor.Cancel(ctx)
+	} else {
+		producer.TriggerCancelCheck()
 	}
+	producer.testSignals.QueueControlEventTriggered.Signal(&controlEventPayload{
+		Action: controlActionCancel,
+		JobID:  job.ID,
+		Queue:  job.Queue,
+	})
 }
 
 func (c *Client[TTx]) queueUpdate(ctx context.Context, executorTx riverdriver.ExecutorTx, name string, params *QueueUpdateParams) (*rivertype.Queue, *controlEventPayload, error) {