From 09fe952530084cc7ae48b6078baeb6532ae4e69a Mon Sep 17 00:00:00 2001 From: Jorge Manrubia Date: Sat, 19 Sep 2026 11:52:23 +0200 Subject: [PATCH 1/5] Assert the gate and the block, not the machine they ran on MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two tests in internal/resilience were failing on the clock rather than on anything they are named after, and between them they blocked an unrelated pull request's Race Detection run. TestGateQueuesTenParallelWorkersThroughTheDefaults gave eighty child process invocations a budget of DefaultMaxWait, which is one operation's queueing budget and no statement about a run of eighty. The run is bounded below by three seconds of deliberate token refill and above by nothing but the box, so on a contended one it spent the margin on the machine: 3 of 3 red on a saturated core, and red on CI at 10.73s and again at 10.97s with the gate having behaved perfectly every time. The queueing claim never needed a stopwatch. Eighty calls against a fifty-token bucket is the claim: thirty of them cannot be served from the starting bucket at all, so "every call succeeded" is only reachable by waiting for refills. That is now asserted against the config rather than written out as literals, and the wall-clock bound is gone. Its sibling TestGateQueuesOversubscribedInvocationsWithinTheSlotLimit had the same upper bound and loses it too; its lower bound, that the overflow waited for a second round, is a real claim and stays. TestRateLimiterRetryAfterOnlyUpdatesIfLater set a 200ms block, set a 50ms one, and asserted 150ms still remained — an assertion that fewer than fifty milliseconds of real time passed between two statements. It now asserts the stored deadline itself, in both directions: a nearer deadline does not shorten the block, a further one does extend it, so neither half can pass for free. Evidence: under a saturated core with race instrumentation the old gate test failed 10.120778248s > 10s with 80 ok, 0 rejected and peak 10; the new one passes 3 of 3 under the same load at 10.06-10.14s. Mutating the rate limiter to stop queueing for refills still reds it (51 ok, 29 rejected). Mutating SetRetryAfter to accept any deadline reds the retry-after test, and a 60ms stall injected between the two statements reds the old assertion while the new one holds. Co-Authored-By: Claude Opus 5 (1M context) --- internal/resilience/gate_test.go | 24 ++++++++++++---- internal/resilience/rate_limiter_test.go | 36 +++++++++++++++++++----- 2 files changed, 47 insertions(+), 13 deletions(-) diff --git a/internal/resilience/gate_test.go b/internal/resilience/gate_test.go index 69a3ff1a3..ec29a954c 100644 --- a/internal/resilience/gate_test.go +++ b/internal/resilience/gate_test.go @@ -151,15 +151,28 @@ func runWorkers(t *testing.T, workers, calls int, env helperEnv) invocationTally // through the production defaults. Before queueing, the shared 50-token // bucket drained in the first half second and every later call failed with // "rate limit exceeded" while the server had never answered 429. +// +// Eighty calls against a fifty-token bucket is what makes this a test of +// queueing: thirty of them cannot be served from the starting bucket at all, +// so "every call succeeded" is only reachable by waiting for refills. The +// run is therefore bounded below by the refill, never above: it used to +// assert that the whole thing — eighty process spawns and three seconds of +// deliberate refill — finished inside DefaultMaxWait, which is one +// operation's queueing budget and no statement about this run. On a +// contended box that margin is spent on the machine, and it failed on CI at +// 10.73s and again at 10.97s with the gate having behaved perfectly. func TestGateQueuesTenParallelWorkersThroughTheDefaults(t *testing.T) { + const workers, calls = 10, 8 + cfg := DefaultConfig() + require.Greater(t, float64(workers*calls), cfg.RateLimiter.MaxTokens, + "the run has to outlast the bucket or nothing queues") + dir := t.TempDir() - start := time.Now() - tl := runWorkers(t, 10, 8, helperEnv{"BH_STATE_DIR": dir, "BH_MODE": "ops", "BH_OPS": "1", "BH_HOLD": "10ms"}) + tl := runWorkers(t, workers, calls, helperEnv{"BH_STATE_DIR": dir, "BH_MODE": "ops", "BH_OPS": "1", "BH_HOLD": "10ms"}) - assert.Equal(t, 80, tl.ok, "every call succeeds: %v", tl.rejections) + assert.Equal(t, workers*calls, tl.ok, "every call succeeds: %v", tl.rejections) assert.Zero(t, tl.rejected) - assert.LessOrEqual(t, tl.peak, 10, "never more than MaxConcurrent live holders") - assert.Less(t, time.Since(start), DefaultMaxWait) + assert.LessOrEqual(t, tl.peak, cfg.Bulkhead.MaxConcurrent, "never more than MaxConcurrent live holders") } // Fifteen simultaneous invocations against ten slots: nobody fails, nobody @@ -178,7 +191,6 @@ func TestGateQueuesOversubscribedInvocationsWithinTheSlotLimit(t *testing.T) { assert.Equal(t, 15, tl.ok, "every invocation succeeds: %v", tl.rejections) assert.LessOrEqual(t, tl.peak, 10, "never more than MaxConcurrent live holders") assert.GreaterOrEqual(t, elapsed, 2*hold, "the overflow waited for a slot") - assert.Less(t, elapsed, DefaultMaxWait) state, err := NewStore(dir).Load() require.NoError(t, err) diff --git a/internal/resilience/rate_limiter_test.go b/internal/resilience/rate_limiter_test.go index 93154532b..0909b1da2 100644 --- a/internal/resilience/rate_limiter_test.go +++ b/internal/resilience/rate_limiter_test.go @@ -137,6 +137,15 @@ func TestRateLimiterSetRetryAfter(t *testing.T) { assert.False(t, allowed, "expected request to be rejected during retry-after period") } +// A Retry-After block only ever moves later: a second 429 carrying a nearer +// deadline must not shorten the one already stored. +// +// This is a decision about two timestamps, so it is asserted on the stored +// timestamp and not on a stopwatch. It used to set a 200ms block, set a 50ms +// one, and assert that 150ms still remained — which is an assertion that +// fewer than fifty milliseconds of real time passed between two statements. +// Under race instrumentation on a two-core runner that is not safe, and it +// failed on CI having said nothing about the behavior it is named after. func TestRateLimiterRetryAfterOnlyUpdatesIfLater(t *testing.T) { dir := t.TempDir() store := NewStore(dir) @@ -146,15 +155,28 @@ func TestRateLimiterRetryAfterOnlyUpdatesIfLater(t *testing.T) { TokensPerRequest: 1, }) - // Set initial retry-after - rl.SetRetryAfterDuration(200 * time.Millisecond) + base := time.Now() + later, earlier := base.Add(time.Minute), base.Add(time.Second) - // Try to set an earlier time - rl.SetRetryAfterDuration(50 * time.Millisecond) + require.NoError(t, rl.SetRetryAfter(later)) + require.NoError(t, rl.SetRetryAfter(earlier)) - // Should still have ~200ms remaining (only updated if later) - remaining, _ := rl.RetryAfterRemaining() - assert.True(t, remaining >= 150*time.Millisecond, "expected ~200ms remaining, got %v", remaining) + state, err := store.Load() + require.NoError(t, err) + assert.True(t, state.RateLimiter.RetryAfterUntil.Equal(later), + "the nearer deadline shortened the block: stored %v, want %v", + state.RateLimiter.RetryAfterUntil, later) + + // And the other direction, so that "unchanged" is not passing for free: + // a deadline further out does move the block. + furtherOut := base.Add(2 * time.Minute) + require.NoError(t, rl.SetRetryAfter(furtherOut)) + + state, err = store.Load() + require.NoError(t, err) + assert.True(t, state.RateLimiter.RetryAfterUntil.Equal(furtherOut), + "a later deadline did not extend the block: stored %v, want %v", + state.RateLimiter.RetryAfterUntil, furtherOut) } func TestRateLimiterReset(t *testing.T) { From 2ae9830e2f7681fa5baa3f0684b58561bac5bd23 Mon Sep 17 00:00:00 2001 From: Jorge Manrubia Date: Sat, 19 Sep 2026 12:10:23 +0200 Subject: [PATCH 2/5] Release the gate's children from a barrier, and watch them queue MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Dropping the wall-clock bound left both multi-process gate tests able to pass without the gate ever having queued anything, which is the same defect wearing the other face: the assertion runs over nothing and is green either way. The cause is that the children raced each other into the gate, so how hard the gate was pushed was decided by how fast the box could fork. On a slow one the ten workers arrive spread out, the bucket refills between them, and every call is served from a full bucket; in the oversubscribed test the first ten invocations finish before the last five start, and nothing is oversubscribed. Both would then pass with a gate that failed fast instead of queueing. The children now come off a barrier. Each announces itself and holds before its first operation, and nobody is released until the last one has arrived, so the load the gate sees is the load the test asked for. Where the eighty calls used to be eighty process spawns paced by the machine, they are now ten already-running processes making eight each. Queueing is then observed rather than inferred. Each child reads the bucket before it gates and reports how many of its operations were admitted out of an empty one, which is only reachable by waiting for a refill — a gate that failed fast would have rejected that same operation. The count is a floor, not an accounting, since a call that reads the bucket just after a refill lands is served without waiting; what it rules out is a run where the wait path was never entered. The oversubscribed test asserts the slot table filled to exactly MaxConcurrent, which says the overflow waited, where the old evidence was that the run took two holds — something slow spawning satisfies on its own. Evidence, all on the saturated single core with race instrumentation that made the old test fail 10.12s > 10s: 3 of 3 green as written, and 3 of 3 red with the rate limiter mutated to stop queueing for refills and the bulkhead mutated to stop queueing for slots. Removing the concurrency limit altogether puts the peak at 15 against 10. Off the barrier, the oversubscription is now exact: the fail-fast bulkhead rejects precisely five of fifteen. The package is also 4 seconds faster under -race, 19.4s against 23.7s, from the seventy process spawns that are no longer needed. Co-Authored-By: Claude Opus 5 (1M context) --- internal/resilience/gate_test.go | 222 +++++++++++++++++++++++-------- 1 file changed, 163 insertions(+), 59 deletions(-) diff --git a/internal/resilience/gate_test.go b/internal/resilience/gate_test.go index ec29a954c..e2f4d1828 100644 --- a/internal/resilience/gate_test.go +++ b/internal/resilience/gate_test.go @@ -2,11 +2,11 @@ package resilience import ( "bufio" - "bytes" "context" "errors" "fmt" "io" + "maps" "os" "os/exec" "strconv" @@ -25,8 +25,13 @@ import ( // each child is one CLI invocation against the shared store in BH_STATE_DIR. // // BH_MODE=ops runs BH_OPS gated operations, each held for BH_HOLD, and prints -// OK or REJECT per operation plus PEAK, the most live slot holders it saw -// while holding one. BH_MODE=linger churns BH_OPS acquire/release pairs, prints +// OK or REJECT per operation, plus PEAK, the most live slot holders it saw +// while holding one, and QUEUED, how many of its operations were admitted +// out of an empty bucket and so can only have got there by waiting for a +// refill. 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_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. func TestHelperProcess(t *testing.T) { @@ -45,13 +50,27 @@ func TestHelperProcess(t *testing.T) { case "ops": hooks := NewGatingHooksFromConfig(store, cfg) op := basecamp.OperationInfo{Service: "Projects", Operation: "List"} - peak := 0 + if os.Getenv("BH_BARRIER") == "1" { + fmt.Println("READY") + _, _ = bufio.NewReader(os.Stdin).ReadString('\n') + } + peak, queued := 0, 0 for range ops { + // Read the bucket before gating: an operation admitted when + // there was nothing to admit it with waited for a refill, which + // is the queueing this test exists to see. A gate that failed + // fast would have rejected that same operation instead. + tokens, tokensErr := hooks.rateLimiter.Tokens() + starved := tokensErr == nil && tokens < cfg.RateLimiter.TokensPerRequest + ctx, err := hooks.OnOperationGate(context.Background(), op) if err != nil { fmt.Printf("REJECT %v\n", err) continue } + if starved { + queued++ + } if inUse, err := hooks.bulkhead.InUse(); err == nil { peak = max(peak, inUse) } @@ -60,6 +79,7 @@ func TestHelperProcess(t *testing.T) { fmt.Println("OK") } fmt.Printf("PEAK %d\n", peak) + fmt.Printf("QUEUED %d\n", queued) case "linger": bh := NewBulkhead(store, cfg.Bulkhead) rl := NewRateLimiter(store, cfg.RateLimiter) @@ -74,6 +94,17 @@ func TestHelperProcess(t *testing.T) { os.Exit(0) } +// isReportLine picks the helper's tallied output out of the test binary's +// own chatter. +func isReportLine(l string) bool { + for _, prefix := range []string{"OK", "REJECT", "PEAK", "QUEUED"} { + if strings.HasPrefix(l, prefix) { + return true + } + } + return false +} + type helperEnv map[string]string func helperCommand(t *testing.T, env helperEnv) *exec.Cmd { @@ -86,25 +117,9 @@ func helperCommand(t *testing.T, env helperEnv) *exec.Cmd { return cmd } -// runInvocation runs one child to completion and returns its report lines. -func runInvocation(t *testing.T, env helperEnv) []string { - t.Helper() - cmd := helperCommand(t, env) - var out bytes.Buffer - cmd.Stdout, cmd.Stderr = &out, &out - require.NoError(t, cmd.Run(), out.String()) - var lines []string - for _, l := range strings.Split(out.String(), "\n") { - if strings.HasPrefix(l, "OK") || strings.HasPrefix(l, "REJECT") || strings.HasPrefix(l, "PEAK") { - lines = append(lines, l) - } - } - return lines -} - type invocationTally struct { - ok, rejected, peak int - rejections []string + ok, rejected, peak, queued int + rejections []string } func tally(lines []string) invocationTally { @@ -119,78 +134,167 @@ func tally(lines []string) invocationTally { case strings.HasPrefix(l, "PEAK"): n, _ := strconv.Atoi(strings.TrimPrefix(l, "PEAK ")) tl.peak = max(tl.peak, n) + case strings.HasPrefix(l, "QUEUED"): + n, _ := strconv.Atoi(strings.TrimPrefix(l, "QUEUED ")) + tl.queued += n } } return tl } -// runWorkers runs `workers` goroutines that each perform `calls` sequential -// child invocations, the shape of a parallel smoke test, and tallies the lot. -func runWorkers(t *testing.T, workers, calls int, env helperEnv) invocationTally { +// runBarrieredInvocations starts n children and waits for every one of them +// to be up and holding before releasing them together. +// +// Spawning a process takes as long as the machine takes, so a test that +// lets the children race each other into the gate is really measuring how +// fast they start: on a slow box they arrive one at a time, the bucket +// refills between them, and the gate is never asked to queue anything. The +// barrier removes the machine from that question. Every child is already +// running when the first one gates, so the load the gate sees is the load +// the test asked for and not the load the box could deliver. +func runBarrieredInvocations(t *testing.T, n int, env helperEnv) invocationTally { t.Helper() + barriered := helperEnv{"BH_BARRIER": "1"} + maps.Copy(barriered, env) + + type child struct { + cmd *exec.Cmd + stdin io.WriteCloser + stdout io.ReadCloser + } + kids := make([]child, 0, n) + ready := make(chan error, n) + for range n { + cmd := helperCommand(t, barriered) + stdin, err := cmd.StdinPipe() + require.NoError(t, err) + stdout, err := cmd.StdoutPipe() + require.NoError(t, err) + require.NoError(t, cmd.Start()) + kids = append(kids, child{cmd, stdin, stdout}) + } + + // Each child announces itself on its own line; nobody is released until + // the last one has. + scanners := make([]*bufio.Scanner, len(kids)) + for i, k := range kids { + scanners[i] = bufio.NewScanner(k.stdout) + go func(sc *bufio.Scanner) { + for sc.Scan() { + if sc.Text() == "READY" { + ready <- nil + return + } + } + ready <- errors.New("child exited before reaching the barrier") + }(scanners[i]) + } + for range kids { + require.NoError(t, <-ready) + } + + for _, k := range kids { + _, err := io.WriteString(k.stdin, "go\n") + require.NoError(t, err) + require.NoError(t, k.stdin.Close()) + } + var mu sync.Mutex var all []string var wg sync.WaitGroup - for range workers { + for i := range kids { wg.Add(1) - go func() { + go func(sc *bufio.Scanner) { defer wg.Done() - for range calls { - lines := runInvocation(t, env) - mu.Lock() - all = append(all, lines...) - mu.Unlock() + var lines []string + for sc.Scan() { + if l := sc.Text(); isReportLine(l) { + lines = append(lines, l) + } } - }() + mu.Lock() + all = append(all, lines...) + mu.Unlock() + }(scanners[i]) } wg.Wait() + for _, k := range kids { + require.NoError(t, k.cmd.Wait()) + } return tally(all) } -// The smoke-test shape: ten workers each making eight sequential calls -// through the production defaults. Before queueing, the shared 50-token +// 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 // "rate limit exceeded" while the server had never answered 429. // -// Eighty calls against a fifty-token bucket is what makes this a test of -// queueing: thirty of them cannot be served from the starting bucket at all, -// so "every call succeeded" is only reachable by waiting for refills. The -// run is therefore bounded below by the refill, never above: it used to -// assert that the whole thing — eighty process spawns and three seconds of -// deliberate refill — finished inside DefaultMaxWait, which is one -// operation's queueing budget and no statement about this run. On a -// contended box that margin is spent on the machine, and it failed on CI at -// 10.73s and again at 10.97s with the gate having behaved perfectly. +// The claim is that the gate queues, and the test now watches that happen +// rather than inferring it from arithmetic. Each child reports how many of +// its operations were admitted out of an empty bucket — only reachable by +// waiting for a refill, since a gate that failed fast would have rejected +// that same operation — and the run cannot pass with that count at zero. +// It is a floor and not an accounting: a call that reads the bucket just +// after a refill lands is served without waiting and is not counted, so the +// number comes out a little under the thirty calls the starting bucket +// cannot cover. What it rules out is the case that matters, a run where the +// wait path was never entered and every assertion here was free. +// +// The children are released from a barrier so they issue their calls at the +// rate the test asked for rather than at the rate the machine can fork +// processes; left to race each other in, on a slow box they arrive spread +// out, the bucket refills between them, and nothing ever queues. +// +// What is deliberately not asserted is how long the whole thing took. It +// used to require the run to finish inside DefaultMaxWait, which is one +// operation's queueing budget and no statement at all about a run of eighty: +// bounded below by three seconds of deliberate refill and above by nothing +// but the box. On a contended one that margin goes to the machine, and it +// failed on CI at 10.73s and again at 10.97s with the gate having behaved +// perfectly both times. func TestGateQueuesTenParallelWorkersThroughTheDefaults(t *testing.T) { - const workers, calls = 10, 8 + const workers, callsEach = 10, 8 cfg := DefaultConfig() - require.Greater(t, float64(workers*calls), cfg.RateLimiter.MaxTokens, + require.Greater(t, float64(workers*callsEach), cfg.RateLimiter.MaxTokens, "the run has to outlast the bucket or nothing queues") dir := t.TempDir() - tl := runWorkers(t, workers, calls, helperEnv{"BH_STATE_DIR": dir, "BH_MODE": "ops", "BH_OPS": "1", "BH_HOLD": "10ms"}) + tl := runBarrieredInvocations(t, workers, helperEnv{ + "BH_STATE_DIR": dir, "BH_MODE": "ops", + "BH_OPS": strconv.Itoa(callsEach), "BH_HOLD": "10ms", + }) - assert.Equal(t, workers*calls, tl.ok, "every call succeeds: %v", tl.rejections) + assert.Equal(t, workers*callsEach, tl.ok, "every call succeeds: %v", tl.rejections) assert.Zero(t, tl.rejected) + assert.Positive(t, tl.queued, + "no call was ever admitted out of an empty bucket, so the wait path was never entered and this test measured nothing") assert.LessOrEqual(t, tl.peak, cfg.Bulkhead.MaxConcurrent, "never more than MaxConcurrent live holders") } -// Fifteen simultaneous invocations against ten slots: nobody fails, nobody -// sees more than ten holders, and the wall clock shows the overflow waited -// for a second round rather than being admitted alongside the first. +// Fifteen simultaneous invocations against ten slots: nobody fails, and the +// 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 peak says it +// directly. func TestGateQueuesOversubscribedInvocationsWithinTheSlotLimit(t *testing.T) { dir := t.TempDir() - hold := 150 * time.Millisecond - start := time.Now() - tl := runWorkers(t, 15, 1, helperEnv{ + cfg := DefaultConfig() + const invocations = 15 + tl := runBarrieredInvocations(t, invocations, helperEnv{ "BH_STATE_DIR": dir, "BH_MODE": "ops", "BH_OPS": "1", - "BH_HOLD": hold.String(), "BH_MAX_TOKENS": "100", + "BH_HOLD": (150 * time.Millisecond).String(), "BH_MAX_TOKENS": "100", }) - elapsed := time.Since(start) - assert.Equal(t, 15, tl.ok, "every invocation succeeds: %v", tl.rejections) - assert.LessOrEqual(t, tl.peak, 10, "never more than MaxConcurrent live holders") - assert.GreaterOrEqual(t, elapsed, 2*hold, "the overflow waited for a slot") + assert.Equal(t, invocations, tl.ok, "every invocation succeeds: %v", tl.rejections) + assert.Zero(t, tl.rejected) + assert.Equal(t, cfg.Bulkhead.MaxConcurrent, tl.peak, + "the slot table filled to exactly MaxConcurrent, so the overflow waited") state, err := NewStore(dir).Load() require.NoError(t, err) From 0ca078a100d5602a94ea5e04d28275236ed32628 Mon Sep 17 00:00:00 2001 From: Jorge Manrubia Date: Sat, 19 Sep 2026 12:23:40 +0200 Subject: [PATCH 3/5] Keep a dead child's account of why it died The barriered runner drives the children through pipes rather than running each to completion with its output in hand, and in the move the children's stderr stopped being kept. A child that panics now fails the parent with "exit status 2" and nothing else, where before the whole transcript came back with the error. Its stderr is buffered again and handed to the failure. Co-Authored-By: Claude Opus 5 (1M context) --- internal/resilience/gate_test.go | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/internal/resilience/gate_test.go b/internal/resilience/gate_test.go index e2f4d1828..312456ae4 100644 --- a/internal/resilience/gate_test.go +++ b/internal/resilience/gate_test.go @@ -2,6 +2,7 @@ package resilience import ( "bufio" + "bytes" "context" "errors" "fmt" @@ -161,6 +162,7 @@ func runBarrieredInvocations(t *testing.T, n int, env helperEnv) invocationTally cmd *exec.Cmd stdin io.WriteCloser stdout io.ReadCloser + stderr *bytes.Buffer } kids := make([]child, 0, n) ready := make(chan error, n) @@ -170,8 +172,12 @@ func runBarrieredInvocations(t *testing.T, n int, env helperEnv) invocationTally require.NoError(t, err) stdout, err := cmd.StdoutPipe() require.NoError(t, err) + // Kept so that a child which dies says why: its panic or its + // testing output is the only account of what went wrong there. + stderr := &bytes.Buffer{} + cmd.Stderr = stderr require.NoError(t, cmd.Start()) - kids = append(kids, child{cmd, stdin, stdout}) + kids = append(kids, child{cmd, stdin, stdout, stderr}) } // Each child announces itself on its own line; nobody is released until @@ -186,7 +192,7 @@ func runBarrieredInvocations(t *testing.T, n int, env helperEnv) invocationTally return } } - ready <- errors.New("child exited before reaching the barrier") + ready <- errors.New("a child exited before reaching the barrier") }(scanners[i]) } for range kids { @@ -219,7 +225,7 @@ func runBarrieredInvocations(t *testing.T, n int, env helperEnv) invocationTally } wg.Wait() for _, k := range kids { - require.NoError(t, k.cmd.Wait()) + require.NoError(t, k.cmd.Wait(), k.stderr.String()) } return tally(all) } From 7415ad2532e0bcc1c2e04022135fa06faea00fda Mon Sep 17 00:00:00 2001 From: Jorge Manrubia Date: Sat, 19 Sep 2026 12:39:22 +0200 Subject: [PATCH 4/5] Let the gate report its own waits, and prove the witness can say no MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The queue signal added in the last commit was inferred from outside: the child read the bucket just before gating and counted the calls that found it empty. A refill landing between the look and the take makes that a wait that never happened, so the witness could report queueing where the gate had returned immediately — the assertion still running over nothing, now with a witness that can lie about it. The rate limiter and the bulkhead each get an onWait seam, called at the point where the caller is actually turned away and settles in to sleep. The clock is already injected here for the same kind of reason; this is the same seam for the same kind of question. Nothing in production sets either one. A counter that always fires would satisfy the queueing test just as well as an honest one, so the witness now has to say no as well as yes: against a bucket nobody can exhaust and a slot for every caller, the same eighty calls go through and both counters must read zero. With that in place the oversubscribed test can say the thing it is named after directly — five of the fifteen found every slot taken and polled for one, not four and not six. The queueing count is thirty calls the starting bucket cannot cover, less the odd one that arrives in the instant after a refill lands and is served without waiting: 29 idle, 28 on a saturated core, over eight runs. One free pickup per worker is the allowance. Evidence, all mutations against the committed tests: a witness rewired to fire on every call, including the ones served immediately, fails the new test at 80 where it wants 0. A rate limiter that stops queueing for refills fails the queueing test at 0 where it wants 20 or more. A bulkhead that stops queueing for slots fails the oversubscribed test with exactly five rejections out of fifteen. All three gate tests pass 3 of 3 on the saturated single core with race instrumentation that made the original fail 10.12s > 10s. Co-Authored-By: Claude Opus 5 (1M context) --- internal/resilience/bulkhead.go | 8 ++ internal/resilience/gate_test.go | 111 ++++++++++++++++++++-------- internal/resilience/rate_limiter.go | 10 +++ 3 files changed, 98 insertions(+), 31 deletions(-) diff --git a/internal/resilience/bulkhead.go b/internal/resilience/bulkhead.go index 0ae3ce2f5..a375c9ba9 100644 --- a/internal/resilience/bulkhead.go +++ b/internal/resilience/bulkhead.go @@ -14,6 +14,11 @@ import ( type Bulkhead struct { config BulkheadConfig store *Store + // onWait, when set, is called each time a caller finds every slot taken + // and settles in to poll for one. Like the rate limiter's, it is there + // so a test can hear the wait path from inside rather than guess at it + // from the slot table. Nothing in production sets it. + onWait func() } // NewBulkhead creates a new bulkhead with the given config. @@ -118,6 +123,9 @@ func (b *Bulkhead) waitSince(ctx context.Context, start, deadline time.Time) err if acquired, _ := b.Acquire(); acquired { //nolint:contextcheck // lock acquisition is context-independent by design return nil } + if b.onWait != nil { + b.onWait() + } if err := pause(ctx, min(jittered(slotPoll), remaining)); err != nil { return err } diff --git a/internal/resilience/gate_test.go b/internal/resilience/gate_test.go index 312456ae4..39e0cd791 100644 --- a/internal/resilience/gate_test.go +++ b/internal/resilience/gate_test.go @@ -27,9 +27,13 @@ import ( // // BH_MODE=ops runs BH_OPS gated operations, each held for BH_HOLD, and prints // OK or REJECT per operation, plus PEAK, the most live slot holders it saw -// while holding one, and QUEUED, how many of its operations were admitted -// out of an empty bucket and so can only have got there by waiting for a -// refill. BH_BARRIER=1 makes it print READY and hold before its first +// while holding one, QUEUED, how many of its operations were turned away by +// the bucket and had to sleep for a refill, and SLOTWAIT, how many found +// every slot taken and had to poll for one. The last two are counted by the +// gate itself through its onWait seams, not inferred from outside: a look +// at the bucket before the call can be overtaken by a refill landing before +// the take, and would report a wait that never happened. +// 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_MODE=linger churns BH_OPS acquire/release pairs, prints @@ -55,23 +59,27 @@ func TestHelperProcess(t *testing.T) { fmt.Println("READY") _, _ = bufio.NewReader(os.Stdin).ReadString('\n') } - peak, queued := 0, 0 + // 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. + var sleptOnBucket, sleptOnSlot bool + hooks.rateLimiter.onWait = func() { sleptOnBucket = true } + hooks.bulkhead.onWait = func() { sleptOnSlot = true } + + peak, queued, slotWaits := 0, 0, 0 for range ops { - // Read the bucket before gating: an operation admitted when - // there was nothing to admit it with waited for a refill, which - // is the queueing this test exists to see. A gate that failed - // fast would have rejected that same operation instead. - tokens, tokensErr := hooks.rateLimiter.Tokens() - starved := tokensErr == nil && tokens < cfg.RateLimiter.TokensPerRequest - + sleptOnBucket, sleptOnSlot = false, false ctx, err := hooks.OnOperationGate(context.Background(), op) if err != nil { fmt.Printf("REJECT %v\n", err) continue } - if starved { + if sleptOnBucket { queued++ } + if sleptOnSlot { + slotWaits++ + } if inUse, err := hooks.bulkhead.InUse(); err == nil { peak = max(peak, inUse) } @@ -81,6 +89,7 @@ func TestHelperProcess(t *testing.T) { } fmt.Printf("PEAK %d\n", peak) fmt.Printf("QUEUED %d\n", queued) + fmt.Printf("SLOTWAIT %d\n", slotWaits) case "linger": bh := NewBulkhead(store, cfg.Bulkhead) rl := NewRateLimiter(store, cfg.RateLimiter) @@ -98,7 +107,7 @@ func TestHelperProcess(t *testing.T) { // isReportLine picks the helper's tallied output out of the test binary's // own chatter. func isReportLine(l string) bool { - for _, prefix := range []string{"OK", "REJECT", "PEAK", "QUEUED"} { + for _, prefix := range []string{"OK", "REJECT", "PEAK", "QUEUED", "SLOTWAIT"} { if strings.HasPrefix(l, prefix) { return true } @@ -119,8 +128,8 @@ func helperCommand(t *testing.T, env helperEnv) *exec.Cmd { } type invocationTally struct { - ok, rejected, peak, queued int - rejections []string + ok, rejected, peak, queued, slotWaits int + rejections []string } func tally(lines []string) invocationTally { @@ -138,6 +147,9 @@ func tally(lines []string) invocationTally { case strings.HasPrefix(l, "QUEUED"): n, _ := strconv.Atoi(strings.TrimPrefix(l, "QUEUED ")) tl.queued += n + case strings.HasPrefix(l, "SLOTWAIT"): + n, _ := strconv.Atoi(strings.TrimPrefix(l, "SLOTWAIT ")) + tl.slotWaits += n } } return tl @@ -235,21 +247,27 @@ func runBarrieredInvocations(t *testing.T, n int, env helperEnv) invocationTally // bucket drained in the first half second and every later call failed with // "rate limit exceeded" while the server had never answered 429. // -// The claim is that the gate queues, and the test now watches that happen -// rather than inferring it from arithmetic. Each child reports how many of -// its operations were admitted out of an empty bucket — only reachable by -// waiting for a refill, since a gate that failed fast would have rejected -// that same operation — and the run cannot pass with that count at zero. -// It is a floor and not an accounting: a call that reads the bucket just -// after a refill lands is served without waiting and is not counted, so the -// number comes out a little under the thirty calls the starting bucket -// cannot cover. What it rules out is the case that matters, a run where the -// wait path was never entered and every assertion here was free. +// The claim is that the gate queues, and the gate is what says so. Its +// onWait seam fires when a caller is turned away by the bucket and settles +// in to sleep for a refill, and each child counts the operations that had +// to. Thirty of the eighty calls cannot be served from the starting bucket, +// so thirty is what the count comes to, and the run cannot pass with it at +// zero — which is the case that matters, a run where the wait path was +// never entered and every assertion here was free. +// +// Inferring the same thing from outside does not work, and the first +// attempt at this did: it read the bucket before each call and counted the +// ones that found it empty. A refill landing between the look and the take +// makes that a wait that never happened. Only the gate knows. // // The children are released from a barrier so they issue their calls at the // rate the test asked for rather than at the rate the machine can fork // processes; left to race each other in, on a slow box they arrive spread -// out, the bucket refills between them, and nothing ever queues. +// out, the bucket refills between them, and nothing ever queues. The +// barrier only lines up the first of each child's eight calls, so a box +// slow enough to pace all eighty across the three seconds of refill would +// still see nothing queue — and would fail here on the zero count rather +// than pass over it. // // What is deliberately not asserted is how long the whole thing took. It // used to require the run to finish inside DefaultMaxWait, which is one @@ -272,11 +290,40 @@ func TestGateQueuesTenParallelWorkersThroughTheDefaults(t *testing.T) { assert.Equal(t, workers*callsEach, tl.ok, "every call succeeds: %v", tl.rejections) assert.Zero(t, tl.rejected) - assert.Positive(t, tl.queued, - "no call was ever admitted out of an empty bucket, so the wait path was never entered and this test measured nothing") + // Thirty calls are not covered by the starting bucket. Nearly all of + // them sleep; the odd one arrives in the instant after a refill lands + // and is served without waiting, so the count comes in a little under — + // 29 idle here, 28 on a saturated core. One free pickup per worker is + // the allowance. A zero is a run that never entered the wait path. + uncovered := workers*callsEach - int(cfg.RateLimiter.MaxTokens) + assert.GreaterOrEqual(t, tl.queued, uncovered-workers, + "the calls the starting bucket could not cover slept for a refill; %d of %d did", tl.queued, uncovered) assert.LessOrEqual(t, tl.peak, cfg.Bulkhead.MaxConcurrent, "never more than MaxConcurrent live holders") } +// The other side of the queue counter: with a bucket nobody can exhaust and +// a slot for every caller, the same eighty calls go through without the +// gate ever sleeping, and the counters say zero. +// +// Without this, a counter wired to fire on every call would satisfy the +// test above and prove nothing — the witness has to be able to say no. +func TestGateReportsNoQueueingWhenNothingHadToWait(t *testing.T) { + const workers, callsEach = 10, 8 + cfg := DefaultConfig() + require.LessOrEqual(t, workers, cfg.Bulkhead.MaxConcurrent, "nobody may have to wait for a slot") + + dir := t.TempDir() + tl := runBarrieredInvocations(t, workers, helperEnv{ + "BH_STATE_DIR": dir, "BH_MODE": "ops", + "BH_OPS": strconv.Itoa(callsEach), "BH_HOLD": "10ms", + "BH_MAX_TOKENS": strconv.Itoa(workers * callsEach * 10), + }) + + assert.Equal(t, workers*callsEach, tl.ok, "every call succeeds: %v", tl.rejections) + assert.Zero(t, tl.queued, "the bucket never ran out, so nothing should have slept for a refill") + assert.Zero(t, tl.slotWaits, "there was a slot for everyone, so nobody should have polled for one") +} + // Fifteen simultaneous invocations against ten slots: nobody fails, and the // 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. @@ -286,8 +333,8 @@ func TestGateQueuesTenParallelWorkersThroughTheDefaults(t *testing.T) { // 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 peak says it -// directly. +// holds, which slow spawning satisfies all by itself. The bulkhead now says +// it itself: five of the fifteen found every slot taken and polled for one. func TestGateQueuesOversubscribedInvocationsWithinTheSlotLimit(t *testing.T) { dir := t.TempDir() cfg := DefaultConfig() @@ -300,7 +347,9 @@ func TestGateQueuesOversubscribedInvocationsWithinTheSlotLimit(t *testing.T) { assert.Equal(t, invocations, tl.ok, "every invocation succeeds: %v", tl.rejections) assert.Zero(t, tl.rejected) assert.Equal(t, cfg.Bulkhead.MaxConcurrent, tl.peak, - "the slot table filled to exactly MaxConcurrent, so the overflow waited") + "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") state, err := NewStore(dir).Load() require.NoError(t, err) diff --git a/internal/resilience/rate_limiter.go b/internal/resilience/rate_limiter.go index 08286be01..7e9931d04 100644 --- a/internal/resilience/rate_limiter.go +++ b/internal/resilience/rate_limiter.go @@ -15,6 +15,13 @@ type RateLimiter struct { // clock is the time the limiter reads. A field so that a test can put // the deadline boundary where it wants it instead of racing a real one. clock func() time.Time + // onWait, when set, is called each time a caller is turned away by the + // bucket and settles in to sleep for a refill or a block. A test that + // wants to know the wait path was taken has to hear it from in here: a + // look at the bucket from outside can be overtaken by a refill landing + // between the look and the take, and would report a wait that never + // happened. Nothing in production sets it. + onWait func() } // NewRateLimiter creates a new rate limiter with the given config. @@ -165,6 +172,9 @@ func (rl *RateLimiter) waitSince(ctx context.Context, start, deadline time.Time) return rl.gateError(blocked, wait, rl.now().Sub(start)) } sleptOnServerBlock = blocked + if rl.onWait != nil { + rl.onWait() + } if err := pause(ctx, sleepWithin(wait, remaining)); err != nil { return err } From 33eececdf73d77f8a16f4f5f028939e51a516ed8 Mon Sep 17 00:00:00 2001 From: Jorge Manrubia Date: Sat, 19 Sep 2026 12:50:45 +0200 Subject: [PATCH 5/5] Stop the comment promising a count the assertion does not make The doc comment said thirty calls sleep for a refill and the assertion two paragraphs down explains that it is 29 idle and 28 on a saturated core. Say roughly, and leave the exact figures where they are measured. Co-Authored-By: Claude Opus 5 (1M context) --- internal/resilience/gate_test.go | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/internal/resilience/gate_test.go b/internal/resilience/gate_test.go index 39e0cd791..145eed250 100644 --- a/internal/resilience/gate_test.go +++ b/internal/resilience/gate_test.go @@ -251,9 +251,9 @@ func runBarrieredInvocations(t *testing.T, n int, env helperEnv) invocationTally // onWait seam fires when a caller is turned away by the bucket and settles // in to sleep for a refill, and each child counts the operations that had // to. Thirty of the eighty calls cannot be served from the starting bucket, -// so thirty is what the count comes to, and the run cannot pass with it at -// zero — which is the case that matters, a run where the wait path was -// never entered and every assertion here was free. +// so that is roughly what the count comes to, and the run cannot pass with +// it at zero — which is the case that matters, a run where the wait path +// was never entered and every assertion here was free. // // Inferring the same thing from outside does not work, and the first // attempt at this did: it read the bucket before each call and counted the @@ -333,8 +333,9 @@ func TestGateReportsNoQueueingWhenNothingHadToWait(t *testing.T) { // 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 now says -// it itself: five of the fifteen found every slot taken and polled for one. +// 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. func TestGateQueuesOversubscribedInvocationsWithinTheSlotLimit(t *testing.T) { dir := t.TempDir() cfg := DefaultConfig()