Module

worker

Worker — the heart of pounce's request handling.

Runs a single asyncio event loop that accepts connections on a socket and processes requests through the protocol → bridge → ASGI pipeline.

Each worker is self-contained: it owns its event loop, its set of active connections, and its per-worker metrics. The supervisor spawns workers as threads (nogil) or processes (GIL) — the worker does not know which.

Connection flow (HTTP/1.1): socket.accept() → H1Protocol.receive_data() → build_scope() → negotiate_compression() → app(scope, receive, send) → access_log()

HTTP/2 and WebSocket connections are delegated to dedicated handler modules (_h2_handler and _ws_handler) to keep this file focused on core lifecycle and HTTP/1.1 handling.

1Class6Functions

Classes

Worker 24 ▼
Single-threaded async worker that serves HTTP requests. Accepts connections from the provided sock…

Single-threaded async worker that serves HTTP requests.

Accepts connections from the provided socket and handles them using asyncio streams. Each connection is processed in its own task.

Methods

set_lifespan_state 1 ▼
Set the lifespan state dict to be shared with all requests.
def set_lifespan_state(self, state: dict[str, Any]) -> None
Parameters
Name Type Description
state —

The state dict populated during lifespan startup.
start_draining 0 ▼
Mark this worker as draining. When draining, the worker finishes its existing …
def start_draining(self) -> None

Mark this worker as draining.

When draining, the worker finishes its existing connections but no longer serves new work. Used for both graceful reload (zero-downtime rolling restart) and full shutdown (SIGTERM).

Setting_drainingmakes every brand-new connection that is still accepted take the bounded clean 503 path at the top of _handle_connectioninstead of being served. On a FULL shutdown the accept loop is deliberately kept open during the bounded drain window (see_serve) so those new connections get that 503 rather than hanging unanswered in the listen backlog (#104). Waking_servevia _async_shutdown begins that window; the actual server.close() is deferred until the worker is idle or the window elapses.

is_idle 0 bool ▼
Check if worker has finished all connections and is idle.
def is_idle(self) -> bool
Returns
bool True if the worker has no active connections.
run 0 ▼
Start the worker's event loop (blocking).
def run(self) -> None
shutdown 0 ▼
Signal the worker to stop accepting connections. Safe to call from any thread.…
def shutdown(self) -> None

Signal the worker to stop accepting connections.

Safe to call from any thread. In multi-worker mode the supervisor sets the sharedthreading.Eventwhich the bridge task picks up. In single-worker mode we use call_soon_threadsafeto safely set the asyncio event from an external thread.

Internal Methods 19 ▼
__init__ 10 ▼
def __init__(self, config: ServerConfig, app: ASGIApp, sock: socket.socket, *, worker_id: int = 0, generation: int = 0, shutdown_event: threading.Event | None = None, max_connections: int = 0, ssl_context: ssl.SSLContext | None = None, lifecycle_collector: LifecycleCollector | None = None, startup_status_queue: Any | None = None) -> None
Parameters
Name Type Description
config —
app —
sock —
worker_id — Default:0
generation — Default:0
shutdown_event — Default:None
max_connections — Default:0
ssl_context — Default:None
lifecycle_collector — Default:None
startup_status_queue — Default:None
_active_connection_count 0 int ▼
Return the active connection count under its free-threading lock.
def _active_connection_count(self) -> int
Returns
int
_await_drain_idle 1 ▼
Keep accepting (503ing new conns) until idle or *timeout* elapses. Runs only o…
async
async def _await_drain_idle(self, timeout: float) -> None

Keep accepting (503ing new conns) until idle or timeout elapses.

Runs only on a FULL shutdown while_drainingis set. The asyncio server stays open so in-flight connections finish AND brand-new connections still land on_handle_connectionand receive the bounded clean 503 (#104). Returns as soon as the worker is idle (no active connections) so a worker with nothing in flight exits promptly, or once the bounded window elapses — after which_servecloses the server and aborts any stragglers.

Parameters
Name Type Description
timeout —
_serve 0 ▼
Accept connections until shutdown is signaled.
async
async def _serve(self) -> None
_report_startup_status 2 ▼
Report initial startup state to the owning supervisor.
def _report_startup_status(self, status: str, detail: str | None = None) -> None
Parameters
Name Type Description
status —
detail — Default:None
_run_worker_startup_hook 0 bool ▼
Run the ``pounce.worker.startup`` hook. Return True on success. Generic ASGI a…
async
async def _run_worker_startup_hook(self) -> bool

Run thepounce.worker.startuphook. Return True on success.

Generic ASGI apps that don't recognise the scope typically raise (e.g.KeyError on scope['method']) — that is normal and returnsFalsewithout being fatal unless worker_startup_failure='shutdown' (see _serve()). A timeout is also treated as a failure.

Returns
bool
_run_worker_draining_hook 2 ▼
Notify the app that active streams should finish promptly.
async
async def _run_worker_draining_hook(self, reason: str, timeout: float) -> None
Parameters
Name Type Description
reason —
timeout —
_bridge_shutdown 1 ▼
Poll an external ``threading.Event`` and set the async shutdown. Runs as a bac…
async
async def _bridge_shutdown(self, ext_event: threading.Event) -> None

Poll an externalthreading.Eventand set the async shutdown.

Runs as a background task inside the worker's event loop. Polls the threading event every 50 ms for responsive shutdown without busy-waiting.

Parameters
Name Type Description
ext_event —
_start_connection 2 ▼
Retain an accepted connection task through transport detach.
def _start_connection(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None
Parameters
Name Type Description
reader —
writer —
_run_connection 2 ▼
Run one handler with unconditional writer ownership.
async
async def _run_connection(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None
Parameters
Name Type Description
reader —
writer —
_handle_connection 2 ▼
Handle a single TCP connection through request-response cycles. After TLS hand…
async
async def _handle_connection(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None

Handle a single TCP connection through request-response cycles.

After TLS handshake, checks ALPN to determine protocol:

  • "h2" → HTTP/2 multiplexed connection handler
  • "http/1.1" or None → HTTP/1.1 keep-alive loop

HTTP/1.1 also supports WebSocket upgrade mid-connection.

Parameters
Name Type Description
reader —
writer —
_handle_request 10 ▼
Process a single HTTP request through the ASGI pipeline.
async
async def _handle_request(self, request: RequestReceived, proto: H1Protocol, reader: asyncio.StreamReader, writer: asyncio.StreamWriter, client: tuple[str, int], server: tuple[str, int], client_str: str, *, initial_body: list[BodyReceived] | None = None, connection_id: int = 0, profile_ctx: RequestProfile | None = None) -> None
Parameters
Name Type Description
request —
proto —
reader —
writer —
client —
server —
client_str —
initial_body — Default:None
connection_id — Default:0
profile_ctx — Default:None
_run_app_with_error_handling 6 ▼
Run the ASGI app with branded traceback and error response handling.
async
async def _run_app_with_error_handling(self, scope: dict, receive: Receive, send: Send, send_state: SendState, writer: asyncio.StreamWriter, proto: H1Protocol) -> None
Parameters
Name Type Description
scope —
receive —
send —
send_state —
writer —
proto —
_run_with_disconnect_monitor 9 ▼
Run the ASGI app with concurrent client disconnect monitoring. For bodyless re…
async
async def _run_with_disconnect_monitor(self, scope: dict, receive: Receive, send: Send, send_state: SendState, reader: asyncio.StreamReader, writer: asyncio.StreamWriter, proto: H1Protocol, disconnect: asyncio.Event, *, connection_id: int = 0) -> None

Run the ASGI app with concurrent client disconnect monitoring.

For bodyless requests (GET/HEAD) where the body is already complete. Spawns a monitor task that reads from the socket to detect client disconnect, and cancels the app task when the client drops.

Mirrors the WebSocket handler's concurrent-task pattern.

Parameters
Name Type Description
scope —
receive —
send —
send_state —
reader —
writer —
proto —
disconnect —
connection_id — Default:0
_run_with_body_reader 9 ▼
Run the ASGI app while concurrently reading request body. Used when the reques…
async
async def _run_with_body_reader(self, scope: dict, receive: Receive, send: Send, send_state: SendState, body_queue: asyncio.Queue[BodyReceived], proto: H1Protocol, reader: asyncio.StreamReader, writer: asyncio.StreamWriter, *, disconnect: asyncio.Event | None = None) -> None

Run the ASGI app while concurrently reading request body.

Used when the request body spans multiple socket reads (large POSTs, chunked uploads). Follows the same concurrent-task pattern as the WebSocket handler.

The actual HTTP status is captured in send_state by the send callable; this method only sets 500 as a fallback when the app raises without having started a response.

Parameters
Name Type Description
scope —
receive —
send —
send_state —
body_queue —
proto —
reader —
writer —
disconnect —

Optional event to set when the reader detects client disconnect (EOF or connection error). When provided, this signals the disconnect-aware receive callable so the ASGI app receiveshttp.disconnect.

Default:None
_body_exceeds_limit 2 bool ▼
Return True when a request body is known to exceed max_request_size.
def _body_exceeds_limit(self, request: RequestReceived, initial_body: list[BodyReceived] | None) -> bool
Parameters
Name Type Description
request —
initial_body —
Returns
bool
_monitor_disconnect 2 ▼
Monitor the TCP connection for client disconnect. Reads from the socket to det…
async staticmethod
async def _monitor_disconnect(reader: asyncio.StreamReader, disconnect: asyncio.Event) -> None

Monitor the TCP connection for client disconnect.

Reads from the socket to detect when the client closes the connection (EOF or error). Sets disconnect to signal the ASGIreceive()callable, which unblocks any app waiting forhttp.disconnect.

This mirrors the WebSocket handler's frame-reader pattern but only watches for connection close — no data is expected.

Parameters
Name Type Description
reader —
disconnect —
_send_error 6 ▼
Send a plain-text error response. When *code* is set, adds the ``X-Pounce-Erro…
async
async def _send_error(self, writer: asyncio.StreamWriter, proto: H1Protocol, status: int, message: str, *, code: str | None = None, hint: str | None = None) -> None

Send a plain-text error response.

When code is set, adds theX-Pounce-Error-Coderesponse header so callers can disambiguate identical-status failures. When config.debug is True, hint(and the code) are appended to the response body for easier debugging.

Parameters
Name Type Description
writer —
proto —
status —
message —
code — Default:None
hint — Default:None
_send_debug_error 6 ▼
Send a rich debug error response with traceback.
async
async def _send_debug_error(self, writer: asyncio.StreamWriter, proto: H1Protocol, exc_info: tuple[type[BaseException], BaseException, Any], *, request_method: str = 'GET', request_path: str = '/', request_headers: list[tuple[bytes, bytes]] | None = None) -> None
Parameters
Name Type Description
writer —

Stream writer for sending response.
proto —

Protocol handler.
exc_info —

Exception info tuple (type, value, traceback).
request_method —

HTTP method.

Default:'GET'
request_path —

Request path.

Default:'/'
request_headers —

Request headers.

Default:None

Functions

_make_asyncio_server_wakeup_idempotent 1 None ▼
Guard CPython's server wakeup against a late second transport detach. CPython …
def _make_asyncio_server_wakeup_idempotent(server: asyncio.Server) -> None

Guard CPython's server wakeup against a late second transport detach.

CPython 3.14 tracks transports in aWeakSet. An abandoned transport can disappear from that set before its destructor calls_detach(), so close() and the delayed detach can both call _wakeup(). The first call sets_waiters to None; the unguarded second call tries to iterate it. Patch only the affected standard-library implementation, not third-party event-loop server objects.

Remove this compatibility guard after python/cpython#130141 is resolved.

Parameters
Name Type Description
server asyncio.Server
_create_h1_protocol 1 H1Protocol ▼
Create an HTTP/1.1 protocol handler.
def _create_h1_protocol(*, max_incomplete_event_size: int | None = None) -> H1Protocol
Parameters
Name Type Description
max_incomplete_event_size int | None Default:None
Returns
H1Protocol
_declared_content_length 1 int | None ▼
Return the declared Content-Length, if present and parseable.
def _declared_content_length(headers: Sequence[tuple[bytes, bytes]]) -> int | None
Parameters
Name Type Description
headers Sequence[tuple[bytes, bytes]]
Returns
int | None
_request_body_expected 1 bool ▼
Return True when headers indicate a request body may still arrive.
def _request_body_expected(headers: Sequence[tuple[bytes, bytes]]) -> bool
Parameters
Name Type Description
headers Sequence[tuple[bytes, bytes]]
Returns
bool
_worker_lifecycle_receive 0 dict[str, Any] ▼
Receive callable for worker lifecycle scopes. Returns ``http.disconnect`` imme…
async
async def _worker_lifecycle_receive() -> dict[str, Any]

Receive callable for worker lifecycle scopes.

Returnshttp.disconnectimmediately so that apps which pass unrecognised scope types to their HTTP handler (and call receive()) unblock and return quickly instead of hanging.

Returns
dict[str, Any]
_worker_lifecycle_send 1 None ▼
No-op send for worker lifecycle scopes.
async
async def _worker_lifecycle_send(message: dict[str, Any]) -> None
Parameters
Name Type Description
message dict[str, Any]