Two gate tests were failing on the clock, not on the gate

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
 		}

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)