diff --git a/CHANGELOG.md b/CHANGELOG.md index 5ced190a..09d691c2 100644 --- a/CHANGELOG.md +++ b/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). diff --git a/internal/jobcompleter/job_completer.go b/internal/jobcompleter/job_completer.go index 271ff798..57b827cb 100644 --- a/internal/jobcompleter/job_completer.go +++ b/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], diff --git a/internal/jobcompleter/job_completer_test.go b/internal/jobcompleter/job_completer_test.go index e4ec74f8..fc499dbf 100644 --- a/internal/jobcompleter/job_completer_test.go +++ b/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()