Function Test_Client_JobRetry

client_test.go:5191–5264

(t *testing.T)

}

func Test_Client_JobRetry(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()

		var (
			dbPool = riversharedtest.DBPool(ctx, t)
			driver = riverpgxv5.New(dbPool)
			schema = riverdbtest.TestSchema(ctx, t, driver, nil)
			config = newTestConfig(t, schema)
			client = newTestClient(t, dbPool, config)
		)

		return client, &testBundle{dbPool: dbPool}
	}

	t.Run("UpdatesAJobScheduledInTheFutureToBeImmediatelyAvailable", 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)

		job, err := client.JobRetry(ctx, insertRes.Job.ID)
		require.NoError(t, err)
		require.NotNil(t, job)

		require.Equal(t, rivertype.JobStateAvailable, job.State)
		require.WithinDuration(t, time.Now().UTC(), job.ScheduledAt, 5*time.Second)
	})

	t.Run("TxVariantAlsoUpdatesJobToAvailable", 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.JobRetryTx(ctx, tx, insertRes.Job.ID)
			return err
		})
		require.NoError(t, err)
		require.NotNil(t, jobAfter)