HTTP Services & Proxies

The Pattern

Problem: You need a web service that does more than serve requests — it has a background loop reconciling state, it proxies WebSocket connections to a backend process, and it must start reliably even when the previous instance didn't exit cleanly.

Approach: FastAPI lifespan for startup/shutdown, asyncio.create_task for background loops, bidirectional WebSocket relay for proxying, and pre-bind port cleanup to prevent systemd crash-loops.

Pattern proven in production across multiple Python CLI tools and web services.

Key Design Decisions

1. Pre-bind port cleanup — preventing the restart crash-loop

When systemd restarts a service, the old process may still hold the port in TIME_WAIT. The new process fails to bind, exits with status=1, systemd restarts it, repeat. In one production deployment, 2,075+ systemd restarts occurred before manual intervention.

The fix runs before uvicorn.run():

def _kill_stale_port_holder(port: int) -> None:
    """Kill any existing process on *port* to prevent EADDRINUSE crash-loops."""
    try:
        result = subprocess.run(
            ["lsof", "-ti", f":{port}"],
            capture_output=True, text=True, timeout=5,
        )
        if result.returncode == 0 and result.stdout.strip():
            my_pid = os.getpid()
            for pid_str in result.stdout.strip().split("\n"):
                pid = int(pid_str.strip())
                if pid != my_pid:
                    os.kill(pid, signal.SIGTERM)
            time.sleep(1)  # Brief wait for the port to be released
    except Exception:
        pass  # lsof not available — proceed; uvicorn will fail naturally

Called right before server start:

_kill_stale_port_holder(port)

2. FastAPI lifespan with background poll loop

Use the FastAPI lifespan pattern to start background tasks at startup and clean them up at shutdown.

Starting a poll loop and an httpx client:

async def lifespan(app: FastAPI):
    global _poll_task, _http_client
    await kill_orphan_processes()
    _poll_task = asyncio.create_task(_poll_loop())
    _http_client = httpx.AsyncClient(verify=False)
    app.state.http_client = _http_client
    yield
    # Shutdown
    _poll_task.cancel()
    await _http_client.aclose()

Starting both a monitor loop and a watchdog loop:

# Example: dual-loop lifespan for services that need both monitoring and maintenance
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    client = await _try_create_client()     # graceful degradation if unavailable
    app.state.orchestrator = Orchestrator(client=client)
    monitor_instance = asyncio.create_task(monitor_loop(app))
    watchdog_instance = asyncio.create_task(
        app.state.orchestrator.watchdog_loop(app.state.instance_store))
    try:
        yield
    finally:
        watchdog_instance.cancel()
        monitor_instance.cancel()
        if client is not None:
            await client.shutdown()

3. Bidirectional WebSocket relay with auth + auto-spawn

When proxying browser WebSocket connections to a backend process, check auth and verify the backend is alive BEFORE accepting the browser WS:

@app.websocket("/terminal/ws")
async def terminal_ws_proxy(websocket: WebSocket) -> None:
    # Auth check BEFORE accept — middleware doesn't cover WebSocket scope
    if not await _ws_auth_check(websocket):
        return

    # Ensure backend is reachable BEFORE accepting the browser WS
    if not _is_backend_alive():
        # Auto-spawn backend, wait for it to bind
        ...

    await websocket.accept(subprotocol="tty")

    async with websockets.connect(
        f"ws://localhost:{BACKEND_PORT}/ws",
        subprotocols=[Subprotocol("tty")]
    ) as backend_ws:
        # Two concurrent tasks: client→backend and backend→client
        async def client_to_backend():
            while True:

…