Public: anyone can read this; the crew can change it.
One call rebuilds combined/N for every open kept-fork pull into the default branch #10053
Check · Review · Land · Live done — Landed and live in production.
scout/forge:scout/10039-one-call-rebuilds-combined-n-for-every-o → main 1 commit fork 1 thread · 1 unresolved
2 files
changed +231
−0
src/core/recombine.ts added +80
| @@ -0,0 +1,80 @@ | ||
| 1 | /** | |
| 2 | * Bring every open kept-fork pull's `combined/N` up to date with the default | |
| 3 | * branch (#9012). One pass per repository at a time: a call made during a | |
| 4 | * pass sets one "again" flag, and when the pass ends exactly one more runs, | |
| 5 | * reading the tip then. Nothing calls this yet; a hook will. | |
| 6 | */ | |
| 7 | ||
| 8 | import type { Database } from "bun:sqlite"; | |
| 9 | import { combine } from "../git/combined.ts"; | |
| 10 | import { repoDir } from "../git/repos.ts"; | |
| 11 | import { resolveOid } from "../git/merge.ts"; | |
| 12 | import type { Log } from "../kernel/log.ts"; | |
| 13 | import { fromKeptFork } from "./landing_kind.ts"; | |
| 14 | import { listOpenPulls } from "./pulls.ts"; | |
| 15 | import { getRepoById } from "./repos.ts"; | |
| 16 | ||
| 17 | /** What a pass needs of the forge: its rows, where repositories live, a log. */ | |
| 18 | export interface RecombineForge { | |
| 19 | readonly db: Database; | |
| 20 | readonly reposRoot: string; | |
| 21 | readonly log: Log; | |
| 22 | } | |
| 23 | ||
| 24 | const running = new Map<string, { again: boolean; done: Promise<number> }>(); | |
| 25 | ||
| 26 | /** One pass: the tip is read once, each pull combined against it. */ | |
| 27 | async function pass(forge: RecombineForge, repoId: string): Promise<void> { | |
| 28 | const repo = getRepoById(forge.db, repoId); | |
| 29 | if (repo === undefined) return; | |
| 30 | const dir = repoDir(forge.reposRoot, repo.ownerName, repo.name); | |
| 31 | const baseOid = await resolveOid(dir, `refs/heads/${repo.defaultBranch}`); | |
| 32 | if (baseOid === undefined) return; | |
| 33 | const pulls = listOpenPulls(forge.db, repoId).filter( | |
| 34 | (p) => p.baseBranch === repo.defaultBranch && fromKeptFork(forge.db, p), | |
| 35 | ); | |
| 36 | for (const pull of pulls) { | |
| 37 | try { | |
| 38 | const headOid = await resolveOid(dir, `refs/pull/${pull.number}/head`); | |
| 39 | const made = | |
| 40 | headOid === undefined ? undefined : await combine(dir, { number: pull.number, baseOid, headOid }); | |
| 41 | if (made === undefined) | |
| 42 | forge.log.warn("recombine skipped a pull", { repo: repoId, pull: pull.number }); | |
| 43 | } catch (e) { | |
| 44 | forge.log.warn("recombine failed on a pull", { | |
| 45 | repo: repoId, | |
| 46 | pull: pull.number, | |
| 47 | error: e instanceof Error ? e.message : String(e), | |
| 48 | }); | |
| 49 | } | |
| 50 | } | |
| 51 | } | |
| 52 | ||
| 53 | /** | |
| 54 | * Recombine the repository's kept-fork pulls. Answers how many passes this | |
| 55 | * call's run took (1, or 2 when calls arrived meanwhile); a call that joins | |
| 56 | * a running pass answers with that run's count. | |
| 57 | */ | |
| 58 | export function recombine(forge: RecombineForge, repoId: string): Promise<number> { | |
| 59 | const current = running.get(repoId); | |
| 60 | if (current !== undefined) { | |
| 61 | current.again = true; | |
| 62 | return current.done; | |
| 63 | } | |
| 64 | const state = { again: false, done: Promise.resolve(0) }; | |
| 65 | state.done = (async () => { | |
| 66 | let passes = 0; | |
| 67 | try { | |
| 68 | do { | |
| 69 | state.again = false; | |
| 70 | passes++; | |
| 71 | await pass(forge, repoId); | |
| 72 | } while (state.again); | |
| 73 | } finally { | |
| 74 | running.delete(repoId); | |
| 75 | } | |
| 76 | return passes; | |
| 77 | })(); | |
| 78 | running.set(repoId, state); | |
| 79 | return state.done; | |
| 80 | } |
test/combined-recombine.test.ts added +151
| @@ -0,0 +1,151 @@ | ||
| 1 | /** `recombine` (#10039): main moves, every kept-fork pull's combined branch follows, one pass at a time. */ | |
| 2 | import { afterAll, beforeAll, expect, test } from "bun:test"; | |
| 3 | import { mkdtemp, rm } from "node:fs/promises"; | |
| 4 | import { tmpdir } from "node:os"; | |
| 5 | import { join } from "node:path"; | |
| 6 | import { createApp, type App } from "../src/app.ts"; | |
| 7 | import { serve, type Servers } from "../src/http/server.ts"; | |
| 8 | import { gitClientEnv } from "./helpers/git-client.ts"; | |
| 9 | import { proveAll } from "./helpers/verified.ts"; | |
| 10 | import { promote } from "./helpers/members.ts"; | |
| 11 | import { recombine } from "../src/core/recombine.ts"; | |
| 12 | import { getRepo } from "../src/core/repos.ts"; | |
| 13 | import { repoDir } from "../src/git/repos.ts"; | |
| 14 | import { tokenFor } from "./helpers/tokens.ts"; | |
| 15 | ||
| 16 | const gitEnv = gitClientEnv({ | |
| 17 | GIT_AUTHOR_NAME: "Jay Moran", | |
| 18 | GIT_AUTHOR_EMAIL: "[email protected]", | |
| 19 | GIT_COMMITTER_NAME: "Jay Moran", | |
| 20 | GIT_COMMITTER_EMAIL: "[email protected]", | |
| 21 | }); | |
| 22 | ||
| 23 | let dataDir = ""; | |
| 24 | let work = ""; | |
| 25 | let app: App; | |
| 26 | let servers: Servers; | |
| 27 | let base = ""; | |
| 28 | let alice = ""; | |
| 29 | let bob = ""; | |
| 30 | let repoId = ""; | |
| 31 | let bare = ""; | |
| 32 | const warnings: string[] = []; | |
| 33 | ||
| 34 | async function git(args: string[], cwd: string): Promise<string> { | |
| 35 | const p = Bun.spawn(["git", ...args], { cwd, env: gitEnv, stdout: "pipe", stderr: "pipe" }); | |
| 36 | const [out, err, code] = await Promise.all([ | |
| 37 | new Response(p.stdout).text(), | |
| 38 | new Response(p.stderr).text(), | |
| 39 | p.exited, | |
| 40 | ]); | |
| 41 | if (code !== 0) throw new Error(`git ${args.join(" ")}: ${err}`); | |
| 42 | return out.trim(); | |
| 43 | } | |
| 44 | const call = (path: string, token: string, body?: unknown): Promise<Response> => | |
| 45 | fetch(`${base}${path}`, { | |
| 46 | method: body === undefined ? "GET" : "POST", | |
| 47 | headers: { authorization: `Bearer ${token}`, "content-type": "application/json" }, | |
| 48 | ...(body === undefined ? {} : { body: JSON.stringify(body) }), | |
| 49 | }); | |
| 50 | const authed = (user: string, token: string) => base.replace("http://", `http://${user}:${token}@`); | |
| 51 | async function signup(username: string): Promise<string> { | |
| 52 | await fetch(`${base}/signup`, { | |
| 53 | method: "POST", | |
| 54 | body: new URLSearchParams({ username, email: `${username}@f.dev`, password: "kept-fork-pass" }), | |
| 55 | redirect: "manual", | |
| 56 | }); | |
| 57 | return tokenFor(app, username, "admin", "k"); | |
| 58 | } | |
| 59 | const forge = () => ({ | |
| 60 | db: app.db, | |
| 61 | reposRoot: app.reposRoot, | |
| 62 | log: { ...app.log, warn: (m: string) => void warnings.push(m) }, | |
| 63 | }); | |
| 64 | const parents = async (ref: string) => (await git(["log", "-1", "--format=%P", ref], bare)).split(" "); | |
| 65 | ||
| 66 | let a = ""; | |
| 67 | let b = ""; | |
| 68 | let first = 0; | |
| 69 | let second = 0; | |
| 70 | ||
| 71 | beforeAll(async () => { | |
| 72 | dataDir = await mkdtemp(join(tmpdir(), "forge-rc-")); | |
| 73 | work = await mkdtemp(join(tmpdir(), "forge-rc-work-")); | |
| 74 | app = createApp({ dataDir, port: 0 }); | |
| 75 | servers = await serve(app); | |
| 76 | base = servers.url; | |
| 77 | alice = await signup("alice"); | |
| 78 | bob = await signup("bob"); | |
| 79 | proveAll(app); | |
| 80 | promote(app.db); | |
| 81 | await call("/api/v1/repos", alice, { name: "app", visibility: "public" }); | |
| 82 | a = join(work, "a"); | |
| 83 | b = join(work, "b"); | |
| 84 | await git(["clone", `${authed("alice", alice)}/alice/app.git`, "a"], work); | |
| 85 | await Bun.write(join(a, "one.txt"), "one\n"); | |
| 86 | await git(["add", "."], a); | |
| 87 | await git(["commit", "-m", "seed"], a); | |
| 88 | await git(["push", "origin", "main"], a); | |
| 89 | await call("/api/v1/repos/alice/app/forks", bob, {}); | |
| 90 | await git(["clone", `${authed("bob", bob)}/bob/app.git`, "b"], work); | |
| 91 | for (const [branch, file] of [["one", "b1.txt"], ["two", "b2.txt"]] as const) { | |
| 92 | await git(["checkout", "-q", "-b", branch, "origin/main"], b); | |
| 93 | await Bun.write(join(b, file), `${branch}\n`); | |
| 94 | await git(["add", "."], b); | |
| 95 | await git(["commit", "-m", `bob ${branch}`], b); | |
| 96 | await git(["push", "origin", branch], b); | |
| 97 | } | |
| 98 | const open = async (head: string): Promise<number> => { | |
| 99 | const r = await call("/api/v1/repos/alice/app/pulls", bob, { | |
| 100 | title: head, | |
| 101 | base: "main", | |
| 102 | head, | |
| 103 | headRepo: "bob/app", | |
| 104 | }); | |
| 105 | expect(r.status).toBe(201); | |
| 106 | return ((await r.json()) as { number: number }).number; | |
| 107 | }; | |
| 108 | first = await open("one"); | |
| 109 | second = await open("two"); | |
| 110 | repoId = getRepo(app.db, "alice", "app")!.id; | |
| 111 | bare = repoDir(app.reposRoot, "alice", "app"); | |
| 112 | }); | |
| 113 | ||
| 114 | afterAll(async () => { | |
| 115 | servers.stop(); | |
| 116 | app.close(); | |
| 117 | await rm(dataDir, { recursive: true, force: true }); | |
| 118 | await rm(work, { recursive: true, force: true }); | |
| 119 | }); | |
| 120 | ||
| 121 | async function moveMain(file: string): Promise<string> { | |
| 122 | await Bun.write(join(a, file), `${file}\n`); | |
| 123 | await git(["add", "."], a); | |
| 124 | await git(["commit", "-m", `main adds ${file}`], a); | |
| 125 | await git(["push", "origin", "main"], a); | |
| 126 | return git(["rev-parse", "HEAD"], a); | |
| 127 | } | |
| 128 | ||
| 129 | test("main moves, and every combined branch has the new main as a parent", async () => { | |
| 130 | const main = await moveMain("m1.txt"); | |
| 131 | expect(await recombine(forge(), repoId)).toBe(1); | |
| 132 | for (const n of [first, second]) expect((await parents(`refs/heads/combined/${n}`))[1]).toBe(main); | |
| 133 | }); | |
| 134 | ||
| 135 | test("one pull that cannot combine is logged and skipped, the others still follow", async () => { | |
| 136 | await git(["update-ref", "-d", `refs/pull/${first}/head`], bare); | |
| 137 | const before = await git(["rev-parse", `refs/heads/combined/${first}`], bare); | |
| 138 | const main = await moveMain("m2.txt"); | |
| 139 | warnings.length = 0; | |
| 140 | await recombine(forge(), repoId); | |
| 141 | expect(warnings.length).toBe(1); | |
| 142 | expect(await git(["rev-parse", `refs/heads/combined/${first}`], bare)).toBe(before); | |
| 143 | expect((await parents(`refs/heads/combined/${second}`))[1]).toBe(main); | |
| 144 | }); | |
| 145 | ||
| 146 | test("calls made during a pass share it: three at once run at most two passes", async () => { | |
| 147 | await moveMain("m3.txt"); | |
| 148 | const counts = await Promise.all([recombine(forge(), repoId), recombine(forge(), repoId), recombine(forge(), repoId)]); | |
| 149 | expect(Math.max(...counts)).toBeLessThanOrEqual(2); | |
| 150 | expect(Math.max(...counts)).toBe(2); | |
| 151 | }); |
The comment cites #9012; this work is #10039. Fix the reference.