| 1 | import { WorkflowEntrypoint, type WorkflowEvent, type WorkflowStep } from "cloudflare:workers"; |
| 2 | import type { Env } from "../env.js"; |
| 3 | import { ImportService, PAGED_STEPS, type StepCursor } from "./service.js"; |
| 4 | import { plainSteps } from "../lib/rpc.js"; |
| 5 | |
| 6 | export interface ImportParams { |
| 7 | importId: string; |
| 8 | } |
| 9 | |
| 10 | const RETRY = { retries: { limit: 5, delay: "15 seconds", backoff: "exponential" }, timeout: "15 minutes" } as const; |
| 11 | /** How long the workflow waits between checks on the Git transfer, and for how many checks (6 hours). */ |
| 12 | const GIT_CHECK = "10 minutes"; |
| 13 | const GIT_CHECKS = 36; |
| 14 | const GIT_TRIES = 2; |
| 15 | /** An overall cap on how many times one paged step may be called, so a hostile or broken source cannot run the workflow forever. */ |
| 16 | const MAX_STEP_CALLS = 2000; |
| 17 | /** How many rate-limit sleeps (each at most an hour) the repository lookup may take. */ |
| 18 | const MAX_RATE_LIMIT_WAITS = 48; |
| 19 | |
| 20 | /** |
| 21 | * Durable execution of one GitHub import (instance id `<importId>-<attempt>`). |
| 22 | * A repository import runs its steps in order, each retried on failure and |
| 23 | * paged so that no step runs long; rate limits become sleeps. A bulk import |
| 24 | * fans out one repository import per repository, a few at a time: each one |
| 25 | * starts the next when it ends, and the last one wakes the bulk import. |
| 26 | */ |
| 27 | export class ImportWorkflow extends WorkflowEntrypoint<Env, ImportParams> { |
| 28 | override async run(event: WorkflowEvent<ImportParams>, rpcStep: WorkflowStep): Promise<unknown> { |
| 29 | const step = plainSteps(rpcStep); |
| 30 | const id = event.payload.importId; |
| 31 | const svc = () => new ImportService(this.env); |
| 32 | const begin = await step.do("begin", RETRY, () => svc().begin(id)); |
| 33 | if (begin.stop) return { stopped: true }; |
| 34 | try { |
| 35 | if (begin.kind === "bulk") await this.bulk(id, step, svc); |
| 36 | else await this.repository(id, step, svc); |
| 37 | } catch (e) { |
| 38 | const message = e instanceof Error ? e.message : String(e); |
| 39 | await step.do("fail", RETRY, () => svc().fail(id, message)); |
| 40 | } |
| 41 | await step.do("after", RETRY, () => svc().afterChild(id)); |
| 42 | return { done: true }; |
| 43 | } |
| 44 | |
| 45 | private async repository(id: string, step: WorkflowStep, svc: () => ImportService): Promise<void> { |
| 46 | // GitHub's rate limit is slept off, never a failure (the first call keeps its original step name). |
| 47 | for (let i = 0; ; i++) { |
| 48 | if (i >= MAX_RATE_LIMIT_WAITS) throw new Error("GitHub kept rate limiting the repository lookup"); |
| 49 | const r = await step.do(i === 0 ? "repository" : `repository ${i}`, RETRY, () => svc().stepRepository(id)); |
| 50 | // `?.`: an instance that ran this step before the change replays it with no result. |
| 51 | if (!r?.wait) break; |
| 52 | await step.sleep(`repository rate limit ${i}`, `${r.wait} seconds`); |
| 53 | } |
| 54 | |
| 55 | for (let attempt = 1; ; attempt++) { |
| 56 | await step.do(`git start ${attempt}`, RETRY, () => svc().startGit(id)); |
| 57 | let status: Awaited<ReturnType<ImportService["gitStatus"]>> = { state: "running" }; |
| 58 | for (let i = 0; i < GIT_CHECKS; i++) { |
| 59 | status = await step.do(`git check ${attempt}.${i}`, RETRY, () => svc().gitStatus(id)); |
| 60 | if (status.state !== "running") break; |
| 61 | try { |
| 62 | await step.waitForEvent(`git wait ${attempt}.${i}`, { type: "git-done", timeout: GIT_CHECK }); |
| 63 | } catch { |
| 64 | // timed out: check again |
| 65 | } |
| 66 | } |
| 67 | if (status.state === "done") break; |
| 68 | if (status.state === "running") throw new Error("the Git transfer did not finish within 6 hours"); |
| 69 | if (!status.retryable || attempt >= GIT_TRIES) throw new Error(status.error ?? "the Git transfer failed"); |
| 70 | const error = status.error ?? ""; |
| 71 | await step.do(`git reset ${attempt}`, RETRY, () => svc().resetGit(id, error)); |
| 72 | } |
| 73 | |
| 74 | for (const name of PAGED_STEPS) { |
| 75 | let cursor: StepCursor = null; |
| 76 | for (let i = 0; ; i++) { |
| 77 | if (i >= MAX_STEP_CALLS) throw new Error(`the ${name} step did not finish within ${MAX_STEP_CALLS} pages`); |
| 78 | const r = await step.do(`${name} ${i}`, RETRY, () => svc().runStep(id, name, cursor)); |
| 79 | if (r.wait) await step.sleep(`${name} rate limit ${i}`, `${r.wait} seconds`); |
| 80 | if (r.done) break; |
| 81 | cursor = r.cursor; |
| 82 | } |
| 83 | } |
| 84 | |
| 85 | await step.do("finalize", RETRY, () => svc().finalize(id)); |
| 86 | } |
| 87 | |
| 88 | private async bulk(id: string, step: WorkflowStep, svc: () => ImportService): Promise<void> { |
| 89 | for (let i = 0; ; i++) { |
| 90 | const r = await step.do(`organization ${i}`, RETRY, () => svc().bulkOrganization(id)); |
| 91 | if (!r.wait) break; |
| 92 | await step.sleep(`organization rate limit ${i}`, `${r.wait} seconds`); |
| 93 | } |
| 94 | await step.do("start repositories", RETRY, () => svc().startChildren(id)); |
| 95 | // Repository imports start each other; this only waits for the last one (and checks hourly, up to a week). |
| 96 | for (let i = 0; i < 168; i++) { |
| 97 | const { open } = await step.do(`progress ${i}`, RETRY, () => svc().bulkProgress(id)); |
| 98 | if (!open) break; |
| 99 | try { |
| 100 | await step.waitForEvent(`wait ${i}`, { type: "children-done", timeout: "1 hour" }); |
| 101 | } catch { |
| 102 | // timed out: restart anything stuck in the queue, then check again |
| 103 | await step.do(`restart ${i}`, RETRY, () => svc().startChildren(id)); |
| 104 | } |
| 105 | } |
| 106 | await step.do("finalize", RETRY, () => svc().bulkFinish(id)); |
| 107 | } |
| 108 | } |