From 25f605c4343dfefe8fe74afc5646bbd6bcd34535 Mon Sep 17 00:00:00 2001 From: Brandur Date: Thu, 16 Jul 2026 14:45:50 -0500 Subject: [PATCH] Prevent SQLite driver from adding `nil` placeholders for missing jobs This one's a follow up to #1308 which modifies the completers to handle a rare condition where no rows are returned from `JobSetStateIfRunningMany`. Codex caught it while I was looking at the other PR. Here, also modify `JobSetStateIfRunningMany` in the SQLite driver so that if a job is missing from query return, in omits the element completely instead of adding a `nil` element in the return slice, bringing its behavior in line with Postgres. We add a check in the driver test suite to verify the consistent behavior. --- riverdriver/riverdrivertest/job_update.go | 5 ++++- riverdriver/riversqlite/river_sqlite_driver.go | 7 ++++--- 2 files changed, 8 insertions(+), 4 deletions(-) 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