From 9e8d68e712becbbcebe145e71e5511d6ff4cf935 Mon Sep 17 00:00:00 2001 From: Andreas Wrede Date: Mon, 21 Sep 2026 09:19:48 -0400 Subject: [PATCH] feat(client): per-protocol connectivity probe Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01Uc9V1v7iXruubpoLiSE24M --- hbd/client/connectivity.py | 202 +++++++++++++++++++++++++++++++++++++ tests/test_connectivity.py | 158 +++++++++++++++++++++++++++++ 2 files changed, 360 insertions(+) create mode 100644 hbd/client/connectivity.py create mode 100644 tests/test_connectivity.py diff --git a/hbd/client/connectivity.py b/hbd/client/connectivity.py new file mode 100644 index 0000000..1250213 --- /dev/null +++ b/hbd/client/connectivity.py @@ -0,0 +1,202 @@ +"""Per-protocol internet presence probe. + +Tier 1 (every tier1_interval seconds, sends no packets): a UDP connect() to the probe +address reveals whether a route exists and which source address would be used. +Tier 2 (every probe_interval seconds, sooner after a failure or a change of address or +gateway): one ICMP echo to the probe address. Absence needs fail_limit consecutive +failures (or no route at all); presence needs one success. + +The result is read by the heartbeat sender (p4/p6 flags on every HTB) and by the +network_monitor plugin (connectivity metric) via the shared instance. +""" + +import asyncio +import logging +import socket +import sys +import time +from dataclasses import asdict, dataclass, replace +from typing import Any, Awaitable, Callable, Dict, Optional, Set, Tuple + +from hbd.client import gateway as gateway_mod +from hbd.client.pingutil import PingUnavailable, run_ping + +logger = logging.getLogger("hbc.connectivity") + +FAMILIES = ("ipv4", "ipv6") +FLAG_KEYS = {"ipv4": "p4", "ipv6": "p6"} +DEFAULT_PROBES = {"ipv4": "1.1.1.1", "ipv6": "2606:4700:4700::1111"} + + +class ProbeUnavailable(Exception): + """The address family cannot be probed on this host (e.g. no IPv6 support).""" + + +@dataclass +class FamilyState: + present: Optional[bool] = None # None = unknown (probe cannot run yet/at all) + addr: Optional[str] = None + gateway: Optional[str] = None + rtt: Optional[float] = None + + +def local_address(family: str, target: str) -> Optional[str]: + """Source address the kernel would use to reach target, or None if there is no route. + + A UDP connect() sends nothing. Raises ProbeUnavailable if the family is unsupported. + """ + af = socket.AF_INET if family == "ipv4" else socket.AF_INET6 + try: + sock = socket.socket(af, socket.SOCK_DGRAM) + except OSError as e: + raise ProbeUnavailable(str(e)) from e + try: + sock.connect((target, 9)) + return str(sock.getsockname()[0]) + except OSError: + return None + finally: + sock.close() + + +async def ping_rtt(target: str, family: str) -> Optional[float]: + """One ICMP echo; returns RTT in ms or None if no reply.""" + result = await run_ping(target, 1, 2, 4 if family == "ipv4" else 6) + rtt = result["rtt_avg"] + return None if rtt == float("inf") else rtt + + +class Connectivity: + def __init__( + self, + probes: Optional[Dict[str, str]] = None, + probe_interval: float = 60.0, + retry_interval: float = 10.0, + tier1_interval: float = 5.0, + fail_limit: int = 2, + local_address_fn: Callable[[str, str], Optional[str]] = local_address, + ping_fn: Callable[[str, str], Awaitable[Optional[float]]] = ping_rtt, + gateway_fn: Callable[[str], Awaitable[Optional[str]]] = gateway_mod.default_gateway, + clock: Callable[[], float] = time.monotonic, + ): + self.probes = {**DEFAULT_PROBES, **(probes or {})} + self.probe_interval = probe_interval + self.retry_interval = retry_interval + self.tier1_interval = tier1_interval + self.fail_limit = fail_limit + self._local_address = local_address_fn + self._ping = ping_fn + self._gateway_fn = gateway_fn + self._clock = clock + # Reading the gateway is a cheap /proc read on Linux but a subprocess elsewhere, + # so elsewhere it is refreshed only when a tier-2 probe is due. + self._cheap_gateway = sys.platform.startswith("linux") + self.states: Dict[str, FamilyState] = {f: FamilyState() for f in FAMILIES} + self._fails: Dict[str, int] = {f: 0 for f in FAMILIES} + self._next_ping: Dict[str, float] = {f: 0.0 for f in FAMILIES} + self._last_key: Dict[str, Optional[Tuple[Optional[str], Optional[str]]]] = { + f: None for f in FAMILIES} + self._ping_disabled: Set[str] = set() + + @classmethod + def from_config(cls, cfg: Optional[Dict[str, Any]]) -> "Connectivity": + cfg = cfg or {} + probes = {} + if cfg.get("probe_v4"): + probes["ipv4"] = str(cfg["probe_v4"]) + if cfg.get("probe_v6"): + probes["ipv6"] = str(cfg["probe_v6"]) + return cls(probes=probes, probe_interval=float(cfg.get("probe_interval", 60))) + + # ---- readers (cheap; safe to call on every heartbeat) -------------------------- + def snapshot(self) -> Dict[str, FamilyState]: + return {f: replace(s) for f, s in self.states.items()} + + def flags(self) -> Dict[str, bool]: + """HTB flags for families whose presence is known.""" + return {FLAG_KEYS[f]: s.present for f, s in self.states.items() if s.present is not None} + + def as_metric(self) -> Dict[str, Dict[str, Any]]: + return {f: asdict(s) for f, s in self.states.items()} + + # ---- probing ------------------------------------------------------------------- + async def refresh(self) -> None: + for fam in FAMILIES: + try: + await self._refresh_family(fam) + except Exception: + logger.exception("connectivity probe failed for %s", fam) + + async def run(self, stop: asyncio.Event) -> None: + while not stop.is_set(): + await self.refresh() + try: + await asyncio.wait_for(stop.wait(), timeout=self.tier1_interval) + except asyncio.TimeoutError: + pass + + async def _gateway(self, fam: str) -> Optional[str]: + try: + return await self._gateway_fn(fam) + except Exception: + return None + + async def _refresh_family(self, fam: str) -> None: + st = self.states[fam] + try: + addr = self._local_address(fam, self.probes[fam]) + except ProbeUnavailable: + st.present = None + return + + if addr is None: # tier 1: no route => absent immediately + self._fails[fam] = 0 + self._next_ping[fam] = 0.0 + self._last_key[fam] = None + st.present, st.addr, st.gateway, st.rtt = False, None, None, None + return + + now = self._clock() + gw = await self._gateway(fam) if self._cheap_gateway else st.gateway + key = (addr, gw) + due = key != self._last_key[fam] or now >= self._next_ping[fam] + if due and not self._cheap_gateway: + gw = await self._gateway(fam) + key = (addr, gw) + st.addr, st.gateway = addr, gw + self._last_key[fam] = key + if due and fam not in self._ping_disabled: + await self._tier2(fam, st, now) + + async def _tier2(self, fam: str, st: FamilyState, now: float) -> None: + try: + rtt = await self._ping(self.probes[fam], fam) + except PingUnavailable: + logger.warning("ping unavailable; %s presence stays unknown", fam) + self._ping_disabled.add(fam) + st.present = None + return + except (asyncio.TimeoutError, OSError): + rtt = None + if rtt is None: + self._fails[fam] += 1 + self._next_ping[fam] = now + self.retry_interval + if self._fails[fam] >= self.fail_limit: + st.present, st.rtt = False, None + else: + self._fails[fam] = 0 + self._next_ping[fam] = now + self.probe_interval + st.present, st.rtt = True, rtt + + +# ---- shared instance (heartbeat sender and network_monitor plugin) ----------------- +_shared: Optional[Connectivity] = None + + +def set_shared(conn: Optional[Connectivity]) -> None: + global _shared + _shared = conn + + +def get_shared() -> Optional[Connectivity]: + return _shared diff --git a/tests/test_connectivity.py b/tests/test_connectivity.py new file mode 100644 index 0000000..60d2237 --- /dev/null +++ b/tests/test_connectivity.py @@ -0,0 +1,158 @@ +import asyncio + +from hbd.client.connectivity import Connectivity, ProbeUnavailable +from hbd.client.pingutil import PingUnavailable + + +class Net: + """Scriptable network: addr None = no route, 'unavailable' = family unsupported.""" + + def __init__(self): + self.addr = {"ipv4": "192.0.2.5", "ipv6": "2001:db8::5"} + self.gw = {"ipv4": "192.0.2.1", "ipv6": "fe80::1"} + self.rtt = {"ipv4": 10.0, "ipv6": 20.0} # None = ping fails, 'unavailable' = no ping binary + self.pings = [] + self.now = 1000.0 + + def local(self, fam, target): + a = self.addr[fam] + if a == "unavailable": + raise ProbeUnavailable("no such family") + return a + + async def ping(self, target, fam): + self.pings.append(fam) + r = self.rtt[fam] + if r == "unavailable": + raise PingUnavailable("no ping") + return r + + async def gateway(self, fam): + return self.gw[fam] + + def clock(self): + return self.now + + +def make(net, **kw): + c = Connectivity(local_address_fn=net.local, ping_fn=net.ping, gateway_fn=net.gateway, + clock=net.clock, **kw) + c._cheap_gateway = True + return c + + +def refresh(c): + asyncio.run(c.refresh()) + + +def test_present_with_addr_gateway_rtt(): + net = Net() + c = make(net) + refresh(c) + s = c.snapshot()["ipv4"] + assert (s.present, s.addr, s.gateway, s.rtt) == (True, "192.0.2.5", "192.0.2.1", 10.0) + assert c.flags() == {"p4": True, "p6": True} + assert c.as_metric()["ipv6"]["gateway"] == "fe80::1" + + +def test_no_route_is_immediately_absent(): + net = Net() + net.addr["ipv6"] = None + c = make(net) + refresh(c) + assert c.flags() == {"p4": True, "p6": False} + assert net.pings == ["ipv4"] # no ping wasted on a family without a route + assert c.snapshot()["ipv6"].addr is None + + +def test_ping_is_debounced_and_rate_limited(): + net = Net() + c = make(net, probe_interval=60.0, retry_interval=10.0) + refresh(c) + refresh(c) # same instant: cached, no second ping + assert net.pings == ["ipv4", "ipv6"] + net.now += 60 + net.rtt["ipv6"] = None # first failure: still present (debounce) + refresh(c) + assert c.flags()["p6"] is True + net.now += 10 # retry_interval elapsed: second failure flips it + refresh(c) + assert c.flags()["p6"] is False + + +def test_first_probe_failing_once_is_unknown(): + net = Net() + net.rtt["ipv4"] = None + c = make(net) + refresh(c) + assert "p4" not in c.flags() and c.flags()["p6"] is True + + +def test_recovers_on_first_success(): + net = Net() + net.rtt["ipv6"] = None + c = make(net, retry_interval=10.0) + refresh(c) + net.now += 10 + refresh(c) + assert c.flags()["p6"] is False + net.rtt["ipv6"] = 25.0 + net.now += 10 + refresh(c) + assert c.flags()["p6"] is True + + +def test_address_change_invalidates_cache(): + net = Net() + c = make(net) + refresh(c) + net.addr["ipv4"] = "198.51.100.7" # moved networks, no time has passed + refresh(c) + assert net.pings.count("ipv4") == 2 + assert c.snapshot()["ipv4"].addr == "198.51.100.7" + + +def test_gateway_change_invalidates_cache(): + net = Net() + c = make(net) + refresh(c) + net.gw["ipv4"] = "192.0.2.254" + refresh(c) + assert net.pings.count("ipv4") == 2 + + +def test_route_loss_then_return_pings_immediately(): + net = Net() + c = make(net) + refresh(c) + net.addr["ipv6"] = None + refresh(c) + net.addr["ipv6"] = "2001:db8::5" + refresh(c) + assert net.pings.count("ipv6") == 2 and c.flags()["p6"] is True + + +def test_missing_ping_binary_leaves_presence_unknown_and_stops_retrying(): + net = Net() + net.rtt["ipv4"] = "unavailable" + c = make(net) + refresh(c) + net.now += 600 + refresh(c) + assert "p4" not in c.flags() + assert net.pings.count("ipv4") == 1 + + +def test_unsupported_family_is_unknown(): + net = Net() + net.addr["ipv6"] = "unavailable" + c = make(net) + refresh(c) + assert "p6" not in c.flags() and c.flags()["p4"] is True + + +def test_from_config(): + c = Connectivity.from_config({"probe_v4": "9.9.9.9", "probe_interval": 30}) + assert c.probes["ipv4"] == "9.9.9.9" + assert c.probes["ipv6"] == "2606:4700:4700::1111" + assert c.probe_interval == 30.0