JobRescuer can overwrite a job that completed after the stuck-job fetch

Summary

JobRescuer selects stuck jobs while they are running, builds rescue
parameters from that snapshot, and later calls JobRescueMany. The rescue
update only checks the job id. If the original worker completes the job after
JobGetStuck returns but before JobRescueMany runs, the rescue update can
move the completed row back to retryable, or to another rescue outcome such
as discarded or cancelled.

Confirmed against River master at
4aa884e927fc07635605967729a288ac9416ebf1 on 2026年07月10日 with Go 1.26.5
on Linux. I did not find an exact existing public report during a targeted
search.

Why this seems wrong

A successful worker completion should be terminal. Once a worker has committed
the row as completed, a maintenance pass based on an older running snapshot
should not rewrite the job state, append a rescue error, update
river:rescue_count, or clear/set finalized_at and scheduled_at.

The worker completion path already uses current-state guards, so adding a
current-state guard to rescue seems consistent with the surrounding design.

Relevant code

The stuck-job query filters to rows that are running only at read time:

-- name: JobGetStuck :many
SELECT *
FROM river_job
WHERE state = 'running'
 AND attempted_at < @stuck_horizon::timestamptz
ORDER BY id
LIMIT @max;

The Postgres rescue update writes by id only:

-- name: JobRescueMany :exec
UPDATE river_job
SET
 errors = array_append(errors, updated_job.error),
 finalized_at = updated_job.finalized_at,
 scheduled_at = updated_job.scheduled_at,
 metadata = river_job.metadata || jsonb_build_object(...),
 state = updated_job.state
FROM (...) AS updated_job
WHERE river_job.id = updated_job.id;

The SQLite rescue update has the same effective guard:

-- name: JobRescue :exec
UPDATE river_job
SET
 errors = jsonb_insert(...),
 finalized_at = cast(sqlc.narg('finalized_at') as text),
 scheduled_at = @scheduled_at,
 metadata = jsonb_set(...),
 state = @state
WHERE id = @id;

Interleaving

1. A job is running and old enough for rescue.
2. JobRescuer calls JobGetStuck and receives the row.
3. Before JobRescueMany runs, the worker completes the row normally and
   commits it as completed.
4. JobRescuer executes the stale rescue update by id and changes the same row
   to retryable, discarded, or cancelled.

Reproduction

I reproduced the storage-level race on SQLite with a focused shared driver test
that:

1. Creates a running row old enough to be returned by JobGetStuck.
2. Calls JobGetStuck and keeps that stale fetched row id.
3. Completes the row through JobSetStateIfRunningMany, matching the normal
   guarded worker completion write.
4. Calls JobRescueMany with the stale fetched row id.
5. Re-reads the row and expects it to remain completed.

Core test sequence:

stuckJobs, err := exec.JobGetStuck(ctx, &riverdriver.JobGetStuckParams{
 Max: 10,
 StuckHorizon: attemptedAt.Add(time.Hour),
})
require.NoError(t, err)
_, err = exec.JobSetStateIfRunningMany(ctx, setStateManyParams(
 riverdriver.JobSetStateCompleted(job.ID, completedAt, nil),
))
require.NoError(t, err)
_, err = exec.JobRescueMany(ctx, &riverdriver.JobRescueManyParams{
 ID: []int64{stuckJobs[0].ID},
 Error: [][]byte{rescueErrorPayload},
 FinalizedAt: []*time.Time{nil},
 ScheduledAt: []time.Time{rescueScheduledAt},
 State: []string{string(rivertype.JobStateRetryable)},
})
require.NoError(t, err)
jobAfterRescue, err := exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: job.ID})
require.NoError(t, err)
require.Equal(t, rivertype.JobStateCompleted, jobAfterRescue.State)

Run:

go test ./riverdriver/riverdrivertest -run "TestDriverRiverSQLiteModernC/JobSetStateIfRunningMany_JobSetStateCompleted/JobRescueMany_DoesNotOverwriteCompletedJobFromStaleSnapshot" -count=1 -v

Current result:

Error: Not equal:
 expected: "completed"
 actual : "retryable"
Test: TestDriverRiverSQLiteModernC/JobSetStateIfRunningMany_JobSetStateCompleted/JobRescueMany_DoesNotOverwriteCompletedJobFromStaleSnapshot

The same update shape exists in the Postgres SQL, so the SQLite test is a
compact reproduction of the shared state-transition issue.

Expected behavior

The rescue update should only affect rows that are still running when the
rescue write executes. If a row has already completed, failed into another
final state, or otherwise left running, the rescuer should treat it as
already resolved and leave it unchanged.

Suggested fix

Add a current-state predicate to rescue updates, for example:

WHERE river_job.id = updated_job.id
 AND river_job.state = 'running'

For SQLite, add the equivalent AND state = 'running' predicate to
JobRescue.

If the update returns/counts fewer rows than requested, those rows can be
treated as no-ops because they are no longer eligible for rescue.