From 4ce683563cf382bc13814fe59fb7c19841f0007f Mon Sep 17 00:00:00 2001 From: Brandur Date: Mon, 20 Jul 2026 23:53:52 -0500 Subject: [PATCH] Make job rescue paginated to protect against full batch of no-rescue jobs This one's aimed at fixing a long-standing bug accidentally detected while working on another rescue-related feature. The `JobRescuer` works by fetching a full batch of jobs to rescue then going through each one to determine what it should be doing about it. The default batch size is 10k, so this generally works perfectly fine. Some stuck jobs are potentially not rescued if their timeout is configured to be -1 (no timeout). Codex identified a tail bug possibility in which if you had an entire batch worth of jobs with -1 timeouts, they'd block any jobs after them from being rescued. So the rescuer would rescue 10k jobs, determine none of them needed rescue, then go back to sleep, stranding any jobs after that. This is such a tail possibility that it's probably fine, but just since we're making changes to the rescue driver boundary right now anyway, it's not a bad time to just get this one fixed up. --- CHANGELOG.md | 1 + internal/maintenance/job_rescuer.go | 17 ++++++++---- internal/maintenance/job_rescuer_test.go | 26 +++++++++++++++++++ riverdriver/river_driver_interface.go | 1 + .../internal/dbsqlc/river_job.sql.go | 8 +++--- .../river_database_sql_driver.go | 1 + riverdriver/riverdrivertest/job_read.go | 13 ++++++++-- .../riverpgxv5/internal/dbsqlc/river_job.sql | 3 ++- .../internal/dbsqlc/river_job.sql.go | 8 +++--- riverdriver/riverpgxv5/river_pgx_v5_driver.go | 1 + .../riversqlite/internal/dbsqlc/river_job.sql | 1 + .../internal/dbsqlc/river_job.sql.go | 8 +++--- .../riversqlite/river_sqlite_driver.go | 1 + 13 files changed, 72 insertions(+), 17 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b72b93de..dc15626c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,6 +18,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### 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). +- Fixed `JobRescuer` pagination so a full batch of running jobs with disabled or longer worker-specific timeouts can't prevent later stuck jobs from being rescued. [PR #1318](https://github.com/riverqueue/river/pull/1318). ## [0.40.0] - 2026-07-02 diff --git a/internal/maintenance/job_rescuer.go b/internal/maintenance/job_rescuer.go index 170c0785..81a89a84 100644 --- a/internal/maintenance/job_rescuer.go +++ b/internal/maintenance/job_rescuer.go @@ -191,11 +191,14 @@ type metadataWithCancelAttemptedAt struct { } func (s *JobRescuer) runOnce(ctx context.Context) (*rescuerRunOnceResult, error) { + var afterID int64 + res := &rescuerRunOnceResult{} + stuckHorizon := time.Now().Add(-s.Config.RescueAfter) for { - stuckHorizon := time.Now().Add(-s.Config.RescueAfter) - stuckJobs, err := s.getStuckJobs(ctx, stuckHorizon) + batchSize := s.batchSize() + stuckJobs, err := s.getStuckJobs(ctx, afterID, batchSize, stuckHorizon) if err != nil { if errors.Is(err, context.DeadlineExceeded) { s.reducedBatchSizeBreaker.Trip() @@ -207,6 +210,9 @@ func (s *JobRescuer) runOnce(ctx context.Context) (*rescuerRunOnceResult, error) s.reducedBatchSizeBreaker.ResetIfNotOpen() s.TestSignals.FetchedBatch.Signal(struct{}{}) + if len(stuckJobs) > 0 { + afterID = stuckJobs[len(stuckJobs)-1].ID + } now := time.Now().UTC() @@ -277,7 +283,7 @@ func (s *JobRescuer) runOnce(ctx context.Context) (*rescuerRunOnceResult, error) // Number of rows fetched was less than query `LIMIT` which means work is // done for this round: - if len(stuckJobs) < s.batchSize() { + if len(stuckJobs) < batchSize { break } @@ -287,12 +293,13 @@ func (s *JobRescuer) runOnce(ctx context.Context) (*rescuerRunOnceResult, error) return res, nil } -func (s *JobRescuer) getStuckJobs(ctx context.Context, stuckHorizon time.Time) ([]*rivertype.JobRow, error) { +func (s *JobRescuer) getStuckJobs(ctx context.Context, afterID int64, batchSize int, stuckHorizon time.Time) ([]*rivertype.JobRow, error) { ctx, cancelFunc := context.WithTimeout(ctx, riversharedmaintenance.TimeoutDefault) defer cancelFunc() return s.Config.Pilot.JobGetStuck(ctx, s.exec, &riverdriver.JobGetStuckParams{ - Max: s.batchSize(), + AfterID: afterID, + Max: batchSize, Schema: s.Config.Schema, StuckHorizon: stuckHorizon, }) diff --git a/internal/maintenance/job_rescuer_test.go b/internal/maintenance/job_rescuer_test.go index 522c8aaa..a5024a88 100644 --- a/internal/maintenance/job_rescuer_test.go +++ b/internal/maintenance/job_rescuer_test.go @@ -292,6 +292,32 @@ func TestJobRescuer(t *testing.T) { } }) + t.Run("RescuesPastFullBatchOfJobsWithNoTimeout", func(t *testing.T) { + t.Parallel() + + rescuer, bundle := setup(t) + rescuer.Config.Default = 3 + + noTimeoutJobs := make([]*rivertype.JobRow, rescuer.Config.Default+1) + for i := range noTimeoutJobs { + noTimeoutJobs[i] = testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{Kind: ptrutil.Ptr(rescuerJobKindNoTimeout), State: ptrutil.Ptr(rivertype.JobStateRunning), AttemptedAt: ptrutil.Ptr(bundle.rescueHorizon.Add(-24 * time.Hour)), MaxAttempts: ptrutil.Ptr(5)}) + } + jobToRescue := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{Kind: ptrutil.Ptr(rescuerJobKind), State: ptrutil.Ptr(rivertype.JobStateRunning), AttemptedAt: ptrutil.Ptr(bundle.rescueHorizon.Add(-1 * time.Hour)), MaxAttempts: ptrutil.Ptr(5)}) + + _, err := rescuer.runOnce(ctx) + require.NoError(t, err) + + for _, job := range noTimeoutJobs { + jobAfter, err := bundle.exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: job.ID, Schema: rescuer.Config.Schema}) + require.NoError(t, err) + require.Equal(t, rivertype.JobStateRunning, jobAfter.State) + } + + jobToRescueAfter, err := bundle.exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: jobToRescue.ID, Schema: rescuer.Config.Schema}) + require.NoError(t, err) + require.Equal(t, rivertype.JobStateRetryable, jobToRescueAfter.State) + }) + t.Run("CustomizableInterval", func(t *testing.T) { t.Parallel() diff --git a/riverdriver/river_driver_interface.go b/riverdriver/river_driver_interface.go index 271f5d30..64412aad 100644 --- a/riverdriver/river_driver_interface.go +++ b/riverdriver/river_driver_interface.go @@ -438,6 +438,7 @@ type JobGetByKindManyParams struct { } type JobGetStuckParams struct { + AfterID int64 Max int Schema string StuckHorizon time.Time diff --git a/riverdriver/riverdatabasesql/internal/dbsqlc/river_job.sql.go b/riverdriver/riverdatabasesql/internal/dbsqlc/river_job.sql.go index ac01941a..c72d277a 100644 --- a/riverdriver/riverdatabasesql/internal/dbsqlc/river_job.sql.go +++ b/riverdriver/riverdatabasesql/internal/dbsqlc/river_job.sql.go @@ -609,18 +609,20 @@ const jobGetStuck = `-- name: JobGetStuck :many SELECT id, args, attempt, attempted_at, attempted_by, created_at, errors, finalized_at, kind, max_attempts, metadata, priority, queue, state, scheduled_at, tags, unique_key, unique_states FROM /* TEMPLATE: schema */river_job WHERE state = 'running' - AND attempted_at < $1::timestamptz + AND id > $1::bigint + AND attempted_at < $2::timestamptz ORDER BY id -LIMIT $2 +LIMIT $3 ` type JobGetStuckParams struct { + AfterID int64 StuckHorizon time.Time Max int32 } func (q *Queries) JobGetStuck(ctx context.Context, db DBTX, arg *JobGetStuckParams) ([]*RiverJob, error) { - rows, err := db.QueryContext(ctx, jobGetStuck, arg.StuckHorizon, arg.Max) + rows, err := db.QueryContext(ctx, jobGetStuck, arg.AfterID, arg.StuckHorizon, arg.Max) if err != nil { return nil, err } diff --git a/riverdriver/riverdatabasesql/river_database_sql_driver.go b/riverdriver/riverdatabasesql/river_database_sql_driver.go index 211a4374..8320b17a 100644 --- a/riverdriver/riverdatabasesql/river_database_sql_driver.go +++ b/riverdriver/riverdatabasesql/river_database_sql_driver.go @@ -343,6 +343,7 @@ func (e *Executor) JobGetByKindMany(ctx context.Context, params *riverdriver.Job func (e *Executor) JobGetStuck(ctx context.Context, params *riverdriver.JobGetStuckParams) ([]*rivertype.JobRow, error) { jobs, err := dbsqlc.New().JobGetStuck(schemaTemplateParam(ctx, params.Schema), e.dbtx, &dbsqlc.JobGetStuckParams{ + AfterID: params.AfterID, Max: int32(min(params.Max, math.MaxInt32)), //nolint:gosec StuckHorizon: params.StuckHorizon, }) diff --git a/riverdriver/riverdrivertest/job_read.go b/riverdriver/riverdrivertest/job_read.go index 5649d910..942ca446 100644 --- a/riverdriver/riverdrivertest/job_read.go +++ b/riverdriver/riverdrivertest/job_read.go @@ -528,8 +528,8 @@ func exerciseJobRead[TTx any](ctx context.Context, t *testing.T, executorWithTx t.Logf("stuckJob1 full = %s", spew.Sdump(stuckJob1)) - // Not returned because we put a maximum of two. - _ = testfactory.Job(ctx, t, exec, &testfactory.JobOpts{AttemptedAt: &beforeHorizon, State: ptrutil.Ptr(rivertype.JobStateRunning)}) + // Not returned on the first page because we put a maximum of two. + stuckJob3 := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{AttemptedAt: &beforeHorizon, State: ptrutil.Ptr(rivertype.JobStateRunning)}) // Not stuck because not in running state. _ = testfactory.Job(ctx, t, exec, &testfactory.JobOpts{State: ptrutil.Ptr(rivertype.JobStateAvailable)}) @@ -545,6 +545,15 @@ func exerciseJobRead[TTx any](ctx context.Context, t *testing.T, executorWithTx require.NoError(t, err) require.Equal(t, []int64{stuckJob1.ID, stuckJob2.ID}, sliceutil.Map(stuckJobs, func(j *rivertype.JobRow) int64 { return j.ID })) + + stuckJobs, err = exec.JobGetStuck(ctx, &riverdriver.JobGetStuckParams{ + AfterID: stuckJob2.ID, + Max: 2, + StuckHorizon: horizon, + }) + require.NoError(t, err) + require.Equal(t, []int64{stuckJob3.ID}, + sliceutil.Map(stuckJobs, func(j *rivertype.JobRow) int64 { return j.ID })) }) t.Run("JobKindList", func(t *testing.T) { diff --git a/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql b/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql index 5436ab0a..509e479b 100644 --- a/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql +++ b/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql @@ -258,6 +258,7 @@ ORDER BY id; SELECT * FROM /* TEMPLATE: schema */river_job WHERE state = 'running' + AND id > @after_id::bigint AND attempted_at < @stuck_horizon::timestamptz ORDER BY id LIMIT @max; @@ -726,4 +727,4 @@ SET metadata = CASE WHEN @metadata_do_update::boolean THEN @metadata::jsonb ELSE metadata END, state = CASE WHEN @state_do_update::boolean THEN @state::/* TEMPLATE: schema */river_job_state ELSE state END WHERE id = @id -RETURNING *; \ No newline at end of file +RETURNING *; diff --git a/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql.go b/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql.go index 0be29561..a361baac 100644 --- a/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql.go +++ b/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql.go @@ -591,18 +591,20 @@ const jobGetStuck = `-- name: JobGetStuck :many SELECT id, args, attempt, attempted_at, attempted_by, created_at, errors, finalized_at, kind, max_attempts, metadata, priority, queue, state, scheduled_at, tags, unique_key, unique_states FROM /* TEMPLATE: schema */river_job WHERE state = 'running' - AND attempted_at < $1::timestamptz + AND id > $1::bigint + AND attempted_at < $2::timestamptz ORDER BY id -LIMIT $2 +LIMIT $3 ` type JobGetStuckParams struct { + AfterID int64 StuckHorizon time.Time Max int32 } func (q *Queries) JobGetStuck(ctx context.Context, db DBTX, arg *JobGetStuckParams) ([]*RiverJob, error) { - rows, err := db.Query(ctx, jobGetStuck, arg.StuckHorizon, arg.Max) + rows, err := db.Query(ctx, jobGetStuck, arg.AfterID, arg.StuckHorizon, arg.Max) if err != nil { return nil, err } diff --git a/riverdriver/riverpgxv5/river_pgx_v5_driver.go b/riverdriver/riverpgxv5/river_pgx_v5_driver.go index 41b510eb..e34fedcb 100644 --- a/riverdriver/riverpgxv5/river_pgx_v5_driver.go +++ b/riverdriver/riverpgxv5/river_pgx_v5_driver.go @@ -347,6 +347,7 @@ func (e *Executor) JobGetByKindMany(ctx context.Context, params *riverdriver.Job func (e *Executor) JobGetStuck(ctx context.Context, params *riverdriver.JobGetStuckParams) ([]*rivertype.JobRow, error) { jobs, err := dbsqlc.New().JobGetStuck(schemaTemplateParam(ctx, params.Schema), e.dbtx, &dbsqlc.JobGetStuckParams{ + AfterID: params.AfterID, Max: int32(min(params.Max, math.MaxInt32)), //nolint:gosec StuckHorizon: params.StuckHorizon, }) diff --git a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql index b0600dc9..35628049 100644 --- a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql +++ b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql @@ -185,6 +185,7 @@ ORDER BY id; SELECT * FROM /* TEMPLATE: schema */river_job WHERE state = 'running' + AND id > @after_id AND attempted_at < cast(@stuck_horizon AS text) ORDER BY id LIMIT @max; diff --git a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go index 20b64838..1ffe44b5 100644 --- a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go +++ b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go @@ -561,18 +561,20 @@ const jobGetStuck = `-- name: JobGetStuck :many SELECT id, json(args), attempt, attempted_at, json(attempted_by), created_at, json(errors), finalized_at, kind, max_attempts, json(metadata), priority, queue, state, scheduled_at, json(tags), unique_key, unique_states FROM /* TEMPLATE: schema */river_job WHERE state = 'running' - AND attempted_at < cast(?1 AS text) + AND id > ?1 + AND attempted_at < cast(?2 AS text) ORDER BY id -LIMIT ?2 +LIMIT ?3 ` type JobGetStuckParams struct { + AfterID int64 StuckHorizon string Max int64 } func (q *Queries) JobGetStuck(ctx context.Context, db DBTX, arg *JobGetStuckParams) ([]*RiverJob, error) { - rows, err := db.QueryContext(ctx, jobGetStuck, arg.StuckHorizon, arg.Max) + rows, err := db.QueryContext(ctx, jobGetStuck, arg.AfterID, arg.StuckHorizon, arg.Max) if err != nil { return nil, err } diff --git a/riverdriver/riversqlite/river_sqlite_driver.go b/riverdriver/riversqlite/river_sqlite_driver.go index f4cc1971..1b833f2e 100644 --- a/riverdriver/riversqlite/river_sqlite_driver.go +++ b/riverdriver/riversqlite/river_sqlite_driver.go @@ -517,6 +517,7 @@ func (e *Executor) JobGetByKindMany(ctx context.Context, params *riverdriver.Job func (e *Executor) JobGetStuck(ctx context.Context, params *riverdriver.JobGetStuckParams) ([]*rivertype.JobRow, error) { jobs, err := dbsqlc.New().JobGetStuck(schemaTemplateParam(ctx, params.Schema), e.dbtx, &dbsqlc.JobGetStuckParams{ + AfterID: params.AfterID, Max: int64(params.Max), StuckHorizon: timeString(params.StuckHorizon), })