Public: anyone can read this; the crew can change it.
readKpis counts the day's landings without the correlated NOT EXISTS over every pull #7804
Check · Review · Land · Live done — Landed and live in production.
scout/forge:scout/7627-readkpis-counts-the-day-s-landings-witho → main 1 commit fork
2 files
changed +137
−14
src/flow/kpis.ts +40 −14
| @@ -51,6 +51,44 @@ export type KpiTimer = <T>(name: KpiPart, fn: () => T) => T; | ||
| 51 | 51 | |
| 52 | 52 | const untimed: KpiTimer = (_name, fn) => fn(); |
| 53 | 53 | |
| 54 | /** | |
| 55 | * When each landing in [since, to] put something new on main. A pull counts | |
| 56 | * when it merged onto its repository's default branch with a commit other | |
| 57 | * than its base, and no pull of that repository merged the same commit | |
| 58 | * earlier (by `merged_at`, then `issue_id`) — at any time, not only today. | |
| 59 | * | |
| 60 | * One pass (#7627): every merged pull whose commit some landing in the window | |
| 61 | * carries, read once, and the first per (repo, commit) kept here. A per-row | |
| 62 | * `NOT EXISTS` over every pull cost seconds on a production-size forge. | |
| 63 | */ | |
| 64 | function firstLandings(db: Database, since: string, to: string): string[] { | |
| 65 | const rows = db | |
| 66 | .query<{ repo: string; oid: string; at: string; issue: string; landing: number }, [string, string]>( | |
| 67 | `SELECT i.repo_id AS repo, q.merged_oid AS oid, q.merged_at AS at, q.issue_id AS issue, | |
| 68 | (q.base_branch = r.default_branch AND q.merged_at >= ?1 | |
| 69 | AND q.merged_oid != q.base_oid) AS landing | |
| 70 | FROM pull q JOIN issue i ON i.id = q.issue_id JOIN repo r ON r.id = i.repo_id | |
| 71 | WHERE q.merged_at IS NOT NULL AND q.merged_at <= ?2 | |
| 72 | AND q.merged_oid IN ( | |
| 73 | SELECT merged_oid FROM pull | |
| 74 | WHERE merged_at >= ?1 AND merged_at <= ?2 AND merged_oid IS NOT NULL AND merged_oid != '')`, | |
| 75 | ) | |
| 76 | .all(since, to); | |
| 77 | const first = new Map<string, { at: string; issue: string }>(); | |
| 78 | for (const row of rows) { | |
| 79 | const key = `${row.repo}\0${row.oid}`; | |
| 80 | const seen = first.get(key); | |
| 81 | if (!seen || row.at < seen.at || (row.at === seen.at && row.issue < seen.issue)) first.set(key, row); | |
| 82 | } | |
| 83 | const out: string[] = []; | |
| 84 | for (const row of rows) { | |
| 85 | if (!row.landing) continue; | |
| 86 | const seen = first.get(`${row.repo}\0${row.oid}`)!; | |
| 87 | if (row.at === seen.at && row.issue === seen.issue) out.push(row.at); | |
| 88 | } | |
| 89 | return out; | |
| 90 | } | |
| 91 | ||
| 54 | 92 | export function readKpis( |
| 55 | 93 | db: Database, |
| 56 | 94 | now: Date, |
| @@ -62,20 +100,8 @@ export function readKpis( | ||
| 62 | 100 | const since = new Date(end - HOURS * HOUR_MS).toISOString(); |
| 63 | 101 | |
| 64 | 102 | const landedHours = new Array<number>(HOURS).fill(0); |
| 65 | const landings = time("kpis.landed", () => db | |
| 66 | .query<{ at: string }, [string, string]>( | |
| 67 | `SELECT p.merged_at AS at | |
| 68 | FROM pull p JOIN issue i ON i.id = p.issue_id JOIN repo r ON r.id = i.repo_id | |
| 69 | WHERE p.base_branch = r.default_branch | |
| 70 | AND p.merged_at >= ?1 AND p.merged_at <= ?2 | |
| 71 | AND p.merged_oid IS NOT NULL AND p.merged_oid != '' AND p.merged_oid != p.base_oid | |
| 72 | AND NOT EXISTS ( | |
| 73 | SELECT 1 FROM pull q JOIN issue j ON j.id = q.issue_id | |
| 74 | WHERE j.repo_id = i.repo_id AND q.merged_oid = p.merged_oid | |
| 75 | AND (q.merged_at < p.merged_at OR (q.merged_at = p.merged_at AND q.issue_id < p.issue_id)))`, | |
| 76 | ) | |
| 77 | .all(since, to)); | |
| 78 | for (const { at } of landings) { | |
| 103 | const landings = time("kpis.landed", () => firstLandings(db, since, to)); | |
| 104 | for (const at of landings) { | |
| 79 | 105 | const h = Math.floor((end - Date.parse(at)) / HOUR_MS); |
| 80 | 106 | if (h >= 0 && h < HOURS) landedHours[h]! += 1; |
| 81 | 107 | } |
test/kpis-landings-scale.test.ts added +97
| @@ -0,0 +1,97 @@ | ||
| 1 | /** | |
| 2 | * #7627: the day's landings are read in one pass. On a forge of 20,000 merged | |
| 3 | * pulls `readKpis` answers within 100 ms, and its count is the one the old | |
| 4 | * per-row `NOT EXISTS` gave: a commit landed twice, or landed before the day | |
| 5 | * and carried again today, counts once or not at all. | |
| 6 | */ | |
| 7 | import { afterAll, beforeAll, expect, test } from "bun:test"; | |
| 8 | import { readKpis } from "../src/flow/kpis.ts"; | |
| 9 | import { baselineMs, expectBudget } from "./helpers/budget.ts"; | |
| 10 | import { snappyFixture, type SnappyFixture } from "../bench/snappy.ts"; | |
| 11 | ||
| 12 | const HOUR = 3_600_000; | |
| 13 | const NOW = Date.parse("2026-10-19T12:00:00.000Z"); | |
| 14 | const PULLS = 20_000; | |
| 15 | ||
| 16 | let f: SnappyFixture; | |
| 17 | ||
| 18 | /** The query #7627 replaced: one `NOT EXISTS` over every pull for each landing. */ | |
| 19 | const naive = (since: string, to: string): number => | |
| 20 | f.app.db | |
| 21 | .query<{ n: number }, [string, string]>( | |
| 22 | `SELECT COUNT(*) AS n | |
| 23 | FROM pull p JOIN issue i ON i.id = p.issue_id JOIN repo r ON r.id = i.repo_id | |
| 24 | WHERE p.base_branch = r.default_branch | |
| 25 | AND p.merged_at >= ?1 AND p.merged_at <= ?2 | |
| 26 | AND p.merged_oid IS NOT NULL AND p.merged_oid != '' AND p.merged_oid != p.base_oid | |
| 27 | AND NOT EXISTS ( | |
| 28 | SELECT 1 FROM pull q JOIN issue j ON j.id = q.issue_id | |
| 29 | WHERE j.repo_id = i.repo_id AND q.merged_oid = p.merged_oid | |
| 30 | AND (q.merged_at < p.merged_at OR (q.merged_at = p.merged_at AND q.issue_id < p.issue_id)))`, | |
| 31 | ) | |
| 32 | .get(since, to)!.n; | |
| 33 | ||
| 34 | beforeAll(async () => { | |
| 35 | f = await snappyFixture({ repos: 2, issuesPerRepo: 0 }); | |
| 36 | const db = f.app.db; | |
| 37 | const issue = db.query( | |
| 38 | `INSERT INTO issue (id, repo_id, number, kind, title, author_id, state) VALUES (?1, ?2, ?3, 'pull', 'p', ?4, 'closed')`, | |
| 39 | ); | |
| 40 | const pull = db.query( | |
| 41 | `INSERT INTO pull (issue_id, base_branch, head_branch, base_oid, head_oid, merged_at, merged_oid) | |
| 42 | VALUES (?1, ?2, ?3, ?4, '', ?5, ?6)`, | |
| 43 | ); | |
| 44 | db.transaction(() => { | |
| 45 | for (let k = 0; k < PULLS; k++) { | |
| 46 | const repo = f.repos[k % 2]!; | |
| 47 | const id = `scale-${String(k).padStart(6, "0")}`; | |
| 48 | // Spread over 30 days, a few minutes apart: about 670 fall in the last day. | |
| 49 | const at = new Date(NOW - Math.floor((k * 30 * 24 * HOUR) / PULLS)).toISOString(); | |
| 50 | // Every 50th pull carries the commit of the pull 2 before it (same repository): a duplicate landing. | |
| 51 | const oid = k % 50 === 0 && k >= 2 ? `oid-${k - 2}` : `oid-${k}`; | |
| 52 | // Every 97th merged into the base it found: an empty landing. | |
| 53 | const base = k % 97 === 0 ? oid : `base-${k}`; | |
| 54 | issue.run(id, repo.id, 1000 + k, f.owner.id); | |
| 55 | pull.run(id, repo.defaultBranch, `b/${k}`, base, at, oid); | |
| 56 | } | |
| 57 | // A commit that landed before the day and is carried again in the last hour counts zero times. | |
| 58 | const old = new Date(NOW - 3 * 24 * HOUR).toISOString(); | |
| 59 | const late = new Date(NOW - 10 * 60_000).toISOString(); | |
| 60 | for (const [id, at, n] of [["carried-old", old, 900], ["carried-late", late, 901]] as const) { | |
| 61 | issue.run(id, f.repos[0]!.id, n, f.owner.id); | |
| 62 | pull.run(id, f.repos[0]!.defaultBranch, `b/${id}`, "base-x", at, "oid-carried"); | |
| 63 | } | |
| 64 | // The same commit merged twice in one instant counts once. | |
| 65 | for (const [id, n] of [["twin-a", 902], ["twin-b", 903]] as const) { | |
| 66 | issue.run(id, f.repos[1]!.id, n, f.owner.id); | |
| 67 | pull.run(id, f.repos[1]!.defaultBranch, `b/${id}`, "base-y", late, "oid-twin"); | |
| 68 | } | |
| 69 | })(); | |
| 70 | }, 120_000); | |
| 71 | ||
| 72 | afterAll(async () => { | |
| 73 | await f.close(); | |
| 74 | }); | |
| 75 | ||
| 76 | test("readKpis answers within 100 ms over 20,000 merged pulls", async () => { | |
| 77 | readKpis(f.app.db, new Date(NOW)); | |
| 78 | const t = performance.now(); | |
| 79 | readKpis(f.app.db, new Date(NOW)); | |
| 80 | const ms = performance.now() - t; | |
| 81 | expectBudget(ms, { ms: 100, against: await baselineMs(f.base) }); | |
| 82 | }); | |
| 83 | ||
| 84 | test("the count matches the per-row NOT EXISTS, duplicates counted once", () => { | |
| 85 | // Run after the timing, so readKpis is timed on the schema production has: the | |
| 86 | // old query is seconds without an index on merged_oid, and this one is the test's own. | |
| 87 | f.app.db.run("CREATE INDEX test_pull_merged_oid ON pull(merged_oid)"); | |
| 88 | const now = new Date(NOW); | |
| 89 | const since = new Date(NOW - 24 * HOUR).toISOString(); | |
| 90 | const want = naive(since, now.toISOString()); | |
| 91 | const k = readKpis(f.app.db, now); | |
| 92 | expect(k.landed).toBe(want); | |
| 93 | expect(k.landed).toBeGreaterThan(600); | |
| 94 | // The last hour: the twins count once, the carried commit not at all. | |
| 95 | const lastHour = naive(new Date(NOW - HOUR).toISOString(), now.toISOString()); | |
| 96 | expect(k.landedHours[0]).toBe(lastHour); | |
| 97 | }); |
new above
offline