Three bounded request pools for metadata, content, and thumbnails (closes #45)

Adds three independent bounded request pools — metadata (10 in flight), content (5), thumbnails (25), each overridable — with on-demand-before-background priority and in-flight dedup (a shared key runs once); an idle pool never lends slots, and retries run inside a slot. Self-contained module; the content cache wires it into the library later.

Model: opus-4-8

src/library/pools.ts

@@ -0,0 +1,181 @@
+// Three bounded request pools for metadata, content, and thumbnails (issue
+// #45).
+//
+// Ente meters differently by traffic class, so quak keeps three independent
+// pools instead of one global limit: metadata is cheap and chatty, original
+// content is heavy, thumbnails are small but numerous. Each pool is a
+// `BoundedPool` — a plain concurrency limiter — and the three run at the
+// design's caps (10 / 5 / 25) unless the caller overrides them.
+//
+// Two behaviours beyond a bare limiter, both per pool:
+//
+//   - Priority: work waiting for a slot is ordered on-demand before
+//     background/precache, so a slot that frees up serves the request a user is
+//     waiting on ahead of speculative prefetch. Within one priority the order
+//     is first-come-first-served.
+//
+//   - In-flight dedup: a task submitted under a `key` that a still-pending task
+//     already carries is not run a second time; both callers await the one
+//     result. The key is released the moment that task settles — success or
+//     failure — so a later request for the same key runs afresh. Callers key by
+//     the id whose fetch must not be duplicated (a fileID, say).
+//
+// The pool holds a task's slot for that task's entire lifetime. A task that
+// retries internally is doing so inside its slot: the slot is not freed between
+// attempts, which is what keeps a retrying request counted against the cap. The
+// pool knows nothing of the retry policy; it only holds the slot until the
+// task's promise settles.
+//
+// This module is self-contained infrastructure. Routing a given `Client` call
+// to the right pool belongs to the unit that wires the pools into the cache;
+// here there is only the machinery.
+
+export const DEFAULT_METADATA_CONCURRENCY = 10;
+export const DEFAULT_CONTENT_CONCURRENCY = 5;
+export const DEFAULT_THUMBNAIL_CONCURRENCY = 25;
+
+// On-demand work is served before background/precache work waiting in the same
+// pool.
+export type Priority = "on-demand" | "background";
+
+export interface RunOptions {
+    // Defaults to "background": an unmarked request yields to on-demand work.
+    priority?: Priority;
+    // When set, a task already pending under this key is shared instead of run
+    // again. Omit for work that must always execute.
+    key?: string | number;
+}
+
+// One queued submission awaiting a slot. `start` runs the task and holds the
+// slot until it settles.
+interface Waiter {
+    priority: Priority;
+    // Submission order, used to break ties within a priority (FIFO).
+    seq: number;
+    start: () => void;
+}
+
+export class BoundedPool {
+    readonly concurrency: number;
+
+    private active = 0;
+    private nextSeq = 0;
+    private readonly waiting: Waiter[] = [];
+    // Keyed by a caller-supplied dedup key; holds the shared promise for as
+    // long as that task is pending, cleared when it settles.
+    private readonly pending = new Map<string | number, Promise<unknown>>();
+
+    constructor(concurrency: number) {
+        if (!Number.isInteger(concurrency) || concurrency < 1) {
+            throw new RangeError(
+                `concurrency must be a positive integer, got ${concurrency}`,
+            );
+        }
+        this.concurrency = concurrency;
+    }
+
+    // Submit `task` to the pool. It runs once a slot is free, subject to
+    // priority; the returned promise settles with the task's result. With a
+    // `key`, a still-pending submission under the same key is returned instead
+    // of running `task` again.
+    run<T>(task: () => Promise<T>, opts: RunOptions = {}): Promise<T> {
+        const { key } = opts;
+        if (key !== undefined) {
+            const shared = this.pending.get(key);
+            if (shared !== undefined) return shared as Promise<T>;
+        }
+
+        const promise = this.enqueue(task, opts.priority ?? "background");
+
+        if (key !== undefined) {
+            this.pending.set(key, promise);
+            const release = (): void => {
+                // Only clear our own entry: a fresh submission under the same
+                // key after this one settled must not be evicted here.
+                if (this.pending.get(key) === promise) this.pending.delete(key);
+            };
+            promise.then(release, release);
+        }
+        return promise;
+    }
+
+    private enqueue<T>(task: () => Promise<T>, priority: Priority): Promise<T> {
+        return new Promise<T>((resolve, reject) => {
+            const start = (): void => {
+                this.active++;
+                // Hold the slot until the task fully settles — every internal
+                // retry included — then admit the next waiter.
+                void (async () => {
+                    try {
+                        resolve(await task());
+                    } catch (err) {
+                        reject(err);
+                    } finally {
+                        this.active--;
+                        this.pump();
+                    }
+                })();
+            };
+            this.waiting.push({ priority, seq: this.nextSeq++, start });
+            this.pump();
+        });
+    }
+
+    // Admit waiters until the pool is full or the queue is empty.
+    private pump(): void {
+        while (this.active < this.concurrency) {
+            const next = this.takeNext();
+            if (next === undefined) return;
+            next.start();
+        }
+    }
+
+    // Remove and return the highest-priority waiter: on-demand before
+    // background, earliest submission first within a priority.
+    private takeNext(): Waiter | undefined {
+        let bestIndex = -1;
+        let best: Waiter | undefined;
+        for (let i = 0; i < this.waiting.length; i++) {
+            const w = this.waiting[i];
+            if (w === undefined) continue;
+            if (best === undefined || this.precedes(w, best)) {
+                best = w;
+                bestIndex = i;
+            }
+        }
+        if (best === undefined) return undefined;
+        this.waiting.splice(bestIndex, 1);
+        return best;
+    }
+
+    private precedes(a: Waiter, b: Waiter): boolean {
+        if (a.priority !== b.priority) return a.priority === "on-demand";
+        return a.seq < b.seq;
+    }
+}
+
+export interface RequestPoolsOptions {
+    metadataConcurrency?: number;
+    contentConcurrency?: number;
+    thumbnailConcurrency?: number;
+}
+
+// The three pools the design calls for, each independent: an idle pool never
+// lends its slots to a busy one.
+export class RequestPools {
+    readonly metadata: BoundedPool;
+    readonly content: BoundedPool;
+    readonly thumbnails: BoundedPool;
+
+    constructor(opts: RequestPoolsOptions = {}) {
+        this.metadata = new BoundedPool(
+            opts.metadataConcurrency ?? DEFAULT_METADATA_CONCURRENCY,
+        );
+        this.content = new BoundedPool(
+            opts.contentConcurrency ?? DEFAULT_CONTENT_CONCURRENCY,
+        );
+        this.thumbnails = new BoundedPool(
+            opts.thumbnailConcurrency ?? DEFAULT_THUMBNAIL_CONCURRENCY,
+        );
+    }
+}

test/library/pools.test.ts

@@ -0,0 +1,345 @@
+/**
+ * Tests for `src/library/pools.ts` — the three bounded request pools (issue
+ * #45).
+ *
+ * A `BoundedPool` runs submitted tasks with a fixed concurrency cap. Within a
+ * pool, on-demand work runs before background work, and a task submitted with a
+ * key that a still-pending task already carries is not run twice — both callers
+ * share the one result. `RequestPools` bundles the three the design calls for
+ * (metadata 10, content 5, thumbnails 25); the pools are independent, so an
+ * idle pool never lends its slots to a busy one.
+ *
+ * ## How the tasks are controlled
+ *
+ * Every task here is a gate: it reports when it *starts* and then blocks until
+ * the test *releases* it, so the test decides exactly how many run at once and
+ * in what order they finish. A shared tracker counts how many tasks are running
+ * at any instant and records the peak, which is what the concurrency assertions
+ * read. No assertion is about wall-clock time.
+ *
+ * `drain()` returns a promise that settles on a macrotask, which flushes the
+ * microtask queue the pool schedules its starts on; the tests await it to let
+ * the pool react to a submission or a release before they inspect it.
+ */
+
+import { describe, it, expect } from "vitest";
+
+import {
+    BoundedPool,
+    RequestPools,
+    DEFAULT_METADATA_CONCURRENCY,
+    DEFAULT_CONTENT_CONCURRENCY,
+    DEFAULT_THUMBNAIL_CONCURRENCY,
+} from "../../src/library/pools.js";
+
+// Settle on a macrotask so every microtask the pool queued has run.
+const drain = (): Promise<void> =>
+    new Promise((resolve) => setTimeout(resolve, 0));
+
+// A controllable task. `task` blocks until `release()` (resolve) or `fail()`
+// (reject) is called; `startOrder` records the sequence in which tasks began.
+interface Gate<T> {
+    task: () => Promise<T>;
+    release: (value: T) => void;
+    fail: (err: unknown) => void;
+    started: () => boolean;
+    runs: () => number;
+}
+
+// Tracks how many gated tasks are running concurrently across a whole test.
+class Tracker {
+    active = 0;
+    peak = 0;
+    readonly starts: string[] = [];
+
+    gate<T>(label = ""): Gate<T> {
+        let settleResolve!: (value: T) => void;
+        let settleReject!: (err: unknown) => void;
+        const settled = new Promise<T>((resolve, reject) => {
+            settleResolve = resolve;
+            settleReject = reject;
+        });
+        let started = false;
+        let runs = 0;
+        const task = async (): Promise<T> => {
+            started = true;
+            runs++;
+            this.active++;
+            this.peak = Math.max(this.peak, this.active);
+            this.starts.push(label);
+            try {
+                return await settled;
+            } finally {
+                this.active--;
+            }
+        };
+        return {
+            task,
+            release: (value: T) => settleResolve(value),
+            fail: (err: unknown) => settleReject(err),
+            started: () => started,
+            runs: () => runs,
+        };
+    }
+}
+
+describe("BoundedPool concurrency cap", () => {
+    it("never runs more than `concurrency` tasks at once", async () => {
+        const pool = new BoundedPool(3);
+        const t = new Tracker();
+        const gates = Array.from({ length: 5 }, () => t.gate<void>());
+
+        const done = gates.map((g) => pool.run(g.task));
+        await drain();
+
+        // Three started, two queued behind the cap.
+        expect(t.active).toBe(3);
+        expect(gates.slice(0, 3).every((g) => g.started())).toBe(true);
+        expect(gates.slice(3).some((g) => g.started())).toBe(false);
+
+        // Finishing one admits exactly one more; the cap holds.
+        gates[0]!.release();
+        await drain();
+        expect(t.active).toBe(3);
+        expect(gates[3]!.started()).toBe(true);
+        expect(gates[4]!.started()).toBe(false);
+
+        for (const g of gates.slice(1)) g.release();
+        await Promise.all(done);
+        expect(t.peak).toBe(3);
+    });
+
+    it("rejects a non-positive or non-integer concurrency", () => {
+        expect(() => new BoundedPool(0)).toThrow(RangeError);
+        expect(() => new BoundedPool(-1)).toThrow(RangeError);
+        expect(() => new BoundedPool(2.5)).toThrow(RangeError);
+    });
+});
+
+describe("BoundedPool priority ordering", () => {
+    it("runs on-demand work before background, FIFO within a priority", async () => {
+        const pool = new BoundedPool(1);
+        const t = new Tracker();
+        const a = t.gate<void>("a");
+        const b = t.gate<void>("b");
+        const c = t.gate<void>("c");
+        const d = t.gate<void>("d");
+
+        // `a` takes the only slot; the rest queue.
+        void pool.run(a.task, { priority: "background" });
+        await drain();
+        void pool.run(b.task, { priority: "background" });
+        void pool.run(c.task, { priority: "on-demand" });
+        void pool.run(d.task, { priority: "background" });
+        await drain();
+        expect(t.starts).toEqual(["a"]);
+
+        // The on-demand `c` jumps ahead of the earlier-queued background `b`.
+        a.release();
+        await drain();
+        expect(t.starts).toEqual(["a", "c"]);
+
+        // Then background work drains in submission order: `b` before `d`.
+        c.release();
+        await drain();
+        expect(t.starts).toEqual(["a", "c", "b"]);
+        b.release();
+        await drain();
+        expect(t.starts).toEqual(["a", "c", "b", "d"]);
+        d.release();
+    });
+
+    it("defaults to background priority", async () => {
+        const pool = new BoundedPool(1);
+        const t = new Tracker();
+        const a = t.gate<void>("a");
+        const plain = t.gate<void>("plain");
+        const urgent = t.gate<void>("urgent");
+
+        void pool.run(a.task);
+        await drain();
+        void pool.run(plain.task); // no options -> background
+        void pool.run(urgent.task, { priority: "on-demand" });
+        await drain();
+
+        a.release();
+        await drain();
+        expect(t.starts).toEqual(["a", "urgent"]);
+        urgent.release();
+        plain.release();
+    });
+});
+
+describe("BoundedPool in-flight dedup", () => {
+    it("fetches a key once and hands both callers the same result", async () => {
+        const pool = new BoundedPool(5);
+        const t = new Tracker();
+        const g = t.gate<number>();
+
+        const first = pool.run(g.task, { key: 7 });
+        const second = pool.run(g.task, { key: 7 });
+        await drain();
+
+        expect(g.runs()).toBe(1);
+        expect(first).toBe(second);
+
+        g.release(99);
+        expect(await first).toBe(99);
+        expect(await second).toBe(99);
+    });
+
+    it("dedups only while in flight; a settled key runs again", async () => {
+        const pool = new BoundedPool(5);
+        const t = new Tracker();
+        const g1 = t.gate<number>();
+
+        const first = pool.run(g1.task, { key: 7 });
+        await drain();
+        g1.release(1);
+        expect(await first).toBe(1);
+
+        // The key is free again once its task settled.
+        const g2 = t.gate<number>();
+        const third = pool.run(g2.task, { key: 7 });
+        await drain();
+        expect(g2.started()).toBe(true);
+        g2.release(2);
+        expect(await third).toBe(2);
+    });
+
+    it("propagates a rejection to every deduped caller", async () => {
+        const pool = new BoundedPool(5);
+        const t = new Tracker();
+        const g = t.gate<number>();
+
+        const first = pool.run(g.task, { key: 7 });
+        const second = pool.run(g.task, { key: 7 });
+        await drain();
+
+        const boom = new Error("boom");
+        g.fail(boom);
+        await expect(first).rejects.toBe(boom);
+        await expect(second).rejects.toBe(boom);
+
+        // A failed key is also freed, so it may be retried by a fresh submit.
+        const g2 = t.gate<number>();
+        const retry = pool.run(g2.task, { key: 7 });
+        await drain();
+        expect(g2.started()).toBe(true);
+        g2.release(5);
+        expect(await retry).toBe(5);
+    });
+});
+
+describe("BoundedPool slot lifetime", () => {
+    it("holds one slot for a task's whole lifetime, retries included", async () => {
+        const pool = new BoundedPool(1);
+        const t = new Tracker();
+
+        // A task that internally makes two attempts before succeeding — the
+        // shape of a retrying request. It must occupy exactly one slot for the
+        // whole of that, so no other task may start until it finally settles.
+        const attempt1 = t.gate<void>("attempt1");
+        const attempt2 = t.gate<void>("attempt2");
+        const retrying = async (): Promise<void> => {
+            try {
+                await attempt1.task();
+            } catch {
+                await attempt2.task();
+            }
+        };
+        const other = t.gate<void>("other");
+
+        const running = pool.run(retrying);
+        await drain();
+        void pool.run(other.task);
+        await drain();
+
+        // First attempt is in flight and holds the only slot.
+        expect(t.starts).toEqual(["attempt1"]);
+        expect(other.started()).toBe(false);
+
+        // The retry is still the same task in the same slot; `other` waits.
+        attempt1.fail(new Error("transient"));
+        await drain();
+        expect(t.starts).toEqual(["attempt1", "attempt2"]);
+        expect(other.started()).toBe(false);
+
+        // Only when the whole task settles does the slot free.
+        attempt2.release();
+        await running;
+        await drain();
+        expect(other.started()).toBe(true);
+        other.release();
+    });
+});
+
+describe("RequestPools", () => {
+    it("exposes three pools at the design's default caps", () => {
+        expect(DEFAULT_METADATA_CONCURRENCY).toBe(10);
+        expect(DEFAULT_CONTENT_CONCURRENCY).toBe(5);
+        expect(DEFAULT_THUMBNAIL_CONCURRENCY).toBe(25);
+
+        const pools = new RequestPools();
+        expect(pools.metadata.concurrency).toBe(10);
+        expect(pools.content.concurrency).toBe(5);
+        expect(pools.thumbnails.concurrency).toBe(25);
+    });
+
+    it("takes overridden caps", () => {
+        const pools = new RequestPools({
+            metadataConcurrency: 1,
+            contentConcurrency: 2,
+            thumbnailConcurrency: 3,
+        });
+        expect(pools.metadata.concurrency).toBe(1);
+        expect(pools.content.concurrency).toBe(2);
+        expect(pools.thumbnails.concurrency).toBe(3);
+    });
+
+    it("keeps pools independent: an idle pool lends no slots", async () => {
+        const pools = new RequestPools({ contentConcurrency: 1 });
+        const t = new Tracker();
+        const c1 = t.gate<void>();
+        const c2 = t.gate<void>();
+        const c3 = t.gate<void>();
+
+        // The content pool is capped at 1. The thumbnail pool sits idle with 25
+        // free slots — none of which may be borrowed to run a second content
+        // task.
+        void pools.content.run(c1.task);
+        void pools.content.run(c2.task);
+        void pools.content.run(c3.task);
+        await drain();
+        expect(t.active).toBe(1);
+
+        c1.release();
+        await drain();
+        expect(t.active).toBe(1);
+        c2.release();
+        await drain();
+        expect(t.active).toBe(1);
+        c3.release();
+        await drain();
+        expect(t.peak).toBe(1);
+    });
+
+    it("runs different pools concurrently", async () => {
+        const pools = new RequestPools({
+            metadataConcurrency: 1,
+            contentConcurrency: 1,
+        });
+        const t = new Tracker();
+        const m = t.gate<void>();
+        const c = t.gate<void>();
+
+        void pools.metadata.run(m.task);
+        void pools.content.run(c.task);
+        await drain();
+
+        // One slot each, in two independent pools: both run at once.
+        expect(t.active).toBe(2);
+        m.release();
+        c.release();
+    });
+});