add JobDelete / JobDeleteTx APIs to Client

CHANGELOG.md

@@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
 
 - The default max attempts of 25 can now be customized on a per-client basis using `Config.MaxAttempts`. This is in addition to the ability to customize at the job type level with `JobArgs`, or on a per-job basis using `InsertOpts`. [PR #383](https://github.com/riverqueue/river/pull/383).
+- Add `JobDelete` / `JobDeleteTx` APIs on `Client` to allow permanently deleting any job that's not currently running. [PR #390](https://github.com/riverqueue/river/pull/390).
 
 ### Fixed
 

client.go

@@ -1047,6 +1047,23 @@ func (c *Client[TTx]) jobCancel(ctx context.Context, exec riverdriver.Executor,
 	})
 }
 
+// JobDelete deletes the job with the given ID from the database, returning the
+// 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, id)
+}
+
+// JobDelete deletes the job with the given ID from the database, returning the
+// deleted row if it was deleted. Jobs in the running state are not deleted,
+// instead returning rivertype.ErrJobRunning. This variant lets a caller retry a
+// job atomically alongside other database changes. A deleted job isn't deleted
+// 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, id)
+}
+
 // JobGet fetches a single job by its ID. Returns the up-to-date JobRow for the
 // specified jobID if it exists. Returns ErrNotFound if the job doesn't exist.
 func (c *Client[TTx]) JobGet(ctx context.Context, id int64) (*rivertype.JobRow, error) {

client_test.go

@@ -1010,12 +1010,122 @@ func Test_Client_ClientFromContext(t *testing.T) {
 	require.Equal(t, client, clientResult)
 }
 
+func Test_Client_JobDelete(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+
+	type testBundle struct {
+		dbPool *pgxpool.Pool
+	}
+
+	setup := func(t *testing.T) (*Client[pgx.Tx], *testBundle) {
+		t.Helper()
+
+		dbPool := riverinternaltest.TestDB(ctx, t)
+		config := newTestConfig(t, nil)
+		client := newTestClient(t, dbPool, config)
+
+		return client, &testBundle{dbPool: dbPool}
+	}
+
+	t.Run("DeletesANonRunningJob", func(t *testing.T) {
+		t.Parallel()
+
+		client, _ := setup(t)
+
+		insertRes, err := client.Insert(ctx, noOpArgs{}, &InsertOpts{ScheduledAt: time.Now().Add(time.Hour)})
+		require.NoError(t, err)
+		require.Equal(t, rivertype.JobStateScheduled, insertRes.Job.State)
+
+		jobAfter, err := client.JobDelete(ctx, insertRes.Job.ID)
+		require.NoError(t, err)
+		require.NotNil(t, jobAfter)
+		require.Equal(t, rivertype.JobStateScheduled, jobAfter.State)
+
+		_, err = client.JobGet(ctx, insertRes.Job.ID)
+		require.ErrorIs(t, err, ErrNotFound)
+	})
+
+	t.Run("DoesNotDeleteARunningJob", func(t *testing.T) {
+		t.Parallel()
+
+		client, _ := setup(t)
+
+		doneCh := make(chan struct{})
+		startedCh := make(chan int64)
+
+		AddWorker(client.config.Workers, WorkFunc(func(ctx context.Context, job *Job[callbackArgs]) error {
+			close(startedCh)
+			<-doneCh
+			return nil
+		}))
+
+		require.NoError(t, client.Start(ctx))
+		t.Cleanup(func() { require.NoError(t, client.Stop(ctx)) })
+		t.Cleanup(func() { close(doneCh) }) // must close before stopping client
+
+		insertRes, err := client.Insert(ctx, callbackArgs{}, nil)
+		require.NoError(t, err)
+
+		// Wait for the job to start:
+		riverinternaltest.WaitOrTimeout(t, startedCh)
+
+		jobAfter, err := client.JobDelete(ctx, insertRes.Job.ID)
+		require.ErrorIs(t, err, rivertype.ErrJobRunning)
+		require.Nil(t, jobAfter)
+
+		jobFromGet, err := client.JobGet(ctx, insertRes.Job.ID)
+		require.NoError(t, err)
+		require.Equal(t, rivertype.JobStateRunning, jobFromGet.State)
+	})
+
+	t.Run("TxVariantAlsoDeletesANonRunningJob", func(t *testing.T) {
+		t.Parallel()
+
+		client, bundle := setup(t)
+
+		insertRes, err := client.Insert(ctx, noOpArgs{}, &InsertOpts{ScheduledAt: time.Now().Add(time.Hour)})
+		require.NoError(t, err)
+		require.Equal(t, rivertype.JobStateScheduled, insertRes.Job.State)
+
+		var jobAfter *rivertype.JobRow
+
+		err = pgx.BeginFunc(ctx, bundle.dbPool, func(tx pgx.Tx) error {
+			var err error
+			jobAfter, err = client.JobDeleteTx(ctx, tx, insertRes.Job.ID)
+			return err
+		})
+		require.NoError(t, err)
+		require.NotNil(t, jobAfter)
+		require.Equal(t, insertRes.Job.ID, jobAfter.ID)
+		require.Equal(t, rivertype.JobStateScheduled, jobAfter.State)
+
+		jobFromGet, err := client.JobGet(ctx, insertRes.Job.ID)
+		require.ErrorIs(t, ErrNotFound, err)
+		require.Nil(t, jobFromGet)
+	})
+
+	t.Run("ReturnsErrNotFoundIfJobDoesNotExist", func(t *testing.T) {
+		t.Parallel()
+
+		client, _ := setup(t)
+
+		jobAfter, err := client.JobDelete(ctx, 0)
+		require.Error(t, err)
+		require.ErrorIs(t, err, ErrNotFound)
+		require.Nil(t, jobAfter)
+	})
+}
+
 func Test_Client_Insert(t *testing.T) {
 	t.Parallel()
 
 	ctx := context.Background()
 
-	type testBundle struct{}
+	type testBundle struct {
+		dbPool *pgxpool.Pool
+	}
 
 	setup := func(t *testing.T) (*Client[pgx.Tx], *testBundle) {
 		t.Helper()
@@ -1024,7 +1134,7 @@ func Test_Client_Insert(t *testing.T) {
 		config := newTestConfig(t, nil)
 		client := newTestClient(t, dbPool, config)
 
-		return client, &testBundle{}
+		return client, &testBundle{dbPool: dbPool}
 	}
 
 	t.Run("Succeeds", func(t *testing.T) {
@@ -1084,7 +1194,13 @@ func Test_Client_Insert(t *testing.T) {
 		t.Parallel()
 
 		ctx := context.Background()
-		client, _ := setup(t)
+
+		_, bundle := setup(t)
+
+		config := newTestConfig(t, nil)
+		config.FetchCooldown = 5 * time.Second
+		config.FetchPollInterval = 5 * time.Second
+		client := newTestClient(t, bundle.dbPool, config)
 		statusUpdateCh := client.monitor.RegisterUpdates()
 
 		startClient(ctx, t, client)

internal/riverinternaltest/riverdrivertest/riverdrivertest.go

@@ -191,6 +191,84 @@ func ExerciseExecutorFull[TTx any](ctx context.Context, t *testing.T, driver riv
 		require.Equal(t, 2, numJobs)
 	})
 
+	t.Run("JobDelete", func(t *testing.T) {
+		t.Parallel()
+
+		t.Run("DoesNotDeleteARunningJob", func(t *testing.T) {
+			t.Parallel()
+
+			exec, _ := setupExecutor(ctx, t, driver, beginTx)
+
+			job := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{
+				State: ptrutil.Ptr(rivertype.JobStateRunning),
+			})
+
+			jobAfter, err := exec.JobDelete(ctx, job.ID)
+			require.ErrorIs(t, err, rivertype.ErrJobRunning)
+			require.Nil(t, jobAfter)
+
+			jobUpdated, err := exec.JobGetByID(ctx, job.ID)
+			require.NoError(t, err)
+			require.Equal(t, rivertype.JobStateRunning, jobUpdated.State)
+		})
+
+		for _, state := range []rivertype.JobState{
+			rivertype.JobStateAvailable,
+			rivertype.JobStateCancelled,
+			rivertype.JobStateCompleted,
+			rivertype.JobStateDiscarded,
+			rivertype.JobStatePending,
+			rivertype.JobStateRetryable,
+			rivertype.JobStateScheduled,
+		} {
+			state := state
+
+			t.Run(fmt.Sprintf("DeletesA_%s_Job", state), func(t *testing.T) {
+				t.Parallel()
+
+				exec, _ := setupExecutor(ctx, t, driver, beginTx)
+
+				now := time.Now().UTC()
+
+				setFinalized := slices.Contains([]rivertype.JobState{
+					rivertype.JobStateCancelled,
+					rivertype.JobStateCompleted,
+					rivertype.JobStateDiscarded,
+				}, state)
+
+				var finalizedAt *time.Time
+				if setFinalized {
+					finalizedAt = &now
+				}
+
+				job := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{
+					FinalizedAt: finalizedAt,
+					ScheduledAt: ptrutil.Ptr(now.Add(1 * time.Hour)),
+					State:       &state,
+				})
+
+				jobAfter, err := exec.JobDelete(ctx, job.ID)
+				require.NoError(t, err)
+				require.NotNil(t, jobAfter)
+				require.Equal(t, job.ID, jobAfter.ID)
+				require.Equal(t, state, jobAfter.State)
+
+				_, err = exec.JobGetByID(ctx, job.ID)
+				require.ErrorIs(t, err, rivertype.ErrNotFound)
+			})
+		}
+
+		t.Run("ReturnsErrNotFoundIfJobDoesNotExist", func(t *testing.T) {
+			t.Parallel()
+
+			exec, _ := setupExecutor(ctx, t, driver, beginTx)
+
+			jobAfter, err := exec.JobDelete(ctx, 1234567890)
+			require.ErrorIs(t, err, rivertype.ErrNotFound)
+			require.Nil(t, jobAfter)
+		})
+	})
+
 	t.Run("JobDeleteBefore", func(t *testing.T) {
 		t.Parallel()
 
@@ -1007,7 +1085,7 @@ func ExerciseExecutorFull[TTx any](ctx context.Context, t *testing.T, driver riv
 			rivertype.JobStateCancelled,
 			rivertype.JobStateCompleted,
 			rivertype.JobStateDiscarded,
-			// TODO(bgentry): add Pending to this list when it's added:
+			rivertype.JobStatePending,
 			rivertype.JobStateRetryable,
 			rivertype.JobStateScheduled,
 		} {

riverdriver/river_driver_interface.go

@@ -77,6 +77,7 @@ type Executor interface {
 
 	JobCancel(ctx context.Context, params *JobCancelParams) (*rivertype.JobRow, error)
 	JobCountByState(ctx context.Context, state rivertype.JobState) (int, error)
+	JobDelete(ctx context.Context, id int64) (*rivertype.JobRow, error)
 	JobDeleteBefore(ctx context.Context, params *JobDeleteBeforeParams) (int, error)
 	JobGetAvailable(ctx context.Context, params *JobGetAvailableParams) ([]*rivertype.JobRow, error)
 	JobGetByID(ctx context.Context, id int64) (*rivertype.JobRow, error)

riverdriver/riverdatabasesql/river_database_sql.go

@@ -77,6 +77,10 @@ func (e *Executor) JobCountByState(ctx context.Context, state rivertype.JobState
 	return 0, riverdriver.ErrNotImplemented
 }
 
+func (e *Executor) JobDelete(ctx context.Context, id int64) (*rivertype.JobRow, error) {
+	return nil, riverdriver.ErrNotImplemented
+}
+
 func (e *Executor) JobDeleteBefore(ctx context.Context, params *riverdriver.JobDeleteBeforeParams) (int, error) {
 	return 0, riverdriver.ErrNotImplemented
 }

riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql

@@ -83,6 +83,30 @@ SELECT count(*)
 FROM river_job
 WHERE state = @state;
 
+-- name: JobDelete :one
+WITH job_to_delete AS (
+    SELECT id
+    FROM river_job
+    WHERE river_job.id = @id
+    FOR UPDATE
+),
+deleted_job AS (
+    DELETE
+    FROM river_job
+    USING job_to_delete
+    WHERE river_job.id = job_to_delete.id
+        -- Do not touch running jobs:
+        AND river_job.state != 'running'::river_job_state
+    RETURNING river_job.*
+)
+SELECT *
+FROM river_job
+WHERE id = @id::bigint
+    AND id NOT IN (SELECT id FROM deleted_job)
+UNION
+SELECT *
+FROM deleted_job;
+
 -- name: JobDeleteBefore :one
 WITH deleted_jobs AS (
     DELETE FROM river_job