Fairness, Quotas & Noisy Neighbors

"""Fairness for multi-tenant APIs: global limit vs per-tenant quota vs weighted fair queuing.

Run: python3 fairness.py   (stdlib only, instant; simulates one second of traffic)
Scenario from Diagram 2: Tenant A runs a 950 req/s batch burst, Tenant B sends 100 req/s.
"""
from collections import deque


def arrivals():
    """A's batch lands first in the second, then B's normal traffic (the noisy-neighbor case)."""
    return ["A"] * 950 + ["B"] * 100


def global_limit(reqs, limit=1000):
    """Wrong pick: one shared counter. First come, first served, regardless of tenant."""
    served = {"A": 0, "B": 0}
    for i, t in enumerate(reqs):
        if i < limit:
            served[t] += 1
    return served


def per_tenant_quota(reqs, quota):
    """Step 2-4a of Diagram 1: each tenant has its own budget, so a 429 hits only that tenant."""
    used = {t: 0 for t in quota}
    admitted, rejected = [], {t: 0 for t in quota}
    for t in reqs:
        if used[t] < quota[t]:
            used[t] += 1
            admitted.append(t)
        else:
            rejected[t] += 1
    return admitted, rejected


def fifo(queue, capacity):
    """One shared queue: whoever enqueued first gets the workers."""
    served = {"A": 0, "B": 0}
    for t in queue[:capacity]:
        served[t] += 1
    return served


def drr(queue, capacity, weights, quantum=1):
    """Step 5: deficit round robin, the practical form of weighted fair queuing.

    Each round a lane earns quantum * weight credits and spends 1 credit per request
    (use the request cost for uneven work). Empty lanes keep no credit, so idle tenants
    cannot bank capacity and burst later.
    """
    lanes = {t: deque() for t in weights}
    for t in queue:
        lanes[t].append(t)                     # Step 4b: enqueue in the tenant lane
    deficit = {t: 0 for t in weights}
    served = {t: 0 for t in weights}
    while capacity > 0 and any(lanes.values()):
        for t, w in weights.items():
            if not lanes[t]:
                deficit[t] = 0
                continue
            deficit[t] += quantum * w
            while lanes[t] and deficit[t] >= 1 and capacity > 0:
                lanes[t].popleft()
                deficit[t] -= 1
                served[t] += 1                 # Step 6: a worker serves it
                capacity -= 1
    return served


if __name__ == "__main__":
    reqs = arrivals()
    print("global limit 1000/s       ->", global_limit(reqs), "(B loses 50 to A's burst)")
    admitted, rejected = per_tenant_quota(reqs, {"A": 600, "B": 300})
    print("per-tenant quota A600 B300 -> rejected", rejected, "(429 for A only)")
    print("workers 500/s, FIFO        ->", fifo(admitted, 500), "(B starves behind A's queue)")
    print("workers 500/s, DRR A1 B2   ->", drr(admitted, 500, {"A": 1, "B": 2}))
    print("DRR, B idle this second    ->", drr([t for t in admitted if t == "A"], 500, {"A": 1, "B": 2}),
          "(work conserving: A may use the spare capacity)")

/**
 * Fair scheduler for a shared worker pool: critical lane first, then deficit round robin by
 * tier weight, plus a per-tenant concurrency cap. Compared against one FIFO queue.
 * Run: npx tsx fairness.ts   (no deps, about 1 s)
 */
type Job = { tenant: string; enqueued: number; run: () => Promise<void> };

const sleep = (ms: number) => new Promise<void>((r) => setTimeout(r, ms));

class FairScheduler {
  private lanes = new Map<string, Job[]>();
  private deficit = new Map<string, number>();
  private inflight = new Map<string, number>();
  private busy = 0;

  constructor(
    private workers: number,
    private weights: Record<string, number>, // tier weight per tenant
    private maxConcurrent: number,           // per-tenant concurrency quota
    private critical = new Set<string>(),    // health checks, auth: never wait behind batch
    private fair = true,                     // false = one shared FIFO queue
  ) {}

  submit(tenant: string, run: () => Promise<void>): void {
    const key = this.fair ? tenant : "fifo";
    if (!this.lanes.has(key)) this.lanes.set(key, []);
    this.lanes.get(key)!.push({ tenant, enqueued: performance.now(), run }); // Step 4b
    this.pump();
  }

  private pick(): Job | undefined {
    if (!this.fair) return this.lanes.get("fifo")?.shift();
    for (const t of this.critical) {                       // Step 5: critical lane first
      const j = this.lanes.get(t)?.shift();
      if (j) return j;
    }
    // Deficit round robin: top up every eligible lane until one can afford a job.
    for (let round = 0; round < 1000; round++) {
      for (const [t, q] of this.lanes) {
        if (this.critical.has(t)) continue;
        const capped = (this.inflight.get(t) ?? 0) >= this.maxConcurrent;
        if (q.length === 0 || capped) { if (q.length === 0) this.deficit.set(t, 0); continue; }
        const d = this.deficit.get(t) ?? 0;
        if (d >= 1) { this.deficit.set(t, d - 1); return q.shift(); }
      }
      let topped = false;
      for (const [t, q] of this.lanes) {
        if (q.length && (this.inflight.get(t) ?? 0) < this.maxConcurrent) {
          this.deficit.set(t, (this.deficit.get(t) ?? 0) + (this.weights[t] ?? 1));
          topped = true;
        }
      }
      if (!topped) return undefined; // everyone is empty or at their concurrency cap
    }
    return undefined;
  }

  private pump(): void {
    while (this.busy < this.workers) {
      const job = this.pick();
      if (!job) return;
      this.busy++;
      this.inflight.set(job.tenant, (this.inflight.get(job.tenant) ?? 0) + 1);
      job.run().finally(() => {                            // Step 6: worker serves it
        this.busy--;
        this.inflight.set(job.tenant, this.inflight.get(job.tenant)! - 1);
        this.pump();
      });
    }
  }
}

async function scenario(fair: boolean) {
  const s = new FairScheduler(4, { A: 1, B: 2 }, 3, new Set(["health"]), fair);
  const waits: Record<string, number[]> = { A: [], B: [], health: [] };
  const done: Promise<void>[] = [];
  const add = (tenant: string, ms: number) =>
    done.push(new Promise<void>((resolve) => {
      const t0 = performance.now();
      s.submit(tenant, async () => {
        waits[tenant]!.push(performance.now() - t0); // queueing delay before a worker picked it
        await sleep(ms);
        resolve();
      });
    }));
  for (let i = 0; i < 120; i++) add("A", 10); // Tenant A batch dumps 120 jobs at once
  for (let i = 0; i < 12; i++) add("B", 10);  // Tenant B: normal interactive load
  add("health", 1);                           // one health check
  await Promise.all(done);
  const p50 = (xs: number[]) => xs.sort((a, b) => a - b)[Math.floor(xs.length / 2)]!.toFixed(0);
  console.log(`${fair ? "fair" : "fifo"}: wait p50 A=${p50(waits.A!)}ms B=${p50(waits.B!)}ms health=${p50(waits.health!)}ms`);
}

await scenario(false);
await scenario(true);