test(requestlog): make the byte-budget drain test independent of runner speed

My own test, and it broke the release build. CI reported
`assert 239800 == 0` on the very commit that was supposed to ship v1.11.0, so
no image was pushed and the tag and release were never cut.

The test drained a 50-row queue with `batch_size=100` and `flush_ms=10`, then
asserted the byte counter was back to zero. `_collect()` stops at whichever
comes first, `batch_size` rows or the flush deadline - and with a batch size
larger than the row count, the deadline is the only thing that can end it. It
was measuring the scheduler, not the sink.

The arithmetic is exact: a row here weighs 1400 + 4096 + 4096 = 9592 bytes, and
239 800 is 25 of them. `_collect()` returned half the queue because 25
iterations of `asyncio.wait_for` were enough to exhaust 10 ms on that runner.
The workflow builds `linux/amd64,linux/arm64`, so one of the two runs under qemu
emulation; a local `docker build` compiles the native platform only and never
sees that path. I could not reproduce the failure even building both platforms
here - this machine fits 49 iterations inside 10 ms - which is the point: a test
whose result depends on how fast the host is will pass everywhere it is
convenient and fail where it matters.

Fixed structurally rather than by widening the window: `batch_size` now EQUALS
the row count, so the collect loop exits on the count and never consults the
deadline at all. The flush window is generous as a backstop, the drain runs in
a loop instead of a single call, and the row count is asserted on the way in
and on the way out so a future change cannot make it vacuous.

Verified on both platforms the workflow builds: 1667 passed / 152 skipped on
linux/arm64 and on linux/amd64 under emulation.

No production code changes.

Author: taylanbakircioglu

backend/tests/test_request_log_fleet_scale.py

@@ -278,16 +278,40 @@ def test_queue_memory_is_bounded_even_at_the_max_body_size_ceiling():
     )
 
 
+_DRAIN_ROWS = 50
+
+
 def test_the_byte_budget_is_released_as_rows_drain():
-    """A budget that only ever counts up is a slow leak, not a limit."""
+    """A budget that only ever counts up is a slow leak, not a limit.
+
+    Deliberately time-INDEPENDENT. `_collect()` stops at whichever comes first,
+    `batch_size` rows or the `flush_ms` deadline, so a batch size larger than
+    the row count makes the result a function of how fast the runner happens to
+    be. The first version of this test used batch=100/flush=10ms for 50 rows and
+    passed on a native build while failing in CI, which builds
+    linux/amd64 + linux/arm64 and therefore runs one of them under qemu
+    emulation: 25 iterations of `asyncio.wait_for` were enough to exhaust 10 ms
+    there, `_collect()` returned half a batch, and the assertion read
+    `239800 == 0` - measuring the scheduler, not the sink.
+
+    So: batch size EQUAL to the row count, so the loop exits on the count and
+    never consults the deadline; a generous flush window in case it somehow
+    does; and a drain loop rather than a single call. Nothing here depends on
+    wall-clock speed.
+    """
     async def drain():
-        sink = RequestLogSink(2000, 100, 10, max_bytes=8 * 1024 * 1024)
+        sink = RequestLogSink(2000, _DRAIN_ROWS, 5000, max_bytes=8 * 1024 * 1024)
         blob = b"x" * 4096
         set_config(replace(get_config(), capture_agent_success=True))
-        for _ in range(50):
+        for _ in range(_DRAIN_ROWS):
             sink.offer(_row(target=None, request_body_raw=blob, response_body_raw=blob))
-        assert sink.stats["queued_bytes"] > 0
-        await sink._collect()
+        assert sink.stats["queued_bytes"] > 0, "nothing was queued, so nothing is being measured"
+        assert sink._queue.qsize() == _DRAIN_ROWS, "the queue did not take every row"
+
+        drained = 0
+        while not sink._queue.empty():
+            drained += len(await sink._collect())
+        assert drained == _DRAIN_ROWS, f"drained {drained} of {_DRAIN_ROWS} rows"
         return sink.stats["queued_bytes"]
 
     assert asyncio.run(drain()) == 0