fix(api): stop silent clients from wedging the gunicorn thread pool

On 2026-10-09 (sdbees) both :8000 API workers had 7 of 8 gthread pool
threads blocked in tcp_recvmsg -> sk_wait_data; /health and
/api/blocked-ips hung 40s+ while the container reported healthy.

The app makes no in-process outbound socket reads (every ACME/HAProxy
interaction is an already-bounded subprocess). The blocked reads were
gunicorn gthread reading the CLIENT socket: TConn.init() calls
setblocking(True), i.e. timeout None, before parsing the request. Any
peer that sends a few bytes and goes quiet (a TLS ClientHello on this
plain-HTTP port, scanners; :8000 is published on 0.0.0.0) pins a pool
thread forever. gunicorn --timeout does not catch it (gthread only
watches the main thread's heartbeat), --max-requests never triggers on
a worker that finishes nothing, and socket.setdefaulttimeout() is reset
by setblocking(True).

- haproxy_gunicorn_worker.TimeoutThreadWorker (used for :8000 and :8080):
  * per-connection socket timeout HAPROXY_MGR_CLIENT_TIMEOUT (20s),
    re-applied on keep-alive reuse; bounds socket I/O, not handling time
  * watchdog: if every pool thread has been in the same request for more
    than HAPROXY_MGR_WEDGE_SECONDS (1200s, > certbot renew's 900s), dump
    all thread stacks and exit so the master respawns the worker
  * client timeouts logged as one INFO line, not a traceback
- haproxy_manager: socket.setdefaulttimeout (HAPROXY_MGR_SOCKET_DEFAULT_TIMEOUT,
  60s) as defence in depth for any future in-process client; /health now
  reports the answering worker's busy/total threads and oldest request age
- scripts/healthcheck.sh: probes :8000/health (4s) AND :80; image
  HEALTHCHECK uses it (WHP's --health-cmd override must too)
- scripts/test-gunicorn-socket-timeouts.py (12 tests, real gunicorn),
  added to the Docker build gate

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

CLAUDE.md

@@ -10,6 +10,7 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co
 - **Stick-table contract**: `python3 scripts/test-stick-table-contract.py` - offline; holds the templates' `store` clauses, `STICK_TABLE_FIELD_CONTRACT`, and every consumer to each other. Run it after touching any `stick-table` line.
 - **Runtime-map contract**: `python3 scripts/test-runtime-map-contract.py` - offline; asserts the runtime map commands are `@1`-prefixed, reference the map by FILE PATH (never `#<id>`), carry the value `1`, and that every captured rejection is classified as a failure. Run it after touching any `add map`/`del map`/`clear map` path.
 - **Certificate destruction safety**: `python3 scripts/test-cert-write-safety.py` - offline; asserts a live `.pem` is never truncated, removed, or its certbot lineage deleted while any configured domain still references it (one bundle serves many names, so `ssl_cert_path` is routinely shared). Run it after touching any `os.remove`/`certbot delete`/PEM-write path.
+- **API thread-pool safety**: `python3 scripts/test-gunicorn-socket-timeouts.py` - runs REAL gunicorn (~30s); asserts a silent/half-written client cannot pin a gthread pool thread (per-connection socket timeout, `HAPROXY_MGR_CLIENT_TIMEOUT`), slow legit requests are not cut off, the wedge watchdog (`HAPROXY_MGR_WEDGE_SECONDS`) replaces a fully stuck worker, and an in-process call to a stalled upstream times out. Run it after touching `haproxy_gunicorn_worker.py`, start-up.sh's gunicorn lines, or upgrading gunicorn (the worker overrides `ThreadWorker.enqueue_req`).
 - **Manual Testing**: Run `curl` commands against `http://localhost:8000` endpoints as shown in README.md
 
 ### Reading stick tables (and why it is easy to get silently wrong)
@@ -166,6 +167,7 @@ re-reads it on reload, so a failed runtime command must degrade to
 - Exposes ports 80 (HTTP), 443 (HTTPS), and 8000 (management API/UI)
 - Stats page on port 8404 (localhost only inside container)
 - Management interface on port 8000 should be firewall-protected in production
-- Dockerfile HEALTHCHECK verifies both port 8000 (Flask API) and port 80 (HAProxy), with `start-period=60s` and `timeout=10s`
+- Dockerfile HEALTHCHECK runs `scripts/healthcheck.sh`: port 8000 `/health` (4s max) AND port 80 (HAProxy). WHP's `lib-edge-container.sh` passes its own `--health-cmd`, which REPLACES the image's; it must call `/haproxy/scripts/healthcheck.sh` or the fleet never probes :8000
+- gunicorn uses `haproxy_gunicorn_worker.TimeoutThreadWorker` (gthread + client socket timeout + wedge watchdog). Stock gthread leaves client sockets blocking with NO timeout; `--timeout` only watches the main thread's heartbeat
 - Supports deployment on servers with git directory at `/root/whp` and web file sync via rsync to `/docker/whp/web/`
 - HAProxy is version 3.0.11

Dockerfile

@@ -44,6 +44,7 @@ WORKDIR /haproxy
 COPY ./templates /haproxy/templates
 COPY requirements.txt /haproxy/
 COPY haproxy_manager.py /haproxy/
+COPY haproxy_gunicorn_worker.py /haproxy/
 COPY scripts /haproxy/scripts
 COPY trusted_ips.list /etc/haproxy/trusted_ips.list
 COPY trusted_ips.map /etc/haproxy/trusted_ips.map
@@ -92,7 +93,8 @@ RUN python3 /haproxy/scripts/test-wpadmin-gate.py \
  && python3 /haproxy/scripts/test-config-rollback.py \
  && python3 /haproxy/scripts/test-cert-write-safety.py \
  && python3 /haproxy/scripts/test-cert-scripts.py \
- && python3 /haproxy/scripts/validate-rendered-config.py
+ && python3 /haproxy/scripts/validate-rendered-config.py \
+ && python3 /haproxy/scripts/test-gunicorn-socket-timeouts.py
 # Create log directories
 RUN mkdir -p /var/log && touch /var/log/haproxy-manager.log /var/log/haproxy-manager-errors.log
 RUN chmod 755 /var/log/haproxy-manager.log /var/log/haproxy-manager-errors.log
@@ -105,7 +107,10 @@ RUN mkdir -p /var/spool/cron/crontabs && \
 # 443/udp carries HTTP/3 (QUIC). EXPOSE is documentation only — the container
 # must still be run with `-p 443:443/udp` for the UDP listener to be reachable.
 EXPOSE 80 443 443/udp 8000
-# Add health check
+# Health check: management API (:8000, short timeout) AND HAProxy (:80).
+# See scripts/healthcheck.sh. NOTE: WHP's lib-edge-container.sh passes its own
+# --health-cmd, which REPLACES this one on deployed hosts; it must be pointed
+# at /haproxy/scripts/healthcheck.sh for the fleet to get the :8000 probe.
 HEALTHCHECK --interval=30s --timeout=10s --start-period=60s --retries=3 \
-    CMD curl -sf --max-time 5 http://localhost:8000/health && curl -s --max-time 5 -o /dev/null http://localhost/ || exit 1
+    CMD /haproxy/scripts/healthcheck.sh
 CMD ["/haproxy/scripts/start-up.sh"]

VERSION

@@ -1 +1 @@
-2026.09.1
+2026.10.1

haproxy_gunicorn_worker.py

@@ -0,0 +1,209 @@
+"""gunicorn gthread worker that cannot be wedged by a silent client.
+
+Incident 2026-10-09 (sdbees), and very likely the 2026-07-07 whp01 one too
+-------------------------------------------------------------------------
+Both :8000 API workers had 7 of 8 pool threads parked in the kernel in
+``tcp_recvmsg -> sk_wait_data``: a blocking socket read with no timeout.
+/health and /api/blocked-ips hung for 40s+, the container stayed "healthy".
+
+haproxy_manager.py makes NO in-process outbound network reads (no requests,
+urllib, http.client, raw socket or ssl client anywhere; every ACME/HAProxy
+interaction is a subprocess, already bounded since 2026.07.3). The reads that
+blocked were gunicorn's own reads of the CLIENT socket:
+
+    gunicorn/workers/gthread.py  TConn.init():  self.sock.setblocking(True)
+
+Once the main loop sees a connection become readable it hands it to a pool
+thread and turns the socket fully blocking -- timeout None -- before parsing
+the request. Any peer that sends a few bytes and then waits (a TLS ClientHello
+sent to this plain-HTTP port, a port scanner, a half-written request, a body
+shorter than its Content-Length) pins that pool thread forever. :8000 is
+published on 0.0.0.0 by the deploy (`-p 8000:8000`, which bypasses firewalld),
+and on sdbees internet scanners were measured connecting to it. Each such
+connection leaks one thread; 4 per worker exhausts the pool.
+
+gunicorn's --timeout does not catch this: for gthread it only checks that the
+worker's MAIN thread keeps heart-beating, and the main thread keeps polling
+happily while every pool thread is stuck. --max-requests does not help either:
+a worker whose threads never finish a request never reaches the limit.
+socket.setdefaulttimeout() does not help: setblocking(True) explicitly resets
+the timeout to None, overriding any default.
+
+What this worker changes (and nothing else)
+-------------------------------------------
+1. Every client connection gets an explicit socket timeout
+   (HAPROXY_MGR_CLIENT_TIMEOUT, default 20s) at the exact point gthread makes it
+   blocking. It bounds each recv/send on the client socket -- request line,
+   headers, the body the app reads, and response writes. It does NOT bound how
+   long the app takes to handle a request (a long certbot run is a subprocess
+   with its own timeout and touches the client socket only once it is done).
+   A stalled peer now costs one thread for at most that long, then the
+   connection is closed and the thread is returned to the pool.
+
+2. A per-worker watchdog thread (HAPROXY_MGR_WEDGE_SECONDS, default 1200s; 0
+   disables). If EVERY pool thread has been inside the same request for longer
+   than that -- i.e. the pool has not turned over at all for 20 minutes, longer
+   than the slowest legitimate request (certbot renew, 900s) -- the worker logs
+   every thread's Python stack to the error log and exits. The gunicorn master
+   respawns it in about a second, without touching HAProxy or the edge.
+   This is a backstop for blocking sources we have not found yet; the fix for
+   the known one is (1).
+
+3. pool_status() exposes busy/total threads and the oldest in-flight age so
+   /health can report saturation.
+"""
+
+import errno
+import logging
+import os
+import sys
+import threading
+import time
+import traceback
+
+from gunicorn.workers.gthread import ThreadWorker
+
+
+def _env_float(name, default):
+    try:
+        return float(os.environ.get(name, default))
+    except (TypeError, ValueError):
+        return float(default)
+
+
+CLIENT_TIMEOUT = _env_float('HAPROXY_MGR_CLIENT_TIMEOUT', 20)
+WEDGE_SECONDS = _env_float('HAPROXY_MGR_WEDGE_SECONDS', 1200)
+WATCHDOG_INTERVAL = _env_float('HAPROXY_MGR_WATCHDOG_INTERVAL', 15)
+
+# Exit status for a watchdog-killed worker. Must NOT be 3 (WORKER_BOOT_ERROR)
+# or 4 (APP_LOAD_ERROR): on those the gunicorn master halts the whole server
+# instead of respawning the worker.
+WEDGED_EXIT_CODE = 1
+
+_current_worker = None
+
+
+def pool_status():
+    """Busy/total threads of the worker serving the caller, or None when the
+    app is not running under this worker class (e.g. tests, dev server)."""
+    w = _current_worker
+    if w is None:
+        return None
+    return w.pool_status()
+
+
+class _QuietClientTimeout(logging.Filter):
+    """gthread logs any OSError without an errno it recognises as
+    log.exception("Socket error processing request.") -- a full traceback.
+    A client timeout is now an expected, bounded event (scanners hit this port
+    constantly), so reduce it to one line instead of flooding the error log."""
+
+    def filter(self, record):
+        exc = record.exc_info[1] if record.exc_info else None
+        if isinstance(exc, TimeoutError) and \
+                str(record.msg).startswith('Socket error processing request'):
+            record.msg = ('Closed client connection idle for more than %ss '
+                          '(HAPROXY_MGR_CLIENT_TIMEOUT)' % CLIENT_TIMEOUT)
+            record.args = ()
+            record.exc_info = None
+            record.exc_text = None
+            record.levelno = logging.INFO
+            record.levelname = 'INFO'
+        return True
+
+
+class TimeoutThreadWorker(ThreadWorker):
+    """gthread worker whose client sockets always carry a read/write timeout."""
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self._inflight = {}          # thread ident -> monotonic start time
+        self._inflight_lock = threading.Lock()
+
+    # -- (1) client socket timeout ------------------------------------------
+    def enqueue_req(self, conn):
+        # Same three steps as ThreadWorker.enqueue_req, with the timeout set
+        # right after conn.init() -- init() is what calls setblocking(True)
+        # (== settimeout(None)), and it runs again on every keep-alive reuse,
+        # so the timeout must be re-applied here each time, not at accept().
+        conn.init()
+        if CLIENT_TIMEOUT > 0:
+            conn.sock.settimeout(CLIENT_TIMEOUT)
+        fs = self.tpool.submit(self.handle, conn)
+        self._wrap_future(fs, conn)
+
+    # -- busy tracking --------------------------------------------------------
+    def handle(self, conn):
+        ident = threading.get_ident()
+        with self._inflight_lock:
+            self._inflight[ident] = time.monotonic()
+        try:
+            return super().handle(conn)
+        finally:
+            with self._inflight_lock:
+                self._inflight.pop(ident, None)
+
+    def pool_status(self):
+        now = time.monotonic()
+        with self._inflight_lock:
+            starts = list(self._inflight.values())
+        return {
+            'pid': os.getpid(),
+            'threads': self.cfg.threads,
+            'busy': len(starts),
+            'oldest_busy_seconds': round(now - min(starts), 1) if starts else 0.0,
+            'client_timeout_seconds': CLIENT_TIMEOUT,
+        }
+
+    # -- (2) watchdog ---------------------------------------------------------
+    def _wedged_for(self):
+        """Seconds the pool has been completely wedged (every thread busy in
+        its current request), measured from the YOUNGEST in-flight request.
+        0 when at least one thread is free."""
+        with self._inflight_lock:
+            starts = list(self._inflight.values())
+        if len(starts) < self.cfg.threads:
+            return 0.0
+        return time.monotonic() - max(starts)
+
+    def _dump_stacks(self):
+        frames = sys._current_frames()
+        names = {t.ident: t.name for t in threading.enumerate()}
+        out = []
+        for ident, frame in frames.items():
+            out.append('--- thread %s (%s)\n%s' % (
+                ident, names.get(ident, '?'), ''.join(traceback.format_stack(frame))))
+        return '\n'.join(out)
+
+    def _watchdog(self):
+        while self.alive:
+            time.sleep(WATCHDOG_INTERVAL)
+            wedged = self._wedged_for()
+            if wedged > WEDGE_SECONDS:
+                self.log.critical(
+                    'WEDGED WORKER (pid %s): all %s threads busy in the same request '
+                    'for %.0fs > HAPROXY_MGR_WEDGE_SECONDS=%s. Exiting so the master '
+                    'respawns it. Thread stacks:\n%s',
+                    os.getpid(), self.cfg.threads, wedged, WEDGE_SECONDS,
+                    self._dump_stacks())
+                for h in logging.getLogger('gunicorn.error').handlers:
+                    try:
+                        h.flush()
+                    except Exception:
+                        pass
+                # os._exit, not SIGTERM: a graceful exit joins the (stuck,
+                # non-daemon) pool threads at interpreter shutdown and would
+                # hang until the master's --timeout escalates to SIGKILL.
+                os._exit(WEDGED_EXIT_CODE)
+
+    def init_process(self):
+        global _current_worker
+        _current_worker = self
+        try:
+            self.log.error_log.addFilter(_QuietClientTimeout())
+        except AttributeError:
+            pass
+        if WEDGE_SECONDS > 0:
+            threading.Thread(target=self._watchdog, name='wedge-watchdog',
+                             daemon=True).start()
+        super().init_process()   # runs the event loop; does not return

haproxy_manager.py

@@ -51,6 +51,28 @@ def _bounded_subprocess_run(*args, **kwargs):
 
 subprocess.run = _bounded_subprocess_run
 
+# ---------------------------------------------------------------------------
+# Default socket timeout (incident 2026-10-09) -- defence in depth only
+# ---------------------------------------------------------------------------
+# This module makes no outbound network calls today (ACME, HAProxy's admin
+# socket etc. all go through bounded subprocesses above). This default makes
+# sure any socket created in-process later -- a urllib/http.client call added
+# to some handler -- cannot block a worker thread forever by omission.
+#
+# It does NOT protect the API's CLIENT connections: gunicorn's gthread worker
+# calls sock.setblocking(True) on each one, which resets the timeout to None
+# and overrides this default. That is what actually wedged the pool on
+# 2026-10-09; it is fixed by haproxy_gunicorn_worker.TimeoutThreadWorker.
+DEFAULT_SOCKET_TIMEOUT = float(os.environ.get('HAPROXY_MGR_SOCKET_DEFAULT_TIMEOUT', '60'))
+if DEFAULT_SOCKET_TIMEOUT > 0:
+    socket.setdefaulttimeout(DEFAULT_SOCKET_TIMEOUT)
+
+try:
+    from haproxy_gunicorn_worker import pool_status as _gunicorn_pool_status
+except Exception:  # dev server / tests without gunicorn
+    def _gunicorn_pool_status():
+        return None
+
 app = Flask(__name__)
 
 # Default page server (port 8080) — served to HAProxy clients whose request hit
@@ -741,11 +763,19 @@ def health_check():
             cursor.execute('SELECT 1')
             cursor.fetchone()
 
-        return jsonify({
+        body = {
             'status': 'healthy',
             'haproxy_status': 'running' if haproxy_running else 'stopped',
             'database': 'connected'
-        }), 200
+        }
+        # Pool saturation of the worker that answered ('busy' includes this
+        # probe). Leaking threads show as busy/oldest_busy_seconds climbing;
+        # a fully wedged worker cannot answer at all, which is what the
+        # healthcheck's short --max-time catches.
+        pool = _gunicorn_pool_status()
+        if pool is not None:
+            body['worker'] = pool
+        return jsonify(body), 200
     except Exception as e:
         return jsonify({
             'status': 'unhealthy',

scripts/healthcheck.sh

@@ -0,0 +1,31 @@
+#!/usr/bin/env bash
+# Container liveness probe: BOTH planes must answer quickly.
+#
+#   1. Management API (:8000) -- GET /health with a short --max-time. On
+#      2026-10-09 (sdbees) the API workers' thread pools were wedged for long
+#      enough that /health hung 40s+, while the container reported "healthy"
+#      because the deployed --health-cmd only probed :80. A hung API is a dead
+#      control plane (every WHP site update times out), so it must turn the
+#      container unhealthy.
+#   2. HAProxy data plane (:80) -- any HTTP answer at all (the default page,
+#      a redirect, a 4xx/5xx) proves HAProxy accepted and answered. Only a
+#      connect/timeout failure fails it, which is why there is no -f here.
+#
+# 127.0.0.1, not localhost: the API binds IPv4 only, and resolving localhost
+# to ::1 first costs a failed connect on every probe.
+#
+# WHP's lib-edge-container.sh overrides the image HEALTHCHECK with its own
+# --health-cmd; point that at this script so the fleet gets the same probe.
+set -u
+API_TIMEOUT="${HAPROXY_MGR_HEALTH_API_TIMEOUT:-4}"
+EDGE_TIMEOUT="${HAPROXY_MGR_HEALTH_EDGE_TIMEOUT:-5}"
+
+if ! curl -sf -o /dev/null --max-time "$API_TIMEOUT" http://127.0.0.1:8000/health; then
+    echo "management API :8000 /health failed or took longer than ${API_TIMEOUT}s"
+    exit 1
+fi
+if ! curl -s -o /dev/null --max-time "$EDGE_TIMEOUT" http://127.0.0.1/; then
+    echo "HAProxy :80 did not answer within ${EDGE_TIMEOUT}s"
+    exit 1
+fi
+exit 0

scripts/start-up.sh

@@ -77,7 +77,7 @@ HAPROXY_SUPERVISOR_INTERVAL="${HAPROXY_SUPERVISOR_INTERVAL:-15}"
 ) &
 
 # Phase 2: WSGI servers
-# Tunable via env: HAPROXY_MGR_API_WORKERS (default 1), HAPROXY_MGR_API_TIMEOUT
+# Tunable via env: HAPROXY_MGR_API_WORKERS (default 2), HAPROXY_MGR_API_TIMEOUT
 # (default 120 — API can do slow ACME calls), HAPROXY_MGR_MAX_REQUESTS (default
 # 1000 — worker recycle frequency).
 #
@@ -91,12 +91,23 @@ API_TIMEOUT="${HAPROXY_MGR_API_TIMEOUT:-120}"
 MAX_REQ="${HAPROXY_MGR_MAX_REQUESTS:-1000}"
 MAX_REQ_JITTER="${HAPROXY_MGR_MAX_REQUESTS_JITTER:-100}"
 
+# Worker class (incident 2026-10-09): stock gthread makes every client socket
+# fully blocking with NO timeout, so any peer that sends a few bytes and goes
+# quiet (TLS hello on this plain-HTTP port, scanners -- :8000 is published on
+# 0.0.0.0) pins a pool thread forever; 4 of them wedge a worker. This subclass
+# sets a per-connection socket timeout (HAPROXY_MGR_CLIENT_TIMEOUT, default
+# 20s) and runs a watchdog that exits a worker whose threads have ALL been
+# stuck for HAPROXY_MGR_WEDGE_SECONDS (default 1200) so the master respawns it.
+# See haproxy_gunicorn_worker.py. --timeout below does NOT cover either case:
+# for gthread it only watches the main thread's heartbeat.
+WORKER_CLASS="haproxy_gunicorn_worker.TimeoutThreadWorker"
+
 # Default page server on :8080. Stays in the background.
 # --threads 4 lets one worker handle bursts of blocked-IP/default-page hits
 # without forking. --max-requests recycles the worker to bound memory drift.
 gunicorn \
     --bind 0.0.0.0:8080 \
-    --workers 1 --threads 4 --worker-class gthread \
+    --workers 1 --threads 4 --worker-class "${WORKER_CLASS}" \
     --max-requests "${MAX_REQ}" --max-requests-jitter "${MAX_REQ_JITTER}" \
     --timeout 30 \
     --access-logfile - --error-logfile - --log-level info \
@@ -108,7 +119,7 @@ gunicorn \
 # up). Longer --timeout because cert issuance hits ACME and can take a while.
 exec gunicorn \
     --bind 0.0.0.0:8000 \
-    --workers "${API_WORKERS}" --threads 4 --worker-class gthread \
+    --workers "${API_WORKERS}" --threads 4 --worker-class "${WORKER_CLASS}" \
     --max-requests "${MAX_REQ}" --max-requests-jitter "${MAX_REQ_JITTER}" \
     --timeout "${API_TIMEOUT}" \
     --access-logfile - --error-logfile - --log-level info \

scripts/test-gunicorn-socket-timeouts.py

@@ -0,0 +1,356 @@
+#!/usr/bin/env python3
+"""Regression tests: a silent peer must not be able to wedge the API's threads.
+
+Why this file exists
+--------------------
+On 2026-10-09 (sdbees) both :8000 API workers had 7 of 8 gthread pool threads
+blocked in the kernel in tcp_recvmsg -> sk_wait_data, /health hung 40s+, and the
+container stayed "healthy". The app makes no in-process outbound socket reads;
+the blocked reads were gunicorn gthread reading the CLIENT socket, which it
+makes fully blocking with no timeout (TConn.init -> setblocking(True)).
+
+These tests run REAL gunicorn processes (no mocks) against tiny WSGI apps and
+pin:
+
+  * the failure mode itself, on the stock gthread worker (control);
+  * that haproxy_gunicorn_worker.TimeoutThreadWorker frees the thread when a
+    client stalls before / during the request line, mid-body, or on a reused
+    keep-alive connection;
+  * that a legitimately SLOW request (handling time > client timeout) is not
+    cut off -- the timeout bounds socket I/O, not request duration;
+  * the watchdog: a worker whose every thread is stuck is replaced;
+  * the timeout is logged as one line, not a traceback per scanner;
+  * haproxy_manager's socket.setdefaulttimeout: an in-process client call to a
+    stalled upstream (accepts, never replies) times out instead of hanging;
+  * why that default alone is not enough (setblocking(True) clears it).
+
+Running
+-------
+    python3 scripts/test-gunicorn-socket-timeouts.py
+
+stdlib unittest, like the other scripts/test-*.py; needs only the app's own
+requirements (gunicorn, Flask...). Runs in the Docker build gate. ~25s.
+"""
+
+import os
+import socket
+import subprocess
+import sys
+import tempfile
+import textwrap
+import threading
+import time
+import unittest
+import urllib.request
+
+REPO_DIR = os.path.abspath(
+    os.environ.get('HAPROXY_MANAGER_DIR',
+                   os.path.join(os.path.dirname(os.path.abspath(__file__)), '..')))
+
+try:
+    import gunicorn  # noqa: F401
+except ImportError:
+    sys.stderr.write('SKIP: gunicorn not installed (pip install -r requirements.txt)\n')
+    raise SystemExit(77)
+
+WSGI_APP = textwrap.dedent('''\
+    import os, time
+
+    def app(environ, start_response):
+        path = environ.get('PATH_INFO', '/')
+        if path == '/sleep':
+            time.sleep(float(environ.get('QUERY_STRING') or 3))
+        elif path == '/hang':
+            time.sleep(3600)
+        elif path == '/body':
+            n = int(environ.get('CONTENT_LENGTH') or 0)
+            environ['wsgi.input'].read(n)
+        body = ('pid=%d' % os.getpid()).encode()
+        start_response('200 OK', [('Content-Type', 'text/plain'),
+                                  ('Content-Length', str(len(body)))])
+        return [body]
+''')
+
+
+def free_port():
+    s = socket.socket()
+    s.bind(('127.0.0.1', 0))
+    port = s.getsockname()[1]
+    s.close()
+    return port
+
+
+def http_get(port, path='/', timeout=5):
+    with urllib.request.urlopen('http://127.0.0.1:%d%s' % (port, path),
+                                timeout=timeout) as r:
+        return r.read().decode()
+
+
+class Gunicorn:
+    """A real gunicorn master on a free port, with its output captured."""
+
+    def __init__(self, worker_class, threads=1, env=None):
+        self.tmp = tempfile.mkdtemp(prefix='gunicorn-timeout-test-')
+        with open(os.path.join(self.tmp, 'tapp.py'), 'w') as fh:
+            fh.write(WSGI_APP)
+        self.port = free_port()
+        self.log_path = os.path.join(self.tmp, 'gunicorn.log')
+        e = dict(os.environ)
+        e.update(env or {})
+        self.log = open(self.log_path, 'w')
+        self.proc = subprocess.Popen(
+            [sys.executable, '-m', 'gunicorn',
+             '--bind', '127.0.0.1:%d' % self.port,
+             '--workers', '1', '--threads', str(threads),
+             '--worker-class', worker_class,
+             '--timeout', '120', '--graceful-timeout', '2',
+             '--log-level', 'info', '--error-logfile', '-',
+             '--pythonpath', '%s,%s' % (REPO_DIR, self.tmp),
+             'tapp:app'],
+            stdout=self.log, stderr=subprocess.STDOUT, env=e)
+        deadline = time.time() + 15
+        while time.time() < deadline:
+            try:
+                http_get(self.port, '/', timeout=1)
+                return
+            except Exception:
+                time.sleep(0.2)
+        self.stop()
+        raise RuntimeError('gunicorn did not come up:\n' + self.output())
+
+    def output(self):
+        self.log.flush()
+        with open(self.log_path) as fh:
+            return fh.read()
+
+    def stop(self):
+        if self.proc.poll() is None:
+            self.proc.terminate()
+            try:
+                self.proc.wait(5)
+            except subprocess.TimeoutExpired:
+                self.proc.kill()
+                self.proc.wait()
+        self.log.close()
+
+
+def stall_client(port, payload=b'\x16\x03\x01\x02\x00\x01\x00\x01\xfc\x03\x03'):
+    """Connect and send a few bytes that are not a complete HTTP request (by
+    default the start of a TLS ClientHello -- what an HTTPS scanner sends to
+    this plain-HTTP port), then go silent and keep the connection open."""
+    s = socket.create_connection(('127.0.0.1', port))
+    if payload:
+        s.sendall(payload)
+    return s
+
+
+def wait_closed_by_server(sock, within):
+    """True if the server closed (EOF/RST) the connection within `within` s."""
+    sock.settimeout(within)
+    try:
+        while True:
+            data = sock.recv(4096)
+            if not data:
+                return True
+    except (ConnectionResetError, BrokenPipeError):
+        return True
+    except socket.timeout:
+        return False
+
+
+STOCK = 'gthread'
+FIXED = 'haproxy_gunicorn_worker.TimeoutThreadWorker'
+FAST = {'HAPROXY_MGR_CLIENT_TIMEOUT': '1'}
+
+
+class StockGthreadControl(unittest.TestCase):
+    """Proves the failure mode is real, so the fixed-worker tests mean something."""
+
+    def test_stock_gthread_is_wedged_by_one_silent_client(self):
+        g = Gunicorn(STOCK, threads=1)
+        self.addCleanup(g.stop)
+        s = stall_client(g.port)
+        self.addCleanup(s.close)
+        time.sleep(0.3)
+        with self.assertRaises(Exception, msg='stock gthread should be wedged'):
+            http_get(g.port, '/', timeout=3)
+
+
+class TimeoutWorkerTests(unittest.TestCase):
+
+    def start(self, threads=1, env=None):
+        g = Gunicorn(FIXED, threads=threads, env=env or FAST)
+        self.addCleanup(g.stop)
+        return g
+
+    def test_silent_client_is_dropped_and_thread_freed(self):
+        g = self.start()
+        s = stall_client(g.port)
+        self.addCleanup(s.close)
+        t0 = time.time()
+        # Only one thread: this request can only be served once the stalled
+        # connection has released it.
+        self.assertTrue(http_get(g.port, '/', timeout=5).startswith('pid='))
+        self.assertLess(time.time() - t0, 4)
+        self.assertTrue(wait_closed_by_server(s, 3),
+                        'server should have closed the stalled connection')
+
+    def test_connect_and_send_nothing_does_not_hold_a_thread(self):
+        # Never readable -> never handed to a pool thread (main loop only).
+        g = self.start()
+        s = stall_client(g.port, payload=b'')
+        self.addCleanup(s.close)
+        self.assertTrue(http_get(g.port, '/', timeout=3).startswith('pid='))
+
+    def test_half_written_headers(self):
+        g = self.start()
+        s = stall_client(g.port, payload=b'GET /health HTTP/1.1\r\nHost: x\r\n')
+        self.addCleanup(s.close)
+        self.assertTrue(http_get(g.port, '/', timeout=5).startswith('pid='))
+        self.assertTrue(wait_closed_by_server(s, 3))
+
+    def test_body_shorter_than_content_length(self):
+        g = self.start()
+        s = stall_client(g.port, payload=(
+            b'POST /body HTTP/1.1\r\nHost: x\r\nContent-Length: 100\r\n\r\n0123456789'))
+        self.addCleanup(s.close)
+        self.assertTrue(http_get(g.port, '/', timeout=5).startswith('pid='))
+
+    def test_timeout_reapplied_on_keepalive_reuse(self):
+        g = self.start()
+        s = socket.create_connection(('127.0.0.1', g.port))
+        self.addCleanup(s.close)
+        s.sendall(b'GET / HTTP/1.1\r\nHost: x\r\n\r\n')
+        s.settimeout(3)
+        self.assertIn(b'200 OK', s.recv(4096))
+        # Same connection, now stall mid-request. gthread calls conn.init()
+        # (setblocking(True)) again for the reused connection.
+        s.sendall(b'GET / HTT')
+        time.sleep(0.3)
+        self.assertTrue(http_get(g.port, '/', timeout=5).startswith('pid='))
+        self.assertTrue(wait_closed_by_server(s, 3))
+
+    def test_slow_legit_request_is_not_cut_off(self):
+        # Handling takes 2.5s, client timeout is 1s: must still succeed.
+        g = self.start()
+        self.assertTrue(http_get(g.port, '/sleep?2.5', timeout=10).startswith('pid='))
+
+    def test_timeout_logged_as_one_line_not_traceback(self):
+        g = self.start()
+        s = stall_client(g.port)
+        self.addCleanup(s.close)
+        http_get(g.port, '/', timeout=5)
+        time.sleep(0.3)
+        out = g.output()
+        self.assertIn('Closed client connection idle', out)
+        self.assertNotIn('Traceback', out)
+
+    def test_watchdog_replaces_fully_wedged_worker(self):
+        g = self.start(env={'HAPROXY_MGR_CLIENT_TIMEOUT': '1',
+                            'HAPROXY_MGR_WEDGE_SECONDS': '2',
+                            'HAPROXY_MGR_WATCHDOG_INTERVAL': '0.5'})
+        pid_before = http_get(g.port, '/')
+        # Wedge the only thread inside the app (not a socket read).
+        hung = threading.Thread(target=lambda: self._swallow(g.port), daemon=True)
+        hung.start()
+        time.sleep(0.5)
+        deadline = time.time() + 12
+        pid_after = None
+        while time.time() < deadline:
+            try:
+                pid_after = http_get(g.port, '/', timeout=2)
+                if pid_after != pid_before:
+                    break
+            except Exception:
+                pass
+            time.sleep(0.3)
+        self.assertIsNotNone(pid_after)
+        self.assertNotEqual(pid_before, pid_after, 'worker was not replaced')
+        out = g.output()
+        self.assertIn('WEDGED WORKER', out)
+        self.assertIn('tapp.py', out)  # the stuck thread's stack was dumped
+        self.assertIsNone(g.proc.poll(), 'master must keep running')
+
+    def test_watchdog_leaves_partially_busy_pool_alone(self):
+        # 2 threads, one hung forever: the pool still serves, so no restart.
+        g = self.start(threads=2, env={'HAPROXY_MGR_CLIENT_TIMEOUT': '1',
+                                       'HAPROXY_MGR_WEDGE_SECONDS': '1',
+                                       'HAPROXY_MGR_WATCHDOG_INTERVAL': '0.3'})
+        pid_before = http_get(g.port, '/')
+        threading.Thread(target=lambda: self._swallow(g.port), daemon=True).start()
+        time.sleep(3)
+        self.assertEqual(pid_before, http_get(g.port, '/', timeout=3))
+        self.assertNotIn('WEDGED WORKER', g.output())
+
+    @staticmethod
+    def _swallow(port):
+        try:
+            http_get(port, '/hang', timeout=60)
+        except Exception:
+            pass
+
+
+class StalledUpstreamDefaultTimeout(unittest.TestCase):
+    """haproxy_manager sets socket.setdefaulttimeout() at import. An in-process
+    client call to an upstream that accepts and never replies must time out."""
+
+    def test_in_process_client_to_stalled_upstream_times_out(self):
+        srv = socket.socket()
+        srv.bind(('127.0.0.1', 0))
+        srv.listen(5)
+        self.addCleanup(srv.close)
+        port = srv.getsockname()[1]
+        accepted = []
+        threading.Thread(target=lambda: accepted.append(srv.accept()),
+                         daemon=True).start()
+        # Import the real module in a child (it changes process-global socket
+        # state and opens its log files at import), then make a client call
+        # with NO explicit timeout -- what a careless future handler would do.
+        probe = textwrap.dedent('''\
+            import logging, os, sys, tempfile, time, urllib.request
+            d = tempfile.mkdtemp()
+            real = logging.FileHandler
+            logging.FileHandler = lambda fn, *a, **k: real(os.path.join(d, os.path.basename(fn)), *a, **k)
+            os.chdir(sys.argv[1]); sys.path.insert(0, sys.argv[1])
+            import haproxy_manager
+            logging.FileHandler = real
+            import socket
+            print('default', socket.getdefaulttimeout())
+            t0 = time.time()
+            try:
+                urllib.request.urlopen('http://127.0.0.1:%s/' % sys.argv[2]).read()
+                print('result returned')
+            except Exception as e:
+                print('result', type(e).__name__, round(time.time() - t0, 1))
+        ''')
+        env = dict(os.environ, HAPROXY_MGR_SOCKET_DEFAULT_TIMEOUT='1.5')
+        r = subprocess.run([sys.executable, '-c', probe, REPO_DIR, str(port)],
+                           capture_output=True, text=True, timeout=30, env=env)
+        if 'No module named' in r.stderr and 'flask' in r.stderr.lower():
+            self.skipTest('Flask not installed')
+        self.assertIn('default 1.5', r.stdout, r.stderr)
+        line = [l for l in r.stdout.splitlines() if l.startswith('result')][0]
+        self.assertTrue(line.startswith('result TimeoutError '), line)
+        elapsed = float(line.split()[-1])
+        self.assertLess(elapsed, 5)
+        self.assertTrue(accepted, 'upstream never saw the connection')
+        accepted[0][0].close()
+
+    def test_default_timeout_alone_does_not_cover_gthread_client_sockets(self):
+        # Documents why the custom worker exists: gthread's setblocking(True)
+        # resets any timeout -- including the process default -- to None.
+        old = socket.getdefaulttimeout()
+        socket.setdefaulttimeout(60)
+        try:
+            a, b = socket.socketpair()
+        finally:
+            socket.setdefaulttimeout(old)
+        self.addCleanup(a.close)
+        self.addCleanup(b.close)
+        self.assertEqual(a.gettimeout(), 60)
+        a.setblocking(True)
+        self.assertIsNone(a.gettimeout())
+
+
+if __name__ == '__main__':
+    unittest.main(verbosity=2)