speed: bufferbloat probe running concurrently with saturation

src/netcheck/speed.py

@@ -211,3 +211,50 @@ async def tier_ndt7(client: httpx.AsyncClient, cfg: Speedtest, timeout: float) -
         upload_mbps=None,
         server=results[0].get("machine"),
     )
+
+
+from netcheck.config import BufferbloatBands
+from netcheck.interpret import grade_bufferbloat
+from netcheck.stats import rtt_stats
+
+
+async def probe_while(
+    coro: Awaitable[None],
+    probe: Callable[[], Awaitable[float | None]],
+    interval: float,
+) -> list[float]:
+    samples: list[float] = []
+    task = asyncio.ensure_future(coro)
+    while not task.done():
+        sample = await probe()
+        if sample is not None:
+            samples.append(sample)
+        await asyncio.sleep(interval)
+    await task
+    return samples
+
+
+async def measure_with_bufferbloat(
+    result: SpeedResult,
+    idle_rtt_ms: float | None,
+    bands: BufferbloatBands,
+    run_download: Callable[[], Awaitable[None]],
+    run_upload: Callable[[], Awaitable[None]],
+    probe: Callable[[], Awaitable[float | None]],
+    interval: float,
+) -> SpeedResult:
+    # This is the one place a measurement is allowed to overlap another: the
+    # whole point is to see what latency does while the link is saturated.
+    down_samples = await probe_while(run_download(), probe, interval)
+    up_samples = await probe_while(run_upload(), probe, interval)
+    result.idle_rtt_ms = idle_rtt_ms
+    result.loaded_rtt_down_ms = rtt_stats(list(down_samples)).avg_ms
+    result.loaded_rtt_up_ms = rtt_stats(list(up_samples)).avg_ms
+    result.bufferbloat_down_ms = bufferbloat_delta(idle_rtt_ms, down_samples)
+    result.bufferbloat_up_ms = bufferbloat_delta(idle_rtt_ms, up_samples)
+    worst_delta = max(
+        [d for d in (result.bufferbloat_down_ms, result.bufferbloat_up_ms) if d is not None],
+        default=None,
+    )
+    result.bufferbloat_grade = grade_bufferbloat(worst_delta, bands)
+    return result

tests/test_speed_math.py

@@ -169,3 +169,85 @@ def test_the_ndt7_consent_notice_states_what_gets_published():
     assert "CC0" in NDT7_CONSENT_NOTICE
     assert "IP" in NDT7_CONSENT_NOTICE
     assert "public" in NDT7_CONSENT_NOTICE.lower()
+
+
+from netcheck.config import BufferbloatBands
+from netcheck.speed import measure_with_bufferbloat, probe_while
+
+
+async def test_probe_while_collects_samples_for_the_whole_duration_of_the_work():
+    import asyncio
+
+    async def work() -> None:
+        await asyncio.sleep(0.25)
+
+    async def probe() -> float | None:
+        return 42.0
+
+    samples = await probe_while(work(), probe, interval=0.05)
+    assert len(samples) >= 3
+    assert all(s == 42.0 for s in samples)
+
+
+async def test_probe_while_drops_failed_probes_instead_of_recording_none():
+    import asyncio
+
+    async def work() -> None:
+        await asyncio.sleep(0.15)
+
+    async def probe() -> float | None:
+        return None
+
+    assert await probe_while(work(), probe, interval=0.05) == []
+
+
+async def test_measure_with_bufferbloat_fills_in_both_directions_and_the_grade():
+    import asyncio
+
+    loaded = iter([80.0, 90.0, 100.0] * 20)
+
+    async def run_download() -> None:
+        await asyncio.sleep(0.15)
+
+    async def run_upload() -> None:
+        await asyncio.sleep(0.15)
+
+    async def probe() -> float | None:
+        return next(loaded)
+
+    result = await measure_with_bufferbloat(
+        SpeedResult(method="cloudflare", download_mbps=300.0, upload_mbps=40.0),
+        idle_rtt_ms=12.0,
+        bands=BufferbloatBands(),
+        run_download=run_download,
+        run_upload=run_upload,
+        probe=probe,
+        interval=0.05,
+    )
+    assert result.idle_rtt_ms == 12.0
+    assert result.loaded_rtt_down_ms is not None
+    assert result.loaded_rtt_up_ms is not None
+    assert result.bufferbloat_down_ms is not None and result.bufferbloat_down_ms > 60
+    assert result.bufferbloat_grade in ("D", "E", "F")
+
+
+async def test_measure_with_bufferbloat_without_an_idle_baseline_grades_unknown():
+    import asyncio
+
+    async def work() -> None:
+        await asyncio.sleep(0.05)
+
+    async def probe() -> float | None:
+        return 80.0
+
+    result = await measure_with_bufferbloat(
+        SpeedResult(method="cloudflare", download_mbps=300.0),
+        idle_rtt_ms=None,
+        bands=BufferbloatBands(),
+        run_download=work,
+        run_upload=work,
+        probe=probe,
+        interval=0.02,
+    )
+    assert result.bufferbloat_down_ms is None
+    assert result.bufferbloat_grade == "?"