Metadata-Version: 2.4
Name: uhttp-workers
Version: 1.8.0
Summary: Multi-process worker dispatcher built on uhttp-server
Author-email: Pavel Revak <pavelrevak@gmail.com>
License-Expression: MIT
Project-URL: Homepage, https://github.com/pavelrevak/uhttp-workers
Project-URL: Repository, https://github.com/pavelrevak/uhttp-workers
Classifier: Programming Language :: Python :: 3
Classifier: Operating System :: POSIX
Requires-Python: >=3.10
Description-Content-Type: text/markdown
Requires-Dist: uhttp-server>=2.6.0

# uhttp-workers

Multi-process worker dispatcher built on [uhttp-server](https://github.com/pavelrevak/uhttp-server).

Single dispatcher process handles all connections, N worker processes handle business logic in parallel. Communication via `multiprocessing.Queue` with efficient `select()` integration (POSIX only).

## Architecture

```
                                    ┌─────────────┐
                                    │  Worker 0   │
                                    │  Worker 1   │
                 request_queue A    │  Worker 2   │
              ┌────────────────►    │  Worker 3   │  ComputeWorker
              │                     └──────┬──────┘
              │                            │
┌─────────────┤                            │
│             │                     ┌──────┴──────┐
│ Dispatcher  │  request_queue B    │  Worker 0   │
│  (main)     ├────────────────►    │  Worker 1   │  StorageWorker
│             │                     └──────┬──────┘
│  - static   │                            │
│  - sync     │    response_queue          │
│  - auth     │◄───────────────────────────┘
│             │    (shared)
└─────────────┘
```

**Key design decisions:**

- Sockets never leave the dispatcher — only serializable data goes through queues
- Each worker pool has its own request queue, all pools share one response queue
- Workers send heartbeats via response queue — dispatcher detects stuck/dead workers
- Each worker has a private control queue for config updates and stop signals

## Installation

```bash
pip install uhttp-workers
```

## Quick Start

```python
import uhttp.workers as _workers


class ItemWorker(_workers.Worker):
    def setup(self):
        self.items = {}
        self.next_id = 1

    @_workers.api('/api/items', 'GET')
    def list_items(self, request):
        return {'items': list(self.items.values())}

    @_workers.api('/api/item/{id:int}', 'GET')
    def get_item(self, request):
        item = self.items.get(request.path_params['id'])
        if not item:
            return {'error': 'Not found'}, 404
        return item


class MyDispatcher(_workers.Dispatcher):
    def do_check(self, client):
        api_key = client.headers.get('x-api-key')
        if api_key not in VALID_KEYS:
            client.respond({'error': 'unauthorized'}, status=401)
            raise _workers.RejectRequest()

    @_workers.sync('/health')
    def health(self, client, path_params):
        client.respond({'status': 'ok'})


def main():
    dispatcher = MyDispatcher(
        port=8080,
        pools=[
            _workers.WorkerPool(
                ItemWorker, num_workers=4,
                routes=['/api/**'],
                timeout=30,
            ),
        ],
    )
    dispatcher.run()


if __name__ == '__main__':
    main()
```

## Multiple Worker Pools

Route different endpoints to different worker pools with independent scaling:

```python
dispatcher = MyDispatcher(
    port=8080,
    pools=[
        _workers.WorkerPool(
            ComputeWorker, num_workers=4,
            routes=['/api/compute/**'],
            timeout=60,
        ),
        _workers.WorkerPool(
            StorageWorker, num_workers=2,
            routes=['/api/items/**', '/api/item/**'],
            timeout=10,
        ),
        _workers.WorkerPool(
            GeneralWorker, num_workers=1,
        ),  # no routes = default/fallback pool
    ],
)
```

Request routing order:
1. **Static files** — served directly by dispatcher
2. **Sync handlers** — run in dispatcher process
3. **Worker pools** — first pool with matching route prefix, or fallback pool
4. **404** — no match

## API Handlers

Group related endpoints under a common URL prefix using `ApiHandler`:

```python
import uhttp.workers as _workers

class UserHandler(_workers.ApiHandler):
    PATTERN = '/api/user'

    @_workers.api('', 'GET')
    def list_users(self, request):
        return {'users': self.worker.db.list_users()}

    @_workers.api('/{id:int}', 'GET')
    def get_user(self, request):
        return self.worker.db.get_user(request.path_params['id'])

    @_workers.api('/{id:int}', 'DELETE')
    def delete_user(self, request):
        self.worker.db.delete_user(request.path_params['id'])
        return {'deleted': request.path_params['id']}

class OrderHandler(_workers.ApiHandler):
    PATTERN = '/api/order'

    @_workers.api('', 'GET')
    def list_orders(self, request):
        return {'orders': []}

    @_workers.api('/{id:int}', 'GET')
    def get_order(self, request):
        return {'id': request.path_params['id']}

class MyWorker(_workers.Worker):
    HANDLERS = [UserHandler, OrderHandler]

    def setup(self):
        self.db = Database(self.kwargs['db_login'])
```

`@api` patterns on handlers are automatically prefixed with the handler's `PATTERN`. Handlers access the worker instance via `self.worker`.

You can also define `@api` methods directly on the worker class — useful for simple workers that don't need handler grouping.

Handlers support inheritance — a subclass inherits all routes from its parent, using the subclass `PATTERN` as prefix.

## Static Files

```python
dispatcher = Dispatcher(
    port=8080,
    static_routes={
        '/static/': './static/',
        '/images/': '/var/data/images/',
    },
)
```

Static files are served directly by the dispatcher process with path traversal protection. Directory requests automatically serve `index.html` if present.

A mount value may also be a dict for per-mount options:

```python
static_routes={
    '/assets/': {
        'path': '/var/www/assets',
        'headers': {'Cache-Control': 'public, max-age=31536000, immutable'},
        'authoritative': True,
    },
}
```

- **`headers`** — response headers attached to every file served from this mount.
- **`authoritative`** (default `False`) — when `True`, the mount **owns its prefix**:
  a missing file or blocked traversal returns **404** immediately instead of falling
  through to sync handlers / worker pools. Use this when a catch-all pool
  (`routes=['/']`) would otherwise swallow `/assets/missing.png` and answer only after
  a full worker round-trip.

Override **`on_static_served(client, file_path, status)`** for an access log — it fires
on every 200, and on the 404 of an authoritative mount (`status` is `200` or `404`).
Exceptions raised by the hook are logged and swallowed.

## Sync Handlers

Lightweight handlers that run directly in the dispatcher process — no queue overhead. Define them as methods on the dispatcher class with the `@sync` decorator:

```python
import uhttp.workers as _workers

class MyDispatcher(_workers.Dispatcher):
    @_workers.sync('/health')
    def health(self, client, path_params):
        client.respond({
            'status': 'ok',
            'pools': [pool.status() for pool in self._pools],
        })

    @_workers.sync('/version')
    def version(self, client, path_params):
        client.respond({'version': '1.0.0'})
```

Use sync handlers for fast, non-blocking responses only — long operations block the entire dispatcher.

## Worker Lifecycle

### Setup

`setup()` is called once when a worker process starts. Use it to initialize resources that cannot be shared across processes (database connections, models, etc.):

```python
class MyWorker(_workers.Worker):
    def setup(self):
        self.db = Database(self.kwargs['db_login'])
```

Extra keyword arguments from `WorkerPool(...)` are available as `self.kwargs`.

### Memory Limit

Pass `worker_memory_limit_mb=<MB>` (a recognized worker kwarg) to cap each worker's
address space via `RLIMIT_AS`:

```python
WorkerPool(MyWorker, num_workers=4, worker_memory_limit_mb=512)
```

A runaway allocation then hits `ENOMEM` and the worker dies cleanly — the pool restarts
the slot — instead of exhausting host RAM. Defense-in-depth for handling untrusted input.
The cap is applied before `setup()`, so it also bounds any initialization done there.
POSIX only (no-op on Windows); `setrlimit` failures are logged, not fatal.

### Teardown

`teardown()` is called once when the worker process is stopping — use it to release resources:

```python
class MyWorker(_workers.Worker):
    def setup(self):
        self.db = connect_db()

    def teardown(self):
        self.db.close()
```

Called after the run loop exits (clean stop, orphan detection, or pipe close), before the process terminates. Exceptions are logged but do not block shutdown.

### Configuration Updates

Dispatcher can send configuration to workers at runtime via per-worker control queues:

```python
# dispatcher side
for pool in dispatcher._pools:
    pool.send_config({'rate_limit': 100})

# worker side
class MyWorker(_workers.Worker):
    def on_config(self, config):
        self.rate_limit = config['rate_limit']
```

### Health Monitoring

Workers send heartbeats automatically via the shared response queue. When a worker takes a request, it reports which `request_id` it is working on. If a worker stops responding:

- **Dead worker** (segfault, crash) — detected via `is_alive()`, restarted immediately
- **Stuck worker** (infinite loop in C code) — detected via heartbeat timeout, killed and restarted
- **Too many restarts** — pool marked as degraded, returns 503

### Request Handling

```python
@_workers.api('/process/{id:int}', 'POST')
def process(self, request):
    # request.request_id     — internal ID for dispatcher pairing
    # request.method          — 'POST'
    # request.path            — '/process/42'
    # request.path_params     — {'id': 42}
    # request.query           — {'page': '1'} or None
    # request.data            — dict (JSON), bytes (binary), or None
    # request.headers         — dict
    # request.cookies         — dict (lazy-parsed from Cookie header)
    # request.content_type    — 'application/json'
    # request.remote_address  — 'host:port' string (honors X-Forwarded-For
    #                            from trusted proxies; configured on the
    #                            HTTP server side)

    # return data (status 200)
    return {'result': 'ok'}

    # return data with status
    return {'error': 'not found'}, 404

    # return data with status and headers
    return {'ok': True}, 200, {'X-Custom': 'value'}

    # return Response object — full control (headers, cookies)
    return _workers.Response(
        None,  # request_id is set automatically
        data={'ok': True},
        headers={'X-Custom': 'value'},
        cookies={'session': 'abc123'})

    # defer response — worker continues accepting requests
    return _workers.DEFERRED
```

`request.respond()` (for deferred responses) also accepts `headers` and `cookies`:

```python
request.respond(
    data={'result': 'done'},
    headers={'X-Job-Id': '42'},
    cookies={'session': 'abc'})
```

### Deferred Responses

Return `DEFERRED` to skip immediate response. The worker stays in the select loop, accepts new requests, and sends the response later via `request.respond()`:

```python
class MyWorker(_workers.Worker):
    def setup(self):
        self._jobs = {}

    @_workers.api('/process', 'POST')
    def process(self, request):
        job_id = start_background_work(request.data)
        self._jobs[job_id] = request
        return _workers.DEFERRED

    def on_work_done(self, job_id, result):
        request = self._jobs.pop(job_id)
        request.respond(data={'result': result})
```

Note: deferred requests are still subject to dispatcher timeout — call `self.keep_alive()` periodically to prevent 504.

### Keep Alive

Call `self.keep_alive()` during long operations to reset both the request timeout and stuck worker detection:

```python
@_workers.api('/export', 'POST')
def export(self, request):
    for chunk in generate_large_export():
        self.keep_alive()
    return {'status': 'done'}
```

### Server-Sent Events (SSE)

Stream events to clients using the same API as `uhttp.server.Client`:

```python
class MyWorker(_workers.Worker):
    def setup(self):
        self._subscribers = {}

    @_workers.api('/events', 'GET')
    def events(self, request):
        request.response_stream()  # sends headers, keeps connection open
        self._subscribers[request.request_id] = request
        return _workers.DEFERRED

    def notify(self, data):
        for req in self._subscribers.values():
            req.send_event(data=data, event='update')

    def on_disconnect(self, request_id):
        self._subscribers.pop(request_id, None)
```

Available streaming methods on `Request`:

| Method | Description |
|--------|-------------|
| `response_stream(content_type, headers, cookies)` | Start streaming (default: `text/event-stream`) |
| `send_event(data, event, event_id, retry)` | Send SSE event |
| `send_chunk(data)` | Send raw data chunk |
| `response_stream_end()` | End stream and close connection |

Streaming requests are excluded from dispatcher timeout expiration. When the client disconnects, the dispatcher notifies the worker via `on_disconnect(request_id)`.

### NDJSON Streaming

Stream JSON objects line-by-line (`application/x-ndjson`) — one JSON value per line, terminated by `\n`. Useful for incremental APIs that aren't event-shaped (long lists, log tails, progress updates):

```python
class MyWorker(_workers.Worker):
    @_workers.api('/devices/scan', 'GET')
    def scan(self, request):
        request.response_ndjson()
        for device in self.discover_devices():
            request.send_ndjson({'id': device.id, 'name': device.name})
        request.response_stream_end()
        return _workers.DEFERRED
```

NDJSON methods on `Request`:

| Method | Description |
|--------|-------------|
| `response_ndjson(headers, cookies)` | Start NDJSON stream (wrapper over `response_stream` with `application/x-ndjson`) |
| `send_ndjson(obj)` | Send one JSON-serializable value as a line |
| `response_stream_end()` | End stream and close connection (shared with SSE) |

Same lifecycle as SSE: excluded from timeout expiration, client disconnect triggers `on_disconnect(request_id)`.

### Flow Control

Workers can stop accepting new requests when busy. Requests stay in the shared pool queue for other workers to pick up:

```python
class MyWorker(_workers.Worker):
    @_workers.api('/events', 'GET')
    def subscribe(self, request):
        request.response_stream()
        self._subscribers[request.request_id] = request
        if len(self._subscribers) >= 100:
            self.pause()
        return _workers.DEFERRED

    def on_disconnect(self, request_id):
        self._subscribers.pop(request_id, None)
        if not self._accepting and len(self._subscribers) < 100:
            self.resume()
```

`pause()` excludes the request queue from `select()` — the worker continues processing control messages, custom fd events, and `on_idle()`. `resume()` re-enables request acceptance.

## URL Patterns

Dispatcher uses prefix matching to route requests to pools:

```python
_workers.WorkerPool(MyWorker, routes=['/api/users/**'])  # matches /api/users/anything
_workers.WorkerPool(MyWorker, routes=['/api/status'])     # exact match only
_workers.WorkerPool(MyWorker)                              # fallback — catches everything else
```

Workers use full pattern matching with type conversion:

```python
@_workers.api('/user/{id:int}', 'GET')        # id converted to int
@_workers.api('/price/{amount:float}', 'GET') # amount converted to float
@_workers.api('/tag/{name}')                   # name as str, all methods
```

## Authentication

Override `do_check()` on the dispatcher — runs before any request is queued:

```python
class MyDispatcher(_workers.Dispatcher):
    def __init__(self, valid_keys, **kwargs):
        super().__init__(**kwargs)
        self.valid_keys = valid_keys

    def do_check(self, client):
        api_key = client.headers.get('x-api-key')
        if api_key not in self.valid_keys:
            client.respond({'error': 'unauthorized'}, status=401)
            raise _workers.RejectRequest()
```

`do_check()` is only called for requests going to worker pools — static files and sync handlers are not affected.

To gate **sync routes**, override `do_check_sync()` — the sync-handler counterpart, called
after a sync route matches but before its handler runs:

```python
class MyDispatcher(_workers.Dispatcher):
    def do_check_sync(self, client):
        if client.path == '/status' and not _authorized(client):
            client.respond({'error': 'unauthorized'}, status=401)
            raise _workers.RejectRequest()
```

Same contract as `do_check`: respond and `raise RejectRequest` to block the handler (the
request is then considered handled — no fall-through). Default no-op. Keeping the two checks
separate avoids silently gating sync routes for dispatchers that only override `do_check`.

### Worker-Level Validation

Override `do_check()` on the worker — runs before routing to handler:

```python
class MyWorker(_workers.Worker):
    def do_check(self, request):
        token = request.cookies.get('session')
        if not token:
            return {'error': 'unauthorized'}, 401
```

Return `(data, status)` tuple to reject, or `None` to continue. You can also raise `RejectRequest`:

```python
    def do_check(self, request):
        if not self.validate_token(request.cookies.get('session')):
            raise _workers.RejectRequest(
                data={'error': 'forbidden'}, status=403)
```

`RejectRequest` accepts optional `data` (default: `{'error': 'Rejected'}`) and `status` (default: `403`).

`RejectRequest` can also be raised from request handlers — useful for access control within individual endpoints:

```python
    @_workers.api('/admin/users', 'GET')
    def admin_users(self, request):
        if not self.is_admin(request):
            raise _workers.RejectRequest(
                data={'error': 'admin only'}, status=403)
        return {'users': self.list_users()}
```

### Error Handling

Override `on_request_error()` on the worker to customize error handling when a request handler raises an exception:

```python
class MyWorker(_workers.Worker):
    def on_request_error(self, request, err):
        if isinstance(err, DatabaseError):
            self.db.reconnect()
        return super().on_request_error(request, err)
```

Default behavior logs the error with traceback and returns 500.

## Dispatcher Hooks

Override these on a `Dispatcher` subclass to extend behavior — each is detailed in its own
section below. None of them need `super()` except where noted.

| Hook | When it fires | Typical use |
|------|---------------|-------------|
| `do_check_sync(client)` | a sync route matched, before its handler | auth/validate sync routes |
| `do_check(client)` | before dispatching to a worker pool | auth/validate API requests |
| `on_request_accepted(request_id, client, pool)` | just before the request is queued | attach `request.context` for the worker |
| `on_response(response, pending)` | after a worker response is sent | post-process / cross-pool work |
| `on_forward(route, data)` | a worker called `forward()` | route the payload to a sibling pool |
| `on_static_served(client, file_path, status)` | a static mount served 200 / authoritative 404 | access logging |
| `on_pending_removed(request_id, pending, reason)` | a request leaves `_pending` (any outcome) | side-state cleanup |
| `on_worker_died(pool, worker_id, reason, exitcode, victims)` | a worker died/was parked | fail victims, post-mortem (call `super()`) |
| `on_idle()` | each `select()` timeout with no events | periodic maintenance |
| `on_log(name, level, message)` | a log message is emitted | customize log output |

Request flow and where each hook sits:

```
static:   _serve_static ──────────────────────────────────► on_static_served
sync:     match sync route ─► do_check_sync ─► handler
dispatch: do_check ─► on_request_accepted ─► queue ─► [worker]
                                                        ├─ forward()  ─► on_forward ─► sibling pool
                                                        └─ response   ─► on_response ─► on_pending_removed
on death: ──────────────────────────────────────────────► on_worker_died
```

`do_check_sync`, `do_check`, and sync handlers may raise `RejectRequest` to stop processing
(respond first). Any other exception anywhere in this flow is logged and returned as 500 —
it cannot break the dispatcher loop.

## Post-Response Hook

Override `on_response()` on the dispatcher to post-process after a response is sent to the client — e.g., forward data to another worker pool:

```python
class MyDispatcher(_workers.Dispatcher):
    def on_response(self, response, pending):
        if response.status == 200 and pending.pool.name == 'ImageWorker':
            storage_pool = self._find_pool('/internal/storage')
            storage_pool.request_queue.put(_workers.Request(
                request_id=-1,
                method='POST',
                path='/internal/storage/save',
                data={
                    'image': pending.client.data,
                    'result': response.data,
                }))
```

`pending` is a `_PendingRequest` with `client` (original connection) and `pool` (source pool).
Requests with `request_id=-1` are ignored by the dispatcher when the worker responds.

`on_response()` fires only for the happy path (handler returned a response). For lifecycle
cleanup that must run regardless of outcome — timeouts, client disconnects, shutdown — use
`on_pending_removed()` (see below).

## Forwarding to a Sibling Pool

When the producer of the data is the worker itself (not the dispatcher), forward straight
from the handler with `self.forward(route, data)` instead of post-processing in the
dispatcher. The worker returns its client-facing response normally **and** hands off a
separate payload to another pool — the dispatcher stays out of the data path:

```python
class RecognizeWorker(_workers.Worker):
    @_workers.api('/recognize', 'POST')
    def recognize(self, request):
        result = run_recognition(request.data)
        # hand the raw result + image off to the storage pool
        self.forward('/_internal/store', {
            'image': request.data,
            'result': result,
        })
        return {'plate': result['plate']}   # narrow public response to client
```

`forward()` is fire-and-forget: the dispatcher routes it (via `on_forward()`) to the pool
matching `route` as a `request_id=-1` Request, and that worker's response is ignored. The
target pool must be registered **before** any catch-all `routes=['/']` pool, since the
first prefix match wins.

Override `on_forward(route, data)` to change the routing or drop policy. The default routes
to the matching pool with **drop-on-full** (a slow sink can't backpressure the emitting
worker) and logs a warning on drop — never a silent loss.

### Per-Request Context

Override `on_request_accepted(request_id, client, pool)` to attach dispatcher-only state to
a request just before it is dispatched — the return value is serialized onto
`request.context` for the worker to read. Use for state the worker can't derive itself
(e.g. an auth role resolved during `do_check`):

```python
class MyDispatcher(_workers.Dispatcher):
    def on_request_accepted(self, request_id, client, pool):
        return {'role': getattr(client, '_auth_role', None)}

class MyWorker(_workers.Worker):
    @_workers.api('/thing', 'GET')
    def thing(self, request):
        if request.context.get('role') == 'admin':
            ...
```

It is pure state-attach — it must **not** reject (that is `do_check`). Exceptions are logged
and swallowed, leaving `request.context` as `None`. The value must be picklable.

## Request Lifecycle Hook

Override `on_pending_removed(request_id, pending, reason)` on the dispatcher when you keep
side-state keyed by `request_id` and need it cleaned up exactly once, no matter how the
request ended:

```python
class MyDispatcher(_workers.Dispatcher):
    def on_pending_removed(self, request_id, pending, reason):
        self._side_state.pop(request_id, None)
        if reason == _workers.PENDING_TIMEOUT:
            self._metrics.timeouts += 1
```

Reason is one of:

| Constant | When |
|---|---|
| `PENDING_COMPLETED` | Handler returned a response, client got it. `on_response()` runs first. |
| `PENDING_TIMEOUT` | Request exceeded `pool.timeout`; client got 504. Worker may still be processing. |
| `PENDING_DISCONNECTED` | Client disconnected mid-stream; worker was notified via control queue (race possible). |
| `PENDING_STREAM_CLOSED` | Worker ended the SSE stream cleanly. |
| `PENDING_SHUTDOWN` | Dispatcher is shutting down; client got 503. |
| `PENDING_WORKER_DIED` | Worker process died/was killed while owning the request; client got 500. `on_worker_died()` runs first. |

The hook is invoked after the client-facing action (respond / disconnect / control queue put)
so dispatcher state is finalized when it runs. Exceptions raised by the hook are logged at
`LOG_ERROR` and swallowed — they will not crash the dispatcher loop.

Override `on_response()` if you only care about the happy path (e.g. cross-pool forwarding).
Override `on_pending_removed()` if you need exactly-once cleanup. Overriding both is allowed
but discouraged — for the `PENDING_COMPLETED` reason, `on_response()` is called immediately
before `on_pending_removed()`.

## Worker Death Hook

Workers die — segfault in a C extension, OOM-kill, or the dispatcher kills them after
`stuck_timeout`. Override `on_worker_died()` to capture which requests they had in-flight
(useful for forensics when a malformed payload reproduces a crash):

```python
class MyDispatcher(_workers.Dispatcher):
    def on_worker_died(
            self, pool, worker_id, reason, exitcode, victims):
        # `victims` is a list of (request_id, _PendingRequest) for all
        # requests this worker had claimed but not completed.
        # `exitcode` is None for stuck workers, otherwise the process exit
        # code (negative = signal: -9 OOM, -11 SIGSEGV).
        for rid, pending in victims:
            c = pending.client
            self._crash_queue.append({
                'reason': reason,
                'exitcode': exitcode,
                'method': c.method,
                'path': c.path,
                'address': c.address,
                'body': c.body,  # raw bytes — replay this to reproduce
            })
        # Default impl responds 500 to victims and fires
        # on_pending_removed(PENDING_WORKER_DIED). Call it after capture.
        super().on_worker_died(
            pool, worker_id, reason, exitcode, victims)
```

What's a victim: any request the worker had claimed via `MSG_HEARTBEAT`
(`pending.worker_id == worker_id`). Requests still in the queue (`worker_id is None`)
are **not** victims — other workers in the pool will pick them up after restart.

Default behavior (if you don't override) is to log the death + each victim, respond
500 (or close the stream for SSE/NDJSON), and fire `on_pending_removed` for each.
Override only if you want to persist payloads or customize the response status/body.

**500 vs 503:** a victim of a crashed worker gets **500** (processing started, then
the server failed). A new request arriving while the pool has zero alive workers
gets **503 + `Retry-After: 1`** (rejected before processing — try again shortly).
A request to a pool that has exceeded `max_restarts` in `restart_window` gets **503**
permanently (`pool.is_degraded`). `pool.alive_count` is exposed for monitoring and
also appears in `pool.status()`.

By default degraded is **sticky** — once set it stays until restart. Pass
`recovery_interval=<seconds>` to let the pool auto-recover: that many seconds after
entering degraded, the flag and restart counter are cleared and workers get a fresh
chance. Useful for transient failures that resolve on their own. The dispatcher logs
a `WARNING` when a pool enters degraded and an
`INFO` when it recovers.

### Permanent-Failure Parking

Some failures are not transient — a missing license/dongle, bad config, or an
incompatible SDK makes a worker die the same way on every restart, burning the
`max_restarts` budget for nothing. Pass `permanent_failure_exitcode=<code>` and have the
worker `exit(code)` on such a fault:

```python
WorkerPool(MyWorker, num_workers=4, permanent_failure_exitcode=42)
```

A worker that dies with that exact code is **parked**: the slot is not restarted and does
not count toward `max_restarts`. Parked slots drop out of `pool.alive_count` and are
flagged in `pool.status()` (`parked` per worker, `parked_count` on the pool). Any requests
the worker had in flight are still failed with 500 via `on_worker_died` (reason
`parked exit=N`). When **every** slot is parked the pool goes `degraded` — clients get a
chronic **503** (no `Retry-After`, since retrying won't help), and `recovery_interval` will
not clear it (there is nothing left to retry). Default `None` disables parking entirely.

## Dispatcher Idle Hook

Override `on_idle()` on the dispatcher for periodic background tasks — called on each `select()` timeout (every `SELECT_TIMEOUT` seconds, default 1s):

```python
class MyDispatcher(_workers.Dispatcher):
    def on_idle(self):
        # periodic cleanup, monitoring, etc.
        pass
```

Workers have their own `on_idle()` hook, called on each `heartbeat_interval` timeout.

## Graceful Shutdown

On `SIGTERM` or `SIGINT`:

1. Stop accepting new connections
2. Wait for pending responses (up to `shutdown_timeout`)
3. Respond 503 to remaining pending requests
4. Send stop signal to all workers via control queues
5. Wait for workers to finish, kill after timeout

## Monitoring

```python
class MyDispatcher(_workers.Dispatcher):
    @_workers.sync('/monitor')
    def monitor(self, client, path_params):
        client.respond({
            'pools': [pool.status() for pool in self._pools],
            'pending': len(self._pending),
        })
```

Pool status includes per-worker info: alive, last seen, current request ID, queue size.

## Logging

Workers have a built-in `Logger` accessible via `self.log`:

```python
class MyWorker(_workers.Worker):
    @_workers.api('/item/{id:int}', 'GET')
    def get_item(self, request):
        item_id = request.path_params['id']
        # %-style (Python logging compatible)
        self.log.info("Getting item %d", item_id)
        # {}-style (kwargs)
        self.log.info("Getting item {id}", id=item_id)
        return {'id': item_id}
```

Log messages are sent to the dispatcher via the shared response queue and printed in the dispatcher process — no interleaved output from multiple processes.

The dispatcher itself also has a `self.log` Logger that writes directly via `on_log()` (no queue), so it can be used at any time — including before workers are started:

```python
class MyDispatcher(_workers.Dispatcher):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.log.info("Dispatcher starting on port %d", self._port)

    def on_idle(self):
        self.log.debug("idle tick")
```

Set dispatcher log level via constructor: `Dispatcher(log_level=LOG_DEBUG, ...)` (default `LOG_INFO`).

### Logger Names

Each logger is named after its class — `MyWorker[0]` for worker 0, `MyDispatcher`
for the dispatcher. Override the `LOG_NAME` class attribute to customize:

```python
class MyWorker(_workers.Worker):
    LOG_NAME = "api-{pool_name}-{worker_id}"   # → "api-MyWorker-0"

class MyDispatcher(_workers.Dispatcher):
    LOG_NAME = "gateway-{pid}"                  # → "gateway-12345"
```

`LOG_NAME` is a format template. Available placeholders:

| Placeholder | Worker | Dispatcher | Value |
|---|---|---|---|
| `{cls}` | ✓ | ✓ | class name |
| `{worker_id}` | ✓ | — | worker index in pool |
| `{pool_name}` | ✓ | — | owning pool name |
| `{pid}` | ✓ | ✓ | process PID |

The defaults are themselves templates — `'{cls}[{worker_id}]'` for workers and `'{cls}'`
for the dispatcher — so overriding `LOG_NAME` gives you the whole name. The worker index
is only present if your template includes `{worker_id}`. An invalid placeholder logs a
warning and falls back to the built-in default name.

**Log levels:** `LOG_DEBUG` (10), `LOG_INFO` (20), `LOG_WARNING` (30), `LOG_ERROR` (40), `LOG_CRITICAL` (50)

Check current level with `is_*` properties to skip expensive formatting:

```python
if self.log.is_debug:
    self.log.debug("Details: %s", expensive_computation())
```

Available: `is_debug`, `is_info`, `is_warning`, `is_error`.

Set minimum level per pool:

```python
_workers.WorkerPool(
    MyWorker, num_workers=4,
    log_level=_workers.LOG_INFO,  # default: LOG_WARNING
)
```

**Output format** — auto-detected at dispatcher init:
- **Terminal:** ANSI colors — bold red (critical), red (error), yellow (warning), dim (debug)
- **systemd:** Syslog priority prefixes (`<3>`, `<4>`, etc.) — journalctl colors by priority

Override `Dispatcher.on_log(name, level, message)` to customize output or forward to a logging framework.

Errors are logged automatically:
- Handler exceptions → ERROR with full traceback (returns 500 to client)
- `setup()` crash → CRITICAL with traceback (worker exits and restarts)
- Worker restart → ERROR with reason (died/stuck)
- Request timeout → WARNING with request ID and timeout value

## Configuration

### Dispatcher

| Parameter | Default | Description |
|-----------|---------|-------------|
| `port` | 8080 | Listen port |
| `address` | `'0.0.0.0'` | Listen address |
| `pools` | `[]` | List of `WorkerPool` instances |
| `static_routes` | `{}` | URL prefix → filesystem path |
| `shutdown_timeout` | 10 | Seconds to wait on shutdown |
| `max_pending` | 1000 | Max pending requests (503 when exceeded) |
| `ssl_context` | `None` | `ssl.SSLContext` for HTTPS |

### WorkerPool

| Parameter | Default | Description |
|-----------|---------|-------------|
| `worker_class` | — | `Worker` subclass |
| `num_workers` | 1 | Number of worker processes |
| `routes` | `None` | Prefix patterns (`None` = fallback pool) |
| `timeout` | 30 | Request timeout in seconds (504) |
| `stuck_timeout` | 60 | Heartbeat timeout before kill |
| `heartbeat_interval` | 1 | Seconds between worker heartbeats |
| `log_level` | `LOG_WARNING` | Minimum log level for worker loggers |
| `max_restarts` | 10 | Max restarts per `restart_window` |
| `restart_window` | 300 | Time window for restart counting (seconds) |
| `queue_warning` | 100 | Log warning when queue size exceeds this (0 = disable) |
| `recovery_interval` | None | Seconds before auto-recovering from degraded (None = sticky) |
| `permanent_failure_exitcode` | None | Exit code that parks a slot instead of restarting (None = disabled) |

Extra `**kwargs` on `WorkerPool` are passed to worker constructor (accessible as `self.kwargs`).

## Requirements

- Python >= 3.10
- POSIX system (Linux, macOS) — uses `select()` with `queue._reader`
- [uhttp-server](https://github.com/pavelrevak/uhttp-server)
