diff --git a/.gitignore b/.gitignore index 83972fa..8d02975 100644 --- a/.gitignore +++ b/.gitignore @@ -216,3 +216,6 @@ __marimo__/ # Streamlit .streamlit/secrets.toml + +# Hermit local state +.hermit/ diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md new file mode 100644 index 0000000..91ad069 --- /dev/null +++ b/CONTRIBUTING.md @@ -0,0 +1,32 @@ +# Contributing + +Runtime is stdlib-only. Pin Python with Hermit (`./bin/python3`, currently 3.13). + +```bash +./bin/hermit install python3@3.13 +./bin/python3 -m pip install -e ".[dev]" +./bin/python3 -m pytest +./bin/python3 -m ruff check src tests +``` + +## Layout + +| Path | Role | +|---|---| +| `src/llm_queue_proxy/admission.py` | FIFO slot gate | +| `src/llm_queue_proxy/paths.py` | Which POSTs take a slot | +| `src/llm_queue_proxy/server.py` | Reverse proxy + `/__queue/health` | +| `src/llm_queue_proxy/cli.py` | Flags / `LLM_QUEUE_*` env | +| `tests/` | Dummy upstream on `127.0.0.1:0` | + +New admitted path → `DEFAULT_ADMIT_PREFIXES` + `tests/test_paths.py` + README. New CLI flag → env + README + `tests/test_cli.py`. + +## Git + +- Default branch `master` +- Feature branches `YYYY-MM-DD.eizus.description` +- Author `eizus ` +- One worktree per PR; reviewer `@adrw` +- `gh pr create --body-file` + +Do not bind `0.0.0.0`. Do not log `Authorization`. This proxy does not evict engine KV; it only serializes HTTP generations. diff --git a/README.md b/README.md index d365ef2..57b3163 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,111 @@ # llm-queue-proxy -FIFO admission proxy for local LLM servers: one in-flight generation, queue the rest. + +FIFO admission proxy for local LLM HTTP servers. One in-flight generation at a time; extra requests wait in arrival order. + +Built by [Cedar Labs](https://cdr.xyz). Contributing: [CONTRIBUTING.md](CONTRIBUTING.md). + +**This does not shrink model weights or KV.** It only stops two clients from prefilling the same local engine at once. + +## Why + +Local engines such as [MTPLX](https://mtplx.com) already serialize decode, but each concurrent `/v1/chat/completions` still holds a KV bank through prefill. Two Hermes threads against Flash-Next on a full unified-memory Mac abort with: + +> insufficient memory: the request exceeded available GPU memory (sustained critical memory pressure during prefill) + +Point Hermes at this proxy (`:8001`) and leave the engine on `:8000`. Generation POSTs take one slot. Health and `/v1/models` stay concurrent. + +What it **will** do: + +- Queue a second Hermes/WebUI thread behind the first instead of overlapping prefills +- Return `429` when the wait queue is full, `503` when a waiter times out +- Stream SSE through without buffering the whole completion + +What it **will not** do: + +- Evict the other session's KV from the engine +- Fix a single 60k–100k context that already sits at zero memory headroom +- Replace `--context-window` caps, closing other apps, or swapping to 27B + +## Install + +Python 3.11+ (this repo pins **3.13** via [Hermit](https://cashapp.github.io/hermit/)). + +```bash +git clone https://github.com/cdrxyz/llm-queue-proxy.git +cd llm-queue-proxy +./bin/hermit install python3@3.13 # first time; downloads Hermit + CPython +./bin/python3 -m pip install -e . +``` + +Or any 3.11+ interpreter: + +```bash +pip install -e . +``` + +## Run + +```bash +llm-queue-proxy \ + --listen 127.0.0.1:8001 \ + --upstream http://127.0.0.1:8000 \ + --slots 1 \ + --max-queue 32 \ + --queue-timeout 600 +``` + +Env equivalents: `LLM_QUEUE_LISTEN`, `LLM_QUEUE_UPSTREAM`, `LLM_QUEUE_SLOTS`, `LLM_QUEUE_MAX_QUEUE`, `LLM_QUEUE_TIMEOUT`. + +Binds **localhost only**. `0.0.0.0` / `::` are refused. + +### Hermes + MTPLX + +Keep MTPLX on `127.0.0.1:8000`. Point the Hermes provider at the proxy: + +```yaml +providers: + mtplx: + api: http://127.0.0.1:8001/v1 + name: MTPLX + api_key: mtplx + default_model: qwen3.8-flash-next-mtplx +``` + +Same change on `custom_providers[].base_url`. + +### Admitted paths (POST only) + +`/v1/chat/completions`, `/v1/completions`, `/v1/messages`, `/v1/responses`, and the unprefixed `/chat/completions`, `/completions`, `/messages`, `/responses`. + +Override with repeated `--admit /your/path`. + +### Proxy health + +`GET /__queue/health` → `{ ok, in_flight, queued, slots, max_queue, upstream }` (not forwarded). + +## Development + +```bash +. ./bin/activate-hermit # optional; or call ./bin/python3 directly +./bin/python3 -m pip install -e ".[dev]" +./bin/python3 -m pytest +./bin/python3 -m ruff check src tests +``` + +## Architecture + +``` +Hermes / other clients + | + v + llm-queue-proxy :8001 + GET /health, /v1/models ---- pass through ----. + POST /v1/chat/completions -- FIFO slot (1) ----+--> MTPLX :8000 + extra POSTs wait (or 429 / 503) +``` + +Stdlib only at runtime (`http.server` + `http.client`). No third-party proxy framework. + +## License + +Apache-2.0. See [LICENSE](LICENSE). diff --git a/bin/.python3@3.13.pkg b/bin/.python3@3.13.pkg new file mode 120000 index 0000000..383f451 --- /dev/null +++ b/bin/.python3@3.13.pkg @@ -0,0 +1 @@ +hermit \ No newline at end of file diff --git a/bin/README.hermit.md b/bin/README.hermit.md new file mode 100644 index 0000000..e889550 --- /dev/null +++ b/bin/README.hermit.md @@ -0,0 +1,7 @@ +# Hermit environment + +This is a [Hermit](https://github.com/cashapp/hermit) bin directory. + +The symlinks in this directory are managed by Hermit and will automatically +download and install Hermit itself as well as packages. These packages are +local to this environment. diff --git a/bin/activate-hermit b/bin/activate-hermit new file mode 100755 index 0000000..fe28214 --- /dev/null +++ b/bin/activate-hermit @@ -0,0 +1,21 @@ +#!/bin/bash +# This file must be used with "source bin/activate-hermit" from bash or zsh. +# You cannot run it directly +# +# THIS FILE IS GENERATED; DO NOT MODIFY + +if [ "${BASH_SOURCE-}" = "$0" ]; then + echo "You must source this script: \$ source $0" >&2 + exit 33 +fi + +BIN_DIR="$(dirname "${BASH_SOURCE[0]:-${(%):-%x}}")" +if "${BIN_DIR}/hermit" noop > /dev/null; then + eval "$("${BIN_DIR}/hermit" activate "${BIN_DIR}/..")" + + if [ -n "${BASH-}" ] || [ -n "${ZSH_VERSION-}" ]; then + hash -r 2>/dev/null + fi + + echo "Hermit environment $("${HERMIT_ENV}"/bin/hermit env HERMIT_ENV) activated" +fi diff --git a/bin/activate-hermit.fish b/bin/activate-hermit.fish new file mode 100755 index 0000000..0367d23 --- /dev/null +++ b/bin/activate-hermit.fish @@ -0,0 +1,24 @@ +#!/usr/bin/env fish + +# This file must be sourced with "source bin/activate-hermit.fish" from Fish shell. +# You cannot run it directly. +# +# THIS FILE IS GENERATED; DO NOT MODIFY + +if status is-interactive + set BIN_DIR (dirname (status --current-filename)) + + if "$BIN_DIR/hermit" noop > /dev/null + # Source the activation script generated by Hermit + "$BIN_DIR/hermit" activate "$BIN_DIR/.." | source + + # Clear the command cache if applicable + functions -c > /dev/null 2>&1 + + # Display activation message + echo "Hermit environment $($HERMIT_ENV/bin/hermit env HERMIT_ENV) activated" + end +else + echo "You must source this script: source $argv[0]" >&2 + exit 33 +end diff --git a/bin/hermit b/bin/hermit new file mode 100755 index 0000000..31559b7 --- /dev/null +++ b/bin/hermit @@ -0,0 +1,43 @@ +#!/bin/bash +# +# THIS FILE IS GENERATED; DO NOT MODIFY + +set -eo pipefail + +export HERMIT_USER_HOME=~ + +if [ -z "${HERMIT_STATE_DIR}" ]; then + case "$(uname -s)" in + Darwin) + export HERMIT_STATE_DIR="${HERMIT_USER_HOME}/Library/Caches/hermit" + ;; + Linux) + export HERMIT_STATE_DIR="${XDG_CACHE_HOME:-${HERMIT_USER_HOME}/.cache}/hermit" + ;; + esac +fi + +export HERMIT_DIST_URL="${HERMIT_DIST_URL:-https://github.com/cashapp/hermit/releases/download/stable}" +HERMIT_CHANNEL="$(basename "${HERMIT_DIST_URL}")" +export HERMIT_CHANNEL +export HERMIT_EXE=${HERMIT_EXE:-${HERMIT_STATE_DIR}/pkg/hermit@${HERMIT_CHANNEL}/hermit} + +if [ ! -x "${HERMIT_EXE}" ]; then + echo "Bootstrapping ${HERMIT_EXE} from ${HERMIT_DIST_URL}" 1>&2 + INSTALL_SCRIPT="$(mktemp)" + # This value must match that of the install script + INSTALL_SCRIPT_SHA256="09ed936378857886fd4a7a4878c0f0c7e3d839883f39ca8b4f2f242e3126e1c6" + if [ "${INSTALL_SCRIPT_SHA256}" = "BYPASS" ]; then + curl -fsSL "${HERMIT_DIST_URL}/install.sh" -o "${INSTALL_SCRIPT}" + else + # Install script is versioned by its sha256sum value + curl -fsSL "${HERMIT_DIST_URL}/install-${INSTALL_SCRIPT_SHA256}.sh" -o "${INSTALL_SCRIPT}" + # Verify install script's sha256sum + openssl dgst -sha256 "${INSTALL_SCRIPT}" | \ + awk -v EXPECTED="$INSTALL_SCRIPT_SHA256" \ + '$2!=EXPECTED {print "Install script sha256 " $2 " does not match " EXPECTED; exit 1}' + fi + /bin/bash "${INSTALL_SCRIPT}" 1>&2 +fi + +exec "${HERMIT_EXE}" --level=fatal exec "$0" -- "$@" diff --git a/bin/hermit.hcl b/bin/hermit.hcl new file mode 100644 index 0000000..081cbe8 --- /dev/null +++ b/bin/hermit.hcl @@ -0,0 +1,2 @@ +github-token-auth { +} diff --git a/bin/pip b/bin/pip new file mode 120000 index 0000000..b3395eb --- /dev/null +++ b/bin/pip @@ -0,0 +1 @@ +.python3@3.13.pkg \ No newline at end of file diff --git a/bin/pip3 b/bin/pip3 new file mode 120000 index 0000000..b3395eb --- /dev/null +++ b/bin/pip3 @@ -0,0 +1 @@ +.python3@3.13.pkg \ No newline at end of file diff --git a/bin/pip3.13 b/bin/pip3.13 new file mode 120000 index 0000000..b3395eb --- /dev/null +++ b/bin/pip3.13 @@ -0,0 +1 @@ +.python3@3.13.pkg \ No newline at end of file diff --git a/bin/pydoc3 b/bin/pydoc3 new file mode 120000 index 0000000..b3395eb --- /dev/null +++ b/bin/pydoc3 @@ -0,0 +1 @@ +.python3@3.13.pkg \ No newline at end of file diff --git a/bin/pydoc3.13 b/bin/pydoc3.13 new file mode 120000 index 0000000..b3395eb --- /dev/null +++ b/bin/pydoc3.13 @@ -0,0 +1 @@ +.python3@3.13.pkg \ No newline at end of file diff --git a/bin/python b/bin/python new file mode 120000 index 0000000..b3395eb --- /dev/null +++ b/bin/python @@ -0,0 +1 @@ +.python3@3.13.pkg \ No newline at end of file diff --git a/bin/python3 b/bin/python3 new file mode 120000 index 0000000..b3395eb --- /dev/null +++ b/bin/python3 @@ -0,0 +1 @@ +.python3@3.13.pkg \ No newline at end of file diff --git a/bin/python3-config b/bin/python3-config new file mode 120000 index 0000000..b3395eb --- /dev/null +++ b/bin/python3-config @@ -0,0 +1 @@ +.python3@3.13.pkg \ No newline at end of file diff --git a/bin/python3.13 b/bin/python3.13 new file mode 120000 index 0000000..b3395eb --- /dev/null +++ b/bin/python3.13 @@ -0,0 +1 @@ +.python3@3.13.pkg \ No newline at end of file diff --git a/bin/python3.13-config b/bin/python3.13-config new file mode 120000 index 0000000..b3395eb --- /dev/null +++ b/bin/python3.13-config @@ -0,0 +1 @@ +.python3@3.13.pkg \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..d8b8ff2 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,49 @@ +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[project] +name = "llm-queue-proxy" +version = "0.1.0" +description = "FIFO admission proxy for local LLM servers: one in-flight generation, queue the rest." +readme = "README.md" +license = { file = "LICENSE" } +requires-python = ">=3.11" +authors = [{ name = "Cedar Labs", email = "hello@cdr.xyz" }] +keywords = ["llm", "proxy", "openai", "mtplx", "mlx", "queue"] +classifiers = [ + "Development Status :: 4 - Beta", + "License :: OSI Approved :: Apache Software License", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", + "Topic :: Internet :: Proxy Servers", +] +dependencies = [] + +[project.urls] +Homepage = "https://github.com/cdrxyz/llm-queue-proxy" +Issues = "https://github.com/cdrxyz/llm-queue-proxy/issues" +Source = "https://github.com/cdrxyz/llm-queue-proxy" + +[project.scripts] +llm-queue-proxy = "llm_queue_proxy.cli:main" + +[project.optional-dependencies] +dev = ["pytest>=8.0", "ruff>=0.6"] + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.pytest.ini_options] +pythonpath = ["src"] +testpaths = ["tests"] +addopts = "-q" + +[tool.ruff] +target-version = "py311" +line-length = 100 + +[tool.ruff.lint] +select = ["E", "F", "I", "UP", "B"] diff --git a/src/llm_queue_proxy/__init__.py b/src/llm_queue_proxy/__init__.py new file mode 100644 index 0000000..1e7f10e --- /dev/null +++ b/src/llm_queue_proxy/__init__.py @@ -0,0 +1,3 @@ +"""FIFO admission proxy for local LLM HTTP servers.""" + +__version__ = "0.1.0" diff --git a/src/llm_queue_proxy/__main__.py b/src/llm_queue_proxy/__main__.py new file mode 100644 index 0000000..3ac06e7 --- /dev/null +++ b/src/llm_queue_proxy/__main__.py @@ -0,0 +1,4 @@ +from llm_queue_proxy.cli import main + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/llm_queue_proxy/admission.py b/src/llm_queue_proxy/admission.py new file mode 100644 index 0000000..3d0feb1 --- /dev/null +++ b/src/llm_queue_proxy/admission.py @@ -0,0 +1,111 @@ +"""Thread-safe FIFO admission gate. + +Slots cap in-flight work. Extra callers wait in arrival order until a slot +frees, the wait exceeds ``timeout_s``, or ``max_queue`` waiters are already +queued (then ``QueueFull``). +""" + +from __future__ import annotations + +import threading +import time +from collections import deque +from dataclasses import dataclass +from typing import Any + + +class QueueFull(Exception): + """The wait queue already has ``max_queue`` callers.""" + + +class QueueTimeout(Exception): + """A caller waited longer than ``timeout_s`` without a slot.""" + + +@dataclass +class _Waiter: + event: threading.Event + granted: bool = False + cancelled: bool = False + + +@dataclass +class Ticket: + """Holds one admission slot until ``release``.""" + + _gate: Admission + wait_s: float + _released: bool = False + + def release(self) -> None: + if self._released: + return + self._released = True + self._gate._release() + + def __enter__(self) -> Ticket: + return self + + def __exit__(self, *_exc: object) -> None: + self.release() + + +class Admission: + def __init__(self, slots: int = 1, max_queue: int = 32, timeout_s: float = 600.0) -> None: + if slots < 1: + raise ValueError("slots must be >= 1") + if max_queue < 0: + raise ValueError("max_queue must be >= 0") + self.slots = slots + self.max_queue = max_queue + self.timeout_s = timeout_s + self._lock = threading.Lock() + self._in_flight = 0 + self._waiters: deque[_Waiter] = deque() + + def snapshot(self) -> dict[str, Any]: + with self._lock: + return { + "slots": self.slots, + "max_queue": self.max_queue, + "timeout_s": self.timeout_s, + "in_flight": self._in_flight, + "queued": len(self._waiters), + } + + def acquire(self, timeout_s: float | None = None) -> Ticket: + deadline_timeout = self.timeout_s if timeout_s is None else timeout_s + started = time.monotonic() + waiter: _Waiter | None = None + with self._lock: + if self._in_flight < self.slots and not self._waiters: + self._in_flight += 1 + return Ticket(_gate=self, wait_s=0.0) + if len(self._waiters) >= self.max_queue: + raise QueueFull("admission wait queue is full") + waiter = _Waiter(event=threading.Event()) + self._waiters.append(waiter) + + if not waiter.event.wait(timeout=deadline_timeout): + with self._lock: + if waiter.granted: + return Ticket(_gate=self, wait_s=time.monotonic() - started) + waiter.cancelled = True + try: + self._waiters.remove(waiter) + except ValueError: + pass + raise QueueTimeout("timed out waiting for an admission slot") + return Ticket(_gate=self, wait_s=time.monotonic() - started) + + def _release(self) -> None: + with self._lock: + self._in_flight = max(0, self._in_flight - 1) + while self._waiters: + nxt = self._waiters.popleft() + if nxt.cancelled: + continue + nxt.granted = True + self._in_flight += 1 + nxt.event.set() + return diff --git a/src/llm_queue_proxy/cli.py b/src/llm_queue_proxy/cli.py new file mode 100644 index 0000000..805ec7d --- /dev/null +++ b/src/llm_queue_proxy/cli.py @@ -0,0 +1,101 @@ +"""CLI for llm-queue-proxy.""" + +from __future__ import annotations + +import argparse +import os +import sys + +from llm_queue_proxy.paths import DEFAULT_ADMIT_PREFIXES +from llm_queue_proxy.server import ProxyConfig, make_server, parse_listen + + +def _env(name: str, default: str) -> str: + value = os.environ.get(name) + return default if value is None or value == "" else value + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="llm-queue-proxy", + description=( + "FIFO admission proxy for local LLM servers. " + "Generation POSTs share a slot; extra requests wait in order." + ), + ) + parser.add_argument( + "--listen", + default=_env("LLM_QUEUE_LISTEN", "127.0.0.1:8001"), + help="host:port to bind (default 127.0.0.1:8001 / LLM_QUEUE_LISTEN)", + ) + parser.add_argument( + "--upstream", + default=_env("LLM_QUEUE_UPSTREAM", "http://127.0.0.1:8000"), + help="upstream origin (default http://127.0.0.1:8000 / LLM_QUEUE_UPSTREAM)", + ) + parser.add_argument( + "--slots", + type=int, + default=int(_env("LLM_QUEUE_SLOTS", "1")), + help="in-flight generation POSTs allowed (default 1)", + ) + parser.add_argument( + "--max-queue", + type=int, + default=int(_env("LLM_QUEUE_MAX_QUEUE", "32")), + help="max waiters before 429 (default 32)", + ) + parser.add_argument( + "--queue-timeout", + type=float, + default=float(_env("LLM_QUEUE_TIMEOUT", "600")), + help="seconds a waiter may block before 503 (default 600)", + ) + parser.add_argument( + "--admit", + action="append", + dest="admit", + default=None, + help=( + "POST path prefix that takes a slot (repeatable). " + "Defaults to OpenAI/Anthropic generate paths." + ), + ) + return parser + + +def config_from_args(args: argparse.Namespace) -> ProxyConfig: + prefixes = tuple(args.admit) if args.admit else DEFAULT_ADMIT_PREFIXES + parse_listen(args.listen) + return ProxyConfig( + listen=args.listen, + upstream=args.upstream, + slots=args.slots, + max_queue=args.max_queue, + timeout_s=args.queue_timeout, + admit_prefixes=prefixes, + ) + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + config = config_from_args(args) + httpd = make_server(config) + host, port = httpd.server_address[:2] + print( + f"llm-queue-proxy listening on http://{host}:{port} -> {config.upstream} " + f"(slots={config.slots} max_queue={config.max_queue})", + file=sys.stderr, + ) + try: + httpd.serve_forever() + except KeyboardInterrupt: + print("\nshutting down", file=sys.stderr) + finally: + httpd.server_close() + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/llm_queue_proxy/paths.py b/src/llm_queue_proxy/paths.py new file mode 100644 index 0000000..433cc7e --- /dev/null +++ b/src/llm_queue_proxy/paths.py @@ -0,0 +1,28 @@ +"""Which upstream requests consume an admission slot.""" + +from __future__ import annotations + +DEFAULT_ADMIT_PREFIXES: tuple[str, ...] = ( + "/v1/chat/completions", + "/v1/completions", + "/v1/messages", + "/v1/responses", + "/chat/completions", + "/completions", + "/messages", + "/responses", +) + + +def should_admit( + method: str, + path: str, + prefixes: tuple[str, ...] = DEFAULT_ADMIT_PREFIXES, +) -> bool: + if method.upper() != "POST": + return False + route = path.split("?", 1)[0] + for prefix in prefixes: + if route == prefix or route.startswith(prefix + "/"): + return True + return False diff --git a/src/llm_queue_proxy/server.py b/src/llm_queue_proxy/server.py new file mode 100644 index 0000000..0057314 --- /dev/null +++ b/src/llm_queue_proxy/server.py @@ -0,0 +1,277 @@ +"""HTTP reverse proxy with FIFO admission for generation POSTs.""" + +from __future__ import annotations + +import http.client +import http.server +import json +import socket +import sys +import threading +from dataclasses import dataclass +from typing import Any +from urllib.parse import urlparse + +from llm_queue_proxy.admission import Admission, QueueFull, QueueTimeout +from llm_queue_proxy.paths import DEFAULT_ADMIT_PREFIXES, should_admit + +HOP_BY_HOP = { + "connection", + "keep-alive", + "proxy-authenticate", + "proxy-authorization", + "te", + "trailers", + "transfer-encoding", + "upgrade", + "proxy-connection", +} + +QUEUE_HEALTH_PATH = "/__queue/health" + + +@dataclass +class ProxyConfig: + listen: str = "127.0.0.1:8001" + upstream: str = "http://127.0.0.1:8000" + slots: int = 1 + max_queue: int = 32 + timeout_s: float = 600.0 + admit_prefixes: tuple[str, ...] = DEFAULT_ADMIT_PREFIXES + + +@dataclass +class RunningProxy: + httpd: http.server.ThreadingHTTPServer + thread: threading.Thread + port: int + config: ProxyConfig + + def shutdown(self) -> None: + self.httpd.shutdown() + self.httpd.server_close() + self.thread.join(timeout=5) + + +class _Handler(http.server.BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + timeout = 600 + + server: QueueHTTPServer # type: ignore[assignment] + + def log_message(self, format: str, *args: object) -> None: + # Access logs only: never dump headers (Authorization). + sys.stderr.write(f"{self.address_string()} - {format % args}\n") + + def do_GET(self) -> None: # noqa: N802 + self._handle() + + def do_POST(self) -> None: # noqa: N802 + self._handle() + + def do_PUT(self) -> None: # noqa: N802 + self._handle() + + def do_PATCH(self) -> None: # noqa: N802 + self._handle() + + def do_DELETE(self) -> None: # noqa: N802 + self._handle() + + def do_HEAD(self) -> None: # noqa: N802 + self._handle() + + def do_OPTIONS(self) -> None: # noqa: N802 + self._handle() + + def _handle(self) -> None: + path = self.path.split("?", 1)[0] + if path == QUEUE_HEALTH_PATH: + self._write_json(200, self.server.health_payload()) + return + + if should_admit(self.command, self.path, prefixes=self.server.config.admit_prefixes): + try: + ticket = self.server.admission.acquire() + except QueueFull: + self._reject( + 429, + { + "error": { + "message": "admission wait queue is full", + "type": "queue_full", + } + }, + ) + return + except QueueTimeout: + self._reject( + 503, + { + "error": { + "message": "timed out waiting for an admission slot", + "type": "queue_timeout", + } + }, + ) + return + with ticket: + self._forward() + return + self._forward() + + def _forward(self) -> None: + parsed = self.server.upstream + host = parsed.hostname or "127.0.0.1" + port = parsed.port or (443 if parsed.scheme == "https" else 80) + if parsed.scheme == "https": + conn: http.client.HTTPConnection = http.client.HTTPSConnection( + host, port, timeout=self.timeout + ) + else: + conn = http.client.HTTPConnection(host, port, timeout=self.timeout) + + length = int(self.headers.get("Content-Length") or "0") + body = self.rfile.read(length) if length > 0 else None + headers: dict[str, str] = {} + for key, value in self.headers.items(): + if key.lower() in HOP_BY_HOP or key.lower() == "host": + continue + headers[key] = value + headers["Host"] = f"{host}:{port}" if parsed.port else host + if body is not None: + headers["Content-Length"] = str(len(body)) + + try: + conn.request(self.command, self.path, body=body, headers=headers) + upstream = conn.getresponse() + self.send_response(upstream.status, upstream.reason) + has_length = upstream.getheader("Content-Length") is not None + for key, value in upstream.getheaders(): + if key.lower() in HOP_BY_HOP: + continue + self.send_header(key, value) + if not has_length: + self.send_header("Connection", "close") + self.close_connection = True + self.end_headers() + if self.command != "HEAD": + _copy_response_body(upstream, self.wfile) + except OSError: + if not self.wfile.closed: + try: + self._write_json( + 502, + { + "error": { + "message": "upstream request failed", + "type": "bad_gateway", + } + }, + ) + except OSError: + pass + finally: + conn.close() + + def _drain_body(self) -> None: + try: + remaining = int(self.headers.get("Content-Length") or "0") + except ValueError: + remaining = 0 + while remaining > 0: + chunk = self.rfile.read(min(65536, remaining)) + if not chunk: + break + remaining -= len(chunk) + if self.headers.get("Content-Length") is None: + self.close_connection = True + + def _reject(self, status: int, payload: dict[str, Any]) -> None: + self._drain_body() + extra = {"Retry-After": "1"} + if self.close_connection: + extra["Connection"] = "close" + self._write_json(status, payload, extra_headers=extra) + + def _write_json( + self, + status: int, + payload: dict[str, Any], + extra_headers: dict[str, str] | None = None, + ) -> None: + raw = json.dumps(payload).encode("utf-8") + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(raw))) + if extra_headers: + for key, value in extra_headers.items(): + self.send_header(key, value) + self.end_headers() + if self.command != "HEAD": + self.wfile.write(raw) + + +def _copy_response_body(upstream: http.client.HTTPResponse, dest: Any) -> None: + while True: + chunk = upstream.read1(8192) + if not chunk: + break + dest.write(chunk) + dest.flush() + + +class QueueHTTPServer(http.server.ThreadingHTTPServer): + daemon_threads = True + allow_reuse_address = True + + def __init__(self, addr: tuple[str, int], config: ProxyConfig) -> None: + super().__init__(addr, _Handler) + self.config = config + self.admission = Admission( + slots=config.slots, + max_queue=config.max_queue, + timeout_s=config.timeout_s, + ) + self.upstream = urlparse(config.upstream) + if not self.upstream.scheme or not self.upstream.hostname: + raise ValueError(f"invalid upstream URL: {config.upstream}") + + def health_payload(self) -> dict[str, Any]: + snap = self.admission.snapshot() + return { + "ok": True, + "upstream": self.config.upstream, + **snap, + } + + def server_bind(self) -> None: + self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + super().server_bind() + + +def parse_listen(listen: str) -> tuple[str, int]: + if listen.startswith("["): + host, _, rest = listen[1:].partition("]") + if not rest.startswith(":"): + raise ValueError(f"invalid listen address: {listen}") + return host, int(rest[1:]) + if ":" not in listen: + raise ValueError(f"invalid listen address: {listen}") + host, _, port_s = listen.rpartition(":") + return host or "127.0.0.1", int(port_s) + + +def make_server(config: ProxyConfig) -> QueueHTTPServer: + host, port = parse_listen(config.listen) + if host in {"0.0.0.0", "::"}: + raise ValueError("refusing to bind on all interfaces; pass an explicit host") + return QueueHTTPServer((host, port), config) + + +def serve_in_thread(config: ProxyConfig) -> RunningProxy: + httpd = make_server(config) + thread = threading.Thread(target=httpd.serve_forever, daemon=True) + thread.start() + port = httpd.server_address[1] + return RunningProxy(httpd=httpd, thread=thread, port=port, config=config) diff --git a/tests/test_admission.py b/tests/test_admission.py new file mode 100644 index 0000000..dcbc700 --- /dev/null +++ b/tests/test_admission.py @@ -0,0 +1,169 @@ +from __future__ import annotations + +import threading +import time + +import pytest + +from llm_queue_proxy.admission import Admission, QueueFull, QueueTimeout + + +def test_second_waiter_runs_after_first_releases(): + gate = Admission(slots=1, max_queue=8, timeout_s=2.0) + order: list[str] = [] + first_held = threading.Event() + release_first = threading.Event() + + def first(): + ticket = gate.acquire() + order.append("first-in") + first_held.set() + release_first.wait(timeout=2) + ticket.release() + order.append("first-out") + + def second(): + first_held.wait(timeout=2) + ticket = gate.acquire() + order.append("second-in") + ticket.release() + order.append("second-out") + + t1 = threading.Thread(target=first) + t2 = threading.Thread(target=second) + t1.start() + t2.start() + assert first_held.wait(timeout=2) + time.sleep(0.05) + assert order == ["first-in"] + release_first.set() + t1.join(timeout=2) + t2.join(timeout=2) + assert order == ["first-in", "first-out", "second-in", "second-out"] + + +def test_waiters_run_in_arrival_order(): + gate = Admission(slots=1, max_queue=8, timeout_s=2.0) + order: list[str] = [] + held = threading.Event() + release_holder = threading.Event() + + def holder(): + ticket = gate.acquire() + held.set() + release_holder.wait(timeout=2) + ticket.release() + + def waiter(name: str, started: threading.Event): + held.wait(timeout=2) + started.set() + ticket = gate.acquire() + order.append(name) + ticket.release() + + a_started = threading.Event() + b_started = threading.Event() + t_hold = threading.Thread(target=holder) + t_a = threading.Thread(target=waiter, args=("A", a_started)) + t_b = threading.Thread(target=waiter, args=("B", b_started)) + t_hold.start() + assert held.wait(timeout=2) + t_a.start() + assert a_started.wait(timeout=2) + time.sleep(0.05) + t_b.start() + assert b_started.wait(timeout=2) + time.sleep(0.05) + assert gate.snapshot()["queued"] == 2 + release_holder.set() + t_hold.join(timeout=2) + t_a.join(timeout=2) + t_b.join(timeout=2) + assert order == ["A", "B"] + + +def test_queue_full_rejects_without_waiting(): + gate = Admission(slots=1, max_queue=1, timeout_s=5.0) + held = threading.Event() + waiting = threading.Event() + release_holder = threading.Event() + + def holder(): + ticket = gate.acquire() + held.set() + release_holder.wait(timeout=2) + ticket.release() + + def waiter(): + held.wait(timeout=2) + ticket = gate.acquire() + waiting.set() + release_holder.wait(timeout=2) + ticket.release() + + t_hold = threading.Thread(target=holder) + t_wait = threading.Thread(target=waiter) + t_hold.start() + assert held.wait(timeout=2) + t_wait.start() + time.sleep(0.05) + with pytest.raises(QueueFull): + gate.acquire() + release_holder.set() + t_hold.join(timeout=2) + t_wait.join(timeout=2) + assert waiting.is_set() + + +def test_queue_timeout(): + gate = Admission(slots=1, max_queue=8, timeout_s=0.05) + ticket = gate.acquire() + try: + with pytest.raises(QueueTimeout): + gate.acquire() + finally: + ticket.release() + + +def test_ticket_context_manager_releases(): + gate = Admission(slots=1, max_queue=8, timeout_s=2.0) + with gate.acquire(): + assert gate.snapshot()["in_flight"] == 1 + assert gate.snapshot()["in_flight"] == 0 + + +def test_snapshot_counts_in_flight_and_queued(): + gate = Admission(slots=1, max_queue=8, timeout_s=2.0) + held = threading.Event() + queued = threading.Event() + release_holder = threading.Event() + + def holder(): + ticket = gate.acquire() + held.set() + release_holder.wait(timeout=2) + ticket.release() + + def waiter(): + held.wait(timeout=2) + queued.set() + ticket = gate.acquire() + ticket.release() + + t_hold = threading.Thread(target=holder) + t_wait = threading.Thread(target=waiter) + t_hold.start() + assert held.wait(timeout=2) + t_wait.start() + assert queued.wait(timeout=2) + time.sleep(0.05) + snap = gate.snapshot() + assert snap["slots"] == 1 + assert snap["in_flight"] == 1 + assert snap["queued"] == 1 + release_holder.set() + t_hold.join(timeout=2) + t_wait.join(timeout=2) + idle = gate.snapshot() + assert idle["in_flight"] == 0 + assert idle["queued"] == 0 diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..90745f9 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,37 @@ +from llm_queue_proxy.cli import build_parser, config_from_args + + +def test_cli_defaults_listen_localhost(): + args = build_parser().parse_args([]) + cfg = config_from_args(args) + assert cfg.listen == "127.0.0.1:8001" + assert cfg.upstream == "http://127.0.0.1:8000" + assert cfg.slots == 1 + assert cfg.max_queue == 32 + + +def test_cli_overrides(): + args = build_parser().parse_args( + [ + "--listen", + "127.0.0.1:9000", + "--upstream", + "http://127.0.0.1:8000", + "--slots", + "2", + "--max-queue", + "4", + "--queue-timeout", + "12.5", + "--admit", + "/v1/chat/completions", + "--admit", + "/custom", + ] + ) + cfg = config_from_args(args) + assert cfg.listen == "127.0.0.1:9000" + assert cfg.slots == 2 + assert cfg.max_queue == 4 + assert cfg.timeout_s == 12.5 + assert cfg.admit_prefixes == ("/v1/chat/completions", "/custom") diff --git a/tests/test_paths.py b/tests/test_paths.py new file mode 100644 index 0000000..75a4d52 --- /dev/null +++ b/tests/test_paths.py @@ -0,0 +1,23 @@ +from llm_queue_proxy.paths import should_admit + + +def test_admits_openai_and_anthropic_generation_posts(): + assert should_admit("POST", "/v1/chat/completions") + assert should_admit("POST", "/v1/completions") + assert should_admit("POST", "/v1/messages") + assert should_admit("POST", "/v1/responses") + assert should_admit("POST", "/chat/completions") + assert should_admit("POST", "/v1/chat/completions?foo=1") + + +def test_does_not_admit_gets_or_health(): + assert not should_admit("GET", "/v1/chat/completions") + assert not should_admit("GET", "/health") + assert not should_admit("GET", "/v1/models") + assert not should_admit("POST", "/health") + assert not should_admit("OPTIONS", "/v1/chat/completions") + + +def test_custom_admit_prefixes(): + assert should_admit("POST", "/custom/gen", prefixes=("/custom/gen",)) + assert not should_admit("POST", "/v1/chat/completions", prefixes=("/custom/gen",)) diff --git a/tests/test_proxy.py b/tests/test_proxy.py new file mode 100644 index 0000000..59475d3 --- /dev/null +++ b/tests/test_proxy.py @@ -0,0 +1,422 @@ +from __future__ import annotations + +import http.client +import http.server +import json +import socket +import threading +import time +from typing import Any + +import pytest + +from llm_queue_proxy.server import ProxyConfig, make_server, serve_in_thread + + +class _Upstream(http.server.BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + state: dict[str, Any] = {} + + def log_message(self, *_args: object) -> None: + return + + def do_GET(self) -> None: # noqa: N802 + if self.path.startswith("/hold"): + self.state["get_inflight"] = self.state.get("get_inflight", 0) + 1 + self.state["get_peak"] = max( + self.state.get("get_peak", 0), self.state["get_inflight"] + ) + self.state["hold_get"].wait(timeout=2) + self.state["get_inflight"] -= 1 + self._json(200, {"ok": True, "path": self.path}) + return + if self.path.startswith("/health") or self.path.startswith("/v1/models"): + self._json(200, {"ok": True, "path": self.path}) + return + self._json(404, {"error": "nope"}) + + def do_POST(self) -> None: # noqa: N802 + length = int(self.headers.get("Content-Length", "0")) + body = self.rfile.read(length) if length else b"" + if self.path.startswith("/v1/chat/completions"): + self.state["post_inflight"] = self.state.get("post_inflight", 0) + 1 + self.state["post_peak"] = max( + self.state.get("post_peak", 0), self.state["post_inflight"] + ) + delay = self.state.get("post_delay", 0.0) + stream = self.state.get("stream", False) + if stream: + self.send_response(200) + self.send_header("Content-Type", "text/event-stream") + self.send_header("Cache-Control", "no-cache") + self.end_headers() + self.wfile.write(b"data: a\n\n") + self.wfile.flush() + self.state["streamed_a"].set() + self.state["stream_continue"].wait(timeout=2) + self.wfile.write(b"data: b\n\n") + self.wfile.flush() + else: + if delay: + time.sleep(delay) + self._json(200, {"echo": body.decode("utf-8"), "path": self.path}) + self.state["post_inflight"] -= 1 + return + self._json(200, {"echo": body.decode("utf-8"), "path": self.path}) + + def _json(self, status: int, payload: dict[str, Any]) -> None: + raw = json.dumps(payload).encode("utf-8") + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(raw))) + self.end_headers() + self.wfile.write(raw) + + +def _start_upstream(state: dict[str, Any]) -> tuple[http.server.ThreadingHTTPServer, int]: + _Upstream.state = state + httpd = http.server.ThreadingHTTPServer(("127.0.0.1", 0), _Upstream) + thread = threading.Thread(target=httpd.serve_forever, daemon=True) + thread.start() + return httpd, httpd.server_address[1] + + +def _start_proxy(**kwargs: Any): + cfg = ProxyConfig(listen="127.0.0.1:0", **kwargs) + return serve_in_thread(cfg) + + +def _request( + port: int, + method: str, + path: str, + body: bytes | None = None, + timeout: float = 3.0, +) -> tuple[int, dict[str, str], bytes]: + conn = http.client.HTTPConnection("127.0.0.1", port, timeout=timeout) + headers = {} + if body is not None: + headers["Content-Type"] = "application/json" + headers["Content-Length"] = str(len(body)) + conn.request(method, path, body=body, headers=headers) + resp = conn.getresponse() + data = resp.read() + hdrs = {k.lower(): v for k, v in resp.getheaders()} + conn.close() + return resp.status, hdrs, data + + +def test_passthrough_gets_are_concurrent(): + state: dict[str, Any] = {"hold_get": threading.Event()} + up, up_port = _start_upstream(state) + proxy = _start_proxy(upstream=f"http://127.0.0.1:{up_port}", slots=1) + try: + results: list[int] = [] + + def hit(): + status, _, _ = _request(proxy.port, "GET", "/hold") + results.append(status) + + t1 = threading.Thread(target=hit) + t2 = threading.Thread(target=hit) + t1.start() + t2.start() + deadline = time.time() + 2 + while state.get("get_inflight", 0) < 2 and time.time() < deadline: + time.sleep(0.01) + assert state.get("get_inflight") == 2 + state["hold_get"].set() + t1.join(timeout=2) + t2.join(timeout=2) + assert results == [200, 200] + assert state["get_peak"] == 2 + finally: + proxy.shutdown() + up.shutdown() + + +def test_generation_posts_are_serialized(): + state: dict[str, Any] = {"post_delay": 0.15} + up, up_port = _start_upstream(state) + proxy = _start_proxy(upstream=f"http://127.0.0.1:{up_port}", slots=1) + try: + statuses: list[int] = [] + + def hit(): + status, _, _ = _request( + proxy.port, "POST", "/v1/chat/completions", body=b'{"m":1}' + ) + statuses.append(status) + + t1 = threading.Thread(target=hit) + t2 = threading.Thread(target=hit) + t1.start() + t2.start() + t1.join(timeout=3) + t2.join(timeout=3) + assert statuses == [200, 200] + assert state["post_peak"] == 1 + finally: + proxy.shutdown() + up.shutdown() + + +def test_queue_full_returns_429(): + state: dict[str, Any] = {"post_delay": 0.4} + up, up_port = _start_upstream(state) + proxy = _start_proxy( + upstream=f"http://127.0.0.1:{up_port}", + slots=1, + max_queue=1, + timeout_s=5.0, + ) + try: + started = threading.Event() + + def first(): + started.set() + _request(proxy.port, "POST", "/v1/chat/completions", body=b"{}", timeout=5) + + t1 = threading.Thread(target=first) + t1.start() + assert started.wait(timeout=2) + time.sleep(0.05) + + def second(): + _request(proxy.port, "POST", "/v1/chat/completions", body=b"{}", timeout=5) + + t2 = threading.Thread(target=second) + t2.start() + time.sleep(0.05) + status, hdrs, body = _request( + proxy.port, "POST", "/v1/chat/completions", body=b"{}" + ) + assert status == 429 + assert "retry-after" in hdrs + assert b"queue" in body.lower() + t1.join(timeout=3) + t2.join(timeout=3) + finally: + proxy.shutdown() + up.shutdown() + + +def test_queue_timeout_returns_503(): + state: dict[str, Any] = {"post_delay": 0.4} + up, up_port = _start_upstream(state) + proxy = _start_proxy( + upstream=f"http://127.0.0.1:{up_port}", + slots=1, + max_queue=8, + timeout_s=0.05, + ) + try: + + def first(): + _request(proxy.port, "POST", "/v1/chat/completions", body=b"{}", timeout=5) + + t1 = threading.Thread(target=first) + t1.start() + time.sleep(0.05) + status, _, body = _request( + proxy.port, "POST", "/v1/chat/completions", body=b"{}" + ) + assert status == 503 + assert b"timed out" in body.lower() + t1.join(timeout=3) + finally: + proxy.shutdown() + up.shutdown() + + +def test_stream_forwards_chunks_without_buffering_the_whole_body(): + state: dict[str, Any] = { + "stream": True, + "streamed_a": threading.Event(), + "stream_continue": threading.Event(), + } + up, up_port = _start_upstream(state) + proxy = _start_proxy(upstream=f"http://127.0.0.1:{up_port}", slots=1) + try: + conn = http.client.HTTPConnection("127.0.0.1", proxy.port, timeout=3) + conn.request( + "POST", + "/v1/chat/completions", + body=b'{"stream":true}', + headers={"Content-Type": "application/json"}, + ) + resp = conn.getresponse() + assert resp.status == 200 + assert state["streamed_a"].wait(timeout=2) + first = resp.read(8) + assert first == b"data: a\n" + state["stream_continue"].set() + buf = first + deadline = time.time() + 2 + while b"data: b" not in buf and time.time() < deadline: + chunk = resp.read(1) + if not chunk: + break + buf += chunk + assert b"data: b" in buf + conn.close() + finally: + state["stream_continue"].set() + proxy.shutdown() + up.shutdown() + + +def test_queue_health_endpoint(): + up, up_port = _start_upstream({}) + proxy = _start_proxy(upstream=f"http://127.0.0.1:{up_port}", slots=1) + try: + status, _, body = _request(proxy.port, "GET", "/__queue/health") + assert status == 200 + payload = json.loads(body) + assert payload["ok"] is True + assert payload["in_flight"] == 0 + assert payload["slots"] == 1 + assert payload["upstream"].startswith("http://127.0.0.1:") + finally: + proxy.shutdown() + up.shutdown() + + +def test_models_get_does_not_take_a_slot(): + state: dict[str, Any] = {"post_delay": 0.3} + up, up_port = _start_upstream(state) + proxy = _start_proxy(upstream=f"http://127.0.0.1:{up_port}", slots=1) + try: + blocker = threading.Thread( + target=lambda: _request( + proxy.port, "POST", "/v1/chat/completions", body=b"{}", timeout=5 + ) + ) + blocker.start() + time.sleep(0.05) + status, _, body = _request(proxy.port, "GET", "/v1/models") + assert status == 200 + assert json.loads(body)["path"] == "/v1/models" + blocker.join(timeout=3) + finally: + proxy.shutdown() + up.shutdown() + + +def test_make_server_refuses_wildcard_bind(): + with pytest.raises(ValueError, match="all interfaces"): + make_server(ProxyConfig(listen="0.0.0.0:8001")) + with pytest.raises(ValueError, match="all interfaces"): + make_server(ProxyConfig(listen="[::]:8001")) + + +def test_chunked_sse_is_decoded_not_forwarded_as_framing(): + ready = threading.Event() + continue_stream = threading.Event() + bound = threading.Event() + port_box: dict[str, int] = {} + + def raw_upstream() -> None: + srv = socket.socket() + srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + srv.bind(("127.0.0.1", 0)) + srv.listen(1) + port_box["port"] = srv.getsockname()[1] + bound.set() + srv.settimeout(3) + conn, _addr = srv.accept() + buf = b"" + while b"\r\n\r\n" not in buf: + buf += conn.recv(4096) + conn.sendall( + b"HTTP/1.1 200 OK\r\n" + b"Content-Type: text/event-stream\r\n" + b"Transfer-Encoding: chunked\r\n" + b"\r\n" + b"8\r\ndata: a\n\r\n" + ) + ready.set() + continue_stream.wait(timeout=2) + conn.sendall(b"8\r\ndata: b\n\r\n0\r\n\r\n") + conn.close() + srv.close() + + t = threading.Thread(target=raw_upstream, daemon=True) + t.start() + assert bound.wait(timeout=2) + proxy = _start_proxy(upstream=f"http://127.0.0.1:{port_box['port']}", slots=1) + try: + conn = http.client.HTTPConnection("127.0.0.1", proxy.port, timeout=3) + conn.request( + "POST", + "/v1/chat/completions", + body=b'{"stream":true}', + headers={"Content-Type": "application/json"}, + ) + resp = conn.getresponse() + assert resp.status == 200 + assert ready.wait(timeout=2) + first = resp.read(8) + assert first == b"data: a\n" + assert not first.startswith(b"8") + continue_stream.set() + rest = b"" + deadline = time.time() + 2 + while b"data: b" not in first + rest and time.time() < deadline: + chunk = resp.read(1) + if not chunk: + break + rest += chunk + assert b"data: b" in first + rest + conn.close() + finally: + continue_stream.set() + proxy.shutdown() + t.join(timeout=2) + + +def test_429_consumes_body_so_connection_can_be_reused(): + state: dict[str, Any] = {"post_delay": 0.5} + up, up_port = _start_upstream(state) + proxy = _start_proxy( + upstream=f"http://127.0.0.1:{up_port}", + slots=1, + max_queue=1, + timeout_s=5.0, + ) + try: + blocker = threading.Thread( + target=lambda: _request( + proxy.port, "POST", "/v1/chat/completions", body=b"{}", timeout=5 + ) + ) + waiter = threading.Thread( + target=lambda: _request( + proxy.port, "POST", "/v1/chat/completions", body=b"{}", timeout=5 + ) + ) + blocker.start() + time.sleep(0.05) + waiter.start() + time.sleep(0.05) + conn = http.client.HTTPConnection("127.0.0.1", proxy.port, timeout=3) + body = b'{"left":"over"}' + conn.request( + "POST", + "/v1/chat/completions", + body=body, + headers={"Content-Type": "application/json", "Content-Length": str(len(body))}, + ) + resp = conn.getresponse() + assert resp.status == 429 + resp.read() + conn.request("GET", "/v1/models") + follow = conn.getresponse() + assert follow.status == 200 + assert json.loads(follow.read())["path"] == "/v1/models" + conn.close() + blocker.join(timeout=3) + waiter.join(timeout=3) + finally: + proxy.shutdown() + up.shutdown()