From c2c2355a906781db28baaed47618353101ed0f8f Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Fri, 17 Jul 2026 01:58:34 +0300 Subject: [PATCH] test(execution): make TestFanOut_ContextCancelled_BlockedSemaphore deterministic The test intermittently reddened the Unit Tests CI job (observed 2/2 on PR #1277, a cancelled->canceled rename that does not touch fan-out logic). Root cause is a select-both-ready race in the test harness, not a production defect: after cancelling the context the test released the blocking goroutine immediately, freeing the single semaphore slot. That made both cases of the launch loop's `select { case sem <- {}: case <-ctx.Done(): }` ready, and Go's uniform-random pick would occasionally launch "second" (recording a nil error) instead of the ctx.Done branch, failing require.Error on the second item. Fix it deterministically by giving the loop's ctx.Done branch an optional, context-scoped synchronization hook (inert in production: a single ctx.Value lookup on the already-canceled path, no caller sets it). The test now waits for the queued item to be recorded as canceled before releasing the blocker, so the freed slot can never race the loop into running "second". No time.Sleep is used for synchronization; the 10s select failsafe is only a deadline guard that fails fast with a clear message on a genuine regression (never reached on a healthy tree). Verified: -race -count=50 green; 18000 runs at GOMAXPROCS=2 (the config that reproduced the flake pre-fix) green; a mutation to the pre-fix unconditional `sem <- {}` acquire is caught in 10s. golangci-lint v2.10.1 and gocyclo clean. --- internal/execution/fanout.go | 16 ++++++++++++++ internal/execution/fanout_test.go | 35 +++++++++++++++++++++++++++++-- 2 files changed, 49 insertions(+), 2 deletions(-) diff --git a/internal/execution/fanout.go b/internal/execution/fanout.go index f3f36a8b0..fe501905a 100644 --- a/internal/execution/fanout.go +++ b/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) { diff --git a/internal/execution/fanout_test.go b/internal/execution/fanout_test.go index d9b8105b0..ace0efb54 100644 --- a/internal/execution/fanout_test.go +++ b/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