tigzig-concurrency 0.1.0

Bounded wait queue + client-IP forwarding for FastAPI backends

tigzig-concurrency

Bounded wait queue + client-IP forwarding for FastAPI / Starlette backends.

A standing-rule pattern for the tigzig fleet. Replaces ad-hoc per-app concurrency-check and IP-extraction code so all backends behave the same way under load.

Why this exists

Two recurring problems across the fleet:

1. Fast-fail concurrency caps reject legitimate users. A user double-clicks submit, two of their parallel sub-requests collide, and one gets a 429 even though the server is mostly idle. This package replaces the fast-fail check with a small bounded queue: the second request waits up to N seconds for a slot, then either runs or 429s. The server is still protected from sustained overload (queue is bounded), but micro-bursts succeed.

2. Backend-to-backend calls trip downstream per-IP limits. When backend A (e.g. TA) calls backend B (e.g. yfin) on behalf of an end user, B sees the request from A's container IP - so two unrelated end users routed via A look like ONE user at B and trip B's per-IP cap. This package provides forward_client_headers() so A passes the real user IP, and get_real_client_ip() (which B already uses) reads it.

Installation

pip install tigzig-concurrency

Add to your requirements.txt:

tigzig-concurrency>=0.1.0

Quick start

1. Wrap your endpoint with the queue

from fastapi import FastAPI, Request
from tigzig_concurrency import BoundedQueue, get_real_client_ip

app = FastAPI()
queue = BoundedQueue.from_env()  # reads MAX_INFLIGHT_*, MAX_QUEUE_* from env

@app.post("/api/work")
async def do_work(request: Request):
    client_ip = get_real_client_ip(request)
    await queue.acquire(client_ip)
    try:
        # ... your endpoint work ...
        return {"ok": True}
    finally:
        queue.release(client_ip)

2. Forward the real user IP on backend-to-backend calls

import httpx
from tigzig_concurrency import forward_client_headers, get_real_client_ip

@app.post("/api/work")
async def do_work(request: Request):
    client_ip = get_real_client_ip(request)
    await queue.acquire(client_ip)
    try:
        async with httpx.AsyncClient() as http:
            response = await http.get(
                "https://other-backend.tigzig.com/data",
                headers=forward_client_headers(client_ip),
            )
        return response.json()
    finally:
        queue.release(client_ip)

The downstream backend's get_real_client_ip() will read the X-Original-Client-IP header and rate-limit per real end user.

Configuration cheatsheet

Five environment variables. You will forget what these mean in three weeks - this is the cheatsheet. Future-you, hello.

The five knobs

Variable	What it does	When to raise / lower
MAX_INFLIGHT_PER_IP	Active requests per IP at one time.	One real user can have this many running at once. Higher = more permissive for power-users; lower = stricter fairness.
MAX_INFLIGHT_GLOBAL	Active requests across all IPs combined.	Cap on total CPU/memory the server commits at once. Higher = more parallelism if you have headroom; lower if memory is tight.
MAX_QUEUE_PER_IP	Extra requests per IP that may wait in line.	Bigger queue = more forgiveness for "user clicked submit 4 times in 5 seconds". 0 = no queueing per-IP (fast-fail behaviour).
MAX_QUEUE_GLOBAL	Extra requests across all IPs that may wait.	Total system buffer for bursts. Beyond this -> instant 503.
MAX_QUEUE_WAIT_SECONDS	How long any one request waits in queue before 429.	Match to typical request length; wait + work + slack should stay under your edge timeout (Cloudflare free is 100s).

Behaviour

Situation	Response
Slot available	Run immediately (no queue overhead).
Slot taken, queue not full	Wait up to MAX_QUEUE_WAIT_SECONDS for a slot, then run, OR 429 if the wait expired.
Per-IP queue full at arrival	503 immediately (this user is hammering).
Global queue full at arrival	503 immediately (server is overloaded).

Both 429 and 503 include a Retry-After: 10 header so the frontend can show a friendly "server busy, retry in 10s" spinner instead of a hard error.

Tuning profiles for the tigzig fleet

Three workload types cover most apps. Pick the closest profile, set the env vars accordingly, and tune from there if needed.

Profile A: fast IO-bound (1-5s per request)

For yfin, mdtopdf, llama-parse, markitdown, neon-proxy, supabase-proxy, and any wrapper that mostly does network I/O + light pandas.

MAX_INFLIGHT_PER_IP=6
MAX_INFLIGHT_GLOBAL=15
MAX_QUEUE_PER_IP=4
MAX_QUEUE_GLOBAL=10
MAX_QUEUE_WAIT_SECONDS=5

Why: requests are quick, so being aggressive with concurrency is safe (no memory pressure). Short wait window because end users expect fast responses; queue absorbs occasional bursts.

Profile B: slow LLM / compute-bound (10-45s per request)

For TA, FFN/SPR, QuantStats, codeinter (Python sandbox), codeinter-gemini.

MAX_INFLIGHT_PER_IP=2
MAX_INFLIGHT_GLOBAL=6
MAX_QUEUE_PER_IP=3
MAX_QUEUE_GLOBAL=8
MAX_QUEUE_WAIT_SECONDS=30

Why: each request hogs a CPU or holds an LLM connection for a long time. Tighter in-flight cap to avoid memory blowup. Longer wait because users already expect slow; an extra 15-30s of queue is invisible vs the 30s of actual work. Total wait + work stays under Cloudflare 100s.

Profile C: heavy memory / database (DuckDB queries, Flowise)

For duckdb-backend, duckdb-dashboards-backend, flowise-docker-custom.

MAX_INFLIGHT_PER_IP=2
MAX_INFLIGHT_GLOBAL=4
MAX_QUEUE_PER_IP=2
MAX_QUEUE_GLOBAL=4
MAX_QUEUE_WAIT_SECONDS=15

Why: a single heavy DuckDB query can pin a CPU AND eat a few hundred MB of memory. Tight global cap matters more than per-IP fairness. Bump the container memory limit if you raise the global cap.

Container-limit reminder

These knobs only protect the server if the container itself has enough CPU and memory. For the tigzig fleet's standard:

limits_cpus=2.0
limits_memory=1024m   (or 2048m for memory-heavy or many-concurrent profiles)
limits_memory_swap=2048m

A queue with MAX_INFLIGHT_GLOBAL=15 on a container capped at cpus=0.25 will just queue everything and run them slowly. The CPU bump is the bigger lever.

Multiple queues in one app

If one endpoint is much heavier than others, you can run two queues with different settings via the prefix argument:

light_queue = BoundedQueue.from_env(prefix="LIGHT_", name="light")
heavy_queue = BoundedQueue.from_env(prefix="HEAVY_", name="heavy")

Then set LIGHT_MAX_INFLIGHT_PER_IP, HEAVY_MAX_INFLIGHT_PER_IP etc. in your environment.

Health / introspection

The queue exposes its current state via a stats property. Useful for /healthz endpoints:

@app.get("/healthz/queue")
def queue_health():
    return queue.stats

Returns:

{
  "name": "queue",
  "limits": {"inflight_per_ip": 2, "inflight_global": 6, ...},
  "current": {"inflight_global": 3, "queued_global": 1, "queued_per_ip": {"1.2.3.4": 1}}
}

Reference: header chain that get_real_client_ip reads

In order:

1. X-Original-Client-IP (canonical for trusted upstream backends)
2. CF-Connecting-IP (Cloudflare)
3. X-Forwarded-For (first IP if comma-separated)
4. X-Real-IP
5. request.client.host (socket peer; usually docker proxy)

Returns "unknown" if no IP can be resolved.

Why X-Original-Client-IP over X-Forwarded-For

XFF gets rewritten by every hop (Cloudflare, Caddy, nginx). For trusted internal forwarding it's safer to use a custom header that only the trusted upstream sets. X-Original-Client-IP is the convention across the tigzig fleet; the helper sets both to maximize compatibility but the canonical reader prefers the custom header.

Versioning

Semantic versioning. Breaking changes to function signatures or env-var names bump the major version. Tuning-profile recommendations may evolve across minor versions; check the README of the version you're using.

License

MIT.