Function Test_Client_RetryPolicy

client_test.go:6756–6877

(t *testing.T)

}

func Test_Client_RetryPolicy(t *testing.T) {
	t.Parallel()

	ctx := context.Background()

	t.Run("RetryUntilDiscarded", func(t *testing.T) {
		t.Parallel()

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

		// The default policy would work too, but this takes some variability
		// out of it to make comparisons easier.
		config.RetryPolicy = &retrypolicytest.RetryPolicyNoJitter{}

		type JobArgs struct {
			testutil.JobArgsReflectKind[JobArgs]
		}

		AddWorker(config.Workers, WorkFunc(func(ctx context.Context, job *Job[JobArgs]) error {
			return errors.New("job error")
		}))

		client := newTestClient(t, dbPool, config)

		now := client.baseService.Time.StubNow(time.Now().UTC())
		t.Logf("Now: %s", now)

		subscribeChan, cancel := client.Subscribe(EventKindJobCompleted, EventKindJobFailed)
		t.Cleanup(cancel)

		originalJobs := make([]*rivertype.JobRow, rivercommon.MaxAttemptsDefault)
		for i := range originalJobs {
			insertRes, err := client.Insert(ctx, JobArgs{}, nil)
			require.NoError(t, err)

			// regression protection to ensure we're testing the right number of jobs:
			require.Equal(t, rivercommon.MaxAttemptsDefault, insertRes.Job.MaxAttempts)

			updatedJob, err := client.driver.GetExecutor().JobUpdateFull(ctx, &riverdriver.JobUpdateFullParams{
				ID:                  insertRes.Job.ID,
				AttemptedAtDoUpdate: true,
				AttemptedAt:         &now, // we want a value here, but it'll be overwritten as jobs are locked by the producer
				AttemptDoUpdate:     true,
				Attempt:             i, // starts at i, but will be i + 1 by the time it's being worked
				Schema:              schema,

				// Need to find a cleaner way around this, but state is required
				// because sqlc can't encode an empty string to the
				// corresponding enum. This value is not actually used because
				// StateDoUpdate was not supplied.
				State: rivertype.JobStateAvailable,
			})
			require.NoError(t, err)