diff --git a/src/zeroconf/_core.py b/src/zeroconf/_core.py index 1814e86c..f184b6fa 100644 --- a/src/zeroconf/_core.py +++ b/src/zeroconf/_core.py @@ -104,6 +104,13 @@ _REGISTER_BROADCASTS = 3 +# RFC 6762 §8.1 thundering-herd avoidance: wait a random +# 0-250ms before the first probe so simultaneously-started +# responders don't collide. We default to 150-250ms to +# preserve existing timing assumptions; tests on loopback +# may patch this lower via the `quick_timing` fixture. +_PROBE_RANDOM_DELAY_INTERVAL = (150, 250) # ms + def async_send_with_transport( log_debug: bool, @@ -561,7 +568,7 @@ async def async_check_service( # Wait a random amount of time up avoid collisions and avoid # a thundering herd when multiple services are started on the network - await self.async_wait(random.randint(150, 250)) # noqa: S311 + await self.async_wait(random.randint(*_PROBE_RANDOM_DELAY_INTERVAL)) # noqa: S311 next_instance_number = 2 next_time = now = current_time_millis() diff --git a/tests/conftest.py b/tests/conftest.py index c281e2d4..a9c4f590 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -82,17 +82,22 @@ def quick_timing() -> Generator[None]: """Shorten the probe/announce/goodbye/first-query intervals for tests on loopback. The production values (_CHECK_TIME=500ms, _REGISTER_TIME=225ms, - _UNREGISTER_TIME=125ms, _FIRST_QUERY_DELAY_RANDOM_INTERVAL=20-120ms) - exist for RFC 6762 interop on real networks (§8.1 thundering-herd - avoidance for probing, §5.2 for the initial-query delay). Tests on - 127.0.0.1 do not need them and pay 1-2s per register/unregister - cycle and 20-120ms per ServiceBrowser startup without this fixture. - Opt in by adding `quick_timing` to a test's argument list. + _UNREGISTER_TIME=125ms, _PROBE_RANDOM_DELAY_INTERVAL=150-250ms, + _FIRST_QUERY_DELAY_RANDOM_INTERVAL=20-120ms) exist for RFC 6762 + interop on real networks (§8.1 thundering-herd avoidance for + probing, §5.2 for the initial-query delay). Tests on 127.0.0.1 + do not need them and pay 1-2s per register/unregister cycle, + 150-250ms per probe, and 20-120ms per ServiceBrowser startup + without this fixture. Opt in either by adding `quick_timing` + to a test's argument list or via + `@pytest.mark.usefixtures("quick_timing")` on the test or + its class. """ with ( patch.object(_core, "_CHECK_TIME", 10), patch.object(_core, "_REGISTER_TIME", 10), patch.object(_core, "_UNREGISTER_TIME", 10), + patch.object(_core, "_PROBE_RANDOM_DELAY_INTERVAL", (1, 5)), patch.object(service_browser, "_FIRST_QUERY_DELAY_RANDOM_INTERVAL", (1, 5)), ): yield @@ -105,9 +110,11 @@ def quick_request_timing() -> Generator[None]: The 200ms `_LISTENER_TIME` and 20-120ms random jitter (RFC 6762 §5.2) help spread queries from multiple clients on real networks. On loopback they're pure overhead — get_service_info-style tests - wait ~250ms before the first query even fires. Opt in by adding - `quick_request_timing` to a test's argument list, then drop the - test's own timeouts (which had to accommodate that delay). + wait ~250ms before the first query even fires. Opt in either by + adding `quick_request_timing` to a test's argument list or via + `@pytest.mark.usefixtures("quick_request_timing")` on the test + or its class, then drop the test's own timeouts (which had to + accommodate that delay). """ with ( patch.object(service_info, "_LISTENER_TIME", 10), diff --git a/tests/services/test_info.py b/tests/services/test_info.py index 56146852..09d3d989 100644 --- a/tests/services/test_info.py +++ b/tests/services/test_info.py @@ -251,6 +251,7 @@ def test_service_info_rejects_expired_records(self): @unittest.skipIf(not has_working_ipv6(), "Requires IPv6") @unittest.skipIf(os.environ.get("SKIP_IPV6"), "IPv6 tests disabled") + @pytest.mark.usefixtures("quick_request_timing") def test_get_info_partial(self): zc = r.Zeroconf(interfaces=["127.0.0.1"]) diff --git a/tests/test_asyncio.py b/tests/test_asyncio.py index e4449b6c..9c387180 100644 --- a/tests/test_asyncio.py +++ b/tests/test_asyncio.py @@ -608,7 +608,7 @@ async def test_async_wait_unblocks_on_update(quick_timing: None) -> None: @pytest.mark.asyncio -async def test_service_info_async_request(quick_timing: None) -> None: +async def test_service_info_async_request(quick_timing: None, quick_request_timing: None) -> None: """Test registering services broadcasts and query with AsyncServceInfo.async_request.""" if not has_working_ipv6() or os.environ.get("SKIP_IPV6"): pytest.skip("Requires IPv6") @@ -710,13 +710,13 @@ async def test_service_info_async_request(quick_timing: None) -> None: aiozc.zeroconf.out_delay_queue.queue.clear() aiosinfo = AsyncServiceInfo(type_, registration_name) _clear_cache(aiozc.zeroconf) - # Generating the race condition is almost impossible - # without patching since its a TOCTOU race. 1500ms covers - # the initial _LISTENER_TIME + random delay (200-320ms) and - # leaves plenty of margin for the loopback response to land + # Generating the race condition is almost impossible without + # patching since it's a TOCTOU race. Under `quick_request_timing` + # the first QU query fires at ~10ms and the QM follow-up at ~15ms; + # 300ms leaves plenty of margin for the loopback response to land # before the loop times out. with patch("zeroconf.asyncio.AsyncServiceInfo._is_complete", False): - await aiosinfo.async_request(aiozc.zeroconf, 1500) + await aiosinfo.async_request(aiozc.zeroconf, 300) assert aiosinfo is not None assert aiosinfo.addresses == [socket.inet_aton("10.0.1.3")] diff --git a/tests/test_services.py b/tests/test_services.py index 922046e3..218afc2a 100644 --- a/tests/test_services.py +++ b/tests/test_services.py @@ -33,7 +33,7 @@ def teardown_module(): class ListenerTest(unittest.TestCase): - @pytest.mark.usefixtures("quick_timing") + @pytest.mark.usefixtures("quick_timing", "quick_request_timing") def test_integration_with_listener_class(self): sub_service_added = Event() service_added = Event() diff --git a/tests/utils/test_asyncio.py b/tests/utils/test_asyncio.py index 361faa2b..665bc867 100644 --- a/tests/utils/test_asyncio.py +++ b/tests/utils/test_asyncio.py @@ -86,18 +86,22 @@ async def _still_running(): await asyncio.sleep(5) def _run_coro() -> None: - runcoro_thread_ready.set() assert loop is not None + future = asyncio.run_coroutine_threadsafe(_still_running(), loop) + runcoro_thread_ready.set() with contextlib.suppress(concurrent.futures.TimeoutError): - asyncio.run_coroutine_threadsafe(_still_running(), loop).result(1) + future.result(0.1) runcoro_thread = threading.Thread(target=_run_coro, daemon=True) runcoro_thread.start() runcoro_thread_ready.wait() - time.sleep(0.1) assert loop is not None - aioutils.shutdown_loop(loop) + # Patch _TASK_AWAIT_TIMEOUT so the inner `asyncio.wait` returns + # within 50ms instead of blocking the full 1s on the deliberately + # never-completing _still_running() task. + with patch.object(aioutils, "_TASK_AWAIT_TIMEOUT", 0.05): + aioutils.shutdown_loop(loop) for _ in range(5): if not loop.is_running(): break