From 609e42e85192e395e9681b5935914818d071b360 Mon Sep 17 00:00:00 2001 From: Thomas Howe Date: Sun, 11 Oct 2026 09:36:20 -0400 Subject: [PATCH] fix(storage): reuse keep-alive connections in webhook storage Each webhook write called requests.post, which opens a new TCP connection and leaves it in TIME_WAIT. At a few hundred writes/s a single conserver pod ran out of ephemeral ports and every write failed with "Failed to establish a new connection", sending the vCons to the storage DLQ. Use one requests.Session per worker process, shared across the short-lived storage threads, with an HTTPAdapter pool sized by the new "pool-size" option (default 16). The session is rebuilt when the pid changes so a forked worker never shares its parent's sockets. Tests: a local HTTP/1.1 receiver sees five sequential writes arrive on one client port; a forked pid gets a fresh session. Both fail when the change is reverted. Co-Authored-By: Claude Opus 5.5 --- common/storage/webhook/__init__.py | 34 +++++++++++++- common/tests/test_webhook_keepalive.py | 46 +++++++++++++++++++ common/tests/test_webhook_storage_failures.py | 8 ++-- 3 files changed, 84 insertions(+), 4 deletions(-) create mode 100644 common/tests/test_webhook_keepalive.py diff --git a/common/storage/webhook/__init__.py b/common/storage/webhook/__init__.py index 9158350..10b453a 100644 --- a/common/storage/webhook/__init__.py +++ b/common/storage/webhook/__init__.py @@ -1,3 +1,5 @@ +import os +import threading import time from lib.vcon_redis import VconRedis @@ -5,6 +7,7 @@ from lib.metrics import record_histogram import requests +from requests.adapters import HTTPAdapter logger = init_logger(__name__) @@ -12,8 +15,36 @@ "webhook-urls": [], "headers": {}, "timeout": 30, + # Keep-alive connections kept per worker process. Size it to the in-flight + # vCons per worker (CONSERVER_VCON_CONCURRENCY); extra connections still work + # but are closed after use. + "pool-size": 16, } +# One Session per process so writes reuse keep-alive connections. A bare +# requests.post opened (and TIME_WAITed) a new socket per write, which exhausted +# the pod's ephemeral ports at a few hundred writes/s. Storage writes run on +# short-lived threads, so the session is shared across threads; urllib3's pool +# is thread-safe and these POSTs carry no cookies. Rebuilt after a fork so a +# child never shares its parent's sockets. +_session = None +_session_pid = None +_session_lock = threading.Lock() + + +def _get_session(pool_size): + global _session, _session_pid + pid = os.getpid() + if _session is None or _session_pid != pid: + with _session_lock: + if _session is None or _session_pid != pid: + session = requests.Session() + adapter = HTTPAdapter(pool_connections=4, pool_maxsize=pool_size) + session.mount("http://", adapter) + session.mount("https://", adapter) + _session, _session_pid = session, pid + return _session + def save(vcon_uuid, opts=default_options): vcon_redis = VconRedis() @@ -38,7 +69,8 @@ def save(vcon_uuid, opts=default_options): f"webhook storage: posting vcon {vcon_uuid} to webhook url: {url}" ) webhook_start = time.time() - resp = requests.post( + session = _get_session(opts.get("pool-size", default_options["pool-size"])) + resp = session.post( url, json=json_dict, headers=headers, timeout=opts.get("timeout", default_options["timeout"]), ) diff --git a/common/tests/test_webhook_keepalive.py b/common/tests/test_webhook_keepalive.py new file mode 100644 index 0000000..a751541 --- /dev/null +++ b/common/tests/test_webhook_keepalive.py @@ -0,0 +1,46 @@ +"""Webhook storage reuses one keep-alive connection instead of a new socket per write.""" + +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from unittest.mock import patch + +from storage import webhook + + +class _Receiver(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + client_ports = [] + + def do_POST(self): + self.rfile.read(int(self.headers["Content-Length"])) + _Receiver.client_ports.append(self.client_address[1]) + self.send_response(201) + self.send_header("Content-Length", "2") + self.end_headers() + self.wfile.write(b"{}") + + def log_message(self, *args): + pass + + +def test_sequential_writes_share_one_connection(): + _Receiver.client_ports = [] + server = ThreadingHTTPServer(("127.0.0.1", 0), _Receiver) + threading.Thread(target=server.serve_forever, daemon=True).start() + url = f"http://127.0.0.1:{server.server_address[1]}/probe" + try: + with patch.object(webhook, "VconRedis") as bodies: + bodies.return_value.get_vcon.return_value.to_dict.return_value = {"synthetic_retry_probe": True} + for _ in range(5): + webhook.save("synthetic-id", {"webhook-urls": [url]}) + finally: + server.shutdown() + server.server_close() + assert len(_Receiver.client_ports) == 5 + assert len(set(_Receiver.client_ports)) == 1 + + +def test_session_is_rebuilt_after_fork(): + first = webhook._get_session(4) + with patch.object(webhook.os, "getpid", return_value=-1): + assert webhook._get_session(4) is not first diff --git a/common/tests/test_webhook_storage_failures.py b/common/tests/test_webhook_storage_failures.py index 65107da..46e1aff 100644 --- a/common/tests/test_webhook_storage_failures.py +++ b/common/tests/test_webhook_storage_failures.py @@ -21,11 +21,12 @@ def test_http_failure_reaches_storage_dlq(status): ) with ( patch.object(webhook, "VconRedis") as bodies, - patch.object(webhook.requests, "post", return_value=response), + patch.object(webhook, "_get_session") as session, patch("main.Storage") as storage, patch("main.queue") as queue, ): bodies.return_value.get_vcon.return_value.to_dict.return_value = {"synthetic_retry_probe": True} + session.return_value.post.return_value = response storage.return_value.save.side_effect = lambda vid: webhook.save(vid, {"webhook-urls": [response.url]}) request._process_storage("test_webhook") queue.enqueue_storage_dlq.assert_called_once_with("test_webhook", "synthetic-id", retention_seconds=604800) @@ -38,8 +39,9 @@ def test_success_has_bounded_request_timeout(options, expected): response._content = b"{}" with ( patch.object(webhook, "VconRedis") as bodies, - patch.object(webhook.requests, "post", return_value=response) as post, + patch.object(webhook, "_get_session") as session, ): bodies.return_value.get_vcon.return_value.to_dict.return_value = {"synthetic_retry_probe": True} + session.return_value.post.return_value = response webhook.save("synthetic-id", {"webhook-urls": ["https://receiver.invalid/probe"], **options}) - assert post.call_args.kwargs["timeout"] == expected + assert session.return_value.post.call_args.kwargs["timeout"] == expected