From dd2410aa667d4703ef28a0747bf979ee5778fd34 Mon Sep 17 00:00:00 2001 From: Jer Miller Date: Wed, 29 Jul 2026 21:10:17 -0600 Subject: [PATCH] fix(link): bound the PL dial and hold backoff across short-lived sessions The manager loop dialed with no overall bound, so the effective bound was whatever the composed per-attempt sub-bounds happened to produce, and one stall class was unbounded outright. It now computes one absolute monotonic instant per attempt and threads it through, so a stalled dial raises into the existing failure and backoff path instead of hanging. Relay enrollment ran outside the deadline, so on the primary remote path a stalled enrollment left the dial fan-out waiting. It is now covered by the same instant. The worker thread cannot be cancelled and may outlive the timeout, which is acceptable because the loop escapes and retries. Backoff reset on a successful dial rather than on a session that stayed up, so a connect-then-die loop retried near the floor indefinitely. Reset is now gated on the adopted session surviving a stability threshold, measured monotonically. Both bounds are constructor keywords with named defaults; no config key. Co-Authored-By: Claude Opus 5 (1M context) --- solstone/think/link/dialer.py | 30 ++- tests/link/test_dialer_unit.py | 392 +++++++++++++++++++++++++++++++-- tests/test_export_pl.py | 2 +- tests/test_transfer_pl.py | 2 +- 4 files changed, 399 insertions(+), 27 deletions(-) diff --git a/solstone/think/link/dialer.py b/solstone/think/link/dialer.py index 76c9928e4..674e17dca 100644 --- a/solstone/think/link/dialer.py +++ b/solstone/think/link/dialer.py @@ -32,8 +32,13 @@ from solstone.think.link.tls import TlsError _REQUEST_TIMEOUT_SECONDS = 180 _QUEUE_PUT_TIMEOUT_SECONDS = 0.1 _REQUEST_SESSION_WAIT_SECONDS = 5.0 +# Overall dial attempt bound, separate from request timeout and per-recv bounds. +_DIAL_TIMEOUT_SECONDS = 30 _RECONNECT_INITIAL_BACKOFF_SECONDS = 1.0 _RECONNECT_MAX_BACKOFF_SECONDS = 30.0 +# Short-lived connect-then-die sessions can last long enough to pass small +# thresholds; keep the default at a minute so they do not reset backoff. +_SESSION_STABLE_AFTER_SECONDS = 60 _SESSION_POLL_SECONDS = 0.25 STATE_DISCONNECTED = "disconnected" @@ -135,7 +140,12 @@ async def _dial_relay( identity: ClientIdentity, deadline: float | None = None, ) -> TunnelSession: - enrolled = await asyncio.to_thread(client.enroll_device, relay_url, identity) + # The worker thread cannot be cancelled, so it may outlive this timeout until + # requests returns; the manager loop still escapes and retries. + enrolled = await _with_deadline( + asyncio.to_thread(client.enroll_device, relay_url, identity), + deadline, + ) return await _with_deadline(client.dial(relay_url, enrolled), deadline) @@ -214,15 +224,19 @@ class TunnelClient: *, request_timeout: float = _REQUEST_TIMEOUT_SECONDS, request_session_wait: float = _REQUEST_SESSION_WAIT_SECONDS, + dial_timeout: float = _DIAL_TIMEOUT_SECONDS, reconnect_initial_backoff: float = _RECONNECT_INITIAL_BACKOFF_SECONDS, reconnect_max_backoff: float = _RECONNECT_MAX_BACKOFF_SECONDS, + session_stable_after: float = _SESSION_STABLE_AFTER_SECONDS, ) -> None: self._identity = identity self._relay_url = relay_url.rstrip("/") if relay_url else None self._request_timeout = request_timeout self._request_session_wait = request_session_wait + self._dial_timeout = dial_timeout self._reconnect_initial_backoff = reconnect_initial_backoff self._reconnect_max_backoff = reconnect_max_backoff + self._session_stable_after = session_stable_after self._loop: asyncio.AbstractEventLoop | None = None self._loop_thread: threading.Thread | None = None self._session: TunnelSession | None = None @@ -293,7 +307,14 @@ class TunnelClient: while not self._closed: await self._set_lifecycle_state(STATE_CONNECTING) try: - session = await open_tunnel(self._identity, self._relay_url) + # _with_deadline expects an absolute monotonic instant. Passing a + # duration would clamp to zero on machines up longer than the bound. + deadline = time.monotonic() + self._dial_timeout + session = await open_tunnel( + self._identity, + self._relay_url, + deadline=deadline, + ) except asyncio.CancelledError: raise except Exception as exc: @@ -307,11 +328,14 @@ class TunnelClient: backoff = min(self._reconnect_max_backoff, backoff * 2) continue - backoff = self._reconnect_initial_backoff + adopted_at = time.monotonic() await self._adopt_session(session) await _wait_session_closed(session, is_closed=lambda: self._closed) if self._closed: return + # Stability is about how long the adopted session survived, not which path cleared it. + if time.monotonic() - adopted_at >= self._session_stable_after: + backoff = self._reconnect_initial_backoff if self._session is session: failure_reason = getattr(session, "failure_reason", None) if callable(failure_reason): diff --git a/tests/link/test_dialer_unit.py b/tests/link/test_dialer_unit.py index 00bc6def3..1dd00cb75 100644 --- a/tests/link/test_dialer_unit.py +++ b/tests/link/test_dialer_unit.py @@ -152,6 +152,341 @@ def _managed_client(**kwargs) -> TunnelClient: ) +def _spy_lifecycle_failures( + client: TunnelClient, + *, + after: int, +) -> tuple[list[tuple[str, str, float | None]], threading.Event]: + records: list[tuple[str, str, float | None]] = [] + observed = threading.Event() + original = client._record_lifecycle_failure + + async def record_spy( + reason: str, + detail: str, + *, + state: str, + next_retry_in: float | None = None, + ) -> None: + records.append((reason, detail, next_retry_in)) + if len(records) >= after: + observed.set() + await original( + reason, + detail, + state=state, + next_retry_in=next_retry_in, + ) + + client._record_lifecycle_failure = record_spy + return records, observed + + +async def _finish_session_after(session: _ManagedSession, delay: float) -> None: + loop = asyncio.get_running_loop() + done: asyncio.Future[None] = loop.create_future() + + def finish() -> None: + session._finish_closed() + if not done.done(): + done.set_result(None) + + loop.call_later(delay, finish) + await done + + +def _remaining_deadline(deadline: float) -> float: + return max(0.0, deadline - time.monotonic()) + + +def test_connection_manager_passes_default_absolute_dial_deadline(monkeypatch) -> None: + deadlines: list[float | None] = [] + called = threading.Event() + + async def fake_open_tunnel(_identity, _relay_url, *, deadline=None, **_kwargs): + deadlines.append(deadline) + called.set() + return _ManagedSession() + + monkeypatch.setattr(dialer, "open_tunnel", fake_open_tunnel) + client = _managed_client() + try: + before = time.monotonic() + client.start() + assert called.wait(timeout=1) + after = time.monotonic() + finally: + client.close() + + assert deadlines[0] is not None + assert before + dialer._DIAL_TIMEOUT_SECONDS <= deadlines[0] + assert deadlines[0] <= after + dialer._DIAL_TIMEOUT_SECONDS + + +def test_connection_manager_dial_deadline_bounds_retries(monkeypatch) -> None: + seen: list[float | None] = [] + calls = 0 + + async def fake_open_tunnel(_identity, _relay_url, *, deadline=None, **_kwargs): + nonlocal calls + calls += 1 + seen.append(deadline) + if deadline is None: + raise AssertionError("dial received no deadline") + await asyncio.wait_for( + asyncio.sleep(3600), + timeout=_remaining_deadline(deadline), + ) + raise AssertionError("unreachable") + + monkeypatch.setattr(dialer, "open_tunnel", fake_open_tunnel) + client = _managed_client( + dial_timeout=0.01, + reconnect_initial_backoff=0.005, + reconnect_max_backoff=0.005, + request_session_wait=0.01, + ) + records, observed = _spy_lifecycle_failures(client, after=2) + try: + client.start() + assert observed.wait(timeout=1) + finally: + client.close() + + assert calls > 1 + assert all(deadline is not None for deadline in seen) + assert any(reason == "TimeoutError" for reason, _detail, _retry in records) + + +def test_connection_manager_adopts_session_inside_dial_bound(monkeypatch) -> None: + sessions: list[_ManagedSession] = [] + + async def fake_open_tunnel(_identity, _relay_url, *, deadline=None, **_kwargs): + if deadline is None: + raise AssertionError("dial received no deadline") + await asyncio.wait_for( + asyncio.sleep(0.01), + timeout=_remaining_deadline(deadline), + ) + session = _ManagedSession() + sessions.append(session) + return session + + monkeypatch.setattr(dialer, "open_tunnel", fake_open_tunnel) + client = _managed_client(dial_timeout=0.2) + try: + assert client.request("GET", "/inside-bound") == ( + 200, + {"x-session": "fresh"}, + b"ok", + ) + finally: + client.close() + + assert len(sessions) == 1 + assert sessions[0].requests == [("GET", "/inside-bound", {}, b"")] + + +def test_connection_manager_passes_explicit_absolute_dial_deadline(monkeypatch) -> None: + dial_timeout = 0.123 + deadlines: list[float | None] = [] + called = threading.Event() + + async def fake_open_tunnel(_identity, _relay_url, *, deadline=None, **_kwargs): + deadlines.append(deadline) + called.set() + return _ManagedSession() + + monkeypatch.setattr(dialer, "open_tunnel", fake_open_tunnel) + client = _managed_client(dial_timeout=dial_timeout) + try: + before = time.monotonic() + client.start() + assert called.wait(timeout=1) + after = time.monotonic() + finally: + client.close() + + assert deadlines[0] is not None + assert before + dial_timeout <= deadlines[0] + assert deadlines[0] <= after + dial_timeout + + +@pytest.mark.asyncio +async def test_open_tunnel_reports_direct_attempt_timeout_label(monkeypatch) -> None: + async def dial_direct(_host, _enrolled, *, port=7657): + _ = port + await asyncio.sleep(3600) + + monkeypatch.setattr(dialer.Client, "dial_direct", staticmethod(dial_direct)) + identity = _identity(endpoints=({"ip": "10.0.0.1", "port": 7657},)) + + with pytest.raises(TlsError) as exc_info: + await dialer.open_tunnel( + identity, + None, + deadline=time.monotonic() + 0.01, + ) + + message = str(exc_info.value) + assert "lan-direct 10.0.0.1:7657" in message + assert "TimeoutError" in message + + +def test_short_lived_sessions_increase_then_plateau_backoff(monkeypatch) -> None: + async def fake_open_tunnel(_identity, _relay_url, **_kwargs): + session = _ManagedSession() + session._finish_closed() + return session + + monkeypatch.setattr(dialer, "open_tunnel", fake_open_tunnel) + client = _managed_client( + reconnect_initial_backoff=0.005, + reconnect_max_backoff=0.04, + ) + records, observed = _spy_lifecycle_failures(client, after=5) + try: + client.start() + assert observed.wait(timeout=1) + finally: + client.close() + + assert [retry for _reason, _detail, retry in records[:5]] == [ + 0.005, + 0.01, + 0.02, + 0.04, + 0.04, + ] + + +def test_stable_session_resets_recorded_retry_backoff(monkeypatch) -> None: + calls = 0 + held: list[_ManagedSession] = [] + held_ready = threading.Event() + + async def fake_open_tunnel(_identity, _relay_url, **_kwargs): + nonlocal calls + calls += 1 + session = _ManagedSession() + if calls <= 2: + session._finish_closed() + else: + held.append(session) + held_ready.set() + return session + + monkeypatch.setattr(dialer, "open_tunnel", fake_open_tunnel) + client = _managed_client( + reconnect_initial_backoff=0.005, + reconnect_max_backoff=0.04, + session_stable_after=0.02, + ) + records, observed = _spy_lifecycle_failures(client, after=3) + try: + client.start() + assert held_ready.wait(timeout=1) + assert client.request("GET", "/stable") == ( + 200, + {"x-session": "fresh"}, + b"ok", + ) + client._run(_finish_session_after(held[0], 0.03)) + assert observed.wait(timeout=1) + finally: + client.close() + + assert [retry for _reason, _detail, retry in records[:3]] == [ + 0.005, + 0.01, + 0.005, + ] + + +def test_short_session_below_stability_threshold_keeps_backing_off(monkeypatch) -> None: + # Short-lived connect-then-die sessions can last long enough to pass small + # thresholds; keep the production default at a minute so they do not reset backoff. + assert dialer._SESSION_STABLE_AFTER_SECONDS >= 60 + + calls = 0 + held: list[_ManagedSession] = [] + held_ready = threading.Event() + + async def fake_open_tunnel(_identity, _relay_url, **_kwargs): + nonlocal calls + calls += 1 + session = _ManagedSession() + if calls == 1: + session._finish_closed() + else: + held.append(session) + held_ready.set() + return session + + monkeypatch.setattr(dialer, "open_tunnel", fake_open_tunnel) + client = _managed_client( + reconnect_initial_backoff=0.005, + reconnect_max_backoff=0.04, + session_stable_after=0.06, + ) + records, observed = _spy_lifecycle_failures(client, after=2) + try: + client.start() + assert held_ready.wait(timeout=1) + assert client.request("GET", "/short") == ( + 200, + {"x-session": "fresh"}, + b"ok", + ) + # This scaled hold is on the short-lived side of the stability threshold. + client._run(_finish_session_after(held[0], 0.035)) + assert observed.wait(timeout=1) + finally: + client.close() + + assert [retry for _reason, _detail, retry in records[:2]] == [0.005, 0.01] + + +@pytest.mark.asyncio +async def test_relay_enrollment_timeout_reports_relay_attempt(monkeypatch) -> None: + identity = _identity(endpoints=()) + entered = threading.Event() + release = threading.Event() + dial_called = False + + def enroll_device( + _relay_url: str, + enrolled_identity: ClientIdentity, + ) -> EnrolledDevice: + entered.set() + release.wait() + return EnrolledDevice(device_token="token", identity=enrolled_identity) + + async def dial(_relay_url: str, _enrolled: EnrolledDevice) -> object: + nonlocal dial_called + dial_called = True + raise AssertionError("dial should not run after enrollment timeout") + + monkeypatch.setattr(dialer.Client, "enroll_device", staticmethod(enroll_device)) + monkeypatch.setattr(dialer.Client, "dial", staticmethod(dial)) + + try: + with pytest.raises(TlsError) as exc_info: + await dialer.open_tunnel( + identity, + "https://relay.test", + deadline=time.monotonic() + 0.01, + ) + finally: + release.set() + + assert entered.wait(timeout=1) + message = str(exc_info.value) + assert "spl-relay" in message + assert "TimeoutError" in message + assert dial_called is False + + @pytest.mark.asyncio async def test_lan_direct_race_picks_first_and_cancels_loser(monkeypatch) -> None: identity = _identity( @@ -393,7 +728,7 @@ def test_cached_session_drops_on_stream_reset(monkeypatch) -> None: sessions: list[ResetSession] = [] - async def open_tunnel(_identity, _relay_url): + async def open_tunnel(_identity, _relay_url, **_kwargs): session = ResetSession() sessions.append(session) return session @@ -522,7 +857,7 @@ class _HangingSession: def test_connection_manager_redials_after_remote_close(monkeypatch) -> None: sessions: list[_ManagedSession] = [] - async def fake_open_tunnel(_identity, _relay_url): + async def fake_open_tunnel(_identity, _relay_url, **_kwargs): session = _ManagedSession() sessions.append(session) return session @@ -565,7 +900,7 @@ def test_connection_manager_redials_after_remote_close(monkeypatch) -> None: def test_connection_manager_records_liveness_failure(monkeypatch) -> None: sessions: list[_ManagedSession] = [] - async def fake_open_tunnel(_identity, _relay_url): + async def fake_open_tunnel(_identity, _relay_url, **_kwargs): session = _ManagedSession() sessions.append(session) return session @@ -605,30 +940,43 @@ def test_connection_manager_records_liveness_failure(monkeypatch) -> None: def test_request_during_reconnect_fails_with_lifecycle_error(monkeypatch) -> None: + """Request wait and stalled dial timeout are bounded independently.""" started = threading.Event() - async def fake_open_tunnel(_identity, _relay_url): + async def fake_open_tunnel(_identity, _relay_url, *, deadline=None, **_kwargs): started.set() - await asyncio.sleep(3600) + if deadline is None: + raise AssertionError("dial received no deadline") + await asyncio.wait_for( + asyncio.sleep(3600), + timeout=_remaining_deadline(deadline), + ) raise AssertionError("unreachable") monkeypatch.setattr(dialer, "open_tunnel", fake_open_tunnel) - client = _managed_client(request_session_wait=0.02) + client = _managed_client( + dial_timeout=1.0, + request_session_wait=0.05, + reconnect_initial_backoff=0.005, + ) + records, observed = _spy_lifecycle_failures(client, after=1) try: client.start() assert started.wait(timeout=1) with pytest.raises(TunnelLifecycleError) as exc_info: client.request("GET", "/during-reconnect") + assert observed.wait(timeout=2) finally: client.close() assert exc_info.value.state == "connecting" assert exc_info.value.retryable is True assert "no live tunnel session" in exc_info.value.detail + assert records[0][0] == "TimeoutError" def test_dead_manager_status_and_requests_fail_closed(monkeypatch) -> None: - async def fake_open_tunnel(_identity, _relay_url): + async def fake_open_tunnel(_identity, _relay_url, **_kwargs): await asyncio.sleep(3600) raise AssertionError("unreachable") @@ -727,7 +1075,7 @@ def test_connection_manager_opens_session_for_request(monkeypatch) -> None: sessions: list[_ManagedSession] = [] calls = 0 - async def fake_open_tunnel(_identity, _relay_url): + async def fake_open_tunnel(_identity, _relay_url, **_kwargs): nonlocal calls calls += 1 session = _ManagedSession() @@ -753,7 +1101,7 @@ def test_connection_manager_opens_session_for_stream_request(monkeypatch) -> Non sessions: list[_ManagedSession] = [] calls = 0 - async def fake_open_tunnel(_identity, _relay_url): + async def fake_open_tunnel(_identity, _relay_url, **_kwargs): nonlocal calls calls += 1 session = _ManagedSession() @@ -776,7 +1124,7 @@ def test_connection_manager_reuses_live_session_without_redial(monkeypatch) -> N sessions: list[_ManagedSession] = [] calls = 0 - async def fake_open_tunnel(_identity, _relay_url): + async def fake_open_tunnel(_identity, _relay_url, **_kwargs): nonlocal calls calls += 1 session = _ManagedSession() @@ -807,7 +1155,7 @@ def test_connection_manager_reuses_live_session_without_redial(monkeypatch) -> N def test_failed_redial_queues_lifecycle_error(monkeypatch) -> None: - async def fake_open_tunnel(_identity, _relay_url): + async def fake_open_tunnel(_identity, _relay_url, **_kwargs): raise OSError("dial failed") monkeypatch.setattr(dialer, "open_tunnel", fake_open_tunnel) @@ -836,7 +1184,7 @@ def test_request_timeout_argument_can_loosen_constructor_bound(monkeypatch) -> N await asyncio.sleep(0.1) return 200, {"x-timeout": "loosened"}, b"ok" - async def fake_open_tunnel(_identity, _relay_url): + async def fake_open_tunnel(_identity, _relay_url, **_kwargs): return SlowRequestSession() monkeypatch.setattr(dialer, "open_tunnel", fake_open_tunnel) @@ -871,7 +1219,7 @@ def test_request_timeout_argument_can_tighten_large_constructor_bound( sessions: list[HangingManagedSession] = [] - async def fake_open_tunnel(_identity, _relay_url): + async def fake_open_tunnel(_identity, _relay_url, **_kwargs): session = HangingManagedSession() sessions.append(session) return session @@ -897,7 +1245,7 @@ def test_request_timeout_uses_constructor_when_argument_omitted(monkeypatch) -> sessions: list[HangingManagedSession] = [] - async def fake_open_tunnel(_identity, _relay_url): + async def fake_open_tunnel(_identity, _relay_url, **_kwargs): session = HangingManagedSession() sessions.append(session) return session @@ -921,7 +1269,7 @@ def test_bare_stream_request_honors_per_request_timeout(monkeypatch) -> None: async def stream_request(self, *_args, **_kwargs): await asyncio.sleep(3600) - async def fake_open_tunnel(_identity, _relay_url): + async def fake_open_tunnel(_identity, _relay_url, **_kwargs): return HangingManagedSession() monkeypatch.setattr(dialer, "open_tunnel", fake_open_tunnel) @@ -943,7 +1291,7 @@ def test_proxy_stream_request_honors_per_request_timeout(monkeypatch) -> None: sessions: list[HangingManagedSession] = [] - async def fake_open_tunnel(_identity, _relay_url): + async def fake_open_tunnel(_identity, _relay_url, **_kwargs): session = HangingManagedSession() sessions.append(session) return session @@ -983,7 +1331,7 @@ def test_stream_request_timeout_does_not_cover_response_tail(monkeypatch) -> Non self.stream_requests.append((method, path, headers, body)) return 200, {}, b"", SlowTailStream() - async def fake_open_tunnel(_identity, _relay_url): + async def fake_open_tunnel(_identity, _relay_url, **_kwargs): return SlowTailSession() monkeypatch.setattr(dialer, "open_tunnel", fake_open_tunnel) @@ -1013,7 +1361,7 @@ def test_proxy_stream_request_times_out_during_head_and_clears_session() -> None sessions: list[HangingManagedSession] = [] - async def fake_open_tunnel(_identity, _relay_url): + async def fake_open_tunnel(_identity, _relay_url, **_kwargs): session = HangingManagedSession() sessions.append(session) return session @@ -1042,7 +1390,7 @@ def test_bare_stream_request_times_out_during_head(monkeypatch) -> None: async def stream_request(self, *_args, **_kwargs): await asyncio.sleep(3600) - async def fake_open_tunnel(_identity, _relay_url): + async def fake_open_tunnel(_identity, _relay_url, **_kwargs): return HangingManagedSession() monkeypatch.setattr(dialer, "open_tunnel", fake_open_tunnel) @@ -1061,7 +1409,7 @@ def test_request_times_out_during_head_and_clears_session(monkeypatch) -> None: sessions: list[HangingManagedSession] = [] - async def fake_open_tunnel(_identity, _relay_url): + async def fake_open_tunnel(_identity, _relay_url, **_kwargs): session = HangingManagedSession() sessions.append(session) return session @@ -1087,7 +1435,7 @@ def test_request_timeout_is_not_armed_during_body_streaming() -> None: sessions: list[SlowBodySession] = [] - async def fake_open_tunnel(_identity, _relay_url): + async def fake_open_tunnel(_identity, _relay_url, **_kwargs): session = SlowBodySession() sessions.append(session) return session @@ -1115,7 +1463,7 @@ async def test_dead_session_redial_is_single_flight(monkeypatch) -> None: sessions: list[_ManagedSession] = [] calls = 0 - async def fake_open_tunnel(_identity, _relay_url): + async def fake_open_tunnel(_identity, _relay_url, **_kwargs): nonlocal calls calls += 1 await asyncio.sleep(0) diff --git a/tests/test_export_pl.py b/tests/test_export_pl.py index 6336975f9..94e738cfb 100644 --- a/tests/test_export_pl.py +++ b/tests/test_export_pl.py @@ -86,7 +86,7 @@ def _patch_tunnel( async def close(self) -> None: return None - async def fake_open_tunnel(_identity, _relay_url): + async def fake_open_tunnel(_identity, _relay_url, **_kwargs): return FakeTunnelSession() monkeypatch.setattr("solstone.think.link.dialer.open_tunnel", fake_open_tunnel) diff --git a/tests/test_transfer_pl.py b/tests/test_transfer_pl.py index ca9ddb881..8b7f74959 100644 --- a/tests/test_transfer_pl.py +++ b/tests/test_transfer_pl.py @@ -123,7 +123,7 @@ def test_transfer_send_pl_posts_journal_segment_day_path( async def close(self) -> None: return None - async def fake_open_tunnel(_identity, _relay_url): + async def fake_open_tunnel(_identity, _relay_url, **_kwargs): return FakeTunnelSession() monkeypatch.setattr("solstone.think.link.dialer.open_tunnel", fake_open_tunnel) -- 2.51.2