From 3c96aa87a7156b1eafc02c1239ed7864ddcb42bf Mon Sep 17 00:00:00 2001 From: sneak Date: Tue, 22 Sep 2026 13:19:31 +0000 Subject: [PATCH] Three bounded request pools for metadata, content, and thumbnails (closes #45) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A self-contained `src/library/pools.ts`: a generic `BoundedPool` concurrency limiter and a `RequestPools` bundle of three at the design's caps (metadata 10, content 5, thumbnails 25, each overridable). Within a pool, on-demand work is served before background/precache work, first-come-first-served within a priority; a task submitted under a `key` already pending is shared rather than run twice, and the key frees the moment that task settles. The slot is held for the task's whole lifetime, so a request that retries internally stays counted against the cap between attempts. Pools only; nothing wires them into `Library` yet — that is a later unit. Model: opus-4-8 --- src/library/pools.ts | 181 +++++++++++++++++++ test/library/pools.test.ts | 345 +++++++++++++++++++++++++++++++++++++ 2 files changed, 526 insertions(+) create mode 100644 src/library/pools.ts create mode 100644 test/library/pools.test.ts diff --git a/src/library/pools.ts b/src/library/pools.ts new file mode 100644 index 0000000..e1a172d --- /dev/null +++ b/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>(); + + 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(task: () => Promise, opts: RunOptions = {}): Promise { + const { key } = opts; + if (key !== undefined) { + const shared = this.pending.get(key); + if (shared !== undefined) return shared as Promise; + } + + 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(task: () => Promise, priority: Priority): Promise { + return new Promise((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, + ); + } +} diff --git a/test/library/pools.test.ts b/test/library/pools.test.ts new file mode 100644 index 0000000..9fed142 --- /dev/null +++ b/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 => + 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(); + }); +});