fix: pool Hive API connections in a shared keep-alive session

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:

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"

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

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()

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()