Function Test_Client_SoftStopTimeout

client_test.go:2541–2655

(t *testing.T)

}

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

	ctx := context.Background()

	type JobArgs struct {
		testutil.JobArgsReflectKind[JobArgs]
	}

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

		config := newTestConfig(t, "")
		config.SoftStopTimeout = 100 * time.Millisecond

		jobDoneChan := make(chan struct{})
		jobStartedChan := make(chan struct{})
		AddWorker(config.Workers, WorkFunc(func(ctx context.Context, job *Job[JobArgs]) error {
			close(jobStartedChan)
			<-ctx.Done() // only finishes when context is cancelled
			close(jobDoneChan)
			return nil
		}))

		client := runNewTestClient(ctx, t, config)

		_, err := client.Insert(ctx, JobArgs{}, nil)
		require.NoError(t, err)

		riversharedtest.WaitOrTimeout(t, jobStartedChan)

		// Stop initiates a soft stop. The job won't finish on its own, but
		// SoftStopTimeout should escalate to a hard stop after 100ms.
		require.NoError(t, client.Stop(ctx))

		// Verify the job's context was indeed cancelled.
		select {
		case <-jobDoneChan:
		default:
			t.Fatal("expected job to have been cancelled by soft stop timeout")
		}
	})

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

		config := newTestConfig(t, "")
		config.SoftStopTimeout = 5 * time.Second

		jobStartedChan := make(chan struct{})
		AddWorker(config.Workers, WorkFunc(func(ctx context.Context, job *Job[JobArgs]) error {
			close(jobStartedChan)
			return nil // finishes immediately
		}))

		client := runNewTestClient(ctx, t, config)

		_, err := client.Insert(ctx, JobArgs{}, nil)