test(purchase): assert multi-account fan-out parallelism structurally

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)

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)
 }