diff --git a/CHANGELOG.md b/CHANGELOG.md index 11bf7c5..83dcf67 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,19 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [Unreleased] + +### Fixed + +- All Hive API calls in `hive/testing.py` (`HiveTestSuite.start`/`end`, + `HiveTest.start`/`end`, `HiveTest.register_multi_test_client`) now share a + single process-wide keep-alive `requests.Session` with connect-error retries + (`HTTPAdapter` + `urllib3.Retry`) instead of opening a fresh TCP connection + per call. At high simulator throughput (e.g. `consume-enginex` at >100 + tests/s), per-call connections exhausted the client-side ephemeral port range + toward the Hive API endpoint (`OSError: [Errno 99]`, `EADDRNOTAVAIL`), + causing sporadic test failures and silently lost test results. + ## [v0.1.0] - 2025-07-09 Initial release of the Python Hive Simulator API with: diff --git a/pyproject.toml b/pyproject.toml index 0d77864..18f29a8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -27,7 +27,7 @@ classifiers = [ "Operating System :: OS Independent", ] requires-python = ">=3.11" -dependencies = ["requests>=2.31.0,<3"] +dependencies = ["requests>=2.31.0,<3", "urllib3>=1.26.0,<3"] [project.urls] Homepage = "https://github.com/marioevz/hive.py" diff --git a/src/hive/session.py b/src/hive/session.py new file mode 100644 index 0000000..80d83a9 --- /dev/null +++ b/src/hive/session.py @@ -0,0 +1,77 @@ +""" +Shared HTTP session for Hive API requests. + +Every request this package makes targets the single, fixed Hive API +endpoint. Opening a fresh TCP connection per call (bare ``requests.post``) +leaves a socket in TIME_WAIT for 60 seconds after close, and because +ephemeral-port exhaustion is per destination tuple, all of that churn +piles onto the one ``(host, port)`` the Hive API listens on. At high +simulator throughput (e.g. ``consume-enginex`` at >100 tests/s with three +API calls per test) the client-side ephemeral port range toward that +endpoint fills up and ``connect()`` intermittently fails with +``EADDRNOTAVAIL`` (``OSError: [Errno 99]``), losing test results. + +Routing all Hive API calls through one process-wide keep-alive session +collapses the per-call connections into a handful of persistent sockets, +removing the mechanism entirely. The mounted ``Retry`` additionally +retries connection-establishment failures with exponential backoff as a +safety net. Only *connect* errors are retried: they are raised before the +request has been sent, so retrying is safe even for non-idempotent POSTs +(a retried ``start_test`` cannot create a duplicate test case). + +Note: pooling only helps if the server honors keep-alive. Hive's Go +backend serves the simulation API with a default ``net/http`` server, +which keeps connections alive and sets no idle timeout. +""" + +import requests +from requests.adapters import HTTPAdapter +from urllib3.util.retry import Retry + +# Retry connection-establishment failures only. Read errors and HTTP error +# statuses are not retried (the request may already have been processed by +# the server, so a retry could duplicate a non-idempotent call). +CONNECT_RETRIES = 5 +BACKOFF_FACTOR = 0.5 # sleeps 0.5, 1, 2, 4 seconds between attempts + +_session: requests.Session | None = None + + +def get_session() -> requests.Session: + """ + Return the process-wide session used for all Hive API requests. + + The session is created lazily on first use, one per process (with + pytest-xdist, one per worker). ``requests.Session`` connection pooling + is thread-safe via the underlying ``urllib3`` pool. + """ + global _session + if _session is None: + retries = Retry( + total=None, + connect=CONNECT_RETRIES, + read=0, + status=0, + other=0, + backoff_factor=BACKOFF_FACTOR, + ) + adapter = HTTPAdapter(max_retries=retries) + session = requests.Session() + session.mount("http://", adapter) + session.mount("https://", adapter) + _session = session + return _session + + +def close_session() -> None: + """ + Close the shared session and discard it. + + The next call to `get_session` creates a fresh session. Mainly useful + for tests and for long-lived processes that want to release the pooled + connections explicitly. + """ + global _session + if _session is not None: + _session.close() + _session = None diff --git a/src/hive/testing.py b/src/hive/testing.py index 33d9f29..c1d9e89 100644 --- a/src/hive/testing.py +++ b/src/hive/testing.py @@ -1,9 +1,8 @@ from dataclasses import asdict, dataclass -import requests - from .client import Client from .network import Network +from .session import get_session @dataclass(kw_only=True) @@ -16,13 +15,13 @@ class HiveTestSuite: @classmethod def start(cls, url: str, name: str, description: str) -> "HiveTestSuite": req = {"Name": name, "Description": description} - response = requests.post(url, json=req) + response = get_session().post(url, json=req) response.raise_for_status() id = response.json() return cls(url=f"{url}/{id}", name=name, description=description, id=id) def end(self): - response = requests.delete(self.url) + response = get_session().delete(self.url) response.raise_for_status() def start_test(self, name: str, description: str) -> "HiveTest": @@ -55,13 +54,13 @@ class HiveTest: @classmethod def start(cls, *, url: str, name: str, description: str) -> "HiveTest": req = {"Name": name, "Description": description} - response = requests.post(url, json=req) + response = get_session().post(url, json=req) response.raise_for_status() id = response.json() return cls(url=f"{url}/{id}", name=name, description=description, id=id) def end(self, *, result: HiveTestResult): - response = requests.post(self.url, json=result.to_dict()) + response = get_session().post(self.url, json=result.to_dict()) response.raise_for_status() def start_client(self, **kwargs) -> Client | None: @@ -73,5 +72,5 @@ def register_multi_test_client(self, client: Client): if not client.multi_test: return url = f"{client.url}/{client.id}/register/{self.id}" - response = requests.post(url) + response = get_session().post(url) response.raise_for_status() diff --git a/src/hive/tests/test_session.py b/src/hive/tests/test_session.py new file mode 100644 index 0000000..c087c02 --- /dev/null +++ b/src/hive/tests/test_session.py @@ -0,0 +1,144 @@ +""" +Tests for the shared Hive API HTTP session. + +These tests run against a local in-process HTTP server; unlike +`test_sanity.py` they do not require a running hive instance. +""" + +import socket +import threading +import time +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Generator, List, Tuple + +import pytest + +from hive.session import CONNECT_RETRIES, close_session, get_session +from hive.testing import HiveTestResult, HiveTestSuite + + +class KeepAliveHandler(BaseHTTPRequestHandler): + """Minimal keep-alive HTTP/1.1 handler mimicking the hive API.""" + + protocol_version = "HTTP/1.1" + + def _respond(self, body: bytes = b"1") -> None: + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def do_POST(self) -> None: # noqa: N802 + length = int(self.headers.get("Content-Length", 0)) + if length: + self.rfile.read(length) + self._respond() + + def do_DELETE(self) -> None: # noqa: N802 + self._respond() + + def log_message(self, format: str, *args) -> None: + pass + + +class ConnectionCountingServer(ThreadingHTTPServer): + """HTTP server recording each new TCP connection's client address.""" + + daemon_threads = True + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.connections: List[Tuple[str, int]] = [] + self._connections_lock = threading.Lock() + + def get_request(self): + request, client_address = super().get_request() + with self._connections_lock: + self.connections.append(client_address) + return request, client_address + + +@pytest.fixture(autouse=True) +def fresh_session() -> Generator[None, None, None]: + """Give each test its own shared session instance.""" + close_session() + yield + close_session() + + +@pytest.fixture +def server() -> Generator[ConnectionCountingServer, None, None]: + server = ConnectionCountingServer(("127.0.0.1", 0), KeepAliveHandler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + yield server + server.shutdown() + server.server_close() + + +def test_get_session_returns_singleton(): + assert get_session() is get_session() + + +def test_session_mounts_connect_retries(): + adapter = get_session().get_adapter("http://127.0.0.1:3000") + assert adapter.max_retries.connect == CONNECT_RETRIES + # Read errors and HTTP error statuses must not be retried: the request + # may already have been processed, and e.g. a duplicated `start_test` + # would create a duplicate test case. + assert adapter.max_retries.read == 0 + assert adapter.max_retries.status == 0 + + +def test_hive_api_calls_reuse_one_connection(server: ConnectionCountingServer): + """ + The whole point of the shared session: many suite/test API calls must + be multiplexed over a small set of persistent keep-alive connections + instead of opening (and TIME_WAIT-ing) one connection per call. + """ + url = f"http://127.0.0.1:{server.server_address[1]}/testsuite" + suite = HiveTestSuite.start(url=url, name="suite", description="desc") + for _ in range(20): + test = suite.start_test(name="test", description="desc") + test.end(result=HiveTestResult(test_pass=True, details="")) + suite.end() + + # 42 API calls; without pooling the server would see 42 connections. + assert len(server.connections) == 1 + + +def test_connect_errors_are_retried_with_backoff(): + """ + A connect() failure (e.g. EADDRNOTAVAIL under ephemeral-port + exhaustion, or ECONNREFUSED here) must be retried until the endpoint + becomes reachable. + """ + # Reserve a port that is initially not listening. + probe = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + probe.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + probe.bind(("127.0.0.1", 0)) + port = probe.getsockname()[1] + probe.close() + + server_holder: List[ConnectionCountingServer] = [] + + def start_server_late() -> None: + time.sleep(1.0) + server = ConnectionCountingServer(("127.0.0.1", port), KeepAliveHandler) + server_holder.append(server) + server.serve_forever() + + thread = threading.Thread(target=start_server_late, daemon=True) + thread.start() + try: + # First attempts hit ECONNREFUSED; a backoff retry succeeds once + # the server is up (retries span ~15s in total, well past 1s). + suite = HiveTestSuite.start( + url=f"http://127.0.0.1:{port}/testsuite", name="suite", description="desc" + ) + assert suite.id == 1 + finally: + if server_holder: + server_holder[0].shutdown() + server_holder[0].server_close() diff --git a/uv.lock b/uv.lock index ed47118..631659c 100644 --- a/uv.lock +++ b/uv.lock @@ -1,5 +1,5 @@ version = 1 -revision = 2 +revision = 3 requires-python = ">=3.11" [[package]] @@ -172,6 +172,7 @@ name = "ethereum-hive" source = { editable = "." } dependencies = [ { name = "requests" }, + { name = "urllib3" }, ] [package.optional-dependencies] @@ -199,6 +200,7 @@ requires-dist = [ { name = "requests", specifier = ">=2.31.0,<3" }, { name = "setuptools-scm", marker = "extra == 'dev'" }, { name = "types-requests", marker = "extra == 'lint'", specifier = ">=2.25.0,<3" }, + { name = "urllib3", specifier = ">=1.26.0,<3" }, ] provides-extras = ["test", "lint", "dev"]