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), })