From 74135bea9f4f9f79885bc7e6acd5f11ac1922def Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Tue, 20 Oct 2026 15:54:52 +0100 Subject: [PATCH 1/2] test(purchase): assert multi-account fan-out parallelism structurally TestExecuteMultiAccount_RunsAccountsInParallel proved the parallel fan-out by timing it: each PurchaseCommitment mock slept 300ms and the whole executePurchase pipeline had to finish under 500ms, leaving a 200ms margin for scheduling on whatever runner executed it. A loaded CI runner or -race could eat that margin and fail a correct implementation. Replace the wall-time bound with a structural check. Each PurchaseCommitment call bumps an atomic in-flight counter, records the peak, and waits at a barrier until both accounts' calls have arrived. A parallel fan-out opens the barrier immediately and peaks at 2; a serial loop leaves the first call alone at the barrier until its 2s timeout and peaks at 1, so the test fails. The timeout bounds only the failure path; a passing run never waits on it, so the test drops from ~300ms to ~1ms. Closes #38 Co-Authored-By: Claude Opus 5.5 --- internal/purchase/execution_test.go | 66 +++++++++++++++++++---------- 1 file changed, 44 insertions(+), 22 deletions(-) diff --git a/internal/purchase/execution_test.go b/internal/purchase/execution_test.go index 81b6cc931..eaed67615 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,16 +1168,19 @@ 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. Parallelism is +// asserted structurally, not by elapsed wall time (issue #38): each +// PurchaseCommitment call bumps an atomic in-flight counter, records the +// peak, then waits at a barrier until both accounts' calls have arrived. +// Under a parallel fan-out both calls are in flight together, so the barrier +// opens at once and the peak is 2 regardless of runner load. Under a serial +// loop the second call cannot start until the first returns, so the first +// waits out barrierTimeout alone and the peak stays 1, failing the assertion. +// barrierTimeout only bounds that failure path; a passing run never waits on it. func TestExecuteMultiAccount_RunsAccountsInParallel(t *testing.T) { const ( - perCallDelay = 300 * time.Millisecond - serialBoundary = 2 * perCallDelay // serial fan-out lower bound: 600ms - parallelMaxBound = 500 * time.Millisecond + accountCount = 2 + barrierTimeout = 2 * time.Second ) ctx := context.Background() @@ -1238,15 +1242,35 @@ 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. + // Each PurchaseCommitment call counts itself in flight, then waits at the + // barrier for the other account's call. The barrier makes the overlap + // deterministic instead of relying on two sleeps happening to coincide. + var inflight, peak, arrived atomic.Int32 + var barrierTimedOut atomic.Bool + allArrived := make(chan struct{}) + updatePeak := func(cur int32) { + for { + p := peak.Load() + if cur <= p || peak.CompareAndSwap(p, cur) { + return + } + } + } 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) + updatePeak(inflight.Add(1)) + defer inflight.Add(-1) + if arrived.Add(1) == accountCount { + close(allArrived) + } + select { + case <-allArrived: + case <-time.After(barrierTimeout): + barrierTimedOut.Store(true) + } }).Twice() mockStore.On("SavePurchaseHistory", ctx, mock.AnythingOfType("*config.PurchaseHistoryRecord")).Return(nil).Twice() @@ -1265,21 +1289,19 @@ 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' PurchaseCommitment calls + // were in flight at the same time. A serial loop peaks at 1. + assert.GreaterOrEqual(t, peak.Load(), int32(accountCount), + "executeMultiAccount must run accounts in parallel: peak in-flight PurchaseCommitment calls was %d, want %d", + peak.Load(), accountCount) + assert.False(t, barrierTimedOut.Load(), + "a PurchaseCommitment call waited %v at the barrier without the other account's call arriving: accounts ran serially", + barrierTimeout) mockStore.AssertExpectations(t) mockEmail.AssertExpectations(t) From 5c026959fa5a292591fbaa79af066c2b790f67a9 Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Tue, 20 Oct 2026 15:58:57 +0100 Subject: [PATCH 2/2] test(scheduler): require the fan-out to reach its parallelism limit TestFanOutPerAccount_RespectsParallelismLimit only asserted that the peak in-flight count stayed at or below CUDLY_MAX_ACCOUNT_PARALLELISM. A serial fanOutPerAccount (peak 1) satisfied every assertion, so a regression to a serial loop, or a limit that stopped being honored on the low side, stayed green (audit finding A16-011 on #38). The first `limit` calls now wait at a one-shot barrier that opens once `limit` calls are in flight, so a correct fan-out deterministically peaks at exactly the limit. The test asserts peak == limit, keeps the peak <= limit upper bound, and requires that no call timed out at the barrier. A serial or below-limit fan-out leaves the first call alone at the barrier until its 2s timeout, which then opens the barrier for the remaining calls, so the failure is reported after one timeout rather than one per account. A passing run never waits on the timeout. Refs #38 Co-Authored-By: Claude Opus 5.5 --- internal/scheduler/scheduler_test.go | 53 ++++++++++++++++++++++------ 1 file changed, 43 insertions(+), 10 deletions(-) diff --git a/internal/scheduler/scheduler_test.go b/internal/scheduler/scheduler_test.go index 9fcf5169c..08833fa88 100644 --- a/internal/scheduler/scheduler_test.go +++ b/internal/scheduler/scheduler_test.go @@ -6,6 +6,7 @@ import ( "fmt" "net/http" "net/http/httptest" + "sync" "sync/atomic" "testing" "time" @@ -1288,12 +1289,21 @@ 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. +// fanOutPerAccount runs up to CUDLY_MAX_ACCOUNT_PARALLELISM account calls +// in parallel, and no more. The fake fn increments an atomic in-flight +// counter on entry, records the peak and decrements on exit. The first +// `limit` calls also wait at a one-shot barrier that opens once `limit` +// calls have arrived, so a correct fan-out deterministically reaches a peak +// of exactly `limit` regardless of runner load. A serial (or under-limit) +// fan-out never gets `limit` calls in flight: the first call waits out +// barrierTimeout alone and the test fails (issue #38, audit A16-011). +// barrierTimeout only bounds that failure path; a passing run never waits on it. func TestFanOutPerAccount_RespectsParallelismLimit(t *testing.T) { - t.Setenv("CUDLY_MAX_ACCOUNT_PARALLELISM", "3") + const ( + limit = 3 + barrierTimeout = 2 * time.Second + ) + t.Setenv("CUDLY_MAX_ACCOUNT_PARALLELISM", fmt.Sprint(limit)) // 20 accounts to make the bound observable. accounts := make([]config.CloudAccount, 20) @@ -1312,18 +1322,41 @@ func TestFanOutPerAccount_RespectsParallelismLimit(t *testing.T) { } } + var arrived atomic.Int32 + var barrierTimeouts atomic.Int32 + limitReached := make(chan struct{}) + var openBarrier sync.Once + release := func() { openBarrier.Do(func() { close(limitReached) }) } + fn := func(ctx context.Context, acct config.CloudAccount) ([]config.RecommendationRecord, *sweepGap, error) { - cur := inflight.Add(1) - updatePeak(cur) - // Small sleep so concurrent workers genuinely overlap. + updatePeak(inflight.Add(1)) + defer inflight.Add(-1) + if arrived.Add(1) == limit { + release() + } + select { + case <-limitReached: + case <-time.After(barrierTimeout): + // Record the failure, then open the barrier for every later + // call so a serial fan-out fails after one timeout, not twenty. + barrierTimeouts.Add(1) + release() + } + // Short hold so a fan-out that ignored the limit would let extra + // workers overlap the held ones and push the peak above it. time.Sleep(5 * time.Millisecond) - 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.LessOrEqual(t, peak.Load(), int32(limit), "peak in-flight must not exceed CUDLY_MAX_ACCOUNT_PARALLELISM") + assert.Equal(t, int32(limit), peak.Load(), + "fanOutPerAccount must run CUDLY_MAX_ACCOUNT_PARALLELISM accounts in parallel: peak in-flight was %d, want %d", + peak.Load(), limit) + assert.Zero(t, barrierTimeouts.Load(), + "%d account call(s) waited %v at the barrier without %d calls in flight: the fan-out ran serially or below the limit", + barrierTimeouts.Load(), barrierTimeout, limit) assert.Equal(t, len(accounts), outcome.SucceededCount) assert.Zero(t, outcome.FailedCount) }