tweak JobRetry to reset ScheduledAt in most cases (#211)

A job which has already completed will have a ScheduledAt that could be
long in the past. Now that we're re-scheduling it, we should update that
to the current time to slot it in alongside other recently-scheduled
jobs and not skip the line. Also, its wait duration can't be calculated
accurately if we don't reset the ScheduledAt (it could show days of wait
time even if it was picked up immediately after being retried).

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Changed
+
+- Tweaked behavior of `JobRetry` so that it does actually update the `ScheduledAt` time of the job in all cases where the job is actually being rescheduled. As before, jobs which are already available with a past `ScheduledAt` will not be touched by this query so that they retain their place in line. [PR #211](https://github.com/riverqueue/river/pull/211).
+
 ## [0.0.20] - 2024-02-14
 
 ### Added

internal/dbadapter/db_adapter_test.go

@@ -1064,16 +1064,44 @@ func Test_StandardAdapter_JobRetryImmediately(t *testing.T) {
 		})
 	}
 
-	t.Run("DoesNotAlterScheduledAtIfInThePast", func(t *testing.T) {
-		// We don't want to update ScheduledAt if the job was already scheduled
+	t.Run("AltersScheduledAtForAlreadyCompletedJob", func(t *testing.T) {
+		// A job which has already completed will have a ScheduledAt that could be
+		// long in the past. Now that we're re-scheduling it, we should update that
+		// to the current time to slot it in alongside other recently-scheduled jobs
+		// and not skip the line; also, its wait duration can't be calculated
+		// accurately if we don't reset the scheduled_at.
+		t.Parallel()
+
+		adapter, bundle := setupTx(t)
+
+		params := makeFakeJobInsertParams(0, nil)
+		params.ScheduledAt = bundle.baselineTime.Add(-1 * time.Hour)
+		res, err := adapter.JobInsert(ctx, params)
+		require.NoError(t, err)
+		_, err = adapter.queries.JobUpdate(ctx, bundle.ex, dbsqlc.JobUpdateParams{
+			FinalizedAtDoUpdate: true,
+			FinalizedAt:         &bundle.baselineTime,
+			ID:                  res.Job.ID,
+			StateDoUpdate:       true,
+			State:               dbsqlc.JobStateCompleted,
+		})
+		require.NoError(t, err)
+
+		jAfter, err := adapter.JobRetryImmediately(ctx, res.Job.ID)
+		require.NoError(t, err)
+		require.Equal(t, dbsqlc.JobStateAvailable, jAfter.State)
+		require.WithinDuration(t, time.Now().UTC(), jAfter.ScheduledAt, 5*time.Second)
+	})
+
+	t.Run("DoesNotAlterScheduledAtIfInThePastAndJobAlreadyAvailable", func(t *testing.T) {
+		// We don't want to update ScheduledAt if the job was already available
 		// because doing so can make it lose its place in line.
 		t.Parallel()
 
 		adapter, bundle := setupTx(t)
 
 		params := makeFakeJobInsertParams(0, nil)
 		params.ScheduledAt = bundle.baselineTime.Add(-1 * time.Hour)
-		params.State = dbsqlc.JobStateScheduled
 		res, err := adapter.JobInsert(ctx, params)
 		require.NoError(t, err)
 

internal/dbsqlc/river_job.sql

@@ -260,7 +260,7 @@ updated_job AS (
   UPDATE river_job
   SET
     state = 'available'::river_job_state,
-    scheduled_at = CASE WHEN scheduled_at < now() THEN scheduled_at ELSE now() END,
+    scheduled_at = now(),
     max_attempts = CASE WHEN attempt = max_attempts THEN max_attempts + 1 ELSE max_attempts END,
     finalized_at = NULL
   FROM job_to_update