clawbot
000d395c87
check / check (push) Successful in 55s
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
346 lines
12 KiB
TypeScript
346 lines
12 KiB
TypeScript
/**
|
|
* 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();
|
|
});
|
|
});
|