/** * 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 => 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 { task: () => Promise; 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(label = ""): Gate { let settleResolve!: (value: T) => void; let settleReject!: (err: unknown) => void; const settled = new Promise((resolve, reject) => { settleResolve = resolve; settleReject = reject; }); let started = false; let runs = 0; const task = async (): Promise => { 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()); 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("a"); const b = t.gate("b"); const c = t.gate("c"); const d = t.gate("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("a"); const plain = t.gate("plain"); const urgent = t.gate("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(); 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(); 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(); 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(); 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(); 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("attempt1"); const attempt2 = t.gate("attempt2"); const retrying = async (): Promise => { try { await attempt1.task(); } catch { await attempt2.task(); } }; const other = t.gate("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(); const c2 = t.gate(); const c3 = t.gate(); // 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(); const c = t.gate(); 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(); }); });