Public: anyone can read this; the crew can change it.
Sentinel: shared:01m23vygyffx5hg2f5vm8r74sr:3s8z6cbut9 is past its budget: trevato/forge: "migrate: after > (unnamed)" failed on 6 pulls in the last hour — the lane's red, not theirs; they wait … #4519
Check · Review · Land · Live done — Landed and live in production.
scout/forge:scout/4515-sentinel-shared-01m23vygyffx5hg2f5vm8r74 → main 2 commits fork 1 thread · 1 unresolved
3 files
changed +101
−7
src/origins/mirror-out.ts +16 −5
| @@ -8,11 +8,16 @@ | ||
| 8 | 8 | |
| 9 | 9 | import type { App } from "../app.ts"; |
| 10 | 10 | import { getMirrorTarget } from "../core/migrate_mirror.ts"; |
| 11 | import { getRepoById } from "../core/repos.ts"; | |
| 11 | import { getRepoById, type Repo } from "../core/repos.ts"; | |
| 12 | 12 | import { pushToMirrorTarget } from "./push-through.ts"; |
| 13 | 13 | |
| 14 | export function startMirrorOut(app: App): () => void { | |
| 14 | /** Stops listening; the promise settles once pushes already running have ended. */ | |
| 15 | export function startMirrorOut( | |
| 16 | app: App, | |
| 17 | push: (app: App, repo: Repo) => Promise<unknown> = pushToMirrorTarget, | |
| 18 | ): () => Promise<void> { | |
| 15 | 19 | const running = new Set<string>(); |
| 20 | const ending = new Set<Promise<void>>(); | |
| 16 | 21 | const again = new Set<string>(); |
| 17 | 22 | |
| 18 | 23 | async function run(repoId: string): Promise<void> { |
| @@ -25,7 +30,7 @@ export function startMirrorOut(app: App): () => void { | ||
| 25 | 30 | do { |
| 26 | 31 | again.delete(repoId); |
| 27 | 32 | const repo = getRepoById(app.db, repoId); |
| 28 | if (repo) await pushToMirrorTarget(app, repo); | |
| 33 | if (repo) await push(app, repo); | |
| 29 | 34 | } while (again.has(repoId)); |
| 30 | 35 | } catch (error) { |
| 31 | 36 | app.log.error("mirror push failed", { repoId, error: String(error) }); |
| @@ -34,12 +39,18 @@ export function startMirrorOut(app: App): () => void { | ||
| 34 | 39 | } |
| 35 | 40 | } |
| 36 | 41 | |
| 37 | return app.bus.subscribe((event) => { | |
| 42 | const unsubscribe = app.bus.subscribe((event) => { | |
| 38 | 43 | if (event.repoId === null) return; |
| 39 | 44 | if (event.kind !== "pull.merged" && event.kind !== "push") return; |
| 40 | 45 | // The migration's own arrival event: the target already holds that history. |
| 41 | 46 | if (event.payload["via"] === "migrate") return; |
| 42 | 47 | if (!getMirrorTarget(app.db, event.repoId)) return; |
| 43 | void run(event.repoId); | |
| 48 | const p = run(event.repoId).finally(() => ending.delete(p)); | |
| 49 | ending.add(p); | |
| 44 | 50 | }); |
| 51 | ||
| 52 | return async () => { | |
| 53 | unsubscribe(); | |
| 54 | await Promise.allSettled([...ending]); | |
| 55 | }; | |
| 45 | 56 | } |
test/migrate-after.test.ts +4 −2
| @@ -41,7 +41,7 @@ describe("migrate: after", () => { | ||
| 41 | 41 | let alice: User; |
| 42 | 42 | let octo: User; |
| 43 | 43 | let originId = ""; |
| 44 | let stopMirror = () => {}; | |
| 44 | let stopMirror = async () => {}; | |
| 45 | 45 | let n = 0; |
| 46 | 46 | |
| 47 | 47 | const person = async (username: string): Promise<User> => { |
| @@ -116,7 +116,9 @@ describe("migrate: after", () => { | ||
| 116 | 116 | }, 30_000); |
| 117 | 117 | |
| 118 | 118 | afterAll(async () => { |
| 119 | stopMirror(); | |
| 119 | // Pushes still running read the database and the repositories: let them | |
| 120 | // end before either goes, whatever the machine's load. | |
| 121 | await stopMirror(); | |
| 120 | 122 | github.stop(); |
| 121 | 123 | servers.stop(); |
| 122 | 124 | app.close(); |
test/mirror-out.test.ts added +81
| @@ -0,0 +1,81 @@ | ||
| 1 | /** | |
| 2 | * Mirror out's stop function: it waits for a push already running and starts | |
| 3 | * none after it. A push is held open here, so the verdict is the code's, not | |
| 4 | * how long a real git push takes on the check machine. | |
| 5 | */ | |
| 6 | ||
| 7 | import { afterAll, beforeAll, describe, expect, test } from "bun:test"; | |
| 8 | import { mkdtemp, rm } from "node:fs/promises"; | |
| 9 | import { tmpdir } from "node:os"; | |
| 10 | import { join } from "node:path"; | |
| 11 | import { createRepo } from "../src/actions/repos.ts"; | |
| 12 | import { createApp, type App } from "../src/app.ts"; | |
| 13 | import { setMirrorTarget } from "../src/core/migrate_mirror.ts"; | |
| 14 | import { getRepo, type Repo } from "../src/core/repos.ts"; | |
| 15 | import { createUser } from "../src/core/users.ts"; | |
| 16 | import { startMirrorOut } from "../src/origins/mirror-out.ts"; | |
| 17 | import { quietLogs } from "./helpers/globals.ts"; | |
| 18 | ||
| 19 | quietLogs(); | |
| 20 | ||
| 21 | describe("mirror out: stopping", () => { | |
| 22 | let dataDir: string; | |
| 23 | let app: App; | |
| 24 | let repo: Repo; | |
| 25 | ||
| 26 | beforeAll(async () => { | |
| 27 | dataDir = await mkdtemp(join(tmpdir(), "forge-mirror-out-")); | |
| 28 | app = createApp({ dataDir, port: 0 }); | |
| 29 | const made = await createUser(app.db, { | |
| 30 | origin: "seed", | |
| 31 | trusted: true, | |
| 32 | username: "alice", | |
| 33 | email: "[email protected]", | |
| 34 | adminIfFirst: true, | |
| 35 | }); | |
| 36 | if (!made.ok) throw new Error(String(made.error)); | |
| 37 | const created = await createRepo(app, made.value, { name: "gizmo", visibility: "private" }); | |
| 38 | if (!created.ok) throw new Error(created.error); | |
| 39 | repo = getRepo(app.db, "alice", "gizmo")!; | |
| 40 | app.db.run("PRAGMA foreign_keys = OFF"); | |
| 41 | setMirrorTarget(app.db, { | |
| 42 | repoId: repo.id, | |
| 43 | originId: "origin-none", | |
| 44 | fullName: "octo/gizmo", | |
| 45 | cloneUrl: "https://example.test/octo/gizmo.git", | |
| 46 | tips: {}, | |
| 47 | }); | |
| 48 | }); | |
| 49 | ||
| 50 | afterAll(async () => { | |
| 51 | app.close(); | |
| 52 | await rm(dataDir, { recursive: true, force: true, maxRetries: 20, retryDelay: 100 }); | |
| 53 | }); | |
| 54 | ||
| 55 | const landed = () => app.bus.append({ kind: "pull.merged", repoId: repo.id }); | |
| 56 | ||
| 57 | test("the stop promise waits for a running push, and no push starts after it", async () => { | |
| 58 | let pushes = 0; | |
| 59 | let release = () => {}; | |
| 60 | const held = new Promise<void>((resolve) => (release = resolve)); | |
| 61 | const stop = startMirrorOut(app, async () => { | |
| 62 | pushes++; | |
| 63 | await held; | |
| 64 | }); | |
| 65 | ||
| 66 | landed(); | |
| 67 | expect(pushes).toBe(1); | |
| 68 | ||
| 69 | let stopped = false; | |
| 70 | const stopping = stop().then(() => (stopped = true)); | |
| 71 | await Bun.sleep(50); | |
| 72 | expect(stopped).toBe(false); | |
| 73 | ||
| 74 | landed(); | |
| 75 | release(); | |
| 76 | await stopping; | |
| 77 | expect(stopped).toBe(true); | |
| 78 | await Bun.sleep(20); | |
| 79 | expect(pushes).toBe(1); | |
| 80 | }); | |
| 81 | }); |
new above
offline
Untested: add a test that the stop promise waits for an in-flight push and that no push starts after stopping.