From 42bac6eebe401a6cc4cea63420da8219290f6d52 Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Fri, 23 Oct 2026 07:57:29 +0000 Subject: [PATCH] test: wait on goroutines and barriers instead of wall-clock sleeps in Go tests Replace the Go test sleeps that stood in for async synchronization with waits that cannot lose to a slow scheduler: - internal/testutil/goroutines: a dependency-free helper that reads runtime.Stack(all) to list goroutines started after a mark. A go statement is in the dump as soon as it executes, so a negative check ("this read spawned nothing") no longer needs a sleep window. - scheduler: the Lambda and unmarked-read "no background refresh" tests check the goroutine snapshot plus the GetGlobalConfig call log instead of sleeping 20ms; the fan-out limit test parks every worker and reads the in-flight count once errgroup.Go is blocked on the limit, instead of a 5ms "overlap" sleep that passed on a serial loop. - api: the RI-utilization SWR tests gate the fetcher on a channel and wait with require.Eventually (no elapsed-time bound, no 250ms/200ms sleeps); the singleflight test releases only after all 10 refreshes are inside Group.Do. The DB rate limiter cleanup tests wait for the cleanup goroutine to exit instead of sleeping. - purchase: the multi-account parallelism test uses a two-party barrier instead of a 300ms sleep and an elapsed-time bound. - server: the scheduled-task integration test counts the collect call instead of sleeping 100ms. Sleeps that test real elapsed time (TTL windows, DB-clock ages) or poll an external system with a generous deadline are kept. The frontend setTimeout waits are left for the UI owner. Refs #36 Co-Authored-By: Claude Opus 5.5 --- internal/api/db_rate_limiter_test.go | 58 +++--- .../ri_utilization_cache_integration_test.go | 42 ++-- internal/api/ri_utilization_cache_test.go | 119 ++++++++---- internal/purchase/execution_test.go | 52 ++--- .../scheduler/background_collection_test.go | 27 ++- internal/scheduler/scheduler_test.go | 91 +++++++-- internal/server/integration_test.go | 11 +- internal/testutil/goroutines/goroutines.go | 182 ++++++++++++++++++ .../testutil/goroutines/goroutines_test.go | 76 ++++++++ 9 files changed, 526 insertions(+), 132 deletions(-) create mode 100644 internal/testutil/goroutines/goroutines.go create mode 100644 internal/testutil/goroutines/goroutines_test.go diff --git a/internal/api/db_rate_limiter_test.go b/internal/api/db_rate_limiter_test.go index ab15e52c7..00aecfc16 100644 --- a/internal/api/db_rate_limiter_test.go +++ b/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") } diff --git a/internal/api/ri_utilization_cache_integration_test.go b/internal/api/ri_utilization_cache_integration_test.go index c63a89c6d..5ee539623 100644 --- a/internal/api/ri_utilization_cache_integration_test.go +++ b/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 diff --git a/internal/api/ri_utilization_cache_test.go b/internal/api/ri_utilization_cache_test.go index f66b1eeb7..78084a12d 100644 --- a/internal/api/ri_utilization_cache_test.go +++ b/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) { diff --git a/internal/purchase/execution_test.go b/internal/purchase/execution_test.go index 81b6cc931..1198caa06 100644 --- a/internal/purchase/execution_test.go +++ b/internal/purchase/execution_test.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "sync" + "sync/atomic" "testing" "time" @@ -1167,17 +1168,15 @@ func TestExecuteMultiAccount_PartialFailure_IsolatesAccounts(t *testing.T) { // proves per-account error isolation but would still pass if executeMultiAccount // were refactored to a serial for-loop. This test fails on a serial loop. // -// Mechanics: two accounts, each with valid credentials, each blocking inside -// PurchaseCommitment for perCallDelay. Parallel fan-out completes in roughly -// perCallDelay; a serial loop would require ~2 * perCallDelay. The threshold -// is set comfortably between the two so a slow CI runner does not flake on -// parallel scheduling overhead but a serial-loop regression is caught. +// Mechanics: two accounts, each with valid credentials. Each account's +// PurchaseCommitment call waits at a two-party barrier until the other +// account's call has started too. A parallel fan-out meets there at once; a +// serial loop never does, so its first call gives up after barrierLimit and +// the test fails on the overlap count. This replaces a sleep-and-measure +// check against an elapsed-time bound (issue #36): barrierLimit only bounds +// how long the serial case takes to fail, it is not a timing assertion. func TestExecuteMultiAccount_RunsAccountsInParallel(t *testing.T) { - const ( - perCallDelay = 300 * time.Millisecond - serialBoundary = 2 * perCallDelay // serial fan-out lower bound: 600ms - parallelMaxBound = 500 * time.Millisecond - ) + const barrierLimit = 10 * time.Second ctx := context.Background() mockStore := new(MockConfigStore) @@ -1238,15 +1237,25 @@ func TestExecuteMultiAccount_RunsAccountsInParallel(t *testing.T) { mockFactory.On("CreateAndValidateProvider", mock.MatchedBy(hasPerRecDeadline(30*time.Second)), "aws", mock.Anything).Return(mockProviderInst, nil).Twice() mockProviderInst.On("GetServiceClient", mock.MatchedBy(hasPerRecDeadline(30*time.Second)), common.ServiceEC2, "us-east-1").Return(mockServiceClient, nil).Twice() - // Each PurchaseCommitment call blocks for perCallDelay. Under serial - // execution the total elapsed time would exceed serialBoundary; under - // errgroup-style parallelism it stays close to perCallDelay. + var arrived sync.WaitGroup + arrived.Add(2) + bothStarted := make(chan struct{}) + go func() { + arrived.Wait() + close(bothStarted) + }() + var overlapped atomic.Int32 mockServiceClient.On("PurchaseCommitment", mock.MatchedBy(hasPerRecDeadline(30*time.Second)), mock.AnythingOfType("common.Recommendation"), mock.AnythingOfType("common.PurchaseOptions"), ).Return(common.PurchaseResult{Success: true, CommitmentID: "ri-parallel"}, nil). Run(func(_ mock.Arguments) { - time.Sleep(perCallDelay) + arrived.Done() + select { + case <-bothStarted: + overlapped.Add(1) + case <-time.After(barrierLimit): + } }).Twice() mockStore.On("SavePurchaseHistory", ctx, mock.AnythingOfType("*config.PurchaseHistoryRecord")).Return(nil).Twice() @@ -1265,21 +1274,16 @@ func TestExecuteMultiAccount_RunsAccountsInParallel(t *testing.T) { credStore: credStore, } - start := time.Now() err := manager.executePurchase(ctx, exec) - elapsed := time.Since(start) require.NoError(t, err, "both accounts have valid credentials and should succeed") require.Len(t, savedExecs, 2, "exactly two SavePurchaseExecution calls expected — one per account") - // The core parallelism assertion: a serial loop would take >= serialBoundary - // (2 * perCallDelay = 600ms). Parallel fan-out completes in roughly - // perCallDelay (300ms) plus modest scheduling overhead. The boundary is - // set with margin so CI scheduler jitter does not flake the test, while - // any serial-loop regression is caught. - assert.Less(t, elapsed, parallelMaxBound, - "executeMultiAccount must run accounts in parallel — serial would take ~%v, parallel ~%v, observed %v", - serialBoundary, perCallDelay, elapsed) + // The core parallelism assertion: both accounts' purchase calls were in + // flight at the same time. A serial loop leaves the first call alone at + // the barrier until it gives up, so at most one call sees the other. + assert.Equal(t, int32(2), overlapped.Load(), + "executeMultiAccount must run accounts in parallel: both accounts' PurchaseCommitment calls must be in flight together") mockStore.AssertExpectations(t) mockEmail.AssertExpectations(t) diff --git a/internal/scheduler/background_collection_test.go b/internal/scheduler/background_collection_test.go index 6d513fa78..8e53697f5 100644 --- a/internal/scheduler/background_collection_test.go +++ b/internal/scheduler/background_collection_test.go @@ -7,6 +7,7 @@ import ( "time" "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/LeanerCloud/cloud-commitments-platform/internal/testutil/goroutines" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" @@ -31,14 +32,13 @@ func TestListRecommendations_UnmarkedColdStart_NeverMarksOrCollects(t *testing.T s := &Scheduler{config: mockStore, isLambda: false} + mark := goroutines.NewMark() recs, err := s.ListRecommendations(ctx, config.RecommendationFilter{}) require.NoError(t, err) assert.Empty(t, recs) - // A wrongly spawned collect would hit GetGlobalConfig first. - time.Sleep(20 * time.Millisecond) + assertNoBackgroundCollect(t, mark, mockStore) mockStore.AssertNotCalled(t, "MarkCollectionStarted", mock.Anything) - mockStore.AssertNotCalled(t, "GetGlobalConfig", mock.Anything) assert.False(t, s.collecting.Load(), "an unmarked read must not claim the collection slot") } @@ -57,15 +57,32 @@ func TestListRecommendations_UnmarkedStaleCache_NeverRefreshes(t *testing.T) { s := &Scheduler{config: mockStore, isLambda: false, cacheTTL: time.Nanosecond} + mark := goroutines.NewMark() _, err := s.ListRecommendations(ctx, config.RecommendationFilter{}) require.NoError(t, err) - time.Sleep(20 * time.Millisecond) - mockStore.AssertNotCalled(t, "GetGlobalConfig", mock.Anything) + assertNoBackgroundCollect(t, mark, mockStore) mockStore.AssertNotCalled(t, "MarkCollectionStarted", mock.Anything) assert.False(t, s.collecting.Load(), "an unmarked read must not claim the collection slot") } +// assertNoBackgroundCollect checks that the read run since mark started no +// goroutine and made no GetGlobalConfig call, without a wall-clock wait +// (issue #36). A go statement puts its goroutine in the runtime's stack dump +// as soon as it executes, scheduled or not, and the goroutine stays there +// until it returns. So a collect the read wrongly spawned is either still +// listed by mark.Spawned, or it has finished, and then it went through +// CollectRecommendations, whose first store call is GetGlobalConfig (the mock +// records every call, expected or not). The goroutine snapshot is taken first +// so a collect that finishes in between is caught by the call log. +func assertNoBackgroundCollect(t *testing.T, mark goroutines.Mark, store *MockConfigStore) { + t.Helper() + for _, g := range mark.Spawned() { + t.Errorf("the read started a background goroutine:\n%s", g.Stack) + } + store.AssertNotCalled(t, "GetGlobalConfig", mock.Anything) +} + // A marked read of a stale cache still kicks the background refresh: the // allowed path keeps working. func TestListRecommendations_MarkedStaleCache_KicksRefresh(t *testing.T) { diff --git a/internal/scheduler/scheduler_test.go b/internal/scheduler/scheduler_test.go index 9fcf5169c..3489f33fc 100644 --- a/internal/scheduler/scheduler_test.go +++ b/internal/scheduler/scheduler_test.go @@ -6,6 +6,8 @@ import ( "fmt" "net/http" "net/http/httptest" + "strings" + "sync" "sync/atomic" "testing" "time" @@ -18,6 +20,7 @@ import ( "github.com/LeanerCloud/cloud-commitments-platform/internal/email" "github.com/LeanerCloud/cloud-commitments-platform/internal/mocks" "github.com/LeanerCloud/cloud-commitments-platform/internal/purchase" + "github.com/LeanerCloud/cloud-commitments-platform/internal/testutil/goroutines" "github.com/aws/aws-sdk-go-v2/aws" "github.com/aws/aws-sdk-go-v2/service/sts" "github.com/stretchr/testify/assert" @@ -1087,14 +1090,14 @@ func TestScheduler_ListRecommendations_LambdaSkipsBackgroundRefresh(t *testing.T cacheTTL: time.Nanosecond, } + mark := goroutines.NewMark() _, err := scheduler.ListRecommendations(ctx, config.RecommendationFilter{}) require.NoError(t, err) - // Give any (wrongly-spawned) goroutine time to hit the store; none - // should fire, so no GetGlobalConfig call should be observed. - time.Sleep(20 * time.Millisecond) - - mockStore.AssertNotCalled(t, "GetGlobalConfig", mock.Anything) + // No wall-clock wait (issue #36): a wrongly spawned refresh is either + // still alive and in the goroutine snapshot, or already done and in the + // GetGlobalConfig call log. + assertNoBackgroundCollect(t, mark, mockStore) assert.False(t, scheduler.collecting.Load(), "collecting flag must stay unset on Lambda") } @@ -1289,9 +1292,12 @@ func TestScheduler_ListRecommendations_ColdStart_NonLambda_KicksBackgroundCollec } // fanOutPerAccount bounds parallel in-flight calls to -// CUDLY_MAX_ACCOUNT_PARALLELISM. Test with a fake fn that increments an -// atomic counter on entry and decrements on exit, assert peak stays at -// or below the limit. +// CUDLY_MAX_ACCOUNT_PARALLELISM. Every fake fn call parks on release, so the +// fan-out fills its slots and then blocks in errgroup.Go waiting for one. Once +// it is parked there and every worker it started has entered fn, nothing else +// can start until the test releases, so the in-flight count read then is the +// bound itself, with no wall-clock wait (issue #36): a missing limit never +// parks (the Eventually times out) and a wrong one parks at another count. func TestFanOutPerAccount_RespectsParallelismLimit(t *testing.T) { t.Setenv("CUDLY_MAX_ACCOUNT_PARALLELISM", "3") @@ -1312,20 +1318,73 @@ func TestFanOutPerAccount_RespectsParallelismLimit(t *testing.T) { } } - fn := func(ctx context.Context, acct config.CloudAccount) ([]config.RecommendationRecord, *sweepGap, error) { + release := make(chan struct{}) + var releaseOnce sync.Once + releaseAll := func() { releaseOnce.Do(func() { close(release) }) } + t.Cleanup(releaseAll) + + //nolint:unparam // signature is fixed by fanOutPerAccount; these calls never report a gap or error + fn := func(_ context.Context, acct config.CloudAccount) ([]config.RecommendationRecord, *sweepGap, error) { cur := inflight.Add(1) updatePeak(cur) - // Small sleep so concurrent workers genuinely overlap. - time.Sleep(5 * time.Millisecond) + <-release inflight.Add(-1) return []config.RecommendationRecord{{ID: acct.ID}}, nil, nil } - out, outcome := fanOutPerAccount(context.Background(), "Test", accounts, fn) - assert.Len(t, out, len(accounts), "all accounts contribute one record") - assert.LessOrEqual(t, peak.Load(), int32(3), "peak in-flight must not exceed CUDLY_MAX_ACCOUNT_PARALLELISM") - assert.Equal(t, len(accounts), outcome.SucceededCount) - assert.Zero(t, outcome.FailedCount) + type fanOutResult struct { + out []config.RecommendationRecord + outcome accountOutcome + } + done := make(chan fanOutResult, 1) + mark := goroutines.NewMark() + go func() { + out, outcome := fanOutPerAccount(context.Background(), "Test", accounts, fn) + done <- fanOutResult{out, outcome} + }() + + var atBound int32 + require.Eventually(t, func() bool { + parked, workers := fanOutParkedOnLimit(mark) + n := inflight.Load() + atBound = n + return parked && int(n) == workers + }, 10*time.Second, time.Millisecond, "fanOutPerAccount never blocked waiting for a free slot: no parallelism limit") + assert.Equal(t, int32(3), atBound, "in-flight calls while the fan-out waits for a slot must equal CUDLY_MAX_ACCOUNT_PARALLELISM") + + releaseAll() + var res fanOutResult + select { + case res = <-done: + case <-time.After(10 * time.Second): + t.Fatal("fanOutPerAccount did not finish after its calls were released") + } + assert.Len(t, res.out, len(accounts), "all accounts contribute one record") + assert.Equal(t, int32(3), peak.Load(), "peak in-flight must reach, and never exceed, CUDLY_MAX_ACCOUNT_PARALLELISM") + assert.Equal(t, len(accounts), res.outcome.SucceededCount) + assert.Zero(t, res.outcome.FailedCount) +} + +// fanOutParkedOnLimit reports whether the goroutine the test started to run +// fanOutPerAccount is blocked in errgroup.Go on the limit's semaphore, and how +// many of the workers it started are alive. It runs inside require.Eventually, +// so it identifies goroutines by creator rather than with mark.Spawned. +func fanOutParkedOnLimit(mark goroutines.Mark) (parked bool, workers int) { + started := mark.Started() + var runner int64 + for _, g := range started { + if strings.Contains(g.CreatedBy, "scheduler.TestFanOutPerAccount_RespectsParallelismLimit") && + strings.Contains(g.Stack, "scheduler.fanOutPerAccount(") { + runner = g.ID + parked = strings.Contains(g.Stack, "[chan send") && strings.Contains(g.Stack, "errgroup.(*Group).Go(") + } + } + for _, g := range started { + if runner != 0 && g.CreatorID == runner { + workers++ + } + } + return parked, workers } // TestFanOutPerAccount_AllAccountsFail pins the new accountOutcome diff --git a/internal/server/integration_test.go b/internal/server/integration_test.go index 722680b1d..f39c251a8 100644 --- a/internal/server/integration_test.go +++ b/internal/server/integration_test.go @@ -7,8 +7,8 @@ import ( "context" "net/http" "net/http/httptest" + "sync/atomic" "testing" - "time" "github.com/LeanerCloud/cloud-commitments-platform/internal/scheduler" "github.com/LeanerCloud/cloud-commitments-platform/internal/testutil" @@ -81,11 +81,13 @@ func TestScheduledTaskIntegration(t *testing.T) { ctx := testutil.TestContext(t) - // Create application with mock dependencies + // Create application with mock dependencies. The task runs synchronously, + // so counting calls replaces the old 100ms "simulate actual work" sleep, + // which slowed the test without checking anything (issue #36). + var collects atomic.Int32 mockScheduler := &testutil.MockScheduler{ CollectRecommendationsFunc: func(ctx context.Context, ownerToken string) (*scheduler.CollectResult, error) { - // Simulate actual work - time.Sleep(100 * time.Millisecond) + collects.Add(1) return &scheduler.CollectResult{}, nil }, } @@ -98,6 +100,7 @@ func TestScheduledTaskIntegration(t *testing.T) { result, err := app.HandleScheduledTask(ctx, TaskCollectRecommendations, ScheduledTaskParams{}) testutil.AssertNoError(t, err) testutil.AssertTrue(t, result != nil, "Result should not be nil") + testutil.AssertEqual(t, int32(1), collects.Load()) t.Logf("Scheduled task completed successfully") } diff --git a/internal/testutil/goroutines/goroutines.go b/internal/testutil/goroutines/goroutines.go new file mode 100644 index 000000000..ce680b6aa --- /dev/null +++ b/internal/testutil/goroutines/goroutines.go @@ -0,0 +1,182 @@ +// Package goroutines lets tests wait on, or rule out, background goroutines +// without a wall-clock sleep (#36). It reads runtime.Stack(all=true), which +// stops the world, so each snapshot is consistent: a goroutine started by a go +// statement that has already executed is in the snapshot even if the scheduler +// has not run it yet, and it stays there until its function returns. +// +// It is a separate package from testutil, with no dependencies, so packages +// that testutil imports (such as scheduler and purchase) can use it from their +// internal tests without an import cycle. +package goroutines + +import ( + "runtime" + "strconv" + "strings" +) + +// Goroutine is one live goroutine from a stack dump. +type Goroutine struct { + // Stack is the goroutine's block of the dump: the header, every frame + // and the "created by" line. + Stack string + // ID is the runtime's goroutine ID. + ID int64 + // CreatedBy is the function that ran the go statement, from the dump's + // "created by" line, or "" when the dump does not name one (the main + // goroutine, runtime workers). + CreatedBy string + // CreatorID is the ID of the goroutine that ran the go statement, or 0 + // when the dump does not name one. + CreatorID int64 +} + +// Snapshot returns every live goroutine, the caller included. +func Snapshot() []Goroutine { + blocks := strings.Split(string(stackDump(true)), "\n\n") + out := make([]Goroutine, 0, len(blocks)) + for _, block := range blocks { + if g, ok := parse(block); ok { + out = append(out, g) + } + } + return out +} + +// CurrentID returns the calling goroutine's ID. +func CurrentID() int64 { + g, _ := parse(string(stackDump(false))) + return g.ID +} + +// Mark remembers which goroutines were alive, and which goroutine took the +// mark, so goroutines started later can be told apart from older ones. +type Mark struct { + alive map[int64]bool + self int64 +} + +// NewMark records the goroutines alive now. Goroutine IDs are never reused, +// so a goroutine missing from the mark was started after it. +func NewMark() Mark { + m := Mark{alive: map[int64]bool{}, self: CurrentID()} + for _, g := range Snapshot() { + m.alive[g.ID] = true + } + return m +} + +// Started returns the live goroutines that were not alive when the mark was +// taken and whose stack contains every one of substrs. Until a goroutine has +// first run, its frames may name only the compiler's wrapper for the go +// statement, so match on frames it blocks in (or use Spawned) rather than on +// its entry function. +func (m Mark) Started(substrs ...string) []Goroutine { + var out []Goroutine + for _, g := range Snapshot() { + if !m.alive[g.ID] && containsAll(g.Stack, substrs) { + out = append(out, g) + } + } + return out +} + +// StartedBy returns the live goroutines that were not alive when the mark was +// taken and whose go statement ran in a function whose name contains creator, +// for example "api.(*DBRateLimiter).StartCleanupWorker". The creator is known +// from the moment the goroutine exists, whether or not it has run yet. +func (m Mark) StartedBy(creator string) []Goroutine { + var out []Goroutine + for _, g := range m.Started() { + if strings.Contains(g.CreatedBy, creator) { + out = append(out, g) + } + } + return out +} + +// Spawned returns the live goroutines started after the mark by the goroutine +// that took it, directly or through another goroutine in the result. A +// goroutine that has already returned is not in the result; a caller ruling +// out background work must also check that work's side effects. Call it from +// the goroutine's own code, not from a callback another helper runs on a +// goroutine of its own (require.Eventually runs its condition that way): +// that helper goroutine would be in the result. Use StartedBy there. +func (m Mark) Spawned() []Goroutine { + fresh := m.Started() + owned := map[int64]bool{m.self: true} + var out []Goroutine + for grew := true; grew; { + grew = false + for _, g := range fresh { + if !owned[g.ID] && owned[g.CreatorID] { + owned[g.ID] = true + out = append(out, g) + grew = true + } + } + } + return out +} + +func stackDump(all bool) []byte { + buf := make([]byte, 64<<10) + for { + n := runtime.Stack(buf, all) + if n < len(buf) { + return buf[:n] + } + buf = make([]byte, 2*len(buf)) + } +} + +// parse reads one dump block: "goroutine 7 [chan receive]:" then frames, +// with "created by pkg.fn in goroutine 1" near the end for a goroutine +// started by a go statement. +func parse(block string) (Goroutine, bool) { + block = strings.TrimSpace(block) + rest, ok := strings.CutPrefix(block, "goroutine ") + if !ok { + return Goroutine{}, false + } + idText, _, ok := strings.Cut(rest, " ") + if !ok { + return Goroutine{}, false + } + id, err := strconv.ParseInt(idText, 10, 64) + if err != nil { + return Goroutine{}, false + } + g := Goroutine{ID: id, Stack: block} + g.CreatedBy, g.CreatorID = creator(block) + return g, true +} + +// creator reads "created by pkg.fn in goroutine 1" from a dump block. +func creator(block string) (fn string, id int64) { + for _, line := range strings.Split(block, "\n") { + rest, ok := strings.CutPrefix(line, "created by ") + if !ok { + continue + } + fn, idText, ok := strings.Cut(rest, " in goroutine ") + if !ok { + return rest, 0 + } + id, err := strconv.ParseInt(idText, 10, 64) + if err != nil { + return fn, 0 + } + return fn, id + } + return "", 0 +} + +func containsAll(s string, substrs []string) bool { + for _, sub := range substrs { + if !strings.Contains(s, sub) { + return false + } + } + return true +} diff --git a/internal/testutil/goroutines/goroutines_test.go b/internal/testutil/goroutines/goroutines_test.go new file mode 100644 index 000000000..d179fe6ac --- /dev/null +++ b/internal/testutil/goroutines/goroutines_test.go @@ -0,0 +1,76 @@ +package goroutines + +import ( + "testing" +) + +func parkedChild(release <-chan struct{}) { <-release } + +func spawnGrandchild(release <-chan struct{}, started chan<- struct{}) { + go parkedChild(release) + close(started) + <-release +} + +func TestMarkSpawnedSeesUnscheduledAndNestedGoroutines(t *testing.T) { + release := make(chan struct{}) + defer close(release) + + mark := NewMark() + if got := mark.Spawned(); len(got) != 0 { + t.Fatalf("nothing started yet, got %d spawned goroutines", len(got)) + } + + // Read the snapshot right after the go statement, before anything can + // have scheduled the child: it must already be in the dump. (Its frames + // may only name the compiler's go-statement wrapper until it runs, so + // match on the creator rather than on the function name.) + go parkedChild(release) + if got := mark.Spawned(); len(got) != 1 { + t.Fatalf("want the just-started child in the snapshot, got %d", len(got)) + } + + started := make(chan struct{}) + go spawnGrandchild(release, started) + <-started + + if by := mark.StartedBy("goroutines.spawnGrandchild"); len(by) != 1 { + t.Fatalf("want the grandchild by its creator, got %d", len(by)) + } + + got := mark.Spawned() + if len(got) != 3 { + t.Fatalf("want child, middle and grandchild, got %d: %+v", len(got), got) + } + self := CurrentID() + for _, g := range got { + if g.ID == self || g.CreatorID == 0 { + t.Fatalf("unexpected entry %+v", g) + } + } +} + +func TestMarkStartedIgnoresGoroutinesAliveAtTheMark(t *testing.T) { + release := make(chan struct{}) + defer close(release) + go parkedChild(release) + + mark := NewMark() + if got := mark.Spawned(); len(got) != 0 { + t.Fatalf("a goroutine alive at the mark must not count as started after it, got %d", len(got)) + } +} + +func TestParse(t *testing.T) { + block := "goroutine 42 [chan receive]:\nmain.work()\n\t/x/main.go:9 +0x1d\ncreated by main.main in goroutine 1\n\t/x/main.go:5 +0x25" + g, ok := parse(block) + if !ok || g.ID != 42 || g.CreatorID != 1 || g.CreatedBy != "main.main" { + t.Fatalf("parse = %+v, %v", g, ok) + } + if _, ok := parse("not a goroutine"); ok { + t.Fatal("parse accepted a non-goroutine block") + } + if g, _ := parse("goroutine 1 [running]:\nmain.main()"); g.CreatorID != 0 || g.CreatedBy != "" { + t.Fatalf("main goroutine has no creator, got %+v", g) + } +}