diff --git a/internal/maintenance/job_rescuer.go b/internal/maintenance/job_rescuer.go index 81a89a84..92df3fee 100644 --- a/internal/maintenance/job_rescuer.go +++ b/internal/maintenance/job_rescuer.go @@ -273,7 +273,7 @@ func (s *JobRescuer) runOnce(ctx context.Context) (*rescuerRunOnceResult, error) } if len(rescueManyParams.ID) > 0 { - _, err = s.Config.Pilot.JobRescueMany(ctx, s.exec, &rescueManyParams) + _, err = s.rescueMany(ctx, &rescueManyParams) if err != nil { return nil, fmt.Errorf("error rescuing stuck jobs: %w", err) } @@ -297,12 +297,21 @@ func (s *JobRescuer) getStuckJobs(ctx context.Context, afterID int64, batchSize ctx, cancelFunc := context.WithTimeout(ctx, riversharedmaintenance.TimeoutDefault) defer cancelFunc() - return s.Config.Pilot.JobGetStuck(ctx, s.exec, &riverdriver.JobGetStuckParams{ + params := &riverdriver.JobGetStuckParams{ AfterID: afterID, Max: batchSize, Schema: s.Config.Schema, StuckHorizon: stuckHorizon, - }) + } + + if pilot, ok := s.Config.Pilot.(riverpilot.PilotJobRescuer); ok { + return pilot.JobGetStuck(ctx, s.exec, params) + } + + // Compatibility fallback for Pilot implementations from before + // PilotJobRescuer. Once Pilot embeds PilotJobRescuer, replace the assertion + // above and this fallback with a direct call to s.Config.Pilot.JobGetStuck. + return s.exec.JobGetStuck(ctx, params) } // jobRetryDecision is a signal from makeRetryDecision as to what to do with a @@ -347,3 +356,14 @@ func (s *JobRescuer) makeRetryDecision(ctx context.Context, job *rivertype.JobRo return jobRetryDecisionDiscard, time.Time{} } + +func (s *JobRescuer) rescueMany(ctx context.Context, params *riverdriver.JobRescueManyParams) (*struct{}, error) { + if pilot, ok := s.Config.Pilot.(riverpilot.PilotJobRescuer); ok { + return pilot.JobRescueMany(ctx, s.exec, params) + } + + // Compatibility fallback for Pilot implementations from before + // PilotJobRescuer. Once Pilot embeds PilotJobRescuer, replace the assertion + // above and this fallback with a direct call to s.Config.Pilot.JobRescueMany. + return s.exec.JobRescueMany(ctx, params) +} diff --git a/internal/maintenance/job_rescuer_test.go b/internal/maintenance/job_rescuer_test.go index a5024a88..7c37371f 100644 --- a/internal/maintenance/job_rescuer_test.go +++ b/internal/maintenance/job_rescuer_test.go @@ -80,6 +80,10 @@ func (p *jobRescuerPilotSpy) JobRescueMany(ctx context.Context, exec riverdriver return p.StandardPilot.JobRescueMany(ctx, exec, params) } +type jobRescuerPilotWithoutJobRescuer struct { + riverpilot.Pilot +} + func TestJobRescuer(t *testing.T) { t.Parallel() @@ -381,6 +385,25 @@ func TestJobRescuer(t *testing.T) { require.Equal(t, pilot.jobGetStuckParams.StuckHorizon, pilot.jobRescueManyParams.StuckHorizon) }) + t.Run("UsesExecutorFallbackForPilotWithoutJobRescuer", func(t *testing.T) { + t.Parallel() + + rescuer, bundle := setup(t) + rescuer.Config.Pilot = &jobRescuerPilotWithoutJobRescuer{Pilot: rescuer.Config.Pilot} + + _, implementsJobRescuer := rescuer.Config.Pilot.(riverpilot.PilotJobRescuer) + require.False(t, implementsJobRescuer) + + job := 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) + + jobAfter, err := bundle.exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: job.ID, Schema: rescuer.Config.Schema}) + require.NoError(t, err) + require.Equal(t, rivertype.JobStateRetryable, jobAfter.State) + }) + t.Run("CanRunMultipleTimes", func(t *testing.T) { t.Parallel() diff --git a/rivershared/riverpilot/pilot.go b/rivershared/riverpilot/pilot.go index c345b460..69d48d3b 100644 --- a/rivershared/riverpilot/pilot.go +++ b/rivershared/riverpilot/pilot.go @@ -35,16 +35,12 @@ type Pilot interface { params *riverdriver.JobGetAvailableParams, ) ([]*rivertype.JobRow, error) - JobGetStuck(ctx context.Context, exec riverdriver.Executor, params *riverdriver.JobGetStuckParams) ([]*rivertype.JobRow, error) - JobInsertMany( ctx context.Context, exec riverdriver.Executor, params *riverdriver.JobInsertFastManyParams, ) ([]*riverdriver.JobInsertFastResult, error) - JobRescueMany(ctx context.Context, exec riverdriver.Executor, params *riverdriver.JobRescueManyParams) (*struct{}, error) - JobRetry(ctx context.Context, exec riverdriver.Executor, params *riverdriver.JobRetryParams) (*rivertype.JobRow, error) JobSetStateIfRunningMany(ctx context.Context, exec riverdriver.Executor, params *riverdriver.JobSetStateIfRunningManyParams) ([]*rivertype.JobRow, error) @@ -100,6 +96,22 @@ func (p *PilotInitParams) Validate() *PilotInitParams { return p } +// PilotJobRescuer contains optional Pilot functionality related to rescuing +// stuck jobs. Pilots that don't implement it fall back to the standard +// executor-backed behavior. +// +// This is temporarily separate from Pilot so implementations built against +// older River versions remain compatible. It can be embedded into Pilot after +// downstream implementations have had a release cycle to adopt it. +// +// Once all supported River Pro releases implement PilotJobRescuer, embed this +// interface into Pilot, replace JobRescuer's capability assertions with direct +// Pilot calls, and remove its executor fallbacks. +type PilotJobRescuer interface { + JobGetStuck(ctx context.Context, exec riverdriver.Executor, params *riverdriver.JobGetStuckParams) ([]*rivertype.JobRow, error) + JobRescueMany(ctx context.Context, exec riverdriver.Executor, params *riverdriver.JobRescueManyParams) (*struct{}, error) +} + // PilotPeriodicJob contains pilot functions related to periodic jobs. This is // extracted as its own interface so there's less surface area to mock in places // like the periodic job enqueuer where that's needed.