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 69a3ff1a3..145eed250 100644 --- a/internal/resilience/gate_test.go +++ b/internal/resilience/gate_test.go @@ -7,6 +7,7 @@ import ( "errors" "fmt" "io" + "maps" "os" "os/exec" "strconv" @@ -25,8 +26,17 @@ 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, 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 // 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 +55,31 @@ 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') + } + // 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 { + sleptOnBucket, sleptOnSlot = false, false ctx, err := hooks.OnOperationGate(context.Background(), op) if err != nil { fmt.Printf("REJECT %v\n", err) continue } + if sleptOnBucket { + queued++ + } + if sleptOnSlot { + slotWaits++ + } if inUse, err := hooks.bulkhead.InUse(); err == nil { peak = max(peak, inUse) } @@ -60,6 +88,8 @@ func TestHelperProcess(t *testing.T) { fmt.Println("OK") } 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) @@ -74,6 +104,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", "SLOTWAIT"} { + 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 +127,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, slotWaits int + rejections []string } func tally(lines []string) invocationTally { @@ -119,66 +144,213 @@ 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 + case strings.HasPrefix(l, "SLOTWAIT"): + n, _ := strconv.Atoi(strings.TrimPrefix(l, "SLOTWAIT ")) + tl.slotWaits += 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 + stderr *bytes.Buffer + } + 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) + // 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, stderr}) + } + + // 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("a 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(), k.stderr.String()) + } 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. +// +// 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 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 +// 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. 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 +// 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, callsEach = 10, 8 + cfg := DefaultConfig() + require.Greater(t, float64(workers*callsEach), 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 := runBarrieredInvocations(t, workers, helperEnv{ + "BH_STATE_DIR": dir, "BH_MODE": "ops", + "BH_OPS": strconv.Itoa(callsEach), "BH_HOLD": "10ms", + }) - assert.Equal(t, 80, 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.LessOrEqual(t, tl.peak, 10, "never more than MaxConcurrent live holders") - assert.Less(t, time.Since(start), DefaultMaxWait) + // 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, 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 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() - 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.Less(t, elapsed, DefaultMaxWait) + 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") + 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 } 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) {