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()