From e555d3555d3ca9d994b2560526dfab44af8effa9 Mon Sep 17 00:00:00 2001 From: Brandur Date: Sat, 25 Jul 2026 12:16:44 -0500 Subject: [PATCH] Fix bug where jobs would be improperly rescued with client-level timeout of -1 I was doing some research for an active rescue job post, and Codex found a bug in the job rescue code around cases where the client is configured with a global `Config.JobTimeout` of -1 (no timeout). Consider this case with a -1 global timeout and a worker using default: config.JobTimeout = -1 // disable the client-level timeout type MyWorker struct { river.WorkerDefaults[MyArgs] // Timeout returns 0 } In `JobExecutor`, we were handling this correctly, which is for the client timeout of -1 to supersede the worker timeout default: jobTimeout := cmp.Or(workerTimeout, clientJobTimeout) However, in `JobRescuer` we weren't performing this check, and we'd end up getting a timeout of 0: timeout := workUnit.Timeout() // returns 0 This made jobs eligible for rescue after `RescueStuckJobsAfter` has elapsed, which was wrong. Here, do something similar in `JobRescuer` as we had in `JobRescuer` by accounting for client-level timeout. --- CHANGELOG.md | 4 ++++ client.go | 1 + client_test.go | 22 ++++++++++++++++++++++ internal/maintenance/job_rescuer.go | 7 ++++++- internal/maintenance/job_rescuer_test.go | 16 ++++++++++++++++ 5 files changed, 49 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 713e2d73..ca81e606 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed + +- Fixed `JobRescuer` incorrectly rescuing jobs whose worker timeout of zero inherits a client-level `Config.JobTimeout` of `-1` (no timeout). [PR #1331](https://github.com/riverqueue/river/pull/1331). + ## [0.41.0] - 2026-07-23 ### Added diff --git a/client.go b/client.go index 422eb4d7..2177ce9c 100644 --- a/client.go +++ b/client.go @@ -959,6 +959,7 @@ func NewClient[TTx any](driver riverdriver.Driver[TTx], config *Config) (*Client { jobRescuer := maintenance.NewRescuer(archetype, &maintenance.JobRescuerConfig{ + ClientJobTimeout: config.JobTimeout, ClientRetryPolicy: config.RetryPolicy, Pilot: client.pilot, RescueAfter: config.RescueStuckJobsAfter, diff --git a/client_test.go b/client_test.go index 0274f5ef..83deea9e 100644 --- a/client_test.go +++ b/client_test.go @@ -5859,6 +5859,28 @@ func Test_Client_Maintenance(t *testing.T) { requireJobHasState(jobNotYetStuck3.ID, jobNotYetStuck3.State) }) + t.Run("JobRescuerDoesNotRescueWorkerInheritingDisabledClientTimeout", func(t *testing.T) { + t.Parallel() + + config := newTestConfig(t, "") + config.JobTimeout = -1 + config.RescueStuckJobsAfter = 5 * time.Minute + + client, bundle := setup(t, config) + + job := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{Kind: ptrutil.Ptr("noOp"), Schema: bundle.schema, State: ptrutil.Ptr(rivertype.JobStateRunning), AttemptedAt: ptrutil.Ptr(time.Now().Add(-time.Hour))}) + + startAndWaitForQueueMaintainer(ctx, t, client) + + rescuer := maintenance.GetService[*maintenance.JobRescuer](client.queueMaintainer) + rescuer.TestSignals.FetchedBatch.WaitOrTimeout() + rescuer.TestSignals.UpdatedBatch.WaitOrTimeout() + + jobAfter, err := bundle.exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: job.ID, Schema: bundle.schema}) + require.NoError(t, err) + require.Equal(t, rivertype.JobStateRunning, jobAfter.State) + }) + t.Run("JobScheduler", func(t *testing.T) { t.Parallel() diff --git a/internal/maintenance/job_rescuer.go b/internal/maintenance/job_rescuer.go index 586b7893..f35ae51a 100644 --- a/internal/maintenance/job_rescuer.go +++ b/internal/maintenance/job_rescuer.go @@ -45,6 +45,10 @@ func (ts *JobRescuerTestSignals) Init(tb testutil.TestingTB) { type JobRescuerConfig struct { riversharedmaintenance.BatchSizes + // ClientJobTimeout is the default job timeout used when a worker returns a + // timeout of zero. + ClientJobTimeout time.Duration + // ClientRetryPolicy is the default retry policy to use for workers that don't // override NextRetry. ClientRetryPolicy jobexecutor.ClientRetryPolicy @@ -119,6 +123,7 @@ func NewRescuer(archetype *baseservice.Archetype, config *JobRescuerConfig, exec return baseservice.Init(archetype, &JobRescuer{ Config: (&JobRescuerConfig{ BatchSizes: batchSizes, + ClientJobTimeout: config.ClientJobTimeout, ClientRetryPolicy: config.ClientRetryPolicy, Interval: cmp.Or(config.Interval, JobRescuerIntervalDefault), Pilot: pilot, @@ -349,7 +354,7 @@ func (s *JobRescuer) makeRetryDecision(ctx context.Context, job *rivertype.JobRo return jobRetryDecisionDiscard, time.Time{} } - timeout := workUnit.Timeout() + timeout := cmp.Or(workUnit.Timeout(), s.Config.ClientJobTimeout) if timeout < 0 || timeout > 0 && now.Sub(*job.AttemptedAt) < timeout { return jobRetryDecisionIgnore, time.Time{} } diff --git a/internal/maintenance/job_rescuer_test.go b/internal/maintenance/job_rescuer_test.go index acd09b9b..66d324d0 100644 --- a/internal/maintenance/job_rescuer_test.go +++ b/internal/maintenance/job_rescuer_test.go @@ -156,6 +156,22 @@ func TestJobRescuer(t *testing.T) { return rescuer, bundle } + t.Run("DoesNotRescueWorkerInheritingDisabledClientTimeout", func(t *testing.T) { + t.Parallel() + + rescuer, bundle := setup(t) + rescuer.Config.ClientJobTimeout = -1 + + job := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{Kind: ptrutil.Ptr(rescuerJobKind), State: ptrutil.Ptr(rivertype.JobStateRunning), AttemptedAt: ptrutil.Ptr(bundle.rescueHorizon.Add(-24 * time.Hour)), MaxAttempts: ptrutil.Ptr(5)}) + + _, err := rescuer.runOnce(ctx) + require.NoError(t, err) + + 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) + }) + t.Run("Defaults", func(t *testing.T) { t.Parallel()