Prevent SQLite driver from adding nil placeholders for missing jobs

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)

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