Skip events for soft stop failures

Author: brandur

CHANGELOG.md

@@ -11,6 +11,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - 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
@@ -80,7 +84,6 @@ river migrate-get --database-url sqlite:// --version 7 --down > river7.down.sql
 - Convert SQLite JSON columns to JSONB (including migration). [PR #1224](https://github.com/riverqueue/river/pull/1224).
 - Change SQLite driver operations over to use bulk inserts where possible now that sqlc has better support for `json_each`. [PR #1276](https://github.com/riverqueue/river/pull/1276)
 - Detect duplicate step names across `river.ResumableStep` and return a validation error. [PR #1281](https://github.com/riverqueue/river/pull/1281)
-- 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)
 - Earlier backpressure from `BatchCompleter` when it's throughput is saturated with fewer warnings to console. [PR #1292](https://github.com/riverqueue/river/pull/1292)
 - Series of minor optimizations in `BatchCompleter` raising throughput ~20% when it's the bottleneck in job processing (e.g. in benchmarks). [PR #1293](https://github.com/riverqueue/river/pull/1293)
 

client_test.go

@@ -2750,6 +2750,21 @@ func Test_Client_SoftStopTimeout(t *testing.T) {
 		testutil.JobArgsReflectKind[JobArgs]
 	}
 
+	requireNoJobEvents := func(t *testing.T, subscribeChan <-chan *Event) {
+		t.Helper()
+
+		var unexpectedEvent *Event
+		require.Eventually(t, func() bool {
+			select {
+			case unexpectedEvent = <-subscribeChan:
+				return unexpectedEvent == nil
+			default:
+				return true
+			}
+		}, 500*time.Millisecond, 10*time.Millisecond)
+		require.Nil(t, unexpectedEvent)
+	}
+
 	t.Run("EscalatesToHardStopAfterTimeout", func(t *testing.T) {
 		t.Parallel()
 
@@ -2766,6 +2781,8 @@ func Test_Client_SoftStopTimeout(t *testing.T) {
 		}))
 
 		client := runNewTestClient(ctx, t, config)
+		subscribeChan, cancelSubscribe := client.Subscribe(EventKindJobCancelled, EventKindJobCompleted, EventKindJobFailed, EventKindJobSnoozed)
+		t.Cleanup(cancelSubscribe)
 
 		_, err := client.Insert(ctx, JobArgs{}, nil)
 		require.NoError(t, err)
@@ -2782,6 +2799,7 @@ func Test_Client_SoftStopTimeout(t *testing.T) {
 		default:
 			t.Fatal("expected job to have been cancelled by soft stop timeout")
 		}
+		requireNoJobEvents(t, subscribeChan)
 	})
 
 	t.Run("ErroringJobGetsFreshAttempt", func(t *testing.T) {
@@ -2893,6 +2911,8 @@ func Test_Client_SoftStopTimeout(t *testing.T) {
 
 		client, err := NewClient(driver, config)
 		require.NoError(t, err)
+		subscribeChan, cancelSubscribe := client.Subscribe(EventKindJobCancelled, EventKindJobCompleted, EventKindJobFailed, EventKindJobSnoozed)
+		t.Cleanup(cancelSubscribe)
 
 		startCtx, startCtxCancel := context.WithCancel(ctx)
 		defer startCtxCancel()
@@ -2915,6 +2935,7 @@ func Test_Client_SoftStopTimeout(t *testing.T) {
 		default:
 			t.Fatal("expected job to have been cancelled by soft stop timeout")
 		}
+		requireNoJobEvents(t, subscribeChan)
 	})
 }
 

internal/jobcompleter/job_completer.go

@@ -95,8 +95,9 @@ func (c *InlineCompleter) JobSetStateIfRunning(ctx context.Context, stats *jobst
 
 	// The driver intentionally returns 0 rows when a job is deleted while the
 	// completer is finalizing it (see UnknownJobIgnored shared driver test).
-	// Guard against an index-out-of-range panic in that case.
-	if len(jobs) < 1 {
+	// Guard against an index-out-of-range panic in that case, and also skip
+	// publishing an event when requested by the caller.
+	if params.SkipEvent || len(jobs) < 1 {
 		return nil
 	}
 
@@ -210,8 +211,9 @@ func (c *AsyncCompleter) JobSetStateIfRunning(ctx context.Context, stats *jobsta
 
 		// The driver intentionally returns 0 rows when a job is deleted while the
 		// completer is finalizing it (see UnknownJobIgnored shared driver test).
-		// Guard against an index-out-of-range panic in that case.
-		if len(jobs) < 1 {
+		// Guard against an index-out-of-range panic in that case, and also skip
+		// publishing an event when requested by the caller.
+		if params.SkipEvent || len(jobs) < 1 {
 			return nil
 		}
 
@@ -503,19 +505,24 @@ 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]
+		if setState.Params.SkipEvent {
+			continue
+		}
 		setState.Stats.CompleteDuration = completeTime.Sub(setState.StartTime)
-		events[i] = CompleterJobUpdated{
+		events = append(events, CompleterJobUpdated{
 			Job:      jobRow,
 			JobStats: setState.Stats,
 			Snoozed:  setState.Params.Snoozed,
-		}
+		})
 	}
 
-	c.subscribeCh <- events
+	if len(events) > 0 {
+		c.subscribeCh <- events
+	}
 
 	func() {
 		c.setStateParamsMu.Lock()

internal/jobexecutor/job_executor.go

@@ -485,6 +485,7 @@ func (e *JobExecutor) reportError(ctx context.Context, jobRow *rivertype.JobRow,
 
 	if softStopped {
 		params := riverdriver.JobSetStateErrorAvailable(jobRow.ID, now, ptrutil.Ptr(max(jobRow.Attempt-1, 0)), nil, metadataUpdates)
+		params.SkipEvent = true
 		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...)
 		}

internal/jobexecutor/job_executor_test.go

@@ -383,7 +383,11 @@ func TestJobExecutor_Execute(t *testing.T) {
 		executor.WorkUnit = newWorkUnitFactoryWithCustomRetry(func() error { return context.Canceled }, nil).MakeUnit(bundle.jobRow)
 
 		executor.Execute(workCtx)
-		riversharedtest.WaitOrTimeout(t, bundle.updateCh)
+		select {
+		case update := <-bundle.updateCh:
+			t.Fatalf("unexpected completion update for soft-stop cancel: %#v", update)
+		default:
+		}
 
 		job, err := bundle.exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{
 			ID:     bundle.jobRow.ID,

riverdriver/river_driver_interface.go

@@ -565,8 +565,10 @@ type JobSetStateIfRunningParams struct {
 	MetadataUpdates []byte
 	ScheduledAt     *time.Time
 	Schema          string // added by completer
-	Snoozed         bool
-	State           rivertype.JobState
+	// SkipEvent suppresses event publication for this transition.
+	SkipEvent bool
+	Snoozed   bool
+	State     rivertype.JobState
 }
 
 func JobSetStateCancelled(id int64, finalizedAt time.Time, errData []byte, metadataUpdates []byte) *JobSetStateIfRunningParams {