From 15f10b0bdf7f2d9f7f4f9bad16580286fd7b9192 Mon Sep 17 00:00:00 2001 From: Blake Gentry Date: Mon, 12 Oct 2026 18:56:11 -0700 Subject: [PATCH] add pilot hooks for job deletion `Client.JobDelete`, `Client.JobDeleteMany`, their `Tx` variants, and the job cleaner's `JobDeleteBefore` call the driver executor directly, so a pilot can't take part in deletes the way it can in cancels, retries, and completions. Add an optional `PilotJobDeleter` interface with `JobDelete`, `JobDeleteMany`, and `JobDeleteBefore`. Each hook receives the executor and the driver params and performs the delete itself, like `JobCancel` and `JobRetry`, so a pilot can substitute its own delete statement or run additional writes in the same transaction. The job cleaner now takes the client's pilot in its config. The interface is temporarily kept separate from `Pilot`, following the `PilotJobRescuer` pattern, so existing pilot implementations still compile; it's meant to be embedded into `Pilot`, and the fallbacks removed, once supported implementations have adopted it. The client and job cleaner use the hooks when the pilot implements them and otherwise fall back to the executor. `StandardPilot` implements them by calling the executor, so behavior is unchanged by default. Tests cover the client and cleaner routing deletes through the pilot, the executor fallback for pilots without the new interface, and the standard pilot's pass-through. --- CHANGELOG.md | 4 + client.go | 36 ++++-- client_pilot_test.go | 112 ++++++++++++++++++ internal/maintenance/job_cleaner.go | 25 +++- internal/maintenance/job_cleaner_test.go | 59 +++++++++ rivershared/riverpilot/pilot.go | 24 ++++ rivershared/riverpilot/standard_pilot.go | 12 ++ rivershared/riverpilot/standard_pilot_test.go | 72 +++++++++++ 8 files changed, 336 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2a8c85b59..b347dea0d 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] +### Changed + +- Job deletion through `Client.JobDelete`, `Client.JobDeleteMany`, their `Tx` variants, and the job cleaner now goes through optional pilot hooks, so a pilot can take part in deletes as it already can in cancels and retries. Behavior with the standard pilot is unchanged. [PR #1500](https://github.com/riverqueue/river/pull/1500). + ### Fixed - Fixed `AddWorkerSafely` leaving a worker's primary kind and earlier aliases registered when a later kind alias conflicts with an existing registration. Failed registrations now leave the worker registry unchanged. [PR #1440](https://github.com/riverqueue/river/pull/1440). diff --git a/client.go b/client.go index 72c798361..e06af4b0c 100644 --- a/client.go +++ b/client.go @@ -1003,6 +1003,7 @@ func NewClient[TTx any](driver riverdriver.Driver[TTx], config *Config) (*Client CancelledJobRetentionPeriod: config.CancelledJobRetentionPeriod, CompletedJobRetentionPeriod: config.CompletedJobRetentionPeriod, DiscardedJobRetentionPeriod: config.DiscardedJobRetentionPeriod, + Pilot: client.pilot, QueuesExcluded: client.pilot.JobCleanerQueuesExcluded(), Schema: config.Schema, Timeout: config.JobCleanerTimeout, @@ -1601,10 +1602,7 @@ func (c *Client[TTx]) jobCancel(ctx context.Context, exec riverdriver.Executor, // deleted row if it was deleted. Jobs in the running state are not deleted, // instead returning rivertype.ErrJobRunning. func (c *Client[TTx]) JobDelete(ctx context.Context, id int64) (*rivertype.JobRow, error) { - return c.driver.GetExecutor().JobDelete(ctx, &riverdriver.JobDeleteParams{ - ID: id, - Schema: c.config.Schema, - }) + return c.jobDelete(ctx, c.driver.GetExecutor(), id) } // JobDeleteTx deletes the job with the given ID from the database, returning the @@ -1614,10 +1612,23 @@ func (c *Client[TTx]) JobDelete(ctx context.Context, id int64) (*rivertype.JobRo // until the transaction commits, and if the transaction rolls back, so too is // the deleted job. func (c *Client[TTx]) JobDeleteTx(ctx context.Context, tx TTx, id int64) (*rivertype.JobRow, error) { - return c.driver.UnwrapExecutor(tx).JobDelete(ctx, &riverdriver.JobDeleteParams{ + return c.jobDelete(ctx, c.driver.UnwrapExecutor(tx), id) +} + +func (c *Client[TTx]) jobDelete(ctx context.Context, exec riverdriver.Executor, id int64) (*rivertype.JobRow, error) { + params := &riverdriver.JobDeleteParams{ ID: id, Schema: c.config.Schema, - }) + } + + if pilot, ok := c.pilot.(riverpilot.PilotJobDeleter); ok { + return pilot.JobDelete(ctx, exec, params) + } + + // Temporary compatibility fallback for Pilot implementations from before + // PilotJobDeleter. Once Pilot embeds PilotJobDeleter, replace the assertion + // above and this fallback with a direct call to c.pilot.JobDelete. + return exec.JobDelete(ctx, params) } // JobGet fetches a single job by its ID. Returns the up-to-date JobRow for the @@ -2484,7 +2495,18 @@ func (c *Client[TTx]) jobDeleteMany(ctx context.Context, exec riverdriver.Execut return nil, err } - jobs, err := exec.JobDeleteMany(ctx, (*riverdriver.JobDeleteManyParams)(listParams)) + driverParams := (*riverdriver.JobDeleteManyParams)(listParams) + + var jobs []*rivertype.JobRow + if pilot, ok := c.pilot.(riverpilot.PilotJobDeleter); ok { + jobs, err = pilot.JobDeleteMany(ctx, exec, driverParams) + } else { + // Temporary compatibility fallback for Pilot implementations from before + // PilotJobDeleter. Once Pilot embeds PilotJobDeleter, replace the + // assertion above and this fallback with a direct call to + // c.pilot.JobDeleteMany. + jobs, err = exec.JobDeleteMany(ctx, driverParams) + } if err != nil { return nil, err } diff --git a/client_pilot_test.go b/client_pilot_test.go index d52654054..6b7f1ed35 100644 --- a/client_pilot_test.go +++ b/client_pilot_test.go @@ -26,6 +26,8 @@ type pilotSpy struct { jobCancelCalls atomic.Int64 jobCleanerQueuesExcludedCalls atomic.Int64 + jobDeleteCalls atomic.Int64 + jobDeleteManyCalls atomic.Int64 jobInsertManyCalls atomic.Int64 jobRetryCalls atomic.Int64 pilotInitCalls atomic.Int64 @@ -69,6 +71,16 @@ func (p *pilotSpy) JobCleanerQueuesExcluded() []string { return p.StandardPilot.JobCleanerQueuesExcluded() } +func (p *pilotSpy) JobDelete(ctx context.Context, exec riverdriver.Executor, params *riverdriver.JobDeleteParams) (*rivertype.JobRow, error) { + p.jobDeleteCalls.Add(1) + return p.StandardPilot.JobDelete(ctx, exec, params) +} + +func (p *pilotSpy) JobDeleteMany(ctx context.Context, exec riverdriver.Executor, params *riverdriver.JobDeleteManyParams) ([]*rivertype.JobRow, error) { + p.jobDeleteManyCalls.Add(1) + return p.StandardPilot.JobDeleteMany(ctx, exec, params) +} + func (p *pilotSpy) JobGetAvailable(ctx context.Context, exec riverdriver.Executor, state riverpilot.ProducerState, params *riverdriver.JobGetAvailableParams) (*riverdriver.JobGetAvailableResult, error) { p.testSignals.JobGetAvailable.Signal(struct{}{}) return p.StandardPilot.JobGetAvailable(ctx, exec, state, params) @@ -130,6 +142,12 @@ func (p *pilotSpy) QueueMetadataChanged(ctx context.Context, exec riverdriver.Ex return p.StandardPilot.QueueMetadataChanged(ctx, exec, params) } +// pilotWithoutJobDeleter hides PilotJobDeleter to simulate a pilot built +// against a River version from before it existed. +type pilotWithoutJobDeleter struct { + riverpilot.Pilot +} + func Test_Client_PilotUsage(t *testing.T) { t.Parallel() @@ -224,6 +242,100 @@ func Test_Client_PilotUsage(t *testing.T) { }) }) + t.Run("JobDeleteManyTxUsesPilot", func(t *testing.T) { + t.Parallel() + + client, pilot := setup(t, nil) + + insertRes, err := client.Insert(ctx, noOpArgs{}, nil) + require.NoError(t, err) + + withClientTx(t, client, func(tx pgx.Tx) { + deleteRes, err := client.JobDeleteManyTx(ctx, tx, NewJobDeleteManyParams().IDs(insertRes.Job.ID)) + require.NoError(t, err) + require.Len(t, deleteRes.Jobs, 1) + require.Equal(t, int64(1), pilot.jobDeleteManyCalls.Load()) + }) + }) + + t.Run("JobDeleteManyUsesPilot", func(t *testing.T) { + t.Parallel() + + client, pilot := setup(t, nil) + + insertRes, err := client.Insert(ctx, noOpArgs{}, nil) + require.NoError(t, err) + + deleteRes, err := client.JobDeleteMany(ctx, NewJobDeleteManyParams().IDs(insertRes.Job.ID)) + require.NoError(t, err) + require.Len(t, deleteRes.Jobs, 1) + require.Equal(t, int64(1), pilot.jobDeleteManyCalls.Load()) + }) + + t.Run("JobDeleteTxUsesPilot", func(t *testing.T) { + t.Parallel() + + client, pilot := setup(t, nil) + + insertRes, err := client.Insert(ctx, noOpArgs{}, nil) + require.NoError(t, err) + + withClientTx(t, client, func(tx pgx.Tx) { + _, err = client.JobDeleteTx(ctx, tx, insertRes.Job.ID) + require.NoError(t, err) + require.Equal(t, int64(1), pilot.jobDeleteCalls.Load()) + }) + }) + + t.Run("JobDeleteUsesExecutorFallbackForPilotWithoutJobDeleter", func(t *testing.T) { + t.Parallel() + + var ( + dbPool = riversharedtest.DBPool(ctx, t) + driver = riverpgxv5.New(dbPool) + schema = riverdbtest.TestSchema(ctx, t, driver, nil) + pilot = &pilotWithoutJobDeleter{Pilot: &pilotSpy{}} + pluginDriver = newDriverWithPlugin(t, dbPool) + ) + pluginDriver.pilot = pilot + + _, implementsJobDeleter := riverpilot.Pilot(pilot).(riverpilot.PilotJobDeleter) + require.False(t, implementsJobDeleter) + + client, err := NewClient(pluginDriver, newTestConfig(t, schema)) + require.NoError(t, err) + + insertRes1, err := client.Insert(ctx, noOpArgs{}, nil) + require.NoError(t, err) + insertRes2, err := client.Insert(ctx, noOpArgs{}, nil) + require.NoError(t, err) + + _, err = client.JobDelete(ctx, insertRes1.Job.ID) + require.NoError(t, err) + + deleteRes, err := client.JobDeleteMany(ctx, NewJobDeleteManyParams().IDs(insertRes2.Job.ID)) + require.NoError(t, err) + require.Len(t, deleteRes.Jobs, 1) + + _, err = client.JobGet(ctx, insertRes1.Job.ID) + require.ErrorIs(t, err, ErrNotFound) + _, err = client.JobGet(ctx, insertRes2.Job.ID) + require.ErrorIs(t, err, ErrNotFound) + }) + + t.Run("JobDeleteUsesPilot", func(t *testing.T) { + t.Parallel() + + client, pilot := setup(t, nil) + + insertRes, err := client.Insert(ctx, noOpArgs{}, nil) + require.NoError(t, err) + + _, err = client.JobDelete(ctx, insertRes.Job.ID) + require.NoError(t, err) + require.Equal(t, int64(1), pilot.jobDeleteCalls.Load()) + }) + t.Run("JobInsertManyUsesPilot", func(t *testing.T) { t.Parallel() diff --git a/internal/maintenance/job_cleaner.go b/internal/maintenance/job_cleaner.go index e70fe3004..185dc53cf 100644 --- a/internal/maintenance/job_cleaner.go +++ b/internal/maintenance/job_cleaner.go @@ -11,6 +11,7 @@ import ( "github.com/riverqueue/river/riverdriver" "github.com/riverqueue/river/rivershared/baseservice" "github.com/riverqueue/river/rivershared/circuitbreaker" + "github.com/riverqueue/river/rivershared/riverpilot" "github.com/riverqueue/river/rivershared/riversharedmaintenance" "github.com/riverqueue/river/rivershared/startstop" "github.com/riverqueue/river/rivershared/testsignal" @@ -54,6 +55,9 @@ type JobCleanerConfig struct { // Interval is the amount of time to wait between runs of the cleaner. Interval time.Duration + // Pilot controls driver-level behavior that can be customized by plugins. + Pilot riverpilot.Pilot + // QueuesExcluded are queues that'll be excluded from cleaning. QueuesExcluded []string @@ -80,6 +84,9 @@ func (c *JobCleanerConfig) mustValidate() *JobCleanerConfig { if c.Interval <= 0 { panic("JobCleanerConfig.Interval must be above zero") } + if c.Pilot == nil { + panic("JobCleanerConfig.Pilot must be set") + } if c.Timeout <= 0 { panic("JobCleanerConfig.Timeout must be above zero") } @@ -110,6 +117,10 @@ type JobCleaner struct { func NewJobCleaner(archetype *baseservice.Archetype, config *JobCleanerConfig, exec riverdriver.Executor) *JobCleaner { batchSizes := config.WithDefaults() + pilot := config.Pilot + if pilot == nil { + pilot = &riverpilot.StandardPilot{} + } return baseservice.Init(archetype, &JobCleaner{ Config: (&JobCleanerConfig{ @@ -119,6 +130,7 @@ func NewJobCleaner(archetype *baseservice.Archetype, config *JobCleanerConfig, e DiscardedJobRetentionPeriod: cmp.Or(config.DiscardedJobRetentionPeriod, riversharedmaintenance.DiscardedJobRetentionPeriodDefault), QueuesExcluded: config.QueuesExcluded, Interval: cmp.Or(config.Interval, riversharedmaintenance.JobCleanerIntervalDefault), + Pilot: pilot, Schema: config.Schema, Timeout: cmp.Or(config.Timeout, riversharedmaintenance.JobCleanerTimeoutDefault), }).mustValidate(), @@ -176,6 +188,17 @@ func (s *JobCleaner) batchSize() int { return s.Config.Default } +func (s *JobCleaner) deleteBefore(ctx context.Context, params *riverdriver.JobDeleteBeforeParams) (int, error) { + if pilot, ok := s.Config.Pilot.(riverpilot.PilotJobDeleter); ok { + return pilot.JobDeleteBefore(ctx, s.exec, params) + } + + // Temporary compatibility fallback for Pilot implementations from before + // PilotJobDeleter. Once Pilot embeds PilotJobDeleter, replace the assertion + // above and this fallback with a direct call to s.Config.Pilot.JobDeleteBefore. + return s.exec.JobDeleteBefore(ctx, params) +} + type jobCleanerRunOnceResult struct { NumJobsDeleted int } @@ -193,7 +216,7 @@ func (s *JobCleaner) runOnce(ctx context.Context) (*jobCleanerRunOnceResult, err return 0, nil } - numDeleted, err := s.exec.JobDeleteBefore(ctx, &riverdriver.JobDeleteBeforeParams{ + numDeleted, err := s.deleteBefore(ctx, &riverdriver.JobDeleteBeforeParams{ CancelledDoDelete: s.Config.CancelledJobRetentionPeriod != -1, CancelledFinalizedAtHorizon: time.Now().Add(-s.Config.CancelledJobRetentionPeriod), CompletedDoDelete: s.Config.CompletedJobRetentionPeriod != -1, diff --git a/internal/maintenance/job_cleaner_test.go b/internal/maintenance/job_cleaner_test.go index fd4751134..338e2d713 100644 --- a/internal/maintenance/job_cleaner_test.go +++ b/internal/maintenance/job_cleaner_test.go @@ -2,6 +2,7 @@ package maintenance import ( "context" + "sync/atomic" "testing" "time" @@ -10,6 +11,7 @@ import ( "github.com/riverqueue/river/riverdbtest" "github.com/riverqueue/river/riverdriver" "github.com/riverqueue/river/riverdriver/riverpgxv5" + "github.com/riverqueue/river/rivershared/riverpilot" "github.com/riverqueue/river/rivershared/riversharedmaintenance" "github.com/riverqueue/river/rivershared/riversharedtest" "github.com/riverqueue/river/rivershared/startstoptest" @@ -17,6 +19,23 @@ import ( "github.com/riverqueue/river/rivertype" ) +type jobCleanerPilotSpy struct { + riverpilot.StandardPilot + + jobDeleteBeforeCalls atomic.Int64 + jobDeleteBeforeParams *riverdriver.JobDeleteBeforeParams +} + +func (p *jobCleanerPilotSpy) JobDeleteBefore(ctx context.Context, exec riverdriver.Executor, params *riverdriver.JobDeleteBeforeParams) (int, error) { + p.jobDeleteBeforeCalls.Add(1) + p.jobDeleteBeforeParams = params + return p.StandardPilot.JobDeleteBefore(ctx, exec, params) +} + +type jobCleanerPilotWithoutJobDeleter struct { + riverpilot.Pilot +} + func TestJobCleaner(t *testing.T) { t.Parallel() @@ -436,4 +455,44 @@ func TestJobCleaner(t *testing.T) { } } }) + + t.Run("UsesPilot", func(t *testing.T) { + t.Parallel() + + cleaner, bundle := setup(t) + pilot := &jobCleanerPilotSpy{} + cleaner.Config.Pilot = pilot + + job := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{State: new(rivertype.JobStateCancelled), FinalizedAt: new(bundle.cancelledDeleteHorizon.Add(-1 * time.Hour))}) + + res, err := cleaner.runOnce(ctx) + require.NoError(t, err) + require.Equal(t, 1, res.NumJobsDeleted) + + require.Equal(t, int64(1), pilot.jobDeleteBeforeCalls.Load()) + require.Equal(t, cleaner.Config.Default, pilot.jobDeleteBeforeParams.Max) + require.True(t, pilot.jobDeleteBeforeParams.CancelledDoDelete) + + _, err = bundle.exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: job.ID, Schema: cleaner.Config.Schema}) + require.ErrorIs(t, err, rivertype.ErrNotFound) + }) + + t.Run("UsesExecutorFallbackForPilotWithoutJobDeleter", func(t *testing.T) { + t.Parallel() + + cleaner, bundle := setup(t) + cleaner.Config.Pilot = &jobCleanerPilotWithoutJobDeleter{Pilot: cleaner.Config.Pilot} + + _, implementsJobDeleter := cleaner.Config.Pilot.(riverpilot.PilotJobDeleter) + require.False(t, implementsJobDeleter) + + job := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{State: new(rivertype.JobStateCancelled), FinalizedAt: new(bundle.cancelledDeleteHorizon.Add(-1 * time.Hour))}) + + res, err := cleaner.runOnce(ctx) + require.NoError(t, err) + require.Equal(t, 1, res.NumJobsDeleted) + + _, err = bundle.exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: job.ID, Schema: cleaner.Config.Schema}) + require.ErrorIs(t, err, rivertype.ErrNotFound) + }) } diff --git a/rivershared/riverpilot/pilot.go b/rivershared/riverpilot/pilot.go index 25108d29e..a35f62656 100644 --- a/rivershared/riverpilot/pilot.go +++ b/rivershared/riverpilot/pilot.go @@ -101,6 +101,30 @@ func (p *PilotInitParams) Validate() *PilotInitParams { return p } +// PilotJobDeleter contains Pilot functionality related to deleting jobs. Each +// function performs the deletion itself so that implementations can make +// related writes atomically with it. Pilots that don't implement it fall back +// to the standard executor-backed behavior. +// +// The executor may or may not be a transaction. An implementation that issues +// more than one statement should use dbutil.WithTxV to run them in a +// transaction, reusing the caller's if there is one. +// +// This is temporarily separate from Pilot so implementations built against +// older River versions remain compatible. It's not meant to stay optional: +// once all supported River Pro releases implement PilotJobDeleter, embed this +// interface into Pilot, replace capability assertions in the client and +// JobCleaner with direct Pilot calls, and remove their executor fallbacks. +type PilotJobDeleter interface { + JobDelete(ctx context.Context, exec riverdriver.Executor, params *riverdriver.JobDeleteParams) (*rivertype.JobRow, error) + + // JobDeleteBefore deletes finalized jobs on behalf of the job cleaner, + // returning the number of jobs deleted. + JobDeleteBefore(ctx context.Context, exec riverdriver.Executor, params *riverdriver.JobDeleteBeforeParams) (int, error) + + JobDeleteMany(ctx context.Context, exec riverdriver.Executor, params *riverdriver.JobDeleteManyParams) ([]*rivertype.JobRow, error) +} + // PilotJobRescuer contains optional Pilot functionality related to rescuing // stuck jobs. Pilots that don't implement it fall back to the standard // executor-backed behavior. diff --git a/rivershared/riverpilot/standard_pilot.go b/rivershared/riverpilot/standard_pilot.go index 2e9b9aec5..643dc0588 100644 --- a/rivershared/riverpilot/standard_pilot.go +++ b/rivershared/riverpilot/standard_pilot.go @@ -17,6 +17,18 @@ type StandardPilot struct { func (p *StandardPilot) JobCleanerQueuesExcluded() []string { return nil } +func (p *StandardPilot) JobDelete(ctx context.Context, exec riverdriver.Executor, params *riverdriver.JobDeleteParams) (*rivertype.JobRow, error) { + return exec.JobDelete(ctx, params) +} + +func (p *StandardPilot) JobDeleteBefore(ctx context.Context, exec riverdriver.Executor, params *riverdriver.JobDeleteBeforeParams) (int, error) { + return exec.JobDeleteBefore(ctx, params) +} + +func (p *StandardPilot) JobDeleteMany(ctx context.Context, exec riverdriver.Executor, params *riverdriver.JobDeleteManyParams) ([]*rivertype.JobRow, error) { + return exec.JobDeleteMany(ctx, params) +} + func (p *StandardPilot) JobGetAvailable(ctx context.Context, exec riverdriver.Executor, state ProducerState, params *riverdriver.JobGetAvailableParams) (*riverdriver.JobGetAvailableResult, error) { if params.MaxToLock <= 0 { return &riverdriver.JobGetAvailableResult{}, nil diff --git a/rivershared/riverpilot/standard_pilot_test.go b/rivershared/riverpilot/standard_pilot_test.go index 09e57796f..df6be669d 100644 --- a/rivershared/riverpilot/standard_pilot_test.go +++ b/rivershared/riverpilot/standard_pilot_test.go @@ -8,14 +8,30 @@ import ( "github.com/stretchr/testify/require" "github.com/riverqueue/river/riverdriver" + "github.com/riverqueue/river/rivertype" ) type standardPilotExecutorMock struct { riverdriver.Executor + jobDeleteBeforeFunc func(ctx context.Context, params *riverdriver.JobDeleteBeforeParams) (int, error) + jobDeleteFunc func(ctx context.Context, params *riverdriver.JobDeleteParams) (*rivertype.JobRow, error) + jobDeleteManyFunc func(ctx context.Context, params *riverdriver.JobDeleteManyParams) ([]*rivertype.JobRow, error) jobGetAvailableFunc func(ctx context.Context, params *riverdriver.JobGetAvailableParams) (*riverdriver.JobGetAvailableResult, error) } +func (m *standardPilotExecutorMock) JobDelete(ctx context.Context, params *riverdriver.JobDeleteParams) (*rivertype.JobRow, error) { + return m.jobDeleteFunc(ctx, params) +} + +func (m *standardPilotExecutorMock) JobDeleteBefore(ctx context.Context, params *riverdriver.JobDeleteBeforeParams) (int, error) { + return m.jobDeleteBeforeFunc(ctx, params) +} + +func (m *standardPilotExecutorMock) JobDeleteMany(ctx context.Context, params *riverdriver.JobDeleteManyParams) ([]*rivertype.JobRow, error) { + return m.jobDeleteManyFunc(ctx, params) +} + func (m *standardPilotExecutorMock) JobGetAvailable(ctx context.Context, params *riverdriver.JobGetAvailableParams) (*riverdriver.JobGetAvailableResult, error) { return m.jobGetAvailableFunc(ctx, params) } @@ -48,6 +64,62 @@ func (m *standardPilotExecutorTxMock) Rollback(context.Context) error { return nil } +func TestStandardPilot_JobDelete(t *testing.T) { + t.Parallel() + + ctx := context.Background() + + t.Run("JobDelete", func(t *testing.T) { + t.Parallel() + + params := &riverdriver.JobDeleteParams{ID: 123} + job := &rivertype.JobRow{ID: 123} + exec := &standardPilotExecutorMock{ + jobDeleteFunc: func(ctx context.Context, innerParams *riverdriver.JobDeleteParams) (*rivertype.JobRow, error) { + require.Same(t, params, innerParams) + return job, nil + }, + } + + deletedJob, err := (&StandardPilot{}).JobDelete(ctx, exec, params) + require.NoError(t, err) + require.Same(t, job, deletedJob) + }) + + t.Run("JobDeleteBefore", func(t *testing.T) { + t.Parallel() + + params := &riverdriver.JobDeleteBeforeParams{Max: 10} + exec := &standardPilotExecutorMock{ + jobDeleteBeforeFunc: func(ctx context.Context, innerParams *riverdriver.JobDeleteBeforeParams) (int, error) { + require.Same(t, params, innerParams) + return 7, nil + }, + } + + numDeleted, err := (&StandardPilot{}).JobDeleteBefore(ctx, exec, params) + require.NoError(t, err) + require.Equal(t, 7, numDeleted) + }) + + t.Run("JobDeleteMany", func(t *testing.T) { + t.Parallel() + + params := &riverdriver.JobDeleteManyParams{Max: 10} + jobs := []*rivertype.JobRow{{ID: 123}} + exec := &standardPilotExecutorMock{ + jobDeleteManyFunc: func(ctx context.Context, innerParams *riverdriver.JobDeleteManyParams) ([]*rivertype.JobRow, error) { + require.Same(t, params, innerParams) + return jobs, nil + }, + } + + deletedJobs, err := (&StandardPilot{}).JobDeleteMany(ctx, exec, params) + require.NoError(t, err) + require.Equal(t, jobs, deletedJobs) + }) +} + func TestStandardPilot_JobGetAvailable(t *testing.T) { t.Parallel()