# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/across-what-pre-ping-validation-and-idle-recycle-gates-must-an-instance-govern-connection-pools-to-prevent-half-open-socket-crashes-and-session-exhaustion.md # Remnant Warrant: https://corpusfetch.com/c/across-what-pre-ping-validation-and-idle-recycle-gates-must-an-instance-govern-connection-pools-to-prevent-half-open-socket-crashes-and-session-exhaustion.warrant.json # Markdown SHA-256: dd09d3d79fcb7f3b34599f6074118ad2fd264e3e8bfebaedc5fd7a1bb7031629 # Code AST SHA-256: 693696ad6396db4b953f998eeb3262dbbc5e268a80bc380aa4c975e328e06c9e # Verified Classes: PooledConnectionProtocol, PoolBudgetExceededError, PoolCheckoutTimeoutError, _PooledSlot, ResilientConnectionPool # Defensive Guards: 4 fail-closed invariants (zero external deps) import sqlite3 import threading import time from collections import deque from contextlib import contextmanager from dataclasses import dataclass from typing import Callable, Generic, Iterator, Protocol, TypeVar class PooledConnectionProtocol(Protocol): def execute(self, sql: str) -> object: ... def rollback(self) -> None: ... def close(self) -> None: ... C = TypeVar("C", bound=PooledConnectionProtocol) class PoolBudgetExceededError(ValueError): '''Raised when configured pool limits across replicas exceed 80% of DB max_connections.''' pass class PoolCheckoutTimeoutError(TimeoutError): '''Raised when no connection becomes available within `checkout_timeout_sec`.''' pass @dataclass class _PooledSlot(Generic[C]): conn: C created_at_monotonic: float last_used_monotonic: float class ResilientConnectionPool(Generic[C]): ''' Production-hardened connection pool enforcing: 1. Boot-time cluster connection budget validation (`<= 80%` of `db_max_connections`). 2. Proactive `max_idle_sec` (default 240s, beating 300s cloud NAT drops) and `max_lifetime_sec`. 3. Transparent `pre_ping` (`SELECT 1`) on checkout with automatic dead-socket replacement. 4. Mandatory `conn.rollback()` reset on checkin so dirty transactions never bleed across callers. ''' def __init__( self, connect_factory: Callable[[], C], *, pool_size: int = 5, max_overflow: int = 5, max_replicas: int = 1, workers_per_replica: int = 1, db_max_connections: int = 100, max_idle_sec: float = 240.0, max_lifetime_sec: float = 1800.0, checkout_timeout_sec: float = 3.0, ) -> None: total_per_worker = pool_size + max_overflow cluster_max_demand = max_replicas * workers_per_replica * total_per_worker safe_db_ceiling = int(db_max_connections * 0.80) if cluster_max_demand > safe_db_ceiling: raise PoolBudgetExceededError( f"Cluster connection demand ({max_replicas} replicas * {workers_per_replica} workers * " f"{total_per_worker} conns = {cluster_max_demand}) exceeds 80% safe ceiling " f"({safe_db_ceiling}/{db_max_connections})" ) self._connect_factory = connect_factory self.pool_size = max(1, pool_size) self.max_total = total_per_worker self.max_idle_sec = max(10.0, max_idle_sec) self.max_lifetime_sec = max(self.max_idle_sec, max_lifetime_sec) self.checkout_timeout_sec = max(0.1, checkout_timeout_sec) self._cond = threading.Condition() self._idle_slots: deque[_PooledSlot[C]] = deque() self._total_allocated: int = 0 def _is_expired(self, slot: _PooledSlot[C], now: float) -> bool: age = now - slot.created_at_monotonic idle = now - slot.last_used_monotonic return age >= self.max_lifetime_sec or idle >= self.max_idle_sec @staticmethod def _pre_ping_alive(conn: C) -> bool: try: conn.execute("SELECT 1") return True except Exception: return False def _safe_close(self, conn: C) -> None: try: conn.close() except Exception: pass def _acquire_slot(self) -> _PooledSlot[C]: deadline = time.monotonic() + self.checkout_timeout_sec while True: slot_to_test: _PooledSlot[C] | None = None need_fresh_create = False with self._cond: while True: now = time.monotonic() if self._idle_slots: candidate = self._idle_slots.popleft() if self._is_expired(candidate, now): self._total_allocated -= 1 self._safe_close(candidate.conn) continue slot_to_test = candidate break if self._total_allocated < self.max_total: self._total_allocated += 1 need_fresh_create = True break remaining = deadline - now if remaining <= 0: raise PoolCheckoutTimeoutError( f"Timed out after {self.checkout_timeout_sec}s waiting for a connection " f"(max_total={self.max_total} exhausted)" ) self._cond.wait(timeout=remaining) if slot_to_test is not None: # Gate 3: Pre-ping outside the lock so slow pings don't block other threads if self._pre_ping_alive(slot_to_test.conn): return slot_to_test # Dead half-open socket detected: close it, decrement count, and loop immediately self._safe_close(slot_to_test.conn) with self._cond: self._total_allocated -= 1 self._cond.notify() continue if need_fresh_create: try: fresh_conn = self._connect_factory() now = time.monotonic() return _PooledSlot( conn=fresh_conn, created_at_monotonic=now, last_used_monotonic=now, ) except BaseException: with self._cond: self._total_allocated -= 1 self._cond.notify() raise def _release_slot(self, slot: _PooledSlot[C], *, discard: bool = False) -> None: if not discard: try: # Gate 4: Deterministic ROLLBACK on return so dirty state never leaks slot.conn.rollback() except Exception: discard = True with self._cond: if discard or len(self._idle_slots) >= self.pool_size: self._total_allocated -= 1 self._cond.notify() should_close = True else: slot.last_used_monotonic = time.monotonic() self._idle_slots.append(slot) self._cond.notify() should_close = False if should_close: self._safe_close(slot.conn) @contextmanager def connection(self) -> Iterator[C]: ''' Checks out a pre-pinged connection and guarantees `rollback()` + checkin on exit. ''' slot = self._acquire_slot() discard_on_return = False try: yield slot.conn except BaseException: discard_on_return = False # rollback() in _release_slot will verify health raise finally: self._release_slot(slot, discard=discard_on_return)