Commita33c82a
perf(dashboards): stop serialising per-row enrichment on /issues and /pulls
perf(dashboards): stop serialising per-row enrichment on /issues and /pulls
Both dashboards enriched their candidate list with
`for (const c of candidates) { await enrich(c) }`. enrichIssue issues 2
queries per issue and the candidate set is capped at 300, so /issues made up
to 600 SEQUENTIAL round-trips to Neon on every page load; /pulls the same
shape at up to 300. Page latency was N × RTT and dominated by network waits,
which is why the nav-audit clocks both around 5s.
Same queries, same results — just no longer serialised. mapWithConcurrency
keeps at most DB_FANOUT_LIMIT (12) in flight, so a 300-item serial walk
collapses to ~25 waves. A plain Promise.all would have fixed the wall-clock
while firing all 300 at once, which on a pooled managed Postgres relocates
the queue into the connection pool or exhausts it.
Results are returned in input order regardless of completion order, because
both callers now zip the enrichment back against `candidates` by index — the
tests pin that down with deliberately inverted durations. The limit is
clamped to [1, items.length], so a zero limit degrades to serial instead of
spawning no workers and hanging.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>4 files changed+171−4a33c82a874dd6d91c4ee4dfe9de7f4cf26f39db2
4 changed files+171−4
Addedsrc/__tests__/concurrency.test.ts+96−0View fileUnifiedSplit
@@ -0,0 +1,96 @@
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
15import { describe, expect, it } from "bun:test";
16import { mapWithConcurrency, DB_FANOUT_LIMIT } from "../lib/concurrency";
17
18const tick = () => new Promise((r) => setTimeout(r, 1));
19
20describe("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
45describe("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
91describe("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});
Addedsrc/lib/concurrency.ts+46−0View fileUnifiedSplit
@@ -0,0 +1,46 @@
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 */
17export 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 */
46export const DB_FANOUT_LIMIT = 12;
Modifiedsrc/routes/issues-dashboard.tsx+16−2View fileUnifiedSplit
@@ -25,6 +25,7 @@
2525 */
2626
2727import { Hono } from "hono";
28import { mapWithConcurrency, DB_FANOUT_LIMIT } from "../lib/concurrency";
2829import {
2930 and,
3031 desc,
@@ -569,9 +570,22 @@ issuesDashboard.get("/issues", requireAuth, async (c) => {
569570 // Enrich + filter to the active tab. The two AI-shaped filters
570571 // (ai-working, needs-triage) only make sense after we know the
571572 // per-issue label set, so the SQL stays generic.
573 // enrichIssue issues 2 queries per issue and `candidates` holds up to 300,
574 // so awaiting inside a for-loop meant up to 600 SEQUENTIAL round-trips to
575 // Neon on every page load — the dominant cost of this route. Same queries,
576 // same results, just no longer serialised; the cap keeps us well inside the
577 // connection pool. Order is preserved, so the sort applied upstream still
578 // holds.
579 const enrichedAll = await mapWithConcurrency(
580 candidates,
581 DB_FANOUT_LIMIT,
582 (cand) => enrichIssue(cand.issue.id)
583 );
584
572585 const rows: IssueRow[] = [];
573 for (const cand of candidates) {
574 const enriched = await enrichIssue(cand.issue.id);
586 for (let i = 0; i < candidates.length; i++) {
587 const cand = candidates[i];
588 const enriched = enrichedAll[i];
575589 const matches =
576590 filter === "mine" ||
577591 filter === "assigned" ||
Modifiedsrc/routes/pulls-dashboard.tsx+13−2View fileUnifiedSplit
@@ -18,6 +18,7 @@
1818 */
1919
2020import { Hono } from "hono";
21import { mapWithConcurrency, DB_FANOUT_LIMIT } from "../lib/concurrency";
2122import { eq, and, desc, inArray, or, isNotNull } from "drizzle-orm";
2223import { db } from "../db";
2324import {
@@ -540,9 +541,19 @@ pullsDashboard.get("/pulls", requireAuth, async (c) => {
540541 }
541542
542543 // Enrich + filter to the active tab.
544 // Same fix as /issues: enrichPr round-trips per PR and `candidates` holds
545 // up to 300, so awaiting in a for-loop serialised the whole page load.
546 // Bounded fan-out, input order preserved.
547 const enrichedAll = await mapWithConcurrency(
548 candidates,
549 DB_FANOUT_LIMIT,
550 (cand) => enrichPr(cand.pr)
551 );
552
543553 const rows: PrRow[] = [];
544 for (const cand of candidates) {
545 const enriched = await enrichPr(cand.pr);
554 for (let i = 0; i < candidates.length; i++) {
555 const cand = candidates[i];
556 const enriched = enrichedAll[i];
546557 const matches =
547558 filter === "mine" ||
548559 filter === "reviewing" ||
549560