fix: --rps reaped at most one completion per rate-limiter tick

[agourlay]

--rps reported latencies that were mostly self-inflicted: at --rps 2000 bfb
reported a p50 of 1.66 s against a server reporting 80 ms.

The gap was bfb's own collection backlog, not the server.

What the loop is supposed to do

With --rps, process_with_rps runs an open-loop generator. One loop does two
jobs at once via tokio::select!:

• Job A: every tick of the rate limiter, fire off a new request.
• Job B: collect requests that have come back, and report any that failed.

in_flight is the bag of requests currently in the air. Job B is "take one out
of the bag."

The bug

Job B was written like this:

Some(Err(err)) = in_flight.next(), if !in_flight.is_empty() => { ...handle error... }

The left side is a pattern, not a variable. It only matches a request that
failed. A request that succeeded comes back as Some(Ok(())), which doesn't
match.

Here's the part that bites: when a select! branch's pattern doesn't match,
tokio doesn't skip it and try again — it switches that branch off for the
rest of that select!, until the loop comes back around and enters it fresh.

So:

1. Enter select!. Check the bag. Pull out one request. It succeeded.
2. Some(Err(...)) doesn't match a success → Job B is switched off.
3. Only Job A is left, so the loop sits there waiting for the next tick.
4. Tick fires, send a request, loop around, select! starts over, Job B is back on.

Every trip around the loop takes exactly one item out of the bag, and every trip
is gated on a tick. So one request collected per tick, no matter how many are
actually sitting there done. The code was accidentally fast at handling errors
and slow at handling successes.

FuturesUnordered::poll_next compounds it: it returns at the first ready child
and leaves the rest of the ready queue unpolled.

Sending is also one per tick, so the bag drains at exactly the rate it fills.
That sounds balanced, but it's the worst place to sit — any hiccup adds a backlog
that never gets worked off.

Why the reported latency went upside down

Each request times itself: it reads the clock when it starts, and again when
its future is next polled after the reply arrives.

That second reading isn't when the server answered. It's when our loop got around
to it. A reply that landed instantly but then sat in the bag for a second gets
stamped one second.

So the printed latency was mostly our own collection backlog.

That also explains the inversion. Draining is tied to ticks, so at 2,000/s you
get 2,000 chances a second to empty the bag; at 20,000/s, ten times as many.
Lower rate → slower drain → longer queue → worse reported latency:

offered rate	bfb reported	server reported
2,000/s	1.4 s	8 ms
20,000/s	340 ms	8 ms

Real congestion goes the other way — push harder, get slower. Latency improving
as load increases is the tell that the number was measuring the client.

The fix

Stop using a pattern. Bind the result to a variable and check for the error
inside the body:

res = in_flight.next(), if !in_flight.is_empty() => {
    if let Some(Err(err)) = res { ...handle error... }
}

A plain variable always matches, so the branch never gets switched off. The loop
drains completions as fast as they arrive instead of once per tick. Error
handling is unchanged — it just moved one line inward.

Measurements

50,000 queries, -t 16 -c 2, --rps 2000, against a server whose closed-loop
saturation point is 2,387/s:

	client p50	server p50	gap
before	1663.71 ms	80.80 ms	1582.91 ms
after	3.29 ms	3.09 ms	0.20 ms

The server-side figure moves too, which is consistent: starved of reaping, the
loop delivered its requests in bursts, and the server was reporting the queue
those bursts created.

Above saturation both builds agree (--rps 4000 on the same server: 1.76 s
either way). That's a real queue, and it correctly doesn't move.

Scope

process_with_parallel is unaffected — buffer_unordered bounds concurrency and
its while let reaps every completion.

One file, src/stats.rs; the functional change is two lines, the rest is a
comment explaining the trap so it doesn't get reintroduced.