retry sequence jobs into pending state

Fixes #971.

Author: bgentry

riverdriver/riverdatabasesql/internal/dbsqlc/river_job.sql.go

@@ -2302,6 +2302,32 @@ func Exercise[TTx any](ctx context.Context, t *testing.T,
 			require.Error(t, err)
 			require.ErrorIs(t, err, rivertype.ErrNotFound)
 		})
+
+		t.Run("SetsStatePendingWhenSeqKeyPresent", func(t *testing.T) {
+			t.Parallel()
+
+			exec, bundle := setup(ctx, t)
+
+			now := time.Now().UTC()
+
+			job := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{
+				ScheduledAt: ptrutil.Ptr(now.Add(1 * time.Hour)),
+				State:       ptrutil.Ptr(rivertype.JobStateRetryable),
+				Metadata:    []byte(`{"seq_key":"foo"}`),
+			})
+
+			jobAfter, err := exec.JobRetry(ctx, &riverdriver.JobRetryParams{
+				ID:  job.ID,
+				Now: &now,
+			})
+			require.NoError(t, err)
+			require.Equal(t, rivertype.JobStatePending, jobAfter.State)
+			require.WithinDuration(t, now, jobAfter.ScheduledAt, bundle.driver.TimePrecision())
+
+			jobUpdated, err := exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: job.ID})
+			require.NoError(t, err)
+			require.Equal(t, rivertype.JobStatePending, jobUpdated.State)
+		})
 	})
 
 	t.Run("JobSchedule", func(t *testing.T) {

riverdriver/riverdrivertest/riverdrivertest.go

@@ -418,7 +418,10 @@ WITH job_to_update AS (
 updated_job AS (
     UPDATE /* TEMPLATE: schema */river_job
     SET
-        state = 'available',
+        state = CASE WHEN river_job.metadata ? 'seq_key'
+            THEN 'pending'::/* TEMPLATE: schema */river_job_state
+            ELSE 'available'::/* TEMPLATE: schema */river_job_state
+        END,
         max_attempts = CASE WHEN attempt = max_attempts THEN max_attempts + 1 ELSE max_attempts END,
         finalized_at = NULL,
         scheduled_at = coalesce(sqlc.narg('now')::timestamptz, now())

riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql

@@ -320,7 +320,7 @@ WHERE id = @id;
 -- name: JobRetry :one
 UPDATE /* TEMPLATE: schema */river_job
 SET
-    state = 'available',
+    state = CASE WHEN river_job.metadata -> 'seq_key' IS NOT NULL THEN 'pending' ELSE 'available' END,
     max_attempts = CASE WHEN attempt = max_attempts THEN max_attempts + 1 ELSE max_attempts END,
     finalized_at = NULL,
     scheduled_at = coalesce(cast(sqlc.narg('now') AS text), datetime('now', 'subsec'))