Make jobs cancelled due to a soft stop immediately available

CHANGELOG.md

@@ -9,8 +9,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Added `EventKindJobInterrupted`, emitted when a running job is interrupted because its client is shutting down, the job was cancelled, and has been made immediately available to be worked again. [PR #1290](https://github.com/riverqueue/river/pull/1290).
 - Added `JobListParams.TagsAll` and `JobListParams.TagsAny` for filtering jobs that match every or any exact tag, respectively. [PR #1339](https://github.com/riverqueue/river/pull/1339).
 
+### Changed
+
+- Jobs that didn't finish in time organically while a client was stopping and had to have their context cancelled no longer have this cancellation counted as an error. `attempt` is reset to the number it was before the job started working, `errors` is left unchanged, and `state` is made `available` so jobs are eligible to be retried immediately. [PR #1290](https://github.com/riverqueue/river/pull/1290)
+
 ## [0.42.0] - 2026-07-31
 
 ### Added

client_test.go

@@ -342,6 +342,7 @@ func subscribe[TTx any](t *testing.T, client *Client[TTx]) <-chan *Event {
 		EventKindJobCancelled,
 		EventKindJobCompleted,
 		EventKindJobFailed,
+		EventKindJobInterrupted,
 		EventKindJobSnoozed,
 		EventKindQueuePaused,
 		EventKindQueueResumed,
@@ -2766,7 +2767,6 @@ func Test_Client_SoftStopTimeout(t *testing.T) {
 		}))
 
 		client := runNewTestClient(ctx, t, config)
-
 		_, err := client.Insert(ctx, JobArgs{}, nil)
 		require.NoError(t, err)
 
@@ -2784,6 +2784,76 @@ func Test_Client_SoftStopTimeout(t *testing.T) {
 		}
 	})
 
+	t.Run("ErroringJobGetsFreshAttempt", func(t *testing.T) {
+		t.Parallel()
+
+		config := newTestConfig(t, "")
+		config.SoftStopTimeout = 100 * time.Millisecond
+
+		firstRunDoneChan := make(chan struct{})
+		jobStartedChan := make(chan int64, 2)
+		var runCount atomic.Int32
+		AddWorker(config.Workers, WorkFunc(func(ctx context.Context, job *Job[JobArgs]) error {
+			jobStartedChan <- job.ID
+			switch runCount.Add(1) {
+			case 1:
+				<-ctx.Done()
+				close(firstRunDoneChan)
+				return ctx.Err()
+			default:
+				return errors.New("real job error")
+			}
+		}))
+
+		client := runNewTestClient(ctx, t, config)
+		subscribeChan, cancelSubscribe := client.Subscribe(EventKindJobInterrupted)
+		t.Cleanup(cancelSubscribe)
+
+		insertRes, err := client.Insert(ctx, JobArgs{}, &InsertOpts{MaxAttempts: 2})
+		require.NoError(t, err)
+
+		jobID := riversharedtest.WaitOrTimeout(t, jobStartedChan)
+		require.Equal(t, insertRes.Job.ID, jobID)
+
+		require.NoError(t, client.Stop(ctx))
+		riversharedtest.WaitOrTimeout(t, firstRunDoneChan)
+
+		event := riversharedtest.WaitOrTimeout(t, subscribeChan)
+		require.NotNil(t, event)
+		require.Equal(t, EventKindJobInterrupted, event.Kind)
+		require.Equal(t, insertRes.Job.ID, event.Job.ID)
+		require.Equal(t, rivertype.JobStateAvailable, event.Job.State)
+		require.NotNil(t, event.JobStats)
+
+		jobAfter, err := client.driver.GetExecutor().JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: jobID, Schema: client.config.Schema})
+		require.NoError(t, err)
+		require.Equal(t, 0, jobAfter.Attempt)
+		require.Nil(t, jobAfter.FinalizedAt)
+		require.Equal(t, 2, jobAfter.MaxAttempts)
+		require.Equal(t, rivertype.JobStateAvailable, jobAfter.State)
+		require.WithinDuration(t, time.Now(), jobAfter.ScheduledAt, 2*time.Second)
+		require.Empty(t, jobAfter.Errors)
+
+		require.NoError(t, client.Start(ctx))
+
+		jobID = riversharedtest.WaitOrTimeout(t, jobStartedChan)
+		require.Equal(t, insertRes.Job.ID, jobID)
+
+		var jobAfterRealError *rivertype.JobRow
+		require.Eventually(t, func() bool {
+			var err error
+			jobAfterRealError, err = client.driver.GetExecutor().JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: jobID, Schema: client.config.Schema})
+			require.NoError(t, err)
+			return jobAfterRealError.State != rivertype.JobStateRunning
+		}, 5*time.Second, 10*time.Millisecond)
+
+		require.Equal(t, 1, jobAfterRealError.Attempt)
+		require.Equal(t, rivertype.JobStateRetryable, jobAfterRealError.State)
+		require.Len(t, jobAfterRealError.Errors, 1)
+		require.Equal(t, "real job error", jobAfterRealError.Errors[0].Error)
+		require.Less(t, time.Until(jobAfterRealError.ScheduledAt), 3*time.Second)
+	})
+
 	t.Run("SoftStopSucceedsBeforeTimeout", func(t *testing.T) {
 		t.Parallel()
 
@@ -2820,7 +2890,7 @@ func Test_Client_SoftStopTimeout(t *testing.T) {
 			close(jobStartedChan)
 			<-ctx.Done()
 			close(jobDoneChan)
-			return nil
+			return ctx.Err()
 		}))
 
 		var (
@@ -2832,6 +2902,8 @@ func Test_Client_SoftStopTimeout(t *testing.T) {
 
 		client, err := NewClient(driver, config)
 		require.NoError(t, err)
+		subscribeChan, cancelSubscribe := client.Subscribe(EventKindJobInterrupted)
+		t.Cleanup(cancelSubscribe)
 
 		startCtx, startCtxCancel := context.WithCancel(ctx)
 		defer startCtxCancel()
@@ -2854,6 +2926,12 @@ func Test_Client_SoftStopTimeout(t *testing.T) {
 		default:
 			t.Fatal("expected job to have been cancelled by soft stop timeout")
 		}
+
+		event := riversharedtest.WaitOrTimeout(t, subscribeChan)
+		require.NotNil(t, event)
+		require.Equal(t, EventKindJobInterrupted, event.Kind)
+		require.Equal(t, rivertype.JobStateAvailable, event.Job.State)
+		require.NotNil(t, event.JobStats)
 	})
 }
 

event.go

@@ -23,6 +23,11 @@ const (
 	// differentiate each type of occurrence.
 	EventKindJobFailed EventKind = "job_failed"
 
+	// EventKindJobInterrupted occurs when a running job is interrupted because
+	// its client is shutting down and is made immediately available to be worked
+	// again. An interruption does not consume an attempt or add an attempt error.
+	EventKindJobInterrupted EventKind = "job_interrupted"
+
 	// EventKindJobSnoozed occurs when a job is snoozed.
 	EventKindJobSnoozed EventKind = "job_snoozed"
 
@@ -37,12 +42,13 @@ const (
 // exported because end users should have no way of subscribing to all known
 // kinds for forward compatibility reasons.
 var allKinds = map[EventKind]struct{}{ //nolint:gochecknoglobals
-	EventKindJobCancelled: {},
-	EventKindJobCompleted: {},
-	EventKindJobFailed:    {},
-	EventKindJobSnoozed:   {},
-	EventKindQueuePaused:  {},
-	EventKindQueueResumed: {},
+	EventKindJobCancelled:   {},
+	EventKindJobCompleted:   {},
+	EventKindJobFailed:      {},
+	EventKindJobInterrupted: {},
+	EventKindJobSnoozed:     {},
+	EventKindQueuePaused:    {},
+	EventKindQueueResumed:   {},
 }
 
 // Event wraps an event that occurred within a River client, like a job being

internal/jobcompleter/job_completer.go

@@ -45,7 +45,7 @@ type SubscribeFunc func(update CompleterJobUpdated)
 type CompleterJobUpdated struct {
 	Job      *rivertype.JobRow
 	JobStats *jobstats.JobStatistics
-	Snoozed  bool
+	Reason   riverdriver.JobSetStateReason
 }
 
 type InlineCompleter struct {
@@ -104,7 +104,7 @@ func (c *InlineCompleter) JobSetStateIfRunning(ctx context.Context, stats *jobst
 	c.subscribeCh <- []CompleterJobUpdated{{
 		Job:      jobs[0],
 		JobStats: stats,
-		Snoozed:  params.Snoozed,
+		Reason:   params.Reason,
 	}}
 
 	return nil
@@ -219,7 +219,7 @@ func (c *AsyncCompleter) JobSetStateIfRunning(ctx context.Context, stats *jobsta
 		c.subscribeCh <- []CompleterJobUpdated{{
 			Job:      jobs[0],
 			JobStats: stats,
-			Snoozed:  params.Snoozed,
+			Reason:   params.Reason,
 		}}
 
 		return nil
@@ -503,19 +503,21 @@ func (c *BatchCompleter) handleBatch(ctx context.Context) error {
 
 	var (
 		completeTime = c.Time.Now()
-		events       = make([]CompleterJobUpdated, len(jobRows))
+		events       = make([]CompleterJobUpdated, 0, len(jobRows))
 	)
-	for i, jobRow := range jobRows {
+	for _, jobRow := range jobRows {
 		setState := setStateBatch[jobRow.ID]
 		setState.Stats.CompleteDuration = completeTime.Sub(setState.StartTime)
-		events[i] = CompleterJobUpdated{
+		events = append(events, CompleterJobUpdated{
 			Job:      jobRow,
 			JobStats: setState.Stats,
-			Snoozed:  setState.Params.Snoozed,
-		}
+			Reason:   setState.Params.Reason,
+		})
 	}
 
-	c.subscribeCh <- events
+	if len(events) > 0 {
+		c.subscribeCh <- events
+	}
 
 	func() {
 		c.setStateParamsMu.Lock()

internal/jobcompleter/job_completer_test.go

@@ -326,9 +326,9 @@ func testCompleterSubscribe(t *testing.T, constructor func(schema string, exec r
 	completer.Stop() // closes subscribeChan
 
 	updates := riversharedtest.WaitOrTimeoutN(t, jobUpdateChan, 4)
-	for range 4 {
-		require.Equal(t, rivertype.JobStateCompleted, updates[0].Job.State)
-		require.False(t, updates[0].Snoozed)
+	for _, update := range updates {
+		require.Equal(t, rivertype.JobStateCompleted, update.Job.State)
+		require.Equal(t, riverdriver.JobSetStateReasonCompleted, update.Reason)
 	}
 	go completer.Stop()
 	// drain all remaining jobs
@@ -1008,15 +1008,17 @@ func testCompleter[TCompleter JobCompleter](
 			job5 = testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{Schema: bundle.schema, State: ptrutil.Ptr(rivertype.JobStateRunning)})
 			job6 = testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{Schema: bundle.schema, State: ptrutil.Ptr(rivertype.JobStateRunning)})
 			job7 = testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{Schema: bundle.schema, State: ptrutil.Ptr(rivertype.JobStateRunning)})
+			job8 = testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{Schema: bundle.schema, State: ptrutil.Ptr(rivertype.JobStateRunning)})
 		)
 
 		require.NoError(t, completer.JobSetStateIfRunning(ctx, &jobstats.JobStatistics{}, riverdriver.JobSetStateCancelled(job1.ID, time.Now(), []byte("{}"), nil)))
 		require.NoError(t, completer.JobSetStateIfRunning(ctx, &jobstats.JobStatistics{}, riverdriver.JobSetStateCompleted(job2.ID, time.Now(), nil)))
 		require.NoError(t, completer.JobSetStateIfRunning(ctx, &jobstats.JobStatistics{}, riverdriver.JobSetStateDiscarded(job3.ID, time.Now(), []byte("{}"), nil)))
 		require.NoError(t, completer.JobSetStateIfRunning(ctx, &jobstats.JobStatistics{}, riverdriver.JobSetStateErrorAvailable(job4.ID, time.Now(), []byte("{}"), nil)))
 		require.NoError(t, completer.JobSetStateIfRunning(ctx, &jobstats.JobStatistics{}, riverdriver.JobSetStateErrorRetryable(job5.ID, time.Now(), []byte("{}"), nil)))
-		require.NoError(t, completer.JobSetStateIfRunning(ctx, &jobstats.JobStatistics{}, riverdriver.JobSetStateSnoozed(job6.ID, time.Now(), 10, nil)))
-		require.NoError(t, completer.JobSetStateIfRunning(ctx, &jobstats.JobStatistics{}, riverdriver.JobSetStateSnoozedAvailable(job7.ID, time.Now(), 10, nil)))
+		require.NoError(t, completer.JobSetStateIfRunning(ctx, &jobstats.JobStatistics{}, riverdriver.JobSetStateInterrupted(job6.ID, time.Now(), job6.Attempt, nil)))
+		require.NoError(t, completer.JobSetStateIfRunning(ctx, &jobstats.JobStatistics{}, riverdriver.JobSetStateSnoozed(job7.ID, time.Now(), 10, nil)))
+		require.NoError(t, completer.JobSetStateIfRunning(ctx, &jobstats.JobStatistics{}, riverdriver.JobSetStateSnoozedAvailable(job8.ID, time.Now(), 10, nil)))
 
 		completer.Stop()
 
@@ -1025,8 +1027,9 @@ func testCompleter[TCompleter JobCompleter](
 		requireState(t, bundle, job3.ID, rivertype.JobStateDiscarded)
 		requireState(t, bundle, job4.ID, rivertype.JobStateAvailable)
 		requireState(t, bundle, job5.ID, rivertype.JobStateRetryable)
-		requireState(t, bundle, job6.ID, rivertype.JobStateScheduled)
-		requireState(t, bundle, job7.ID, rivertype.JobStateAvailable)
+		requireState(t, bundle, job6.ID, rivertype.JobStateAvailable)
+		requireState(t, bundle, job7.ID, rivertype.JobStateScheduled)
+		requireState(t, bundle, job8.ID, rivertype.JobStateAvailable)
 	})
 
 	t.Run("Subscription", func(t *testing.T) {
@@ -1037,23 +1040,23 @@ func testCompleter[TCompleter JobCompleter](
 		var (
 			job1 = testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{Schema: bundle.schema, State: ptrutil.Ptr(rivertype.JobStateRunning)})
 			job2 = testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{Schema: bundle.schema, State: ptrutil.Ptr(rivertype.JobStateRunning)})
+			job3 = testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{Schema: bundle.schema, State: ptrutil.Ptr(rivertype.JobStateRunning)})
 		)
 
 		require.NoError(t, completer.JobSetStateIfRunning(ctx, &jobstats.JobStatistics{}, riverdriver.JobSetStateCompleted(job1.ID, time.Now(), nil)))
-		require.NoError(t, completer.JobSetStateIfRunning(ctx, &jobstats.JobStatistics{}, riverdriver.JobSetStateSnoozedAvailable(job2.ID, time.Now(), job2.Attempt, nil)))
+		require.NoError(t, completer.JobSetStateIfRunning(ctx, &jobstats.JobStatistics{}, riverdriver.JobSetStateInterrupted(job2.ID, time.Now(), job2.Attempt, nil)))
+		require.NoError(t, completer.JobSetStateIfRunning(ctx, &jobstats.JobStatistics{}, riverdriver.JobSetStateSnoozedAvailable(job3.ID, time.Now(), job3.Attempt, nil)))
 
 		completer.Stop()
 
 		// Unfortunately we have to do this awkward loop to wait for all updates
 		// because updates are sent through the channel as batches. The sync and
-		// async completer don't work in batches and therefore send batches of
-		// one item each while the batch completer will send both. Put
-		// otherwise, expect the sync and async completers to iterate this loop
-		// twice and batch completer to iterate it once.
+		// async completers send batches of one item each, while the batch
+		// completer may group updates.
 		var jobUpdates []CompleterJobUpdated
 		for {
 			jobUpdates = append(jobUpdates, riversharedtest.WaitOrTimeout(t, bundle.subscribeCh)...)
-			if len(jobUpdates) >= 2 {
+			if len(jobUpdates) >= 3 {
 				break
 			}
 		}
@@ -1070,10 +1073,13 @@ func testCompleter[TCompleter JobCompleter](
 
 		job1Update := findUpdate(job1.ID)
 		require.Equal(t, rivertype.JobStateCompleted, job1Update.Job.State)
-		require.False(t, job1Update.Snoozed)
+		require.Equal(t, riverdriver.JobSetStateReasonCompleted, job1Update.Reason)
 		job2Update := findUpdate(job2.ID)
 		require.Equal(t, rivertype.JobStateAvailable, job2Update.Job.State)
-		require.True(t, job2Update.Snoozed)
+		require.Equal(t, riverdriver.JobSetStateReasonInterrupted, job2Update.Reason)
+		job3Update := findUpdate(job3.ID)
+		require.Equal(t, rivertype.JobStateAvailable, job3Update.Job.State)
+		require.Equal(t, riverdriver.JobSetStateReasonSnoozed, job3Update.Reason)
 	})
 
 	t.Run("MultipleCycles", func(t *testing.T) {

internal/jobexecutor/job_executor.go

@@ -18,6 +18,7 @@ import (
 	"github.com/riverqueue/river/internal/jobcompleter"
 	"github.com/riverqueue/river/internal/jobstats"
 	"github.com/riverqueue/river/internal/pluginlookup"
+	"github.com/riverqueue/river/internal/rivercommon"
 	"github.com/riverqueue/river/internal/workunit"
 	"github.com/riverqueue/river/riverdriver"
 	"github.com/riverqueue/river/rivershared/baseservice"
@@ -450,6 +451,7 @@ func (e *JobExecutor) reportError(ctx context.Context, jobRow *rivertype.JobRow,
 		cancelJob bool
 		cancelErr *rivertype.JobCancelError
 	)
+	softStopped := isSoftStopCancelError(ctx, res.Err)
 
 	logAttrs := []any{
 		slog.String("error", res.ErrorStr()),
@@ -461,6 +463,8 @@ func (e *JobExecutor) reportError(ctx context.Context, jobRow *rivertype.JobRow,
 	case errors.As(res.Err, &cancelErr):
 		cancelJob = true
 		e.Logger.DebugContext(ctx, e.Name+": Job cancelled explicitly", logAttrs...)
+	case softStopped:
+		e.Logger.InfoContext(ctx, e.Name+": Job stopped due to client shutdown; retrying", logAttrs...)
 	case res.Err != nil:
 		if jobRow.Attempt >= jobRow.MaxAttempts {
 			e.Logger.InfoContext(ctx, e.Name+": Job errored", logAttrs...)
@@ -471,11 +475,21 @@ func (e *JobExecutor) reportError(ctx context.Context, jobRow *rivertype.JobRow,
 		e.Logger.InfoContext(ctx, e.Name+": Job panicked", logAttrs...)
 	}
 
-	if e.ErrorHandler != nil && !cancelJob {
+	if e.ErrorHandler != nil && !cancelJob && !softStopped {
 		// Error handlers also have an opportunity to cancel the job.
 		cancelJob = e.invokeErrorHandler(ctx, res)
 	}
 
+	now := e.Time.Now()
+
+	if softStopped {
+		params := riverdriver.JobSetStateInterrupted(jobRow.ID, now, max(jobRow.Attempt-1, 0), metadataUpdates)
+		if err := e.Completer.JobSetStateIfRunning(ctx, e.stats, params); err != nil {
+			e.Logger.ErrorContext(ctx, e.Name+": Failed to make soft-stopped job available", logAttrs...)
+		}
+		return
+	}
+
 	attemptErr := rivertype.AttemptError{
 		At:      e.start,
 		Attempt: jobRow.Attempt,
@@ -489,8 +503,6 @@ func (e *JobExecutor) reportError(ctx context.Context, jobRow *rivertype.JobRow,
 		return
 	}
 
-	now := e.Time.Now()
-
 	if cancelJob {
 		if err := e.Completer.JobSetStateIfRunning(ctx, e.stats, riverdriver.JobSetStateCancelled(jobRow.ID, now, errData, metadataUpdates)); err != nil {
 			e.Logger.ErrorContext(ctx, e.Name+": Failed to cancel job and report error", logAttrs...)
@@ -539,6 +551,15 @@ func (e *JobExecutor) reportError(ctx context.Context, jobRow *rivertype.JobRow,
 	}
 }
 
+// isSoftStopCancelError reports whether a worker returned because the client
+// was stopping and cancelled its job context. The context cause distinguishes
+// client stop cancellation from ordinary worker cancellation or timeouts.
+func isSoftStopCancelError(ctx context.Context, err error) bool {
+	return err != nil &&
+		errors.Is(context.Cause(ctx), rivercommon.ErrStop) &&
+		(errors.Is(err, context.Canceled) || errors.Is(err, rivercommon.ErrStop))
+}
+
 type withJobsAndErrorsByID interface {
 	ErrorsByID() map[int64]error
 	Jobs() []*rivertype.JobRow