Public: anyone can read this; the crew can change it.
Land the drift alarm's git-read cache from branch scout/6502-the-drift-alarm-reads-each-pull-s-distan, with the older tests as they are on main #6538
Check · Review · Land · Live done — Landed and live in production.
scout/forge:scout/6535-land-the-drift-alarm-s-git-read-cache-fr → main 5 commits fork 1 thread · 1 unresolved
2 files
changed +137
−6
src/flow/alarm_drift.ts +53 −5
| @@ -52,6 +52,43 @@ export async function readDrift(dir: string, branch: string, headOid: string): P | ||
| 52 | 52 | return { landings: times.length, since: times.length > 0 && Number.isFinite(first) ? first * 1000 : null }; |
| 53 | 53 | } |
| 54 | 54 | |
| 55 | type Measure = Omit<Drift, "number">; | |
| 56 | export type DriftReader = (dir: string, branch: string, headOid: string) => Promise<Measure>; | |
| 57 | ||
| 58 | export const DRIFT_CACHE_MAX = 1000; | |
| 59 | const cache = new Map<string, Measure>(); | |
| 60 | ||
| 61 | export function clearDriftCache(): void { | |
| 62 | cache.clear(); | |
| 63 | } | |
| 64 | ||
| 65 | /** The tip of `branch`, one cheap git call; null when unreadable. */ | |
| 66 | export async function branchTip(dir: string, branch: string): Promise<string | null> { | |
| 67 | if (!isSafeRef(branch)) return null; | |
| 68 | const out = await runGit(["rev-parse", "--verify", "-q", `refs/heads/${branch}`], { cwd: dir }); | |
| 69 | const oid = out.ok ? out.value.trim() : ""; | |
| 70 | return isSafeOid(oid) ? oid : null; | |
| 71 | } | |
| 72 | ||
| 73 | /** `read` answered once per repository, head oid and main's tip; the oldest entries go past DRIFT_CACHE_MAX. */ | |
| 74 | export async function cachedDrift( | |
| 75 | read: DriftReader, | |
| 76 | dir: string, | |
| 77 | branch: string, | |
| 78 | headOid: string, | |
| 79 | tip: string | null, | |
| 80 | max = DRIFT_CACHE_MAX, | |
| 81 | ): Promise<Measure> { | |
| 82 | if (tip === null) return read(dir, branch, headOid); | |
| 83 | const key = `${dir}\0${branch}\0${headOid}\0${tip}`; | |
| 84 | const hit = cache.get(key); | |
| 85 | if (hit) return hit; | |
| 86 | const fresh = await read(dir, branch, headOid); | |
| 87 | cache.set(key, fresh); | |
| 88 | while (cache.size > max) cache.delete(cache.keys().next().value as string); | |
| 89 | return fresh; | |
| 90 | } | |
| 91 | ||
| 55 | 92 | const behind = (d: Drift, now: number): number => (d.since === null ? 0 : now - d.since); |
| 56 | 93 | const drifting = (d: Drift, now: number): boolean => d.landings > DRIFT_LANDINGS || behind(d, now) > DRIFT_MS; |
| 57 | 94 | |
| @@ -71,7 +108,11 @@ export function driftAlarm(drifts: readonly Drift[], now: number): DriftAlarm { | ||
| 71 | 108 | } |
| 72 | 109 | |
| 73 | 110 | /** Reads every open pull's drift from its repository's main and answers the alarm. */ |
| 74 | export async function alarmDrift(app: App, now: Date | number = Date.now()): Promise<DriftAlarm> { | |
| 111 | export async function alarmDrift( | |
| 112 | app: App, | |
| 113 | now: Date | number = Date.now(), | |
| 114 | { read = readDrift, readTip = branchTip }: { read?: DriftReader; readTip?: typeof branchTip } = {}, | |
| 115 | ): Promise<DriftAlarm> { | |
| 75 | 116 | const at = typeof now === "number" ? now : now.getTime(); |
| 76 | 117 | const rows = app.db |
| 77 | 118 | .query<Row, []>( |
| @@ -81,11 +122,18 @@ export async function alarmDrift(app: App, now: Date | number = Date.now()): Pro | ||
| 81 | 122 | WHERE i.kind = 'pull' AND i.state = 'open' AND p.merged_at IS NULL`, |
| 82 | 123 | ) |
| 83 | 124 | .all(); |
| 125 | const tips = new Map<string, Promise<string | null>>(); | |
| 126 | const tipOf = (dir: string, branch: string) => { | |
| 127 | const k = `${dir}\0${branch}`; | |
| 128 | let t = tips.get(k); | |
| 129 | if (!t) tips.set(k, (t = readTip(dir, branch))); | |
| 130 | return t; | |
| 131 | }; | |
| 84 | 132 | const drifts = await Promise.all( |
| 85 | rows.map(async (r) => ({ | |
| 86 | number: r.number, | |
| 87 | ...(await readDrift(repoDir(app.reposRoot, r.owner, r.name), r.base_branch, r.head_oid)), | |
| 88 | })), | |
| 133 | rows.map(async (r) => { | |
| 134 | const dir = repoDir(app.reposRoot, r.owner, r.name); | |
| 135 | return { number: r.number, ...(await cachedDrift(read, dir, r.base_branch, r.head_oid, await tipOf(dir, r.base_branch))) }; | |
| 136 | }), | |
| 89 | 137 | ); |
| 90 | 138 | return driftAlarm(drifts, at); |
| 91 | 139 | } |
test/flow-alarm-drift.test.ts +84 −1
| @@ -9,7 +9,7 @@ import { setIssueState } from "../src/core/issues.ts"; | ||
| 9 | 9 | import { createPull } from "../src/core/pulls.ts"; |
| 10 | 10 | import { createRepo } from "../src/core/repos.ts"; |
| 11 | 11 | import { createUser } from "../src/core/users.ts"; |
| 12 | import { alarmDrift, driftAlarm } from "../src/flow/alarm_drift.ts"; | |
| 12 | import { alarmDrift, clearDriftCache, branchTip, cachedDrift, driftAlarm, readDrift } from "../src/flow/alarm_drift.ts"; | |
| 13 | 13 | import { repoDir } from "../src/git/repos.ts"; |
| 14 | 14 | import { quietLogs } from "./helpers/globals.ts"; |
| 15 | 15 | |
| @@ -126,3 +126,86 @@ test("a rebased head stops counting; a merged pull never counts", async () => { | ||
| 126 | 126 | app.db.query("UPDATE pull SET merged_at = ?1 WHERE issue_id = ?2").run(new Date().toISOString(), pull.id); |
| 127 | 127 | expect((await alarmDrift(app, new Date())).fired).toBe(false); |
| 128 | 128 | }); |
| 129 | ||
| 130 | test("unmoved tips cost one git read per pull; moving main recomputes; the cache is bounded", async () => { | |
| 131 | const ada = ok( | |
| 132 | await createUser(app.db, { origin: "seed", trusted: true, username: "cy", email: "[email protected]", password: "a-good-passphrase" }), | |
| 133 | ); | |
| 134 | const repo = ok(await createRepo(app.db, app.reposRoot, { ownerId: ada.id, name: "cache", visibility: "public" })); | |
| 135 | const dir = repoDir(app.reposRoot, ada.username, repo.name); | |
| 136 | ok(await commitToBranch(app, ada, repo, { message: "seed", files: [{ path: "a.txt", content: "a\n" }] })); | |
| 137 | for (const b of ["p1", "p2"]) | |
| 138 | ok(await commitToBranch(app, ada, repo, { message: b, branch: b, files: [{ path: `${b}.txt`, content: "x\n" }] })); | |
| 139 | for (const b of ["p1", "p2"]) | |
| 140 | ok(await createPull(app.db, dir, { repoId: repo.id, authorId: ada.id, title: b, baseBranch: "main", headBranch: b })); | |
| 141 | clearDriftCache(); | |
| 142 | let calls = 0; | |
| 143 | const counted = (d: string, b: string, h: string) => (calls++, readDrift(d, b, h)); | |
| 144 | await alarmDrift(app, Date.now(), { read: counted }); | |
| 145 | expect(calls).toBe(2); | |
| 146 | await alarmDrift(app, Date.now(), { read: counted }); | |
| 147 | expect(calls).toBe(2); | |
| 148 | ok(await commitToBranch(app, ada, repo, { message: "moves", files: [{ path: "m.txt", content: "m\n" }] })); | |
| 149 | await alarmDrift(app, Date.now(), { read: counted }); | |
| 150 | expect(calls).toBe(4); | |
| 151 | }); | |
| 152 | ||
| 153 | test("past the cap the oldest entry is dropped and re-read; a recent one is not", async () => { | |
| 154 | clearDriftCache(); | |
| 155 | let calls = 0; | |
| 156 | const read = async () => (calls++, { landings: calls, since: null }); | |
| 157 | const at = (n: number) => cachedDrift(read, "/r", "main", `${n}`.padStart(40, "0"), "t", 3); | |
| 158 | for (const n of [1, 2, 3, 4]) await at(n); | |
| 159 | expect(calls).toBe(4); | |
| 160 | await at(4); | |
| 161 | expect(calls).toBe(4); | |
| 162 | await at(1); | |
| 163 | expect(calls).toBe(5); | |
| 164 | }); | |
| 165 | ||
| 166 | test("an unreadable tip is read directly and never cached", async () => { | |
| 167 | clearDriftCache(); | |
| 168 | let calls = 0; | |
| 169 | const read = async () => (calls++, { landings: 0, since: null }); | |
| 170 | await cachedDrift(read, "/r", "main", "a".repeat(40), null); | |
| 171 | await cachedDrift(read, "/r", "main", "a".repeat(40), null); | |
| 172 | expect(calls).toBe(2); | |
| 173 | await cachedDrift(read, "/r", "main", "a".repeat(40), "t"); | |
| 174 | expect(calls).toBe(3); | |
| 175 | }); | |
| 176 | ||
| 177 | test("many pulls in a few repositories cost one tip read per repository", async () => { | |
| 178 | const ada = ok( | |
| 179 | await createUser(app.db, { origin: "seed", trusted: true, username: "di", email: "[email protected]", password: "a-good-passphrase" }), | |
| 180 | ); | |
| 181 | clearDriftCache(); | |
| 182 | let pulls = 0; | |
| 183 | const dirs: string[] = []; | |
| 184 | for (const name of ["m1", "m2", "m3"]) { | |
| 185 | const repo = ok(await createRepo(app.db, app.reposRoot, { ownerId: ada.id, name, visibility: "public" })); | |
| 186 | const dir = repoDir(app.reposRoot, ada.username, repo.name); | |
| 187 | dirs.push(dir); | |
| 188 | ok(await commitToBranch(app, ada, repo, { message: "seed", files: [{ path: "a.txt", content: "a\n" }] })); | |
| 189 | for (let i = 0; i < 4; i++) { | |
| 190 | const b = `b${i}`; | |
| 191 | ok(await commitToBranch(app, ada, repo, { message: b, branch: b, files: [{ path: `${b}.txt`, content: "x\n" }] })); | |
| 192 | ok(await createPull(app.db, dir, { repoId: repo.id, authorId: ada.id, title: b, baseBranch: "main", headBranch: b })); | |
| 193 | pulls++; | |
| 194 | } | |
| 195 | } | |
| 196 | let tipReads = 0; | |
| 197 | let reads = 0; | |
| 198 | const readTip = (d: string, b: string) => (tipReads++, branchTip(d, b)); | |
| 199 | const read = (d: string, b: string, h: string) => (reads++, readDrift(d, b, h)); | |
| 200 | const before = (await alarmDrift(app, Date.now(), { read, readTip })).fired; | |
| 201 | const firstReads = reads; | |
| 202 | const repos = app.db.query<{ n: number }, []>("SELECT COUNT(DISTINCT i.repo_id) AS n FROM issue i JOIN pull p ON p.issue_id = i.id WHERE i.state = 'open' AND p.merged_at IS NULL").get()!.n; | |
| 203 | expect(tipReads).toBe(repos); | |
| 204 | expect(repos).toBeLessThan(reads); | |
| 205 | tipReads = 0; | |
| 206 | const again = await alarmDrift(app, Date.now(), { read, readTip }); | |
| 207 | expect(again.fired).toBe(before); | |
| 208 | expect(tipReads).toBe(repos); | |
| 209 | expect(reads).toBe(firstReads); | |
| 210 | expect(pulls).toBe(12); | |
| 211 | }); |
new above
offline
Not blocking:
readDriftanswers{landings: 0, since: null}whenmerge-basefails, for example when the head object has not arrived yet. That zero is cached until main's tip moves. On a quiet main, a drifting pull could read as not drifting for hours. Consider caching only whenreadactually measured something. One way is forreadDriftto tellnoneapart from a real zero, so a failed read is never stored, just like a null tip.