sidetransport: bump TestPaceUpdateSignalling expectation

This bumps the expectation for "immediate wakeup" to 30ms. This is 30x
what I've observed running this under stress, but only 5ms over the
recent test failure.

The real fix here is going to be to thread our clock source down into
the pacer so that we can completely mock the clock and make the test
deterministic.

While here, I updated the test a bit.

Informs #155199
Release note: None

Author: stevendanna

pkg/kv/kvserver/closedts/sidetransport/BUILD.bazel

@@ -53,11 +53,13 @@ go_test(
         "//pkg/settings/cluster",
         "//pkg/testutils",
         "//pkg/testutils/skip",
+        "//pkg/util/ctxgroup",
         "//pkg/util/hlc",
         "//pkg/util/leaktest",
         "//pkg/util/log",
         "//pkg/util/stop",
         "//pkg/util/syncutil",
+        "@com_github_cockroachdb_crlib//crtime",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_stretchr_testify//require",
         "@io_storj_drpc//:drpc",

pkg/kv/kvserver/closedts/sidetransport/sender_test.go

@@ -9,6 +9,7 @@ import (
 	"context"
 	"fmt"
 	"net"
+	"slices"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -25,11 +26,13 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/skip"
+	"github.com/cockroachdb/cockroach/pkg/util/ctxgroup"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/stop"
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
+	"github.com/cockroachdb/crlib/crtime"
 	"github.com/cockroachdb/errors"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/grpc"
@@ -957,26 +960,25 @@ func TestPaceUpdateSignalling(t *testing.T) {
 	// each other.
 	//
 	// seqNum is the sequence number to wait for.
-	testPacing := func(seqNum ctpb.SeqNum, assertionFunc func(timeSpread time.Duration)) {
+	testPacing := func(t *testing.T, seqNum ctpb.SeqNum, assertionFunc func(timeSpread time.Duration)) {
 		// Track the times when goroutines receive items from the buffer.
-		var receiveTimes []time.Time
+		var receiveTimes []crtime.Mono
 		var mu syncutil.Mutex
 
 		// Create numWaiters goroutines that wait on s.buf.GetBySeq and record
 		// receive times.
-		done := make(chan struct{}, numWaiters)
-		for i := 0; i < numWaiters; i++ {
-			go func() {
+		g := ctxgroup.WithContext(ctx)
+		for range numWaiters {
+			g.Go(func() error {
 				// Wait for the specified sequence number.
 				_, ok := s.buf.GetBySeq(ctx, seqNum)
-				require.Equal(t, true, ok)
-				if ok {
-					mu.Lock()
-					receiveTimes = append(receiveTimes, time.Now())
-					mu.Unlock()
-				}
-				done <- struct{}{}
-			}()
+				require.True(t, ok)
+
+				mu.Lock()
+				defer mu.Unlock()
+				receiveTimes = append(receiveTimes, crtime.NowMono())
+				return nil
+			})
 		}
 
 		// Wait until all goroutines are waiting on the buffer.
@@ -991,47 +993,43 @@ func TestPaceUpdateSignalling(t *testing.T) {
 		s.publish(ctx)
 
 		// Wait for all goroutines to finish.
-		for i := 0; i < numWaiters; i++ {
-			<-done
-		}
+		require.NoError(t, g.Wait())
 
 		// Verify that all goroutines received the message.
 		require.Len(t, receiveTimes, numWaiters)
 
-		// Find min and max receive times.
-		minTime := receiveTimes[0]
-		maxTime := receiveTimes[0]
-		for _, t := range receiveTimes[1:] {
-			if t.Before(minTime) {
-				minTime = t
-			}
-			if t.After(maxTime) {
-				maxTime = t
-			}
-		}
-
 		// Verify that the time spread matches expectations.
+		minTime := slices.Min(receiveTimes)
+		maxTime := slices.Max(receiveTimes)
 		timeSpread := maxTime.Sub(minTime)
 		assertionFunc(timeSpread)
 	}
 
 	// Test with 250ms pacing interval - expect at least 125ms spread just to be
 	// conservative. In practice, it should be closer to 250ms.
-	closedts.SideTransportPacingRefreshInterval.Override(ctx, &st.SV, 250*time.Millisecond)
-	testPacing(1 /* seqNum */, func(timeSpread time.Duration) {
-		require.GreaterOrEqual(t, timeSpread, 125*time.Millisecond)
+	t.Run("pacing_interval=250ms", func(t *testing.T) {
+		closedts.SideTransportPacingRefreshInterval.Override(ctx, &st.SV, 250*time.Millisecond)
+		testPacing(t, 1 /* seqNum */, func(timeSpread time.Duration) {
+			require.GreaterOrEqual(t, timeSpread, 125*time.Millisecond)
+		})
 	})
 
 	// Change to 100ms pacing interval - expect at least 50ms spread.
-	closedts.SideTransportPacingRefreshInterval.Override(ctx, &st.SV, 100*time.Millisecond)
-	testPacing(2 /* seqNum */, func(timeSpread time.Duration) {
-		require.GreaterOrEqual(t, timeSpread, 50*time.Millisecond)
+	t.Run("pacing_interval=100ms", func(t *testing.T) {
+		closedts.SideTransportPacingRefreshInterval.Override(ctx, &st.SV, 100*time.Millisecond)
+		testPacing(t, 2 /* seqNum */, func(timeSpread time.Duration) {
+			require.GreaterOrEqual(t, timeSpread, 50*time.Millisecond)
+		})
 	})
 
 	// Change to 0ms (disabled) pacing interval - expect all goroutines to be
 	// woken within a few milliseconds of each other.
-	closedts.SideTransportPacingRefreshInterval.Override(ctx, &st.SV, 0)
-	testPacing(3 /* seqNum */, func(timeSpread time.Duration) {
-		require.LessOrEqual(t, timeSpread, 25*time.Millisecond)
+	t.Run("pacing_interval=0ms", func(t *testing.T) {
+		closedts.SideTransportPacingRefreshInterval.Override(ctx, &st.SV, 0)
+		testPacing(t, 3 /* seqNum */, func(timeSpread time.Duration) {
+			// The expectation here is many 30x what is typically seen. But, we want
+			// to avoid flakes.
+			require.LessOrEqual(t, timeSpread, 30*time.Millisecond)
+		})
 	})
 }