add JobRetry to Pilot interface

Allow Pro pilot to further customize this behavior.

Author: bgentry

client.go

@@ -1412,11 +1412,7 @@ func (c *Client[TTx]) JobGetTx(ctx context.Context, tx TTx, id int64) (*rivertyp
 // MaxAttempts is also incremented by one if the job has already exhausted its
 // max attempts.
 func (c *Client[TTx]) JobRetry(ctx context.Context, id int64) (*rivertype.JobRow, error) {
-	return c.driver.GetExecutor().JobRetry(ctx, &riverdriver.JobRetryParams{
-		ID:     id,
-		Now:    c.baseService.Time.NowUTCOrNil(),
-		Schema: c.config.Schema,
-	})
+	return c.jobRetry(ctx, c.driver.GetExecutor(), id)
 }
 
 // JobRetryTx updates the job with the given ID to make it immediately available
@@ -1433,7 +1429,11 @@ func (c *Client[TTx]) JobRetry(ctx context.Context, id int64) (*rivertype.JobRow
 // MaxAttempts is also incremented by one if the job has already exhausted its
 // max attempts.
 func (c *Client[TTx]) JobRetryTx(ctx context.Context, tx TTx, id int64) (*rivertype.JobRow, error) {
-	return c.driver.UnwrapExecutor(tx).JobRetry(ctx, &riverdriver.JobRetryParams{
+	return c.jobRetry(ctx, c.driver.UnwrapExecutor(tx), id)
+}
+
+func (c *Client[TTx]) jobRetry(ctx context.Context, exec riverdriver.Executor, id int64) (*rivertype.JobRow, error) {
+	return c.pilot.JobRetry(ctx, exec, &riverdriver.JobRetryParams{
 		ID:     id,
 		Now:    c.baseService.Time.NowUTCOrNil(),
 		Schema: c.config.Schema,

rivershared/riverpilot/pilot.go

@@ -32,6 +32,8 @@ type Pilot interface {
 		params *riverdriver.JobInsertFastManyParams,
 	) ([]*riverdriver.JobInsertFastResult, 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)
 
 	PilotInit(archetype *baseservice.Archetype)

rivershared/riverpilot/standard.go

@@ -28,6 +28,10 @@ func (p *StandardPilot) JobInsertMany(
 	return exec.JobInsertFastMany(ctx, params)
 }
 
+func (p *StandardPilot) JobRetry(ctx context.Context, exec riverdriver.Executor, params *riverdriver.JobRetryParams) (*rivertype.JobRow, error) {
+	return exec.JobRetry(ctx, params)
+}
+
 func (p *StandardPilot) JobSetStateIfRunningMany(ctx context.Context, exec riverdriver.Executor, params *riverdriver.JobSetStateIfRunningManyParams) ([]*rivertype.JobRow, error) {
 	return exec.JobSetStateIfRunningMany(ctx, params)
 }