From 9bb0d3a775183c74e5b9e717100db2ae5877c3bb Mon Sep 17 00:00:00 2001 From: Jorge Manrubia Date: Sat, 19 Sep 2026 18:36:16 +0200 Subject: [PATCH 1/2] Wait for the oversubscription instead of hoping for it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit TestGateQueuesOversubscribedInvocationsWithinTheSlotLimit asserted that five of fifteen callers found every slot taken. CI got one. The assertion was mine, from #774, and it was wrong in the way that card was about: it described a race and called it a fact. A barrier at the start lines up the arrivals, which is what the last change fixed. It does not keep them together afterwards. A child released and then descheduled arrives after an earlier holder has already finished and takes the slot it freed, so it never waits — and on a two-core runner under fifteen processes that gap is wide. So the parent now holds the slot table full. Each child announces the first time the bulkhead turns it away, and announces holding a slot and then waits before releasing it; the parent blocks until ten are held and five have been refused, and only then lets the holders go. The oversubscription is something the test waited for rather than something it hoped for, and five is exact. It waits rather than polls on purpose. A bulkhead that stopped excluding anyone cannot make this test pass with a smaller number — it hangs and fails on the test's own deadline, loudly, instead of reporting one and looking like a flake. Evidence. Spreading the arrivals 50ms apart, which is what a loaded runner does to fifteen forks, fails the old version with peak 4 against 10 and slotWaits 0 against 5 — the CI symptom, reproduced deliberately — and leaves the new one passing, four times out of four. Removing the concurrency limit entirely makes the new test hang and fail on its deadline rather than pass. Five runs on a saturated core under race instrumentation pass. Main has had one run since #774 merged and it was green, so this was a latent flake rather than a broken main; it surfaced on a pull request that had not touched this package. Co-Authored-By: Claude Opus 5 (1M context) --- internal/resilience/gate_test.go | 151 ++++++++++++++++++++++++++++--- 1 file changed, 140 insertions(+), 11 deletions(-) diff --git a/internal/resilience/gate_test.go b/internal/resilience/gate_test.go index 145eed250..cd9442c26 100644 --- a/internal/resilience/gate_test.go +++ b/internal/resilience/gate_test.go @@ -36,6 +36,10 @@ import ( // BH_BARRIER=1 makes it print READY and hold before its first // operation until the parent sends it a line, so the parent can set up the // condition under test with every child already running. +// BH_HOLD_BARRIER=1 additionally prints WAITING the first time the bulkhead +// turns it away, and HELD once it holds a slot, then waits for a second +// line before releasing it — so a parent can hold the slot table full until +// every other child has been refused. // BH_MODE=linger churns BH_OPS acquire/release pairs, prints // DONE, then stays alive until stdin closes so the parent can check that its // releases were not lost while the process still counts as a live holder. @@ -62,9 +66,19 @@ func TestHelperProcess(t *testing.T) { // The gate says when it waits. Each operation raises its own flag at // most once, so the counts are operations that queued and not sleeps // they took getting through. + // + // Under BH_HOLD_BARRIER the first refusal is announced as it + // happens, so the parent can wait for the contention rather than + // hope for it. + holdBarrier := os.Getenv("BH_HOLD_BARRIER") == "1" var sleptOnBucket, sleptOnSlot bool hooks.rateLimiter.onWait = func() { sleptOnBucket = true } - hooks.bulkhead.onWait = func() { sleptOnSlot = true } + hooks.bulkhead.onWait = func() { + if holdBarrier && !sleptOnSlot { + fmt.Println("WAITING") + } + sleptOnSlot = true + } peak, queued, slotWaits := 0, 0, 0 for range ops { @@ -83,6 +97,12 @@ func TestHelperProcess(t *testing.T) { if inUse, err := hooks.bulkhead.InUse(); err == nil { peak = max(peak, inUse) } + // Holding until the parent says so is what makes the slot table + // full for as long as the test needs it full. + if holdBarrier { + fmt.Println("HELD") + _, _ = bufio.NewReader(os.Stdin).ReadString('\n') + } time.Sleep(hold) hooks.OnOperationEnd(ctx, op, nil, hold) fmt.Println("OK") @@ -242,6 +262,108 @@ func runBarrieredInvocations(t *testing.T, n int, env helperEnv) invocationTally return tally(all) } +// runSlotOversubscription starts n children against a bulkhead of `slots` +// and does not let go of the slot table until the oversubscription has +// actually happened: every child has either taken a slot or been turned away +// from one. +// +// The single barrier is not enough for that. It lines up the start, but a +// child released and then descheduled can arrive after an earlier holder has +// finished, take the slot it freed, and never wait — which is how "exactly +// five of fifteen waited" came out as one on a two-core runner. Holding the +// slots until every other child has reported being refused makes the +// contention a fact the test waited for rather than a race it hoped for. +// +// It waits rather than polls, so a bulkhead that stopped excluding anyone +// fails here by never finishing, under the test's own deadline, instead of +// passing with a smaller number. +func runSlotOversubscription(t *testing.T, n, slots int, env helperEnv) invocationTally { + t.Helper() + require.Greater(t, n, slots, "there is no oversubscription to wait for") + barriered := helperEnv{"BH_BARRIER": "1", "BH_HOLD_BARRIER": "1"} + maps.Copy(barriered, env) + + type child struct { + cmd *exec.Cmd + stdin io.WriteCloser + stderr *bytes.Buffer + } + kids := make([]child, 0, n) + lines := make(chan string, 4*n) + var wg sync.WaitGroup + for range n { + cmd := helperCommand(t, barriered) + stdin, err := cmd.StdinPipe() + require.NoError(t, err) + stdout, err := cmd.StdoutPipe() + require.NoError(t, err) + stderr := &bytes.Buffer{} + cmd.Stderr = stderr + require.NoError(t, cmd.Start()) + kids = append(kids, child{cmd, stdin, stderr}) + wg.Add(1) + go func(r io.Reader) { + defer wg.Done() + sc := bufio.NewScanner(r) + for sc.Scan() { + lines <- sc.Text() + } + }(stdout) + } + go func() { wg.Wait(); close(lines) }() + + // Everything the children say, tallied as it arrives; the report lines + // are kept for the caller. + var report []string + await := func(want map[string]int) { + got := map[string]int{} + for { + done := true + for k, n := range want { + if got[k] < n { + done = false + } + } + if done { + return + } + line, ok := <-lines + if !ok { + require.FailNowf(t, "a child stopped early", + "waiting for %v, saw %v; output so far: %v", want, got, report) + } + if isReportLine(line) { + report = append(report, line) + } + got[line]++ + } + } + + release := func(what string) { + for _, k := range kids { + _, err := io.WriteString(k.stdin, what+"\n") + require.NoError(t, err) + } + } + + await(map[string]int{"READY": n}) + release("go") + // Every slot taken, and every child that could not have one saying so. + await(map[string]int{"HELD": slots, "WAITING": n - slots}) + release("release") + + for line := range lines { + if isReportLine(line) { + report = append(report, line) + } + } + for _, k := range kids { + _ = k.stdin.Close() + require.NoError(t, k.cmd.Wait(), k.stderr.String()) + } + return tally(report) +} + // The smoke-test shape: ten parallel workers making eighty calls between // them through the production defaults. Before queueing, the shared 50-token // bucket drained in the first half second and every later call failed with @@ -328,21 +450,27 @@ func TestGateReportsNoQueueingWhenNothingHadToWait(t *testing.T) { // slot table fills to exactly ten and no further, so five of them were made // to wait for a second round rather than being admitted alongside the first. // -// "Simultaneous" is why the children come off a barrier. Left to race each -// other into the gate they arrive as fast as the box can fork, and on a slow -// one the first ten are done before the last five start — at which point -// nothing is oversubscribed and the test passes having measured nothing. The -// old evidence that the overflow waited was that the run took at least two -// holds, which slow spawning satisfies all by itself. The bulkhead reports -// it directly now: five of the fifteen found every slot taken and polled -// for one. +// "Simultaneous" is the whole difficulty, and a barrier at the start does +// not give it. Left to race each other into the gate the children arrive as +// fast as the box can fork, and on a slow one the first ten are done before +// the last five start — nothing is oversubscribed and the test passes having +// measured nothing. Lining up the start is not enough either: a child +// released and then descheduled can arrive after a holder has finished and +// take the slot it freed. That is not a guess. It is what CI did to the +// first version of this test, which asserted that five of fifteen were +// refused and got one. +// +// So the parent holds the slot table full until every child that cannot +// have a slot has said so, and only then lets the holders go. The +// oversubscription is a fact the test waited for, and five is then exact +// rather than likely. func TestGateQueuesOversubscribedInvocationsWithinTheSlotLimit(t *testing.T) { dir := t.TempDir() cfg := DefaultConfig() const invocations = 15 - tl := runBarrieredInvocations(t, invocations, helperEnv{ + tl := runSlotOversubscription(t, invocations, cfg.Bulkhead.MaxConcurrent, helperEnv{ "BH_STATE_DIR": dir, "BH_MODE": "ops", "BH_OPS": "1", - "BH_HOLD": (150 * time.Millisecond).String(), "BH_MAX_TOKENS": "100", + "BH_HOLD": "10ms", "BH_MAX_TOKENS": "100", }) assert.Equal(t, invocations, tl.ok, "every invocation succeeds: %v", tl.rejections) @@ -351,6 +479,7 @@ func TestGateQueuesOversubscribedInvocationsWithinTheSlotLimit(t *testing.T) { "the slot table filled to exactly MaxConcurrent") assert.Equal(t, invocations-cfg.Bulkhead.MaxConcurrent, tl.slotWaits, "the overflow — and only the overflow — found every slot taken and polled for one") + assert.LessOrEqual(t, tl.peak, cfg.Bulkhead.MaxConcurrent, "never more than MaxConcurrent live holders") state, err := NewStore(dir).Load() require.NoError(t, err) From 6bc2372c04c6b1bec06e8b466c8b14b7084276da Mon Sep 17 00:00:00 2001 From: Jorge Manrubia Date: Sat, 19 Sep 2026 18:44:45 +0200 Subject: [PATCH 2/2] The slot count the parent waits on is the one the children run under MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit runSlotOversubscription took a slots argument and did not use it for anything but its own arithmetic: the children went on running under DefaultConfig's ten. Any other value left the parent waiting for a split that cannot happen. That matters more here than an ignored parameter usually would. This helper waits for the oversubscription rather than sampling it, which is the point of the change — but it means a mismatch no longer shows up as a wrong number. It shows up as the test deadline, with a message about a timeout and nothing about the cause. Confirmed before fixing: eight callers against five slots hung for the full forty-five seconds. The count now goes to the children through BH_MAX_CONCURRENT, beside BH_MAX_TOKENS, and the helper refuses a caller environment that would override it. Its own test, because the failure mode is a deadline rather than an assertion: five slots for eight callers, five hold, three are refused. Removing the wiring hangs it. Co-Authored-By: Claude Opus 5 (1M context) --- internal/resilience/gate_test.go | 37 +++++++++++++++++++++++++++++++- 1 file changed, 36 insertions(+), 1 deletion(-) diff --git a/internal/resilience/gate_test.go b/internal/resilience/gate_test.go index cd9442c26..f277d261b 100644 --- a/internal/resilience/gate_test.go +++ b/internal/resilience/gate_test.go @@ -40,6 +40,9 @@ import ( // turns it away, and HELD once it holds a slot, then waits for a second // line before releasing it — so a parent can hold the slot table full until // every other child has been refused. +// BH_MAX_TOKENS and BH_MAX_CONCURRENT override the bucket and the slot +// count; a parent that waits on a split has to be able to set the limit it +// is waiting on. // BH_MODE=linger churns BH_OPS acquire/release pairs, prints // DONE, then stays alive until stdin closes so the parent can check that its // releases were not lost while the process still counts as a live holder. @@ -54,6 +57,9 @@ func TestHelperProcess(t *testing.T) { if tokens, err := strconv.ParseFloat(os.Getenv("BH_MAX_TOKENS"), 64); err == nil { cfg.RateLimiter.MaxTokens = tokens } + if slots, err := strconv.Atoi(os.Getenv("BH_MAX_CONCURRENT")); err == nil { + cfg.Bulkhead.MaxConcurrent = slots + } switch os.Getenv("BH_MODE") { case "ops": @@ -280,8 +286,17 @@ func runBarrieredInvocations(t *testing.T, n int, env helperEnv) invocationTally func runSlotOversubscription(t *testing.T, n, slots int, env helperEnv) invocationTally { t.Helper() require.Greater(t, n, slots, "there is no oversubscription to wait for") - barriered := helperEnv{"BH_BARRIER": "1", "BH_HOLD_BARRIER": "1"} + // The limit the parent waits on and the limit the children run under + // have to be the same number. Passing slots without setting it would + // leave this waiting for a split that cannot happen, and the failure + // would be a test deadline pointing nowhere near the cause. + barriered := helperEnv{ + "BH_BARRIER": "1", "BH_HOLD_BARRIER": "1", + "BH_MAX_CONCURRENT": strconv.Itoa(slots), + } maps.Copy(barriered, env) + require.Equal(t, strconv.Itoa(slots), barriered["BH_MAX_CONCURRENT"], + "the caller's environment must not override the slot count the wait is built on") type child struct { cmd *exec.Cmd @@ -486,6 +501,26 @@ func TestGateQueuesOversubscribedInvocationsWithinTheSlotLimit(t *testing.T) { assert.Empty(t, state.Bulkhead.ActivePIDs, "every slot released") } +// The slot count the parent waits on is the slot count the children run +// under. It is worth its own test because getting it wrong does not produce +// a wrong number any more — the parent waits for a split that cannot +// happen, and the only symptom is the test deadline, pointing nowhere near +// the cause. Five slots for eight callers: five hold, three are refused. +func TestGateSlotLimitIsTheOneTheTestWaitsOn(t *testing.T) { + dir := t.TempDir() + const callers, slots = 8, 5 + + tl := runSlotOversubscription(t, callers, slots, helperEnv{ + "BH_STATE_DIR": dir, "BH_MODE": "ops", "BH_OPS": "1", + "BH_HOLD": "10ms", "BH_MAX_TOKENS": "100", + }) + + assert.Equal(t, callers, tl.ok, "every caller gets through eventually: %v", tl.rejections) + assert.Zero(t, tl.rejected) + assert.Equal(t, slots, tl.peak, "the limit that held is the one that was asked for, not the default") + assert.Equal(t, callers-slots, tl.slotWaits) +} + // A dozen processes churning the lock must not lose each other's releases: // while every child is still alive (so a leaked PID would not be swept as // dead) the slot table is empty.