Blame · Line-by-line history

`concurrency.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.tsBlame46 lines · 1 contributor

Normal view Raw

a33c82accantynz-alt	1	/**
	2	* Bounded-concurrency async map.
	3	*
	4	* The dashboards enriched their candidate lists with a plain
	5	* `for (const c of candidates) { await enrich(c) }`, which serialises one
	6	* round-trip per item — /issues issued up to 600 sequential queries to Neon
	7	* per page load (2 per issue × 300 candidates) and /pulls up to 300. Latency
	8	* was therefore N × RTT rather than N/limit × RTT.
	9	*
	10	* `Promise.all(items.map(...))` would fix the wall-clock but fire all N at
	11	* once, which on a pooled managed Postgres just moves the queue into the
	12	* connection pool (or exhausts it and starts erroring). This keeps at most
	13	* `limit` in flight.
	14	*
	15	* Results are returned in input order regardless of completion order.
	16	*/
	17	export async function mapWithConcurrency<T, R>(
	18	items: readonly T[],
	19	limit: number,
	20	fn: (item: T, index: number) => Promise<R>
	21	): Promise<R[]> {
	22	if (items.length === 0) return [];
	23
	24	const effective = Math.max(1, Math.min(limit, items.length));
	25	const results = new Array<R>(items.length);
	26	let next = 0;
	27
	28	async function worker(): Promise<void> {
	29	for (;;) {
	30	const i = next++;
	31	if (i >= items.length) return;
	32	results[i] = await fn(items[i], i);
	33	}
	34	}
	35
	36	await Promise.all(Array.from({ length: effective }, () => worker()));
	37	return results;
	38	}
	39
	40	/**
	41	* Default in-flight cap for per-row enrichment against the primary database.
	42	*
	43	* Chosen to be comfortably below a managed-Postgres pool size while still
	44	* collapsing a 300-item serial walk into ~25 sequential waves.
	45	*/
	46	export const DB_FANOUT_LIMIT = 12;