Function TestJobExecutor_Execute

internal/jobexecutor/job_executor_test.go:115–1171

(t *testing.T)

}

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

	ctx := context.Background()

	type testBundle struct {
		completer    *jobcompleter.InlineCompleter
		exec         riverdriver.Executor
		errorHandler *testErrorHandler
		jobRow       *rivertype.JobRow
		updateCh     <-chan []jobcompleter.CompleterJobUpdated
	}

	setup := func(t *testing.T) (*JobExecutor, *testBundle) {
		t.Helper()

		var (
			tx        = riverdbtest.TestTxPgx(ctx, t)
			archetype = riversharedtest.BaseServiceArchetype(t)
			exec      = riverpgxv5.New(nil).UnwrapExecutor(tx)
			updateCh  = make(chan []jobcompleter.CompleterJobUpdated, 10)
			completer = jobcompleter.NewInlineCompleter(archetype, "", exec, &riverpilot.StandardPilot{}, updateCh)
		)

		t.Cleanup(completer.Stop)

		workUnitFactory := newWorkUnitFactoryWithCustomRetry(func() error { return nil }, nil)

		now := time.Now().UTC()
		results, err := exec.JobInsertFastMany(ctx, &riverdriver.JobInsertFastManyParams{
			Jobs: []*riverdriver.JobInsertFastParams{
				{
					EncodedArgs: []byte("{}"),
					Kind:        "jobexecutor_test",
					MaxAttempts: rivercommon.MaxAttemptsDefault,
					Priority:    rivercommon.PriorityDefault,
					Queue:       rivercommon.QueueDefault,
					// Needs to be explicitly set to a "now" horizon that's aligned with the
					// JobGetAvailable call. InsertMany applies a default scheduled_at in Go
					// so it can't pick up the Postgres-level `now()` default.
					ScheduledAt: ptrutil.Ptr(now),
					State:       rivertype.JobStateAvailable,
				},
			},
		})
		require.NoError(t, err)

		// Fetch the job to make sure it's marked as running:
		jobs, err := exec.JobGetAvailable(ctx, &riverdriver.JobGetAvailableParams{
			MaxToLock: 1,
			Now:       ptrutil.Ptr(now),
			Queue:     rivercommon.QueueDefault,
		})
		require.NoError(t, err)

		require.Len(t, jobs, 1)
		require.Equal(t, results[0].Job.ID, jobs[0].ID)
		job := jobs[0]