fix(completer): guard against index panic when job deleted during completion

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- Guard against empty job slice returned by `JobSetStateIfRunningMany` when a job has been deleted mid-run. [PR #1308](https://github.com/riverqueue/river/pull/1308).
+
 ### Changed
 
 - Both hooks and middleware should now be registered under a general `Config.Plugins` instead, a change that simpifies things somewhat, and better handles cases where a hook or middleware might be _both_ a hook and a middleware as opposed to only one or the other. `Config.Hooks` and `Config.Middleware` are still available for use, but considered deprecated in favor of the more general `Config.Plugins`. [PR #1284](https://github.com/riverqueue/river/pull/1284).

internal/jobcompleter/job_completer.go

@@ -92,6 +92,13 @@ func (c *InlineCompleter) JobSetStateIfRunning(ctx context.Context, stats *jobst
 		return err
 	}
 
+	// 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 {
+		return nil
+	}
+
 	stats.CompleteDuration = c.Time.Now().Sub(start)
 	c.subscribeCh <- []CompleterJobUpdated{{
 		Job:      jobs[0],
@@ -200,6 +207,13 @@ func (c *AsyncCompleter) JobSetStateIfRunning(ctx context.Context, stats *jobsta
 			return err
 		}
 
+		// 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 {
+			return nil
+		}
+
 		stats.CompleteDuration = c.Time.Now().Sub(start)
 		c.subscribeCh <- []CompleterJobUpdated{{
 			Job:      jobs[0],

internal/jobcompleter/job_completer_test.go

@@ -110,6 +110,28 @@ func TestInlineJobCompleter_Complete(t *testing.T) {
 	require.Equal(t, numRetries, attempt)
 }
 
+func TestInlineJobCompleter_CompleteDeletedJob(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+
+	// The driver intentionally returns 0 rows when a job is deleted while the
+	// completer is finalizing it (UnknownJobIgnored contract). Verify that the
+	// InlineCompleter does not panic in that case.
+	execMock := &partialExecutorMock{}
+	execMock.JobSetStateIfRunningManyFunc = func(ctx context.Context, params *riverdriver.JobSetStateIfRunningManyParams) ([]*rivertype.JobRow, error) {
+		return []*rivertype.JobRow{}, nil
+	}
+
+	subscribeCh := make(chan []CompleterJobUpdated, 10)
+	t.Cleanup(riverinternaltest.DiscardContinuously(subscribeCh))
+
+	completer := NewInlineCompleter(riversharedtest.BaseServiceArchetype(t), "", execMock, &riverpilot.StandardPilot{}, subscribeCh)
+	t.Cleanup(completer.Stop)
+
+	require.NoError(t, completer.JobSetStateIfRunning(ctx, &jobstats.JobStatistics{}, riverdriver.JobSetStateCompleted(1, time.Now(), nil)))
+}
+
 func TestInlineJobCompleter_Subscribe(t *testing.T) {
 	t.Parallel()
 
@@ -228,6 +250,29 @@ func TestAsyncJobCompleter_Complete(t *testing.T) {
 	resultCh <- nil
 }
 
+func TestAsyncJobCompleter_CompleteDeletedJob(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+
+	// The driver intentionally returns 0 rows when a job is deleted while the
+	// completer is finalizing it (UnknownJobIgnored contract). Verify that the
+	// AsyncCompleter does not panic in that case.
+	execMock := &partialExecutorMock{}
+	execMock.JobSetStateIfRunningManyFunc = func(ctx context.Context, params *riverdriver.JobSetStateIfRunningManyParams) ([]*rivertype.JobRow, error) {
+		return []*rivertype.JobRow{}, nil
+	}
+
+	subscribeCh := make(chan []CompleterJobUpdated, 10)
+	t.Cleanup(riverinternaltest.DiscardContinuously(subscribeCh))
+
+	completer := newAsyncCompleterWithConcurrency(riversharedtest.BaseServiceArchetype(t), "", execMock, &riverpilot.StandardPilot{}, 2, subscribeCh)
+	require.NoError(t, completer.Start(ctx))
+	t.Cleanup(completer.Stop)
+
+	require.NoError(t, completer.JobSetStateIfRunning(ctx, &jobstats.JobStatistics{}, riverdriver.JobSetStateCompleted(1, time.Now(), nil)))
+}
+
 func TestAsyncJobCompleter_Subscribe(t *testing.T) {
 	t.Parallel()