diff --git a/src/netcheck/probes/icmp_win.py b/src/netcheck/probes/icmp_win.py index d800069..c5e2b5c 100644 --- a/src/netcheck/probes/icmp_win.py +++ b/src/netcheck/probes/icmp_win.py @@ -145,11 +145,23 @@ def ping_samples_win(host: str, count: int, interval: float, timeout: float) -> return samples +# A single non-responding hop must never be allowed to eat the whole trace budget: +# cap its timeout even if that leaves room for very few hops. +_MAX_HOP_TIMEOUT_MS = 4000 + + def trace_hops_win(dest: str, max_hops: int, timeout_ms: int) -> list[tuple[int, IcmpReply]]: + # timeout_ms is the TOTAL budget for the whole trace (all hops combined), not + # a per-hop timeout - a caller-supplied 60s budget across 30 hops must not turn + # into 30 minutes of blocking IcmpSendEcho2 calls. address = socket.gethostbyname(dest) hops: list[tuple[int, IcmpReply]] = [] + per_hop_timeout_ms = max(1, min(timeout_ms // max(max_hops, 1), _MAX_HOP_TIMEOUT_MS)) + deadline = time.monotonic() + timeout_ms / 1000.0 for ttl in range(1, max_hops + 1): - reply = echo_once(address, ttl=ttl, timeout_ms=timeout_ms) + if time.monotonic() >= deadline: + break + reply = echo_once(address, ttl=ttl, timeout_ms=per_hop_timeout_ms) hops.append((ttl, reply)) if classify_status(reply.status) == "ok": break diff --git a/src/netcheck/probes/traceroute.py b/src/netcheck/probes/traceroute.py index 6c976fa..ddfbcf9 100644 --- a/src/netcheck/probes/traceroute.py +++ b/src/netcheck/probes/traceroute.py @@ -1,15 +1,42 @@ from __future__ import annotations import asyncio +import concurrent.futures import json import platform import shutil -from typing import Awaitable, Callable +import threading +from typing import Awaitable, Callable, TypeVar from netcheck.models import Capabilities, TraceHop, TraceResult from netcheck.probes.icmp_win import IcmpReply, classify_status from netcheck.traceparse import build_trace_result, finalize_hop +_T = TypeVar("_T") + + +def _run_in_daemon_thread(func: Callable[..., _T], *args: object) -> "asyncio.Future[_T]": + # asyncio.to_thread()/loop.run_in_executor(None, ...) queue work onto the + # ThreadPoolExecutor-backed default executor, whose worker threads are non-daemon + # and get joined by concurrent.futures' atexit handler. If the underlying call + # (a blocking WinAPI IcmpSendEcho2) ever outlives its own timeout budget, that + # join blocks process exit indefinitely. A raw daemon thread is never tracked + # by that handler, so it can never hold the process open. + result: concurrent.futures.Future = concurrent.futures.Future() + + def _target() -> None: + if not result.set_running_or_notify_cancel(): + return + try: + value = func(*args) + except BaseException as exc: # noqa: BLE001 - propagated to the awaiting coroutine + result.set_exception(exc) + else: + result.set_result(value) + + threading.Thread(target=_target, name="netcheck-icmp-win", daemon=True).start() + return asyncio.wrap_future(result) + def tier_order(caps: Capabilities, tcp_trace: bool = False) -> list[str]: # An explicit --tcp-trace means the user already knows ICMP is being filtered, @@ -104,7 +131,7 @@ async def _tier_mtr(target: str, binary: str, cycles: int, max_hops: int, timeou async def _tier_icmp_win(target: str, max_hops: int, timeout: float) -> TraceResult: from netcheck.probes.icmp_win import trace_hops_win - replies = await asyncio.to_thread(trace_hops_win, target, max_hops, int(timeout * 1000)) + replies = await _run_in_daemon_thread(trace_hops_win, target, max_hops, int(timeout * 1000)) hops = hops_from_win_replies(replies) return TraceResult( target=target, diff --git a/tests/test_icmp_win_trace.py b/tests/test_icmp_win_trace.py new file mode 100644 index 0000000..5909fd5 --- /dev/null +++ b/tests/test_icmp_win_trace.py @@ -0,0 +1,78 @@ +from __future__ import annotations + +import time +from unittest.mock import patch + +from netcheck.probes.icmp_win import IP_REQ_TIMED_OUT, IcmpReply, trace_hops_win + + +def _fake_gethostbyname(host: str) -> str: + return "203.0.113.1" + + +def test_trace_hops_win_total_elapsed_is_bounded_by_the_total_budget_not_hops_times_timeout(): + # Every hop times out; each fake call sleeps for its per-hop timeout to + # simulate a genuinely unresponsive router. Old (buggy) code handed the + # *entire* 100ms total budget to every one of the 30 hops as its own + # per-hop timeout, i.e. ~3s worst case for this scenario. The fix must + # divide that budget across hops instead, keeping total wall time close + # to the 100ms total budget. + call_timeouts: list[int] = [] + + def fake_echo_once(dest: str, ttl: int, timeout_ms: int, payload: bytes = b"netcheck") -> IcmpReply: + call_timeouts.append(timeout_ms) + time.sleep(timeout_ms / 1000.0) + return IcmpReply(address=None, status=IP_REQ_TIMED_OUT, rtt_ms=None, ttl=0) + + with ( + patch("netcheck.probes.icmp_win.socket.gethostbyname", side_effect=_fake_gethostbyname), + patch("netcheck.probes.icmp_win.echo_once", side_effect=fake_echo_once), + ): + started = time.monotonic() + trace_hops_win("example.invalid", max_hops=30, timeout_ms=100) + elapsed = time.monotonic() - started + + # Comfortably under the ~3s the old per-hop-gets-the-whole-budget bug would + # have produced here, while still generous about scheduling jitter. + assert elapsed < 1.0 + # Every individual hop got a fraction of the total budget, not the whole thing. + assert call_timeouts + assert all(t == 100 // 30 for t in call_timeouts) + + +def test_trace_hops_win_per_hop_timeout_is_capped_even_for_a_huge_total_budget(): + call_timeouts: list[int] = [] + + def fake_echo_once(dest: str, ttl: int, timeout_ms: int, payload: bytes = b"netcheck") -> IcmpReply: + call_timeouts.append(timeout_ms) + return IcmpReply(address=None, status=IP_REQ_TIMED_OUT, rtt_ms=None, ttl=0) + + with ( + patch("netcheck.probes.icmp_win.socket.gethostbyname", side_effect=_fake_gethostbyname), + patch("netcheck.probes.icmp_win.echo_once", side_effect=fake_echo_once), + ): + # A single-hop trace with a huge total budget (e.g. subprocess_seconds=60s + # divided across a small max_hops) must still not hand the whole budget + # to one hop - it's capped at _MAX_HOP_TIMEOUT_MS (4s). + trace_hops_win("example.invalid", max_hops=1, timeout_ms=60_000) + + assert call_timeouts == [4000] + + +def test_trace_hops_win_happy_path_hop_count_and_ok_break_are_unchanged(): + from netcheck.probes.icmp_win import IP_SUCCESS + + def fake_echo_once(dest: str, ttl: int, timeout_ms: int, payload: bytes = b"netcheck") -> IcmpReply: + if ttl < 3: + return IcmpReply(address=f"10.0.0.{ttl}", status=IP_REQ_TIMED_OUT, rtt_ms=None, ttl=0) + return IcmpReply(address="203.0.113.1", status=IP_SUCCESS, rtt_ms=5.0, ttl=64) + + with ( + patch("netcheck.probes.icmp_win.socket.gethostbyname", side_effect=_fake_gethostbyname), + patch("netcheck.probes.icmp_win.echo_once", side_effect=fake_echo_once), + ): + hops = trace_hops_win("example.invalid", max_hops=30, timeout_ms=60_000) + + # Fast-responding hops still stop the loop at the first success, same as before. + assert [ttl for ttl, _ in hops] == [1, 2, 3] + assert hops[-1][1].status == IP_SUCCESS diff --git a/tests/test_traceroute_cascade.py b/tests/test_traceroute_cascade.py index 88c4251..3a747f9 100644 --- a/tests/test_traceroute_cascade.py +++ b/tests/test_traceroute_cascade.py @@ -1,12 +1,13 @@ from __future__ import annotations import asyncio +import threading import pytest from netcheck.models import Capabilities, TraceHop, TraceResult from netcheck.probes.icmp_win import IP_REQ_TIMED_OUT, IP_SUCCESS, IP_TTL_EXPIRED_TRANSIT, IcmpReply -from netcheck.probes.traceroute import hops_from_win_replies, run_cascade, tier_order +from netcheck.probes.traceroute import _run_in_daemon_thread, hops_from_win_replies, run_cascade, tier_order def caps(**kw) -> Capabilities: @@ -161,3 +162,45 @@ def test_win_replies_become_hops_with_ttl_expiry_handled(): assert hops[1].loss_pct == 100.0 assert hops[2].ip == "1.1.1.1" assert hops[2].avg_ms == 12.0 + + +async def test_run_in_daemon_thread_returns_the_function_result(): + result = await _run_in_daemon_thread(lambda a, b: a + b, 2, 3) + assert result == 5 + + +async def test_run_in_daemon_thread_propagates_exceptions(): + def boom(): + raise ValueError("nope") + + with pytest.raises(ValueError, match="nope"): + await _run_in_daemon_thread(boom) + + +def test_run_in_daemon_thread_never_creates_a_non_daemon_worker_thread(): + # This is the crux of the process-exit hang fix: asyncio.to_thread()/the + # default executor spawn non-daemon ThreadPoolExecutor workers that get + # joined by concurrent.futures' atexit handler, which is exactly what let + # an orphaned, uncancellable IcmpSendEcho2 call keep the whole process + # alive. We can't unit-test "process exit is never blocked" directly (that + # needs a real subprocess), but we can pin the one property that makes it + # true: every thread this helper spawns is a plain daemon thread, never + # routed through a ThreadPoolExecutor. + seen: list[threading.Thread] = [] + original_init = threading.Thread.__init__ + + def spy_init(self, *args, **kwargs): + original_init(self, *args, **kwargs) + seen.append(self) + + async def run(): + await _run_in_daemon_thread(lambda: None) + + threading.Thread.__init__ = spy_init + try: + asyncio.run(run()) + finally: + threading.Thread.__init__ = original_init + + assert seen + assert all(t.daemon for t in seen)