test: wait on goroutines and barriers instead of wall-clock sleeps in Go tests

internal/api/db_rate_limiter_test.go

@@ -8,6 +8,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/testutil/goroutines"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -101,42 +102,47 @@ func TestDBRateLimiter_maybeCleanup_TooSoon(t *testing.T) {
 func TestDBRateLimiter_maybeCleanup_ReadyToCleanup(t *testing.T) {
 	rl := NewDBRateLimiter(nil)
 	rl.cleanupInterval = time.Millisecond
-	rl.lastCleanup = time.Now().Add(-time.Minute) // Long ago
+	before := time.Now().Add(-time.Minute) // Long ago
+	rl.lastCleanup = before
 
+	mark := goroutines.NewMark()
 	// Should trigger cleanup (no-op because pool is nil)
 	rl.maybeCleanup()
-	// Brief wait for the goroutine to complete
-	time.Sleep(20 * time.Millisecond)
+
+	// The trigger stamps lastCleanup before it starts the goroutine.
+	rl.cleanupMu.Lock()
+	stamped := rl.lastCleanup.After(before)
+	rl.cleanupMu.Unlock()
+	assert.True(t, stamped, "a due cleanup must record when it ran")
+
+	// Wait for the cleanup goroutine to return and hand the slot back,
+	// rather than sleeping a fixed 20ms (issue #36).
+	require.Eventually(t, func() bool {
+		return len(mark.StartedBy("api.(*DBRateLimiter).maybeCleanup")) == 0 && !rl.cleanupRunning.Load()
+	}, 10*time.Second, time.Millisecond, "the cleanup goroutine did not finish and release cleanupRunning")
 }
 
 // Regression test for 02-M2: StartCleanupWorker must launch a goroutine that
-// stops cleanly on context cancellation. The goroutine is exercised with a
-// fast ticker so we can confirm it fires without sleeping in tests.
+// stops cleanly on context cancellation. The goroutine is found in the
+// runtime's goroutine snapshot, so the test waits for its exit instead of
+// sleeping (issue #36).
+//
+// With a nil pool, cleanup() is a no-op, so a tick is not observable here and
+// the test no longer shortens dbRateLimiterScheduledCleanupInterval: the
+// worker reads that variable on its own goroutine, and restoring it in
+// t.Cleanup is a data race the race detector reports, since nothing orders
+// the worker's read before the restore.
 func TestDBRateLimiter_StartCleanupWorker_StopsOnContextCancel(t *testing.T) {
-	// Override the scheduled interval to a very short duration for testing.
-	orig := dbRateLimiterScheduledCleanupInterval
-	dbRateLimiterScheduledCleanupInterval = 5 * time.Millisecond
-	t.Cleanup(func() { dbRateLimiterScheduledCleanupInterval = orig })
-
 	rl := NewDBRateLimiter(nil) // nil pool; cleanup() is a no-op
 	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
 
-	done := make(chan struct{})
-	// Wrap cleanup in a channel signal to confirm the worker ran.
-	go func() {
-		defer close(done)
-		rl.StartCleanupWorker(ctx)
-	}()
+	mark := goroutines.NewMark()
+	rl.StartCleanupWorker(ctx)
+	const worker = "api.(*DBRateLimiter).StartCleanupWorker"
+	require.Len(t, mark.StartedBy(worker), 1, "StartCleanupWorker must run the worker on its own goroutine")
 
-	// Let the ticker fire at least twice (2 * 5ms = 10ms + margin).
-	time.Sleep(30 * time.Millisecond)
 	cancel()
-
-	// StartCleanupWorker itself returns immediately (goroutine is launched
-	// inside it). The done channel signals the outer goroutine exited.
-	select {
-	case <-done:
-	case <-time.After(100 * time.Millisecond):
-		t.Error("StartCleanupWorker goroutine wrapper did not return after cancel")
-	}
+	require.Eventually(t, func() bool { return len(mark.StartedBy(worker)) == 0 },
+		10*time.Second, time.Millisecond, "the cleanup worker did not exit after its context was canceled")
 }

internal/api/ri_utilization_cache_integration_test.go

@@ -8,6 +8,7 @@ import (
 	"encoding/json"
 	"path/filepath"
 	"runtime"
+	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -105,7 +106,9 @@ func TestRIUtilizationCache_Integration_FreshHitNoFetch(t *testing.T) {
 
 // TestRIUtilizationCache_Integration_StaleNonLambdaServesStaleAndRefreshes
 // verifies SWR end-to-end: stale row served synchronously, background
-// refresh fires, Postgres row is updated within a short window.
+// refresh fires, Postgres row is updated. The fetcher blocks until the test
+// releases it, so getOrFetch returning proves the stale read did not wait for
+// the refresh, without an elapsed-time bound (issue #36).
 func TestRIUtilizationCache_Integration_StaleNonLambdaServesStaleAndRefreshes(t *testing.T) {
 	ctx := context.Background()
 	store, cleanup := setupRICacheIntegration(ctx, t)
@@ -120,30 +123,33 @@ func TestRIUtilizationCache_Integration_StaleNonLambdaServesStaleAndRefreshes(t
 	require.NoError(t, store.UpsertRIUtilizationCache(ctx, "us-east-1", 30, stalePayload, time.Now().Add(-200*time.Millisecond)))
 
 	cache := newRIUtilizationCache(store, false) // non-Lambda
-	fetch := func(ctx context.Context, lookbackDays int) ([]recommendations.RIUtilization, error) {
+	release := make(chan struct{})
+	var releaseOnce sync.Once
+	releaseFetch := func() { releaseOnce.Do(func() { close(release) }) }
+	t.Cleanup(releaseFetch)
+	//nolint:unparam // fetcher signature is fixed by getOrFetch; this refresh path never errors
+	fetch := func(_ context.Context, _ int) ([]recommendations.RIUtilization, error) {
+		<-release
 		return freshData, nil
 	}
 
-	start := time.Now()
-	got, err := cache.getOrFetch(ctx, "us-east-1", 30, 50*time.Millisecond, time.Hour, fetch)
-	require.NoError(t, err)
-	require.Less(t, time.Since(start), 100*time.Millisecond, "stale-read must return synchronously, not wait for background fetch")
+	got := getOrFetchWithin(t, 10*time.Second, func() ([]recommendations.RIUtilization, error) {
+		return cache.getOrFetch(ctx, "us-east-1", 30, 50*time.Millisecond, time.Hour, fetch)
+	}, "stale-read must return synchronously, not wait for background fetch")
+	require.Len(t, got, 1)
 	require.Equal(t, "ri-stale", got[0].ReservedInstanceID, "first call must return the stale copy")
 
-	// Wait for the background refresh to land the fresh data.
-	deadline := time.Now().Add(2 * time.Second)
-	for time.Now().Before(deadline) {
+	// Let the background refresh finish and wait for it to land the fresh data.
+	releaseFetch()
+	require.Eventually(t, func() bool {
 		entry, err := store.GetRIUtilizationCache(ctx, "us-east-1", 30)
-		if err == nil && entry != nil {
-			var round []recommendations.RIUtilization
-			_ = json.Unmarshal(entry.Payload, &round)
-			if len(round) == 1 && round[0].ReservedInstanceID == "ri-fresh" {
-				return
-			}
+		if err != nil || entry == nil {
+			return false
 		}
-		time.Sleep(20 * time.Millisecond)
-	}
-	t.Fatalf("background refresh did not update the ri_utilization_cache row within 2s")
+		var round []recommendations.RIUtilization
+		_ = json.Unmarshal(entry.Payload, &round)
+		return len(round) == 1 && round[0].ReservedInstanceID == "ri-fresh"
+	}, 10*time.Second, 20*time.Millisecond, "background refresh did not update the ri_utilization_cache row")
 }
 
 // TestRIUtilizationCache_Integration_HardExpiredSyncRefetch verifies

internal/api/ri_utilization_cache_test.go

@@ -12,6 +12,8 @@ import (
 
 	"github.com/LeanerCloud/cloud-commitments-go/providers/aws/recommendations"
 	"github.com/LeanerCloud/cloud-commitments-platform/internal/config"
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/testutil/goroutines"
+	"github.com/stretchr/testify/require"
 )
 
 // fakeRIUtilCacheStore is a minimal in-test implementation of
@@ -122,6 +124,10 @@ func TestRIUtilizationCache_HitReturnsCachedData(t *testing.T) {
 	}
 }
 
+// The fetcher blocks until the test releases it, so getOrFetch returning at
+// all proves the stale copy was served without waiting for the refresh. That
+// replaces an elapsed-time bound, and require.Eventually replaces the
+// sleep-poll on the stored row (issue #36).
 func TestRIUtilizationCache_StaleOnNonLambdaServesStaleAndKicksRefresh(t *testing.T) {
 	store := newFakeRIUtilCacheStore()
 	staleData := []recommendations.RIUtilization{{ReservedInstanceID: "ri-stale"}}
@@ -132,43 +138,59 @@ func TestRIUtilizationCache_StaleOnNonLambdaServesStaleAndKicksRefresh(t *testin
 	c := newRIUtilizationCache(store, false) // non-Lambda
 
 	fetchCalled := make(chan struct{}, 1)
-	fetch := func(ctx context.Context, lookbackDays int) ([]recommendations.RIUtilization, error) {
+	release := make(chan struct{})
+	var releaseOnce sync.Once
+	releaseFetch := func() { releaseOnce.Do(func() { close(release) }) }
+	t.Cleanup(releaseFetch)
+	//nolint:unparam // fetcher signature is fixed by getOrFetch; this refresh path never errors
+	fetch := func(_ context.Context, _ int) ([]recommendations.RIUtilization, error) {
 		fetchCalled <- struct{}{}
+		<-release
 		return freshData, nil
 	}
 
-	start := time.Now()
-	got, err := c.getOrFetch(context.Background(), "us-east-1", 30, 50*time.Millisecond, time.Hour, fetch)
-	elapsed := time.Since(start)
-	if err != nil {
-		t.Fatalf("unexpected error: %v", err)
-	}
-	// SWR must return quickly with the stale payload, not wait for
-	// the fetcher.
-	if elapsed > 50*time.Millisecond {
-		t.Fatalf("getOrFetch took %s — should have returned the stale copy synchronously", elapsed)
-	}
+	got := getOrFetchWithin(t, 10*time.Second, func() ([]recommendations.RIUtilization, error) {
+		return c.getOrFetch(context.Background(), "us-east-1", 30, 50*time.Millisecond, time.Hour, fetch)
+	}, "getOrFetch waited for the fetcher; SWR must return the stale copy synchronously")
 	if len(got) != 1 || got[0].ReservedInstanceID != "ri-stale" {
 		t.Fatalf("expected stale copy, got %+v", got)
 	}
 
-	// Background fetch should fire within a short window.
 	select {
 	case <-fetchCalled:
-	case <-time.After(500 * time.Millisecond):
+	case <-time.After(10 * time.Second):
 		t.Fatalf("background fetch never fired")
 	}
+	releaseFetch()
 
-	// Give the storePayload write a beat to land.
-	deadline := time.Now().Add(500 * time.Millisecond)
-	for time.Now().Before(deadline) {
+	require.Eventually(t, func() bool {
 		latest, ok := store.latest("us-east-1", 30)
-		if ok && len(latest) == 1 && latest[0].ReservedInstanceID == "ri-fresh" {
-			return
-		}
-		time.Sleep(10 * time.Millisecond)
+		return ok && len(latest) == 1 && latest[0].ReservedInstanceID == "ri-fresh"
+	}, 10*time.Second, time.Millisecond, "background refresh did not update the stored row")
+}
+
+// getOrFetchWithin runs call on another goroutine and fails the test with msg
+// if it has not returned within limit. The limit only bounds a hang (a call
+// stuck on a fetcher the test never releases); it is not a timing assertion.
+func getOrFetchWithin(t *testing.T, limit time.Duration, call func() ([]recommendations.RIUtilization, error), msg string) []recommendations.RIUtilization {
+	t.Helper()
+	type result struct {
+		err  error
+		data []recommendations.RIUtilization
+	}
+	ret := make(chan result, 1)
+	go func() {
+		data, err := call()
+		ret <- result{data: data, err: err}
+	}()
+	select {
+	case r := <-ret:
+		require.NoError(t, r.err)
+		return r.data
+	case <-time.After(limit):
+		t.Fatal(msg)
+		return nil
 	}
-	t.Fatalf("background refresh did not update the stored row")
 }
 
 func TestRIUtilizationCache_StaleOnLambdaBlocksForSyncRefetch(t *testing.T) {
@@ -228,6 +250,9 @@ func TestRIUtilizationCache_SingleflightCollapsesConcurrentRefreshes(t *testing.
 	// Gate the fetcher so concurrent calls all race to enter the
 	// refresh — singleflight should collapse them.
 	release := make(chan struct{})
+	var releaseOnce sync.Once
+	releaseFetch := func() { releaseOnce.Do(func() { close(release) }) }
+	t.Cleanup(releaseFetch)
 	//nolint:unparam // fetcher signature is fixed by newRIUtilizationCache; this refresh path never errors
 	fetch := func(_ context.Context, _ int) ([]recommendations.RIUtilization, error) {
 		calls.Add(1)
@@ -236,8 +261,10 @@ func TestRIUtilizationCache_SingleflightCollapsesConcurrentRefreshes(t *testing.
 	}
 
 	// Spawn 10 concurrent readers.
+	const readers = 10
+	mark := goroutines.NewMark()
 	var wg sync.WaitGroup
-	for i := 0; i < 10; i++ {
+	for i := 0; i < readers; i++ {
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
@@ -249,25 +276,39 @@ func TestRIUtilizationCache_SingleflightCollapsesConcurrentRefreshes(t *testing.
 	}
 	wg.Wait()
 
-	// All 10 getOrFetch calls returned the stale copy. The 10
-	// background-refresh goroutines are now racing to enter sf.Do —
-	// under CPU pressure (full test suite with -race) they can lag
-	// behind the getOrFetch returns, and a straggler that arrives
-	// after the first singleflight batch finishes would open a
-	// second batch and bump calls to 2. Give the scheduler enough
-	// runway for all 10 to reach sf.Do (where they'll block on
-	// release) before we unblock the in-flight fetch.
-	time.Sleep(250 * time.Millisecond)
-
-	// Unblock the single in-flight fetch and wait for the collapsed
-	// batch's storePayload to complete so calls.Load() reflects the
-	// final state.
-	close(release)
-	time.Sleep(200 * time.Millisecond)
-
+	// All 10 getOrFetch calls returned the stale copy, each kicking a
+	// background refresh. Releasing the fetch before every refresh has
+	// reached sf.Do would let a straggler open a second batch, so wait until
+	// all 10 are accounted for: each one either called the fetcher, or is
+	// parked in sf.Do on the in-flight call's WaitGroup as a duplicate. No
+	// refresh can start a new fetch while the first is blocked, so once the
+	// sum is 10 the fetch count is final (issue #36: this replaces a 250ms
+	// head-start sleep that could still lose to a slow scheduler).
+	require.Eventually(t, func() bool {
+		return int(calls.Load())+len(sfDuplicatesWaiting(mark)) == readers
+	}, 10*time.Second, time.Millisecond, "the background refreshes never all reached singleflight")
 	if n := calls.Load(); n != 1 {
 		t.Fatalf("expected exactly 1 fetch call under singleflight; got %d", n)
 	}
+
+	// Unblock the single in-flight fetch and wait for the collapsed batch to
+	// store the fresh row and every refresh goroutine to return.
+	releaseFetch()
+	require.Eventually(t, func() bool {
+		latest, ok := store.latest("us-east-1", 30)
+		return ok && len(latest) == 1 && latest[0].ReservedInstanceID == "ri-fresh" &&
+			len(mark.Started("riUtilizationCache).kickBackgroundRefresh")) == 0
+	}, 10*time.Second, time.Millisecond, "the collapsed refresh did not store the fresh row and return")
+	if n := calls.Load(); n != 1 {
+		t.Fatalf("expected exactly 1 fetch call under singleflight after release; got %d", n)
+	}
+}
+
+// sfDuplicatesWaiting returns the background refreshes started after mark
+// that singleflight has registered as duplicates of an in-flight call: they
+// are blocked on that call's WaitGroup inside Group.Do.
+func sfDuplicatesWaiting(mark goroutines.Mark) []goroutines.Goroutine {
+	return mark.Started("riUtilizationCache).kickBackgroundRefresh", "singleflight.(*Group).Do(", "sync.(*WaitGroup).Wait(")
 }
 
 func TestRIUtilizationCache_FetchErrorNotCached(t *testing.T) {