Blame · Line-by-line history

`concurrency.test.ts`

Each line is annotated with the commit that last touched it. Click any SHA to jump to that commit and see the surrounding change.

concurrency.test.tsBlame96 lines · 1 contributor

Normal view Raw

a33c82accantynz-alt	1	/**
	2	* mapWithConcurrency — bounded async fan-out.
	3	*
	4	* The /issues and /pulls dashboards enriched their candidate lists with
	5	* `for (const c of candidates) { await enrich(c) }`, serialising one DB
	6	* round-trip per item: up to 600 sequential queries per /issues page load
	7	* (2 per issue × 300 candidates). Latency was N × RTT.
	8	*
	9	* A plain Promise.all would fix wall-clock but fire all N at once, which on a
	10	* pooled managed Postgres just relocates the queue into the connection pool.
	11	* These pin down both properties that matter: order preservation (the rows
	12	* are zipped back against `candidates` by index) and the concurrency cap.
	13	*/
	14
	15	import { describe, expect, it } from "bun:test";
	16	import { mapWithConcurrency, DB_FANOUT_LIMIT } from "../lib/concurrency";
	17
	18	const tick = () => new Promise((r) => setTimeout(r, 1));
	19
	20	describe("order preservation", () => {
	21	it("returns results in INPUT order, not completion order", async () => {
	22	// Deliberately invert the durations so completion order is the reverse
	23	// of input order. Zipping by index in the callers depends on this.
	24	const items = [1, 2, 3, 4, 5];
	25	const out = await mapWithConcurrency(items, 5, async (n) => {
	26	await new Promise((r) => setTimeout(r, (6 - n) * 5));
	27	return n * 10;
	28	});
	29	expect(out).toEqual([10, 20, 30, 40, 50]);
	30	});
	31
	32	it("passes the index through", async () => {
	33	const out = await mapWithConcurrency(["a", "b", "c"], 2, async (v, i) => `${i}:${v}`);
	34	expect(out).toEqual(["0:a", "1:b", "2:c"]);
	35	});
	36
	37	it("handles an empty list without spawning workers", async () => {
	38	let called = 0;
	39	const out = await mapWithConcurrency([], 8, async () => { called++; return 1; });
	40	expect(out).toEqual([]);
	41	expect(called).toBe(0);
	42	});
	43	});
	44
	45	describe("concurrency is actually bounded", () => {
	46	it("never exceeds the limit in flight", async () => {
	47	let inFlight = 0;
	48	let peak = 0;
	49	await mapWithConcurrency(Array.from({ length: 50 }, (_, i) => i), 5, async () => {
	50	inFlight++;
	51	peak = Math.max(peak, inFlight);
	52	await tick();
	53	inFlight--;
	54	return null;
	55	});
	56	expect(peak).toBeLessThanOrEqual(5);
	57	expect(peak).toBeGreaterThan(1); // ...but it IS parallel, not serial
	58	});
	59
	60	it("clamps the limit to the item count", async () => {
	61	let peak = 0;
	62	let inFlight = 0;
	63	await mapWithConcurrency([1, 2], 100, async () => {
	64	inFlight++;
	65	peak = Math.max(peak, inFlight);
	66	await tick();
	67	inFlight--;
	68	return null;
	69	});
	70	expect(peak).toBeLessThanOrEqual(2);
	71	});
	72
	73	it("treats a zero or negative limit as serial rather than deadlocking", async () => {
	74	// A limit of 0 would spawn no workers and hang forever if unclamped.
	75	const out = await mapWithConcurrency([1, 2, 3], 0, async (n) => n);
	76	expect(out).toEqual([1, 2, 3]);
	77	});
	78
	79	it("processes every item exactly once", async () => {
	80	const seen: number[] = [];
	81	await mapWithConcurrency(Array.from({ length: 30 }, (_, i) => i), 7, async (n) => {
	82	await tick();
	83	seen.push(n);
	84	return n;
	85	});
	86	expect(seen.length).toBe(30);
	87	expect(new Set(seen).size).toBe(30);
	88	});
	89	});
	90
	91	describe("the shared limit", () => {
	92	it("is conservative enough for a pooled managed Postgres", () => {
	93	expect(DB_FANOUT_LIMIT).toBeGreaterThan(1);
	94	expect(DB_FANOUT_LIMIT).toBeLessThanOrEqual(20);
	95	});
	96	});