From ad25134ebfa345cf19b5b21169ddc5264450eb7e Mon Sep 17 00:00:00 2001 From: Jer Miller Date: Wed, 29 Jul 2026 17:28:44 -0600 Subject: [PATCH] fix(link): honor the caller's PL request timeout PlHttpSession.post() and .get() accepted a timeout but discarded it. Private-link requests were instead bounded by a constant named for session establishment, and that bound also covered the request-body send. Rename the timeout constant and TunnelClient constructor keyword to describe what they bound: _REQUEST_TIMEOUT_SECONDS and request_timeout=. Move the default from 30s to 180s so a worst-case segment body fits on a slow link while a wedged tunnel remains bounded. All three public entry points, request, stream_request, and proxy_stream_request, now accept a per-request timeout. Precedence is per-request argument, then constructor keyword, then module constant, resolved in one helper. The bound remains; it is now the caller's bound. Response-tail streaming stays outside it. Co-Authored-By: GPT-5 Codex --- solstone/observe/pl_http.py | 18 ++- solstone/think/link/dialer.py | 33 +++++- tests/link/test_dialer_unit.py | 202 +++++++++++++++++++++++++++++++-- tests/test_export_pl.py | 46 +++++++- tests/test_pl_http.py | 43 +++++++ tests/test_transfer_pl.py | 46 +++++++- 6 files changed, 366 insertions(+), 22 deletions(-) create mode 100644 tests/test_pl_http.py diff --git a/solstone/observe/pl_http.py b/solstone/observe/pl_http.py index 9c7ee49a3..af4e7794e 100644 --- a/solstone/observe/pl_http.py +++ b/solstone/observe/pl_http.py @@ -48,7 +48,7 @@ class PlHttpSession: json: Any = None, files: Any = None, headers: dict[str, str] | None = None, - timeout: float | tuple[float, float | None] | None = None, + timeout: float | None = None, ) -> PlHttpResponse: request_headers = self._headers(headers, url) body = b"" @@ -70,16 +70,24 @@ class PlHttpSession: else: body = bytes(data) - return self._request("POST", url, headers=request_headers, body=body) + return self._request( + "POST", url, headers=request_headers, body=body, timeout=timeout + ) def get( self, url: str, *, headers: dict[str, str] | None = None, - timeout: float | tuple[float, float | None] | None = None, + timeout: float | None = None, ) -> PlHttpResponse: - return self._request("GET", url, headers=self._headers(headers, url), body=b"") + return self._request( + "GET", + url, + headers=self._headers(headers, url), + body=b"", + timeout=timeout, + ) def close(self) -> None: self._tunnel.close() @@ -91,6 +99,7 @@ class PlHttpSession: *, headers: dict[str, str], body: bytes, + timeout: float | None = None, ) -> PlHttpResponse: path = _path_from_url(url) status, response_headers, response_body = self._tunnel.request( @@ -98,6 +107,7 @@ class PlHttpSession: path, headers=headers, body=body, + timeout=timeout, ) return PlHttpResponse(status, dict(response_headers), response_body) diff --git a/solstone/think/link/dialer.py b/solstone/think/link/dialer.py index 4b4958f8a..76c9928e4 100644 --- a/solstone/think/link/dialer.py +++ b/solstone/think/link/dialer.py @@ -24,7 +24,12 @@ from solstone.think.link.client import ( ) from solstone.think.link.tls import TlsError -_ESTABLISH_TIMEOUT_SECONDS = 30 +# Request timeout, not session establishment: a single request may include a +# worst-case segment body. The max observed body is 26.4 MB = 211.2 Mbit, +# which needs ~105.6s at 2 Mbit/s; 180s clears that with headroom while +# still bounding a wedged tunnel. The old 30s establishment-shaped bound +# could not fit a request body, which is why it was wrong. +_REQUEST_TIMEOUT_SECONDS = 180 _QUEUE_PUT_TIMEOUT_SECONDS = 0.1 _REQUEST_SESSION_WAIT_SECONDS = 5.0 _RECONNECT_INITIAL_BACKOFF_SECONDS = 1.0 @@ -207,14 +212,14 @@ class TunnelClient: identity: ClientIdentity, relay_url: str | None, *, - establish_timeout: float = _ESTABLISH_TIMEOUT_SECONDS, + request_timeout: float = _REQUEST_TIMEOUT_SECONDS, request_session_wait: float = _REQUEST_SESSION_WAIT_SECONDS, reconnect_initial_backoff: float = _RECONNECT_INITIAL_BACKOFF_SECONDS, reconnect_max_backoff: float = _RECONNECT_MAX_BACKOFF_SECONDS, ) -> None: self._identity = identity self._relay_url = relay_url.rstrip("/") if relay_url else None - self._establish_timeout = establish_timeout + self._request_timeout = request_timeout self._request_session_wait = request_session_wait self._reconnect_initial_backoff = reconnect_initial_backoff self._reconnect_max_backoff = reconnect_max_backoff @@ -455,6 +460,11 @@ class TunnelClient: return self._run(self._close_session_async()) + def _resolve_request_timeout(self, timeout: float | None) -> float: + if timeout is not None: + return timeout + return self._request_timeout + def request( self, method: str, @@ -462,6 +472,7 @@ class TunnelClient: *, headers: dict[str, str] | None = None, body: bytes | BodySource = b"", + timeout: float | None = None, ) -> tuple[int, dict[str, str], bytes]: try: return self._run( @@ -470,6 +481,7 @@ class TunnelClient: path, headers=headers or {}, body=body, + timeout=timeout, ) ) except TunnelLifecycleError: @@ -485,11 +497,12 @@ class TunnelClient: *, headers: dict[str, str], body: bytes | BodySource, + timeout: float | None = None, ) -> tuple[int, dict[str, str], bytes]: await self._begin_active_request() try: session = await self._get_session_async() - async with asyncio.timeout(self._establish_timeout): + async with asyncio.timeout(self._resolve_request_timeout(timeout)): return await session.request(method, path, headers=headers, body=body) finally: await self._end_active_request() @@ -502,6 +515,7 @@ class TunnelClient: headers: dict[str, str] | None = None, body: bytes | BodySource = b"", chunks: queue.Queue[TunnelResponseHead | bytes | Exception | None], + timeout: float | None = None, ) -> Future[None]: """Stream a proxy response to a queue. @@ -517,6 +531,7 @@ class TunnelClient: headers=headers or {}, body=body, chunks=chunks, + timeout=timeout, ), loop, ) @@ -529,6 +544,7 @@ class TunnelClient: headers: dict[str, str] | None = None, body: bytes | BodySource = b"", chunks: queue.Queue[bytes | Exception | None] | None = None, + timeout: float | None = None, ) -> Future[None] | tuple[int, dict[str, str], bytes, Any]: if chunks is None: return self._run( @@ -537,6 +553,7 @@ class TunnelClient: path, headers=headers or {}, body=body, + timeout=timeout, ) ) loop = self._ensure_loop() @@ -547,6 +564,7 @@ class TunnelClient: headers=headers or {}, body=body, chunks=chunks, + timeout=timeout, ), loop, ) @@ -558,9 +576,10 @@ class TunnelClient: *, headers: dict[str, str], body: bytes | BodySource, + timeout: float | None = None, ) -> tuple[int, dict[str, str], bytes, Any]: session = await self._get_session_async() - async with asyncio.timeout(self._establish_timeout): + async with asyncio.timeout(self._resolve_request_timeout(timeout)): return await session.stream_request( method, path, headers=headers, body=body ) @@ -573,6 +592,7 @@ class TunnelClient: headers: dict[str, str], body: bytes | BodySource, chunks: queue.Queue[TunnelResponseHead | bytes | Exception | None], + timeout: float | None = None, ) -> None: stream: Any | None = None cancelled = False @@ -588,6 +608,7 @@ class TunnelClient: path, headers=headers, body=body, + timeout=timeout, ) await _put_queue_item( chunks, TunnelResponseHead(status, dict(resp_headers)) @@ -624,6 +645,7 @@ class TunnelClient: headers: dict[str, str], body: bytes | BodySource, chunks: queue.Queue[bytes | Exception | None], + timeout: float | None = None, ) -> None: stream: Any | None = None cancelled = False @@ -634,6 +656,7 @@ class TunnelClient: path, headers=headers, body=body, + timeout=timeout, ) if status == 200: if initial_body: diff --git a/tests/link/test_dialer_unit.py b/tests/link/test_dialer_unit.py index 35062a693..00bc6def3 100644 --- a/tests/link/test_dialer_unit.py +++ b/tests/link/test_dialer_unit.py @@ -10,6 +10,7 @@ import contextlib import gc import queue import threading +import time from collections.abc import Iterator from concurrent.futures import CancelledError @@ -419,7 +420,7 @@ def test_proxy_stream_request_queues_head_body_and_sentinel(monkeypatch) -> None client = TunnelClient(_identity(endpoints=()), None) calls = [] - async def fake_stream_request_async(method, path, *, headers, body): + async def fake_stream_request_async(method, path, *, headers, body, timeout=None): calls.append((method, path, headers, body)) return 418, {"x-test": "yes"}, b"initial", FakeStream() @@ -448,8 +449,8 @@ def test_proxy_stream_request_queues_head_body_and_sentinel(monkeypatch) -> None def test_proxy_stream_request_queues_tunnel_error_and_sentinel(monkeypatch) -> None: client = TunnelClient(_identity(endpoints=()), None) - async def fake_stream_request_async(_method, _path, *, headers, body): - _ = (headers, body) + async def fake_stream_request_async(_method, _path, *, headers, body, timeout=None): + _ = (headers, body, timeout) raise ConnectionError("down") monkeypatch.setattr(client, "_stream_request_async", fake_stream_request_async) @@ -666,7 +667,7 @@ def test_proxy_stream_request_accepts_body_source(monkeypatch) -> None: source = BodySource(6, (b"ab", b"cd", b"ef")) calls = [] - async def fake_stream_request_async(method, path, *, headers, body): + async def fake_stream_request_async(method, path, *, headers, body, timeout=None): calls.append((method, path, headers, body)) return 200, {}, b"", FakeStream() @@ -704,8 +705,8 @@ def test_proxy_stream_request_cancel_resets_remote_stream(monkeypatch) -> None: client = TunnelClient(_identity(endpoints=()), None) - async def fake_stream_request_async(_method, _path, *, headers, body): - _ = (headers, body) + async def fake_stream_request_async(_method, _path, *, headers, body, timeout=None): + _ = (headers, body, timeout) return 200, {}, b"", CancellableStream() monkeypatch.setattr(client, "_stream_request_async", fake_stream_request_async) @@ -826,6 +827,185 @@ def test_failed_redial_queues_lifecycle_error(monkeypatch) -> None: assert client._session is None +def test_request_timeout_argument_can_loosen_constructor_bound(monkeypatch) -> None: + class SlowRequestSession(_ManagedSession): + async def request(self, method, path, *, headers, body): + self.requests.append((method, path, headers, body)) + assert isinstance(body, bytes) + assert len(body) >= 1024 * 1024 + await asyncio.sleep(0.1) + return 200, {"x-timeout": "loosened"}, b"ok" + + async def fake_open_tunnel(_identity, _relay_url): + return SlowRequestSession() + + monkeypatch.setattr(dialer, "open_tunnel", fake_open_tunnel) + body = b"x" * (1024 * 1024) + + client = _managed_client(request_timeout=0.02) + try: + assert client.request("POST", "/large", body=body, timeout=2.0) == ( + 200, + {"x-timeout": "loosened"}, + b"ok", + ) + finally: + client.close() + + client = _managed_client(request_timeout=0.02) + try: + with pytest.raises(TunnelRequestError) as exc_info: + client.request("POST", "/large", body=body) + finally: + client.close() + + assert exc_info.value.reason == "TimeoutError" + + +def test_request_timeout_argument_can_tighten_large_constructor_bound( + monkeypatch, +) -> None: + class HangingManagedSession(_ManagedSession): + async def request(self, *_args, **_kwargs): + await asyncio.sleep(3600) + + sessions: list[HangingManagedSession] = [] + + async def fake_open_tunnel(_identity, _relay_url): + session = HangingManagedSession() + sessions.append(session) + return session + + monkeypatch.setattr(dialer, "open_tunnel", fake_open_tunnel) + client = _managed_client(request_timeout=10.0) + started = time.monotonic() + try: + with pytest.raises(TunnelRequestError) as exc_info: + client.request("GET", "/hang", timeout=0.05) + finally: + client.close() + + assert time.monotonic() - started < 1.0 + assert exc_info.value.reason == "TimeoutError" + assert sessions[0].close_calls == 1 + + +def test_request_timeout_uses_constructor_when_argument_omitted(monkeypatch) -> None: + class HangingManagedSession(_ManagedSession): + async def request(self, *_args, **_kwargs): + await asyncio.sleep(3600) + + sessions: list[HangingManagedSession] = [] + + async def fake_open_tunnel(_identity, _relay_url): + session = HangingManagedSession() + sessions.append(session) + return session + + monkeypatch.setattr(dialer, "open_tunnel", fake_open_tunnel) + client = _managed_client(request_timeout=0.05) + started = time.monotonic() + try: + with pytest.raises(TunnelRequestError) as exc_info: + client.request("GET", "/hang") + finally: + client.close() + + assert time.monotonic() - started < 1.0 + assert exc_info.value.reason == "TimeoutError" + assert sessions[0].close_calls == 1 + + +def test_bare_stream_request_honors_per_request_timeout(monkeypatch) -> None: + class HangingManagedSession(_ManagedSession): + async def stream_request(self, *_args, **_kwargs): + await asyncio.sleep(3600) + + async def fake_open_tunnel(_identity, _relay_url): + return HangingManagedSession() + + monkeypatch.setattr(dialer, "open_tunnel", fake_open_tunnel) + client = _managed_client(request_timeout=10.0) + started = time.monotonic() + try: + with pytest.raises(TimeoutError): + client.stream_request("GET", "/hang", timeout=0.05) + finally: + client.close() + + assert time.monotonic() - started < 1.0 + + +def test_proxy_stream_request_honors_per_request_timeout(monkeypatch) -> None: + class HangingManagedSession(_ManagedSession): + async def stream_request(self, *_args, **_kwargs): + await asyncio.sleep(3600) + + sessions: list[HangingManagedSession] = [] + + async def fake_open_tunnel(_identity, _relay_url): + session = HangingManagedSession() + sessions.append(session) + return session + + monkeypatch.setattr(dialer, "open_tunnel", fake_open_tunnel) + client = _managed_client(request_timeout=10.0) + chunks: queue.Queue[TunnelResponseHead | bytes | Exception | None] = queue.Queue() + started = time.monotonic() + try: + future = client.proxy_stream_request( + "GET", + "/hang", + chunks=chunks, + timeout=0.05, + ) + future.result(timeout=1) + finally: + client.close() + + assert time.monotonic() - started < 1.0 + error = chunks.get_nowait() + assert isinstance(error, TunnelRequestError) + assert error.reason == "TimeoutError" + assert chunks.get_nowait() is None + assert sessions[0].close_calls == 1 + + +def test_stream_request_timeout_does_not_cover_response_tail(monkeypatch) -> None: + class SlowTailStream: + async def read(self): + for chunk in (b"tail-a", b"tail-b", b"tail-c"): + await asyncio.sleep(0.25) + yield chunk + + class SlowTailSession(_ManagedSession): + async def stream_request(self, method, path, *, headers, body): + self.stream_requests.append((method, path, headers, body)) + return 200, {}, b"", SlowTailStream() + + async def fake_open_tunnel(_identity, _relay_url): + return SlowTailSession() + + monkeypatch.setattr(dialer, "open_tunnel", fake_open_tunnel) + client = _managed_client(request_timeout=10.0) + chunks: queue.Queue[bytes | Exception | None] = queue.Queue() + try: + future = client.stream_request( + "GET", + "/slow-tail", + chunks=chunks, + timeout=0.4, + ) + future.result(timeout=3) + finally: + client.close() + + assert chunks.get_nowait() == b"tail-a" + assert chunks.get_nowait() == b"tail-b" + assert chunks.get_nowait() == b"tail-c" + assert chunks.get_nowait() is None + + def test_proxy_stream_request_times_out_during_head_and_clears_session() -> None: class HangingManagedSession(_ManagedSession): async def stream_request(self, *_args, **_kwargs): @@ -840,7 +1020,7 @@ def test_proxy_stream_request_times_out_during_head_and_clears_session() -> None monkeypatch = pytest.MonkeyPatch() monkeypatch.setattr(dialer, "open_tunnel", fake_open_tunnel) - client = _managed_client(establish_timeout=0.05) + client = _managed_client(request_timeout=0.05) chunks: queue.Queue[TunnelResponseHead | bytes | Exception | None] = queue.Queue() try: future = client.proxy_stream_request("GET", "/hang", chunks=chunks) @@ -866,7 +1046,7 @@ def test_bare_stream_request_times_out_during_head(monkeypatch) -> None: return HangingManagedSession() monkeypatch.setattr(dialer, "open_tunnel", fake_open_tunnel) - client = _managed_client(establish_timeout=0.05) + client = _managed_client(request_timeout=0.05) try: with pytest.raises(TimeoutError): client.stream_request("GET", "/hang") @@ -887,7 +1067,7 @@ def test_request_times_out_during_head_and_clears_session(monkeypatch) -> None: return session monkeypatch.setattr(dialer, "open_tunnel", fake_open_tunnel) - client = _managed_client(establish_timeout=0.05) + client = _managed_client(request_timeout=0.05) try: with pytest.raises(TunnelRequestError) as exc_info: client.request("GET", "/hang") @@ -899,7 +1079,7 @@ def test_request_times_out_during_head_and_clears_session(monkeypatch) -> None: assert client._session is None -def test_establish_timeout_is_not_armed_during_body_streaming() -> None: +def test_request_timeout_is_not_armed_during_body_streaming() -> None: class SlowBodySession(_ManagedSession): async def stream_request(self, method, path, *, headers, body): self.stream_requests.append((method, path, headers, body)) @@ -914,7 +1094,7 @@ def test_establish_timeout_is_not_armed_during_body_streaming() -> None: monkeypatch = pytest.MonkeyPatch() monkeypatch.setattr(dialer, "open_tunnel", fake_open_tunnel) - client = _managed_client(establish_timeout=0.01) + client = _managed_client(request_timeout=0.01) chunks: queue.Queue[bytes | Exception | None] = queue.Queue() try: future = client.stream_request("GET", "/slow", chunks=chunks) diff --git a/tests/test_export_pl.py b/tests/test_export_pl.py index 95c243e43..6336975f9 100644 --- a/tests/test_export_pl.py +++ b/tests/test_export_pl.py @@ -14,9 +14,11 @@ import requests from cryptography.hazmat.primitives import serialization import solstone.think.utils as think_utils +from solstone.observe import export +from solstone.observe.export import UPLOAD_TIMEOUT from solstone.observe.peer_lookup import PeerInfo from solstone.observe.peer_unpair import maybe_prompt_unpair -from solstone.observe.pl_http import PlHttpResponse +from solstone.observe.pl_http import PlHttpResponse, PlHttpSession from solstone.think.link.ca import cert_fingerprint, generate_ca @@ -91,6 +93,48 @@ def _patch_tunnel( monkeypatch.setenv("SOL_LINK_RELAY_URL", "https://relay.test") +def test_query_manifest_forwards_upload_timeout() -> None: + class RecordingTunnel: + def __init__(self) -> None: + self.calls: list[tuple[str, str, dict[str, str], bytes, float | None]] = [] + + def request( + self, + method: str, + path: str, + *, + headers: dict[str, str], + body: bytes, + timeout: float | None = None, + ) -> tuple[int, dict[str, str], bytes]: + self.calls.append((method, path, headers, body, timeout)) + return 200, {}, b"{}" + + def close(self) -> None: + return None + + tunnel = RecordingTunnel() + session = PlHttpSession(tunnel) + + assert ( + export._query_manifest( + session, + "https://pl.peer", + "12345678-1234-1234-1234-123456789abc", + ) + == {} + ) + assert tunnel.calls == [ + ( + "GET", + "/app/import/journal/12345678/manifest/segments", + {"Host": "pl.peer"}, + b"", + UPLOAD_TIMEOUT, + ) + ] + + def test_export_pl_single_area_uses_pl_url_and_no_authorization( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, diff --git a/tests/test_pl_http.py b/tests/test_pl_http.py new file mode 100644 index 000000000..28eec15e2 --- /dev/null +++ b/tests/test_pl_http.py @@ -0,0 +1,43 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +from __future__ import annotations + +from solstone.observe.pl_http import PlHttpSession + + +class RecordingTunnel: + def __init__(self) -> None: + self.calls: list[tuple[str, str, dict[str, str], bytes, float | None]] = [] + self.closed = False + + def request( + self, + method: str, + path: str, + *, + headers: dict[str, str], + body: bytes, + timeout: float | None = None, + ) -> tuple[int, dict[str, str], bytes]: + self.calls.append((method, path, headers, body, timeout)) + return 200, {}, b"{}" + + def close(self) -> None: + self.closed = True + + +def test_pl_http_post_and_get_forward_timeout_to_tunnel() -> None: + tunnel = RecordingTunnel() + session = PlHttpSession(tunnel) + post_timeout = 1.25 + get_timeout = 2.5 + + session.post( + "https://peer.test/app/import", json={"ok": True}, timeout=post_timeout + ) + session.get("https://peer.test/app/manifest?area=segments", timeout=get_timeout) + + assert [call[4] for call in tunnel.calls] == [post_timeout, get_timeout] + assert tunnel.calls[0][0:2] == ("POST", "/app/import") + assert tunnel.calls[1][0:2] == ("GET", "/app/manifest?area=segments") diff --git a/tests/test_transfer_pl.py b/tests/test_transfer_pl.py index 0459cce12..ca9ddb881 100644 --- a/tests/test_transfer_pl.py +++ b/tests/test_transfer_pl.py @@ -11,8 +11,10 @@ import pytest from cryptography.hazmat.primitives import serialization import solstone.think.utils as think_utils +from solstone.observe import transfer from solstone.observe.peer_lookup import PeerInfo -from solstone.observe.transfer import send_segments_pl +from solstone.observe.pl_http import PlHttpSession +from solstone.observe.transfer import UPLOAD_TIMEOUT, send_segments_pl from solstone.think.link.ca import cert_fingerprint, generate_ca @@ -55,6 +57,48 @@ def _write_segment(journal: Path) -> None: (segment / "stream.json").write_bytes(b'{"name": "laptop"}') +def test_query_journal_segments_forwards_upload_timeout() -> None: + class RecordingTunnel: + def __init__(self) -> None: + self.calls: list[tuple[str, str, dict[str, str], bytes, float | None]] = [] + + def request( + self, + method: str, + path: str, + *, + headers: dict[str, str], + body: bytes, + timeout: float | None = None, + ) -> tuple[int, dict[str, str], bytes]: + self.calls.append((method, path, headers, body, timeout)) + return 200, {}, b"{}" + + def close(self) -> None: + return None + + tunnel = RecordingTunnel() + session = PlHttpSession(tunnel) + + assert ( + transfer._query_journal_segments( + session, + "https://pl.peer", + "12345678", + ) + == {} + ) + assert tunnel.calls == [ + ( + "GET", + "/app/import/journal/12345678/manifest/segments", + {"Host": "pl.peer"}, + b"", + UPLOAD_TIMEOUT, + ) + ] + + def test_transfer_send_pl_posts_journal_segment_day_path( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, -- 2.51.2