test(execution): make TestFanOut_ContextCancelled_BlockedSemaphore deterministic

internal/execution/fanout.go

@@ -54,6 +54,15 @@ func FanOut[T any](
 	return FanOutWithConcurrency(ctx, accountIDs, fn, DefaultMaxConcurrency)
 }
 
+// semBlockedHookKey is the context key for an optional test-only
+// synchronization hook. When a func() is stored under this key,
+// FanOutWithConcurrency invokes it each time an item is recorded as canceled at
+// the semaphore-acquire boundary (the <-ctx.Done() branch below). It exists
+// purely so tests can synchronize deterministically with that moment; no
+// production caller sets it, so the branch is inert (a single context lookup on
+// the already-canceled path) outside tests.
+type semBlockedHookKey struct{}
+
 // FanOutWithConcurrency is like FanOut but with an explicit concurrency limit.
 func FanOutWithConcurrency[T any](
 	ctx context.Context,
@@ -82,6 +91,13 @@ func FanOutWithConcurrency[T any](
 		case <-ctx.Done():
 			results[i] = Result[T]{AccountID: id, Err: ctx.Err()}
 			wg.Done()
+			// Optional test-only synchronization hook (unset in production):
+			// signals that a queued item was recorded as canceled at the
+			// semaphore boundary, letting tests order a subsequent unblock
+			// deterministically. See semBlockedHookKey.
+			if hook, ok := ctx.Value(semBlockedHookKey{}).(func()); ok {
+				hook()
+			}
 			continue
 		}
 		go func(idx int, accountID string) {

internal/execution/fanout_test.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"sort"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -105,7 +106,20 @@ func TestFanOut_ContextCancelled_BlockedSemaphore(t *testing.T) {
 	started := make(chan struct{}, 1)
 	release := make(chan struct{}, 1)
 
-	ctx, cancel := context.WithCancel(context.Background())
+	// canceledRecorded is closed by the fan-out launch loop the moment the
+	// queued "second" item is recorded as ctx.Canceled at the semaphore
+	// boundary (via the semBlockedHookKey hook). This gives a deterministic
+	// happens-before edge: the test only releases the blocker AFTER the
+	// cancellation has been recorded, so the freed semaphore slot can never
+	// race the launch loop into running "second". Without it, cancel() and the
+	// slot-release both make the loop's select ready, and Go's uniform-random
+	// pick would intermittently launch "second" (recording a nil error).
+	canceledRecorded := make(chan struct{})
+	ctx, cancel := context.WithCancel(
+		context.WithValue(context.Background(), semBlockedHookKey{}, func() {
+			close(canceledRecorded)
+		}),
+	)
 	defer cancel()
 
 	// Collect results in a separate goroutine so this test goroutine can
@@ -130,9 +144,26 @@ func TestFanOut_ContextCancelled_BlockedSemaphore(t *testing.T) {
 		done <- outcome{res}
 	}()
 
-	// Wait until the first goroutine holds the semaphore, then cancel.
+	// Wait until the first goroutine holds the semaphore, then cancel. Once
+	// canceledRecorded fires, the loop has committed "second" to the
+	// <-ctx.Done() branch, so releasing the blocker is now race-free.
 	<-started
 	cancel()
+	select {
+	case <-canceledRecorded:
+		// Correct behavior: the loop honored ctx cancellation at the
+		// semaphore boundary. Fires within microseconds, so the deadline
+		// below is never reached on a healthy tree (no flakiness).
+	case <-time.After(10 * time.Second):
+		// A regression to unconditional `sem <- struct{}{}` would block the
+		// launch loop on the second item forever (the blocker never releases
+		// because we never send release), so canceledRecorded never fires.
+		// Fail fast with a clear message instead of hanging for the whole
+		// go-test timeout.
+		t.Fatal("timed out waiting for the queued item to be recorded as " +
+			"canceled; FanOutWithConcurrency likely blocked on the semaphore " +
+			"instead of honoring ctx cancellation")
+	}
 	release <- struct{}{} // let the first goroutine finish so FanOut can return
 
 	o := <-done