Move new pilot functions into a compatibility layer so we don't need immediate pro release

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

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()
 

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.