From 163c5f7e19b2e7561bda6a6b74ae326c9fe12b79 Mon Sep 17 00:00:00 2001 From: Brandur Date: Wed, 22 Jul 2026 14:35:26 -0500 Subject: [PATCH] Move new pilot functions into a compatibility layer so we don't need immediate pro release In #1317 we added some new functions to the pilot to make the active job rescue system work. I was thinking this morning that the way this was done is a little suboptimal for our pro customers. Expanding the pilot interface makes the pro pilot immediately incompatible, so once we released this we'd need to make sure to do an _immediate_ fast follow with a pro release, otherwise there'd be no compatible version. And even with that fast follow, if a customer tried to upgrade just River by itself, it'd break their pro version without an upgrade. Here, move the new functions into a special `PilotJobRescuer` interface that's used where available by the rescuer, but allows a fall back in case it's not. This will let us release the pro side at our leisure and there will be multiple compatible River releases by then. We'd want to unwind this eventually, so I left a couple comments in place with instructions on how to do that. --- internal/maintenance/job_rescuer.go | 26 +++++++++++++++++++++--- internal/maintenance/job_rescuer_test.go | 23 +++++++++++++++++++++ rivershared/riverpilot/pilot.go | 20 ++++++++++++++---- 3 files changed, 62 insertions(+), 7 deletions(-) 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.