Priority Queues: Free Users Wait, Paying Users Do Not

Tiered service sounds like a one-line change: put requests in a priority queue and serve the high-priority ones first. That implementation has three failure modes — unbounded queues, permanent starvation of the low tier, and one enthusiastic customer occupying all the capacity of their own tier. Fixing all three takes about eighty lines, and the numbers in it are derivable.

What a plain priority queue gets wrong

Strict priority means: serve any high-priority item before any low-priority one. Under sustained load where high-priority arrivals alone meet or exceed capacity, that has a precise consequence — the low-priority queue is never served at all. Not slowly: never. Little’s law makes this exact, because if the departure rate for a class is zero, its queue grows without bound and its wait diverges.

The second failure is the unbounded queue itself. A queue with no limit converts an overload into a latency collapse: requests are accepted, wait past every client timeout, get computed anyway, and the answer is delivered to a socket that closed minutes ago. You pay full GPU cost for work nobody receives. On a system with expensive computation this is the single most costly mistake in the list.

The third is noisy neighbours. Priority is per tier, but capacity is consumed per request, so one paying customer submitting a thousand concurrent jobs occupies the entire premium tier and every other paying customer waits behind them.

Three defects, three named fixes: a bound, an ageing rule, and a per-tenant concurrency cap.

Bounding the queue

The bound is not a round number, it is a latency budget divided by a service rate — the same Little’s law relationship used for autoscaling, applied to admission instead of to replica count.

Maximum useful queue length per tier:

  L_max = μ × W_max

  μ      total service rate of the pool, requests/second
  W_max  the longest wait that is still useful to that tier

Worked, all assumptions labelled:

  pool: 6 replicas × 0.67 req/s each ....... μ = 4 req/s
  premium tier tolerable wait .............. W_max = 5 s   → L_max = 20
  free tier tolerable wait ................. W_max = 60 s  → L_max = 240

A queued premium request beyond position 20 will wait more than 5 s, so
accepting it is a promise you have already broken. Refuse it instead, with
a Retry-After the client can use.

Sanity check against the client: if the client's own timeout is 30 s, any
queue position implying a wait beyond 30 s is guaranteed waste. Take the
minimum of the two bounds.

Refusing early is kinder than it sounds. A 429 with a Retry-After arrives in milliseconds and lets a client back off deliberately; a request that times out after thirty seconds gives it nothing and has already cost you a slot. Handling 429 is the client side of the same contract, and backpressure is the general principle.

Ageing: the starvation guard

Ageing means a request’s effective priority improves the longer it waits, so every request is eventually served. One line expresses it:

effective_priority = base_priority − (waited_seconds / ageing_rate)

  lower number = served sooner

  base_priority:  premium 0,  standard 10,  free 20
  ageing_rate:    seconds of waiting worth one point of priority

Choose ageing_rate from the guarantee you want to make. To promise that a
free request never waits more than 120 s behind premium work:

  it must climb 20 points in 120 s  →  ageing_rate = 120/20 = 6 s per point

Check the consequence for the premium tier: a free request that has waited
60 s has effective priority 20 − 10 = 10, so it now outranks any standard
request that has just arrived, and after 120 s it ties with fresh premium
work. That is the guarantee working, and it is a cost the premium tier pays
in a bounded, stated way rather than an unbounded, hidden one.

This is a strictly better contract than strict priority for both sides, because it is a contract at all. “Free requests are served within two minutes under load” is something you can publish and alert on. “Free requests are served when there is spare capacity” is not.

Per-tenant caps stop one customer taking it all

Cap concurrent in-flight requests per tenant, not just per tier. The cap is a second admission gate: a request whose tenant is already at its limit stays in the queue and the scheduler skips it, so other tenants of the same tier move ahead.

Size the cap as a share of pool capacity, not as an absolute:

per_tenant_limit = max(1, floor(pool_concurrency × share))

  pool_concurrency  total simultaneous generations the pool sustains
  share             the fraction one tenant may hold at once

  pool_concurrency = 24 (6 replicas × 4 concurrent)
  premium  share 0.25 → 6 concurrent
  standard share 0.10 → 2 concurrent
  free     share 0.04 → 1 concurrent

The floor of 1 matters: a share that rounds to zero silently blocks a
tenant forever, which is the worst possible bug in a fairness mechanism.

Note that this is a concurrency cap, not a rate limit, and it does a different job. A rate limit constrains requests per minute; a concurrency cap constrains simultaneous occupancy of the resource, which is what actually determines whether other tenants can be served. Most services need both — rate limits explained covers the other one, and concurrency control covers this one from the client’s side.

The scheduler, in full

# Priority queueing with ageing, per-tenant concurrency caps and a bound.
# Single-process; for a fleet, keep the same logic and move the state into a
# shared store with atomic operations.
import asyncio, heapq, itertools, time
from dataclasses import dataclass, field

TIERS = {                       # base priority, tolerable wait, tenant share
    "premium":  (0,   5.0, 0.25),
    "standard": (10, 20.0, 0.10),
    "free":     (20, 60.0, 0.04),
}
AGEING_RATE = 6.0               # seconds of waiting per point of priority
POOL_CONCURRENCY = 24           # simultaneous generations the pool sustains
MEAN_SERVICE_SECONDS = 6.0      # measured, at POOL_CONCURRENCY concurrency

@dataclass(order=True)
class Job:
    sort_key: float
    seq: int
    tenant: str = field(compare=False)
    tier: str = field(compare=False)
    enqueued: float = field(compare=False)
    fut: asyncio.Future = field(compare=False, default=None)

class Scheduler:
    def __init__(self):
        self._heap = []
        self._counter = itertools.count()
        self._inflight_total = 0
        self._inflight_by_tenant = {}
        self._queued_by_tier = {t: 0 for t in TIERS}
        self._wake = asyncio.Event()

    def _bound(self, tier: str) -> int:
        _, w_max, _ = TIERS[tier]
        service_rate = POOL_CONCURRENCY / MEAN_SERVICE_SECONDS
        return max(1, int(service_rate * w_max))

    def _tenant_limit(self, tier: str) -> int:
        _, _, share = TIERS[tier]
        return max(1, int(POOL_CONCURRENCY * share))

    async def submit(self, tenant: str, tier: str, work) -> "asyncio.Future":
        if self._queued_by_tier[tier] >= self._bound(tier):
            raise Overloaded(retry_after=self._estimated_wait(tier))
        base, _, _ = TIERS[tier]
        now = time.monotonic()
        job = Job(sort_key=base, seq=next(self._counter), tenant=tenant,
                  tier=tier, enqueued=now, fut=asyncio.get_running_loop().create_future())
        job.work = work
        heapq.heappush(self._heap, job)
        self._queued_by_tier[tier] += 1
        self._wake.set()
        return job.fut

    def _rescore(self, now: float) -> None:
        """Recompute effective priorities. O(n log n); call on a timer, not
        per dispatch, and only when the queue is non-trivially long."""
        for job in self._heap:
            base, _, _ = TIERS[job.tier]
            job.sort_key = base - (now - job.enqueued) / AGEING_RATE
        heapq.heapify(self._heap)

    def _next_runnable(self):
        """Pop the best job whose tenant is under its cap. Jobs that are
        blocked by their tenant cap are set aside and pushed back."""
        held = []
        chosen = None
        while self._heap:
            job = heapq.heappop(self._heap)
            limit = self._tenant_limit(job.tier)
            if self._inflight_by_tenant.get(job.tenant, 0) < limit:
                chosen = job
                break
            held.append(job)
        for j in held:
            heapq.heappush(self._heap, j)
        return chosen

    async def run(self):
        last_rescore = 0.0
        while True:
            now = time.monotonic()
            if now - last_rescore > 1.0 and len(self._heap) > 16:
                self._rescore(now); last_rescore = now
            if self._inflight_total >= POOL_CONCURRENCY or not self._heap:
                self._wake.clear()
                await asyncio.wait_for(self._wake.wait(), timeout=1.0) \
                    if self._heap else await self._wake.wait()
                continue
            job = self._next_runnable()
            if job is None:                       # every job blocked by a cap
                await asyncio.sleep(0.01)
                continue
            self._queued_by_tier[job.tier] -= 1
            self._inflight_total += 1
            self._inflight_by_tenant[job.tenant] = \
                self._inflight_by_tenant.get(job.tenant, 0) + 1
            asyncio.create_task(self._execute(job))

    async def _execute(self, job: Job):
        waited = time.monotonic() - job.enqueued
        metrics.observe("queue_wait_seconds", waited, tier=job.tier)
        try:
            job.fut.set_result(await job.work())
        except Exception as exc:
            job.fut.set_exception(exc)
        finally:
            self._inflight_total -= 1
            n = self._inflight_by_tenant[job.tenant] - 1
            if n: self._inflight_by_tenant[job.tenant] = n
            else: del self._inflight_by_tenant[job.tenant]
            self._wake.set()

Three implementation notes. The rescoring pass is O(n log n), so it runs on a timer and only when the queue is long enough to matter — rescoring a queue of four is wasted work. Jobs blocked by a tenant cap are pushed back rather than dropped, which keeps their accumulated ageing. And the sequence counter breaks ties in arrival order, so equal-priority requests are FIFO, which is what users expect.

For a multi-process fleet the same logic holds but the state must move: in-flight counts per tenant and queue lengths per tier become entries in a shared store with atomic increment and a TTL, so that a crashed worker’s slots are released rather than leaked. The TTL is the important detail — without it, one crash permanently reduces a tenant’s cap.

Where the queue should live

The scheduler above assumes an in-process queue in front of the model server. That is the right default, and it is worth being explicit about why, because the alternative — a message broker in front of a pool of workers — is the shape most people reach for first.

The distinction that decides it is whether a request outlives its client. If it does — a document upload processed later, an evaluation run — the queue must be durable and the broker is right. If it does not, durability buys nothing, because a queued request whose client disconnected is work you should be discarding rather than persisting.

For the synchronous case there is one more decision: whether admission state is per replica or shared. Per replica is simpler and it is what the code above does, but it means the tenant cap is enforced per replica, so a tenant with a cap of two can hold two slots on each of six replicas. If that matters — and it matters as soon as one tenant can be significantly larger than the others — the in-flight counts must move to a shared store with atomic increment and a TTL. Budget a round trip on the admission path for it, and check that the store is on the same network segment, because a slow admission check is latency added to every request including the ones you accept.

Whichever placement you choose, make the load balancer aware of it. Least-outstanding-requests balancing beats round-robin substantially for a workload whose service times vary by an order of magnitude, and it is usually a one-line change — inference load balancing covers why the usual algorithms behave badly here.

What to tell a client you are refusing

A refusal should carry enough for the client to behave well. That is three things: the status, an estimate, and a distinction.

HTTP/1.1 429 Too Many Requests
Retry-After: 12
X-Queue-Position: 241
X-Tier: free

{"error": {"type": "capacity", "message":
  "Queue for the free tier is at its bound. Estimated wait exceeds 60 s.",
  "retry_after_seconds": 12}}

The distinction that matters: a 429 because the tenant exceeded their limit is a different event from a 429 because the system is saturated. The first is the client’s to fix by slowing down; the second is yours to fix by adding capacity, and a client backing off will not help. Use distinct error types so both sides can tell them apart, and count them separately — normalising API errors covers the taxonomy.

What to measure

• Queue wait, p50 and p95, per tier. The service-level objective lives here, not on total latency, because total latency mixes queueing with generation length.
• Age of the oldest queued request, per tier. The starvation alarm. If this exceeds your ageing guarantee, the ageing rule is not working or the pool is simply too small.
• Rejections, split by cause. Tier bound versus tenant cap. Rising tenant-cap rejections mean one customer needs a conversation; rising bound rejections mean you need capacity.
• Concurrency utilisation. In-flight divided by pool concurrency. Persistently below one while requests are queueing means the tenant caps are too tight and you are refusing work you could have done.
• Abandoned requests. Clients that disconnected while queued. Every one is compute you were about to spend on nobody — check for a closed connection at dispatch time, not only at the end.