diff --git a/riverdriver/riverdrivertest/job_update.go b/riverdriver/riverdrivertest/job_update.go index e2d22210..5c1f6a9a 100644 --- a/riverdriver/riverdrivertest/job_update.go +++ b/riverdriver/riverdrivertest/job_update.go @@ -658,8 +658,9 @@ func exerciseJobUpdate[TTx any](ctx context.Context, t *testing.T, executorWithT // The operation doesn't return anything like a "not found" in case // of an unknown job so that it doesn't fail in case a job is // deleted in the interim as a completer is trying to finalize it. - _, err := exec.JobSetStateIfRunningMany(ctx, setStateManyParams(riverdriver.JobSetStateCompleted(0, time.Now().UTC(), nil))) + jobsAfter, err := exec.JobSetStateIfRunningMany(ctx, setStateManyParams(riverdriver.JobSetStateCompleted(0, time.Now().UTC(), nil))) require.NoError(t, err) + require.Empty(t, jobsAfter) }) }) @@ -934,11 +935,13 @@ func exerciseJobUpdate[TTx any](ctx context.Context, t *testing.T, executorWithT job3 := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{State: ptrutil.Ptr(rivertype.JobStateRunning)}) jobsAfter, err := exec.JobSetStateIfRunningMany(ctx, setStateManyParams( + riverdriver.JobSetStateCompleted(0, now, nil), riverdriver.JobSetStateCompleted(job1.ID, now, []byte(`{"a":"b"}`)), riverdriver.JobSetStateErrorRetryable(job2.ID, future, makeErrPayload(t, now), nil), riverdriver.JobSetStateCancelled(job3.ID, now, makeErrPayload(t, now), nil), )) require.NoError(t, err) + require.Len(t, jobsAfter, 3) completedJob := jobsAfter[0] require.Equal(t, rivertype.JobStateCompleted, completedJob.State) require.WithinDuration(t, now, *completedJob.FinalizedAt, time.Microsecond) diff --git a/riverdriver/riversqlite/river_sqlite_driver.go b/riverdriver/riversqlite/river_sqlite_driver.go index c7df4964..f4cc1971 100644 --- a/riverdriver/riversqlite/river_sqlite_driver.go +++ b/riverdriver/riversqlite/river_sqlite_driver.go @@ -833,7 +833,7 @@ func (e *Executor) JobSchedule(ctx context.Context, params *riverdriver.JobSched } func (e *Executor) JobSetStateIfRunningMany(ctx context.Context, params *riverdriver.JobSetStateIfRunningManyParams) ([]*rivertype.JobRow, error) { - setRes := make([]*rivertype.JobRow, len(params.ID)) + setRes := make([]*rivertype.JobRow, 0, len(params.ID)) if err := dbutil.WithTx(ctx, e, func(ctx context.Context, execTx riverdriver.ExecutorTx) error { ctx = schemaTemplateParam(ctx, params.Schema) @@ -880,7 +880,7 @@ func (e *Executor) JobSetStateIfRunningMany(ctx context.Context, params *riverdr }) if err != nil { if errors.Is(err, sql.ErrNoRows) { - return nil + continue } return fmt.Errorf("error setting job metadata: %w", err) @@ -889,10 +889,11 @@ func (e *Executor) JobSetStateIfRunningMany(ctx context.Context, params *riverdr return fmt.Errorf("error setting job state: %w", err) } } - setRes[i], err = jobRowFromInternal(job) + jobRow, err := jobRowFromInternal(job) if err != nil { return err } + setRes = append(setRes, jobRow) } return nil