Module

sync_worker

SyncWorker — blocking I/O worker for request-response workloads.

One request at a time per thread, no asyncio. On 3.14t, runs in a thread with true parallelism. Handles HTTP/1.1 keep-alive in a tight recv/send loop.

When the ASGI app returns a streaming response (more_body=True) or WebSocket upgrade, raises NeedsAsyncError — the supervisor hands off to the async pool (Phase 2). For Phase 1, streaming requests receive 501 Not Implemented.

2Classes6Functions

Classes

_RequestMeta 1 ▼
Pre-extracted header values from a single pass over request headers. All fields populated by ``_cl…

Pre-extracted header values from a single pass over request headers.

All fields populated by _classify_request() — avoids redundant linear scans per request. Header names from _fast_h1 are already lowered, so no.lower()is needed on names.

Methods

Internal Methods 1 ▼
__init__ 0 ▼
def __init__(self) -> None
SyncWorker 24 ▼
Blocking I/O worker — one request at a time, no event loop. On 3.14t, runs in a thread with true p…

Blocking I/O worker — one request at a time, no event loop.

On 3.14t, runs in a thread with true parallelism. Handles HTTP/1.1 keep-alive in a tight recv/send loop. Streaming and WebSocket require handoff to the async pool (Phase 2).

Methods

set_lifespan_state 1 ▼
Set the lifespan state dict shared with all requests.
def set_lifespan_state(self, state: dict[str, Any]) -> None
Parameters
Name Type Description
state —
start_draining 0 ▼
Mark this worker as draining (stop accepting new connections).
def start_draining(self) -> None
is_idle 0 bool ▼
True if no connection is currently being handled.
def is_idle(self) -> bool
Returns
bool
run 0 ▼
Accept connections until shutdown (blocking).
def run(self) -> None
Internal Methods 20 ▼
__init__ 12 ▼
def __init__(self, config: ServerConfig, app: ASGIApp, sock: socket.socket | None, *, worker_id: int = 0, generation: int = 0, shutdown_event: threading.Event | None = None, ssl_context: ssl.SSLContext | None = None, lifecycle_collector: LifecycleCollector | None = None, async_pool: AsyncPool | None = None, conn_queue: queue.Queue[tuple[socket.socket, object]] | None = None, sync_app: SyncApp | 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
ssl_context — Default:None
lifecycle_collector — Default:None
async_pool — Default:None
conn_queue — Default:None
sync_app — 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
_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 ``pounce.worker.startup`` on this sync worker's runner loop.
async
async def _run_worker_startup_hook(self) -> bool
Returns
bool
_run_worker_shutdown_hook 0 ▼
Run ``pounce.worker.shutdown`` on this sync worker's runner loop.
async
async def _run_worker_shutdown_hook(self) -> None
_run_worker_draining_hook 1 ▼
Notify the app before streaming handoffs are retired.
async
async def _run_worker_draining_hook(self, reason: str) -> None
Parameters
Name Type Description
reason —
_run_from_queue 2 ▼
Get connections from distributor queue (no thundering herd).
def _run_from_queue(self, poll_interval: float, runner: asyncio.Runner) -> None
Parameters
Name Type Description
poll_interval —
runner —
_run_accept_loop 2 ▼
Accept directly from socket (SO_REUSEPORT or single worker).
def _run_accept_loop(self, poll_interval: float, runner: asyncio.Runner) -> None
Parameters
Name Type Description
poll_interval —
runner —
_drain_pending_queue 1 ▼
Serve in-flight queued connections on a FULL shutdown; never reset them. Only …
def _drain_pending_queue(self, runner: asyncio.Runner) -> None

Serve in-flight queued connections on a FULL shutdown; never reset them.

Only acts on a full shutdown (_ext_shutdownset). On a graceful reload (_drain_event set, _ext_shutdownunset) the shared conn_queueis still owned by the AcceptDistributor, which keeps enqueuing for the NEW generation — its owndrain_eventis only set on full shutdown, not reload. Touching those entries here would steal connections the new generation is meant to serve, so leave them untouched and let this retiring worker simply exit (issue #102).

Every connection left in the queue was accepted by the distributor before drain began (the distributor 503s anything it accepts after), so each holds an in-flight request the client is waiting on. Closing such a socket — even after a 503 — RSTs it because the client's unread request bytes are still in the receive buffer, surfacing as a ConnectionResetErrorfor an in-flight request (#104). So SERVE each queued connection to completion instead._handle_connectionalready forcesConnection: closewhile draining, so each finishes one request and closes cleanly. The whole drain-serve loop is bounded by shutdown_timeout; any connection still queued past the deadline gets a bounded 503 so the worker thread can still exit promptly.

Parameters
Name Type Description
runner —
_drain_accept_window 0 ▼
For a bounded window, accept new connections and answer them a 503. Only runs …
def _drain_accept_window(self) -> None

For a bounded window, accept new connections and answer them a 503.

Only runs once the accept loop is stopping. The window is bounded by shutdown_timeoutso a draining worker still answers late arrivals with an actionable refusal without blocking process exit.

This window ONLY runs on a full shutdown (SIGTERM —_ext_shutdown set), where the server is going away and a brand-new connection should get a bounded clean 503 (issue #101). On a graceful reload (SIGHUP — _drain_event set but _ext_shutdownNOT set) the new generation owns this listener and serves new connections, so the retiring worker must return early: it must neither steal connections from the new gen nor spin for the fullshutdown_timeoutrefusing them (issue #102).

_send_drain_503 2 ▼
Write the request-aware drain 503 to *conn* and close it.
def _send_drain_503(self, conn: socket.socket, *, timeout: float | None = None) -> None
Parameters
Name Type Description
conn —
timeout — Default:None
_handle_connection 3 ▼
Handle a single TCP connection through request-response cycles.
def _handle_connection(self, conn: socket.socket, addr: tuple[str, int], runner: asyncio.Runner) -> None
Parameters
Name Type Description
conn —
addr —
runner —
_handle_connection_impl 3 bool ▼
Inner connection handling. Returns True if handed off to async pool.
def _handle_connection_impl(self, conn: socket.socket, addr: tuple[str, int], runner: asyncio.Runner) -> bool
Parameters
Name Type Description
conn —
addr —
runner —
Returns
bool
_streaming_handoff 5 bool ▼
Hand a streaming response off to the async pool, or send 501. Returns True whe…
def _streaming_handoff(self, conn: socket.socket, scope: dict[str, Any], body: bytes, request_id: str | None, connection_id: int) -> bool

Hand a streaming response off to the async pool, or send 501.

Returns True when the connection was handed off (caller should return True), False when no pool is configured and a 501 was sent (caller should break the keep-alive loop).

Parameters
Name Type Description
conn —
scope —
body —
request_id —
connection_id —
Returns
bool
_open_connection 1 tuple[tuple[str, int], t… ▼
Resolve peer/local addresses and record ConnectionOpened. Returns (client, ser…
def _open_connection(self, conn: socket.socket) -> tuple[tuple[str, int], tuple[str, int], str, int, int]

Resolve peer/local addresses and record ConnectionOpened.

Returns (client, server, client_str, conn_id, conn_start).

Parameters
Name Type Description
conn —
Returns
tuple[tuple[str, int], tuple[str, int], str, int, int]
_serve_builtin 10 ▼
Serialize and send a protocol-neutral built-in response.
def _serve_builtin(self, conn: socket.socket, path: str, conn_header: bytes, response: BuiltinResponse, *, request_method: bytes, conn_id: int, request_start: int, http_version: str, client_str: str, request_id: str | None) -> None
Parameters
Name Type Description
conn —
path —
conn_header —
response —
request_method —
conn_id —
request_start —
http_version —
client_str —
request_id —
_serialize_send_and_log 12 ▼
Serialize, send, and record/log a single buffered response. Local helper for t…
def _serialize_send_and_log(self, conn: socket.socket, status: int, headers: list[tuple[bytes, bytes]], body_out: bytes, *, conn_id: int, request_start: int, http_version: str, record_method: str, log_method: str, log_path: str, client_str: str, request_id: str | None = None) -> None

Serialize, send, and record/log a single buffered response.

Local helper for the SyncApp fast path and the inline ASGI path — both bypass h11 and share the identical serialize -> sendmsg(/sendall fallback) -> ResponseCompleted -> access-log sequence. Health and dictionary responses keep their own plainsendallpath.

Parameters
Name Type Description
conn —
status —
headers —
body_out —
conn_id —
request_start —
http_version —
record_method —
log_method —
log_path —
client_str —
request_id — Default:None
_recv_request_fast 1 tuple[RequestReceived | … ▼
Read a complete HTTP request using the fast parser. Accumulates data in the re…
def _recv_request_fast(self, conn: socket.socket) -> tuple[RequestReceived | None, bytes]

Read a complete HTTP request using the fast parser.

Accumulates data in the recv buffer until a complete request (headers + body based on Content-Length) is available. Returns (request, body) or (None, b"") on connection close/error.

Parameters
Name Type Description
conn —
Returns
tuple[RequestReceived | None, bytes]
_cached_date_header 0 bytes | None ▼
Return cached Date header bytes, refreshing at most once per second.
def _cached_date_header(self) -> bytes | None
Returns
bytes | None
_send_error 5 ▼
Send a plain-text error response (raw bytes, no h11). When *code* is set, adds…
def _send_error(self, conn: socket.socket, status: int, message: str, *, code: str | None = None, hint: str | None = None) -> None

Send a plain-text error response (raw bytes, no h11).

When code is set, adds theX-Pounce-Error-Coderesponse header. Whenconfig.debugis True, the code and hint are appended to the body to aid debugging.

Parameters
Name Type Description
conn —
status —
message —
code — Default:None
hint — Default:None

Functions

_send_response_parts 3 None ▼
Send an HTTP response iovec completely. ``socket.sendmsg`` may accept only a p…
def _send_response_parts(conn: socket.socket, head: bytes, body: bytes) -> None

Send an HTTP response iovec completely.

socket.sendmsgmay accept only a prefix of the supplied buffers under downstream backpressure. Preserve the allocation-free one-call fast path, then usesendallonly for the unsent suffix.

Parameters
Name Type Description
conn socket.socket
head bytes
body bytes
_worker_lifecycle_receive 0 dict[str, Any] ▼
Return a disconnect message for worker lifecycle extension scopes.
async
async def _worker_lifecycle_receive() -> dict[str, Any]
Returns
dict[str, Any]
_worker_lifecycle_send 1 None ▼
Ignore messages sent by worker lifecycle extension handlers.
async
async def _worker_lifecycle_send(message: dict[str, Any]) -> None
Parameters
Name Type Description
message dict[str, Any]
_wants_100_continue 1 bool ▼
Return True if the header block declares ``Expect: 100-continue``. ``header_bl…
def _wants_100_continue(header_block: bytes) -> bool

Return True if the header block declaresExpect: 100-continue.

header_blockis the raw bytes between the request line and the blank-line terminator. Each CRLF-delimited line is checked so that an expect:substring inside the request target or another header value cannot trigger a false match; full header validation still happens in _fast_h1.parse_request.

Parameters
Name Type Description
header_block bytes
Returns
bool
_classify_request 1 _RequestMeta ▼
Single-pass header extraction for the sync-worker hot path. Replaces separate …
def _classify_request(request: RequestReceived) -> _RequestMeta

Single-pass header extraction for the sync-worker hot path.

Replaces separate calls to_wants_close, _is_websocket_upgrade, andget_header(accept-encoding). Header names are already lowered by _fast_h1.parse_request.

Parameters
Name Type Description
request RequestReceived
Returns
_RequestMeta
_finalize_response_headers 6 tuple[list[tuple[bytes, … ▼
Rewrite response headers and body for the sync-worker response paths. Local he…
def _finalize_response_headers(response_headers: Sequence[tuple[bytes, bytes]], body: bytes, compressor: Compressor | None, dictionary: CompressionDictionary | None, config: ServerConfig, *, apply_min_size: bool) -> tuple[list[tuple[bytes, bytes]], bytes]

Rewrite response headers and body for the sync-worker response paths.

Local helper shared by the SyncApp fast path and the inline ASGI path. Performs the single-pass content-encoding / content-length rewrite, applies the negotiatedcompressor(suppressed when the app pre-set a content-encoding header), re-appends content-length, and — when dcz was negotiated — emits used-dictionary.

apply_min_size gates compression on config.compression_min_size (the SyncApp fast path); the inline ASGI path passesFalsebecause the bridge already governs sub-threshold bodies, so the historical sync-ASGI behaviour compresses regardless of body size.

Returns (headers_list, body_out). To preserve byte-for-byte header order, callers append request-scoped headers (x-request-id), the dictionary-advertisement headers (RFC 9842), and the connection header after this returns.

Parameters
Name Type Description
response_headers Sequence[tuple[bytes, bytes]]
body bytes
compressor Compressor | None
dictionary CompressionDictionary | None
config ServerConfig
apply_min_size bool
Returns
tuple[list[tuple[bytes, bytes]], bytes]