Function Test_Client_Stop_Common

client_test.go:2114–2341

(t *testing.T)

}

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

	ctx := context.Background()

	// Performs continual job insertion on a number of background goroutines.
	// Returns a `finish` function that should be deferred to stop insertion and
	// safely stop goroutines.
	doParallelContinualInsertion := func(ctx context.Context, t *testing.T, client *Client[pgx.Tx], args JobArgs) func() {
		t.Helper()

		ctx, cancel := context.WithCancel(ctx)

		var wg sync.WaitGroup
		for range 10 {
			wg.Go(func() {
				for {
					select {
					case <-ctx.Done():
						return
					default:
					}

					_, err := client.Insert(ctx, args, nil)
					// A cancelled context may produce a variety of underlying
					// errors in pgx, so rather than comparing the return error,
					// first check if context is cancelled, and ignore an error
					// return if it is.
					if ctx.Err() != nil {
						return
					}
					require.NoError(t, err)

					// Sleep a brief time between inserts.
					serviceutil.CancellableSleep(ctx, randutil.DurationBetween(1*time.Microsecond, 10*time.Millisecond))
				}
			})
		}

		return func() {
			cancel()
			wg.Wait()
		}
	}

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

		client := runNewTestClient(ctx, t, newTestConfig(t, ""))

		// Shutdown should still complete promptly, but the client may need a full
		// leadership resign attempt before stopping, which can take about a second
		// under race instrumentation.
		ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
		defer cancel()

		require.NoError(t, client.Stop(ctx))
	})