Request Concurrency Limiting in Browsers

// scheduler.ts — priority queue + semaphore + abort support.
type Priority = "user" | "background";

interface Waiter {
  resolve: () => void;
  reject: (e: Error) => void;
  priority: Priority;
  enqueuedAt: number;
  signal?: AbortSignal;
}

export class ConcurrencyLimiter {
  private inFlight = 0;
  private queue: Waiter[] = [];

  constructor(private max = 6, private maxQueue = 64, private maxWaitMs = 20_000) {}

  private next() {
    // User-initiated work jumps ahead of prefetches; FIFO within a priority.
    const idx = this.queue.findIndex((w) => w.priority === "user");
    const pick = idx >= 0 ? idx : 0;
    const w = this.queue.splice(pick, 1)[0];
    if (!w) return;
    if (w.signal?.aborted) { w.reject(new DOMException("Aborted", "AbortError")); return this.next(); }
    if (Date.now() - w.enqueuedAt > this.maxWaitMs) {
      w.reject(new Error("queue_wait_exceeded"));    // stale: the user has moved on
      return this.next();
    }
    this.inFlight += 1;
    w.resolve();
  }

  async run<T>(fn: () => Promise<T>, opts: { priority?: Priority; signal?: AbortSignal } = {}): Promise<T> {
    if (this.queue.length >= this.maxQueue) throw new Error("client_queue_full");

    if (this.inFlight >= this.max) {
      await new Promise<void>((resolve, reject) => {
        const w: Waiter = { resolve, reject, priority: opts.priority ?? "background",
                            enqueuedAt: Date.now(), signal: opts.signal };
        this.queue.push(w);
        // An abort while queued must free the queue slot immediately.
        opts.signal?.addEventListener("abort", () => {
          const i = this.queue.indexOf(w);
          if (i >= 0) { this.queue.splice(i, 1); reject(new DOMException("Aborted", "AbortError")); }
        }, { once: true });
      });
    } else {
      this.inFlight += 1;
    }

    try {
      return await fn();
    } finally {
      // Release in finally: a thrown request must not leak a slot forever.
      this.inFlight -= 1;
      this.next();
    }
  }

  get state() { return { inFlight: this.inFlight, queued: this.queue.length }; }
}

// transport.ts — rate first, then concurrency: both controls, one call path.
const bucket = new TokenBucket({ ratePerSec: 1.5, capacity: 18 });
const slots = new ConcurrencyLimiter(6);

export async function apiFetch(url: string, init: RequestInit & { priority?: "user" | "background" } = {}) {
  await bucket.acquire();                       // may we send at all, yet?
  return slots.run(() => fetch(url, init), {    // and is there room in flight?
    priority: init.priority ?? "background",
    signal: init.signal ?? undefined,
  });
}

// concurrency.test.ts — assert the cap, the ordering, and the release path.
import { describe, it, expect, vi } from "vitest";
import { ConcurrencyLimiter } from "./scheduler";

const deferred = () => { let r: () => void; const p = new Promise<void>((res) => (r = res as any)); return { p, r: r! }; };

it("never exceeds the configured cap", async () => {
  const lim = new ConcurrencyLimiter(3);
  let peak = 0, active = 0;
  const gates = Array.from({ length: 10 }, deferred);
  const runs = gates.map((g) => lim.run(async () => {
    active += 1; peak = Math.max(peak, active); await g.p; active -= 1;
  }));
  await Promise.resolve();
  expect(peak).toBe(3);
  gates.forEach((g) => g.r());
  await Promise.all(runs);
});

it("releases the slot when the task throws", async () => {
  const lim = new ConcurrencyLimiter(1);
  await expect(lim.run(async () => { throw new Error("boom"); })).rejects.toThrow("boom");
  await expect(lim.run(async () => "ok")).resolves.toBe("ok");   // slot was freed
});

it("serves user-priority work before background work", async () => {
  const lim = new ConcurrencyLimiter(1);
  const order: string[] = [];
  const gate = deferred();
  const blocking = lim.run(async () => { await gate.p; order.push("blocking"); });
  const bg = lim.run(async () => { order.push("background"); }, { priority: "background" });
  const user = lim.run(async () => { order.push("user"); }, { priority: "user" });
  gate.r();
  await Promise.all([blocking, bg, user]);
  expect(order).toEqual(["blocking", "user", "background"]);
});

it("frees a queued slot when the caller aborts", async () => {
  const lim = new ConcurrencyLimiter(1);
  const gate = deferred();
  const blocking = lim.run(async () => { await gate.p; });
  const ac = new AbortController();
  const queued = lim.run(async () => "never", { signal: ac.signal });
  ac.abort();
  await expect(queued).rejects.toThrow(/Abort/);
  expect(lim.state.queued).toBe(0);
  gate.r(); await blocking;
});