Hosted runners on the new image aren't turned away while the repository object catches up with a deploy (#229)
* Hosted runners on the new image aren't turned away while the repository object catches up with a deploy
- CI / Typecheck and bundlesuccess
- CI / Unit tests (1/2)success
- CI / Unit tests (2/2)Failed
3 files changed+40−13
modifiedapps/diffinery/src/http/runner.ts+3−2View file
| @@ -44,7 +44,8 @@ | ||
| 44 | 44 | runner.get("/runner/jobs/:jobId", async (c) => { |
| 45 | 45 | const jobId = c.req.param("jobId"); |
| 46 | 46 | const { token, repoId } = jobAuth(c.req.raw, jobId); |
| 47 | - return c.json(await repoStub(c.env, repoId).runnerSpec(jobId, token, parseRunnerInfo(c.req.header("x-diffinery-runner")))); | |
| 47 | + // This Worker's runner image, not the repository object's: after a deploy the object can run the previous version for minutes. | |
| 48 | + return c.json(await repoStub(c.env, repoId).runnerSpec(jobId, token, parseRunnerInfo(c.req.header("x-diffinery-runner")), c.env.RUNNER_IMAGE ?? null)); | |
| 48 | 49 | }); |
| 49 | 50 | |
| 50 | 51 | runner.post("/runner/jobs/:jobId/logs", async (c) => { |
| @@ -65,7 +66,7 @@ | ||
| 65 | 66 | runner.post("/runner/jobs/:jobId/complete", async (c) => { |
| 66 | 67 | const jobId = c.req.param("jobId"); |
| 67 | 68 | const { token, repoId } = jobAuth(c.req.raw, jobId); |
| 68 | - await repoStub(c.env, repoId).runnerComplete(jobId, token, (await c.req.json()) as JobCompletion); | |
| 69 | + await repoStub(c.env, repoId).runnerComplete(jobId, token, (await c.req.json()) as JobCompletion, c.env.RUNNER_IMAGE ?? null); | |
| 69 | 70 | return c.json({ ok: true }); |
| 70 | 71 | }); |
| 71 | 72 |
modifiedapps/diffinery/src/repo/actions.ts+15−7View file
| @@ -2119,17 +2119,24 @@ | ||
| 2119 | 2119 | return run.event === "pull_request" && this.repo.policy.isForkPull(run.pr_number); |
| 2120 | 2120 | } |
| 2121 | 2121 | |
| 2122 | - /** Turns away a hosted runner on an image other than the deployed one (see STALE_IMAGE_TRIES). */ | |
| 2123 | - private async refuseStaleRunner(job: JobRow, runner: JobRunnerInfo | null): Promise<void> { | |
| 2124 | - const want = this.env.RUNNER_IMAGE; | |
| 2122 | + /** | |
| 2123 | + * Turns away a hosted runner on an image other than the deployed one (see STALE_IMAGE_TRIES). | |
| 2124 | + * `deployed` is the image of the Worker that took the runner's request. After a deploy this | |
| 2125 | + * object can run the previous version for minutes, while the Worker is updated within seconds | |
| 2126 | + * and the containers come up on the new image: going by this object's own RUNNER_IMAGE then | |
| 2127 | + * turned away every runner on the new image until this object was updated too. | |
| 2128 | + */ | |
| 2129 | + private async refuseStaleRunner(job: JobRow, runner: JobRunnerInfo | null, deployed: string | null = null): Promise<void> { | |
| 2130 | + const want = deployed ?? this.env.RUNNER_IMAGE; | |
| 2125 | 2131 | if (!want || job.labels || runner?.image === want || job.placement_tries >= STALE_IMAGE_TRIES) return; |
| 2126 | 2132 | await this.placementFailed(job.id, `a runner started on ${runner?.image ?? "an older image"} while the deploy of ${want} was rolling out`); |
| 2127 | 2133 | throw new HttpError(409, `this runner's image (${runner?.image ?? "unknown"}) was replaced by ${want}; the job starts again on a new runner`); |
| 2128 | 2134 | } |
| 2129 | 2135 | |
| 2130 | - async runnerSpec(jobId: string, token: string, runner: JobRunnerInfo | null = null): Promise<JobSpec> { | |
| 2136 | + /** `deployed`: the runner image of the Worker that took the request (see refuseStaleRunner). */ | |
| 2137 | + async runnerSpec(jobId: string, token: string, runner: JobRunnerInfo | null = null, deployed: string | null = null): Promise<JobSpec> { | |
| 2131 | 2138 | const job = await this.authJob(jobId, token); |
| 2132 | - await this.refuseStaleRunner(job, runner); | |
| 2139 | + await this.refuseStaleRunner(job, runner, deployed); | |
| 2133 | 2140 | const run = this.runRow(job.run_id); |
| 2134 | 2141 | const exp = await this.expanded(run); |
| 2135 | 2142 | const def: Job = exp.workflow.jobs[job.job_key]!; |
| @@ -2258,13 +2265,14 @@ | ||
| 2258 | 2265 | this.repo.broadcast(`run:${job.run_id}`, { action: "job", jobId }); |
| 2259 | 2266 | } |
| 2260 | 2267 | |
| 2261 | - async runnerComplete(jobId: string, token: string, completion: JobCompletion): Promise<void> { | |
| 2268 | + async runnerComplete(jobId: string, token: string, completion: JobCompletion, deployed: string | null = null): Promise<void> { | |
| 2262 | 2269 | const job = await this.authJob(jobId, token); |
| 2263 | 2270 | if (completion.replaced && completion.conclusion !== "success" && !job.labels) { |
| 2264 | 2271 | // The runner can't tell a rollout's SIGTERM from the platform's own; its image can: on the |
| 2265 | 2272 | // deployed one, no deploy replaced it (the platform stopped or moved its container). |
| 2266 | 2273 | const image = job.runner_info ? (JSON.parse(job.runner_info) as JobRunnerInfo).image : undefined; |
| 2267 | - return this.runnerReplaced(job, image && this.env.RUNNER_IMAGE && image === this.env.RUNNER_IMAGE ? "stopped" : "replaced"); | |
| 2274 | + const want = deployed ?? this.env.RUNNER_IMAGE; | |
| 2275 | + return this.runnerReplaced(job, image && want && image === want ? "stopped" : "replaced"); | |
| 2268 | 2276 | } |
| 2269 | 2277 | await this.completeJob(jobId, completion); |
| 2270 | 2278 | } |
modifiedapps/diffinery/test/unit/actions-runtime.test.ts+22−4View file
| @@ -902,11 +902,11 @@ | ||
| 902 | 902 | const actions = new Actions(repo, env); |
| 903 | 903 | const t = new Date().toISOString(); |
| 904 | 904 | sql.exec("INSERT INTO runs (id, run_number, workflow_path, workflow_name, event, ref, sha, actor, status, created_at, updated_at, event_payload) VALUES ('run1', 1, 'w.yml', 'w', 'push', 'refs/heads/main', 'abc', '{}', 'in_progress', ?, ?, '{}')", t, t); |
| 905 | - sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES ('j1', 'run1', 'k', 'k', 'queued'), ('s1', 'run1', 's', 's', 'waiting')"); | |
| 905 | + sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES ('j1', 'run1', 'k', 'k', 'queued'), ('s1', 'run1', 's', 's', 'waiting'), ('j2', 'run1', 'k2', 'k2', 'queued')"); | |
| 906 | 906 | sql.exec("UPDATE jobs SET labels = '[\"self-hosted\"]' WHERE id = 's1'"); |
| 907 | - const a = actions as unknown as { launch(id: string): Promise<void>; refuseStaleRunner(job: unknown, runner: unknown): Promise<void>; jobRow(id: string): unknown }; | |
| 908 | - const refuse = (id: string, image?: string) => a.refuseStaleRunner(a.jobRow(id), image ? { image } : null); | |
| 909 | - const job = () => sql.exec("SELECT status, placement_tries FROM jobs WHERE id = 'j1'").toArray()[0]; | |
| 907 | + const a = actions as unknown as { launch(id: string): Promise<void>; refuseStaleRunner(job: unknown, runner: unknown, deployed?: string): Promise<void>; jobRow(id: string): unknown }; | |
| 908 | + const refuse = (id: string, image?: string, deployed?: string) => a.refuseStaleRunner(a.jobRow(id), image ? { image } : null, deployed); | |
| 909 | + const job = (id = "j1") => sql.exec("SELECT status, placement_tries FROM jobs WHERE id = ?", id).toArray()[0]; | |
| 910 | 910 | |
| 911 | 911 | await a.launch("j1"); |
| 912 | 912 | // The deployed image runs the job. |
| @@ -924,6 +924,14 @@ | ||
| 924 | 924 | await expect(refuse("j1", "diffinery-runner:old")).resolves.toBeUndefined(); |
| 925 | 925 | // Self-hosted runners run any image. |
| 926 | 926 | await expect(refuse("s1", "my-image")).resolves.toBeUndefined(); |
| 927 | + | |
| 928 | + // Right after a deploy this object can still run the previous version (its RUNNER_IMAGE is the | |
| 929 | + // old one) while the Worker that took the runner's request has the new one: the Worker's wins. | |
| 930 | + await a.launch("j2"); | |
| 931 | + await expect(refuse("j2", "diffinery-runner:next", "diffinery-runner:next")).resolves.toBeUndefined(); | |
| 932 | + expect(job("j2")).toMatchObject({ status: "waiting", placement_tries: 0 }); | |
| 933 | + await expect(refuse("j2", "diffinery-runner:new", "diffinery-runner:next")).rejects.toMatchObject({ status: 409 }); | |
| 934 | + expect(job("j2")).toMatchObject({ status: "queued", placement_tries: 1 }); | |
| 927 | 935 | }); |
| 928 | 936 | |
| 929 | 937 | it("a running job whose runner froze is ended at its timeout plus JOB_DEADLINE_GRACE_MS, and a healthy one is left alone", async () => { |
| @@ -1119,6 +1127,16 @@ | ||
| 1119 | 1127 | expect(job(id)).toMatchObject({ status: "completed", conclusion: "failure", restarts: 0 }); |
| 1120 | 1128 | expect(JSON.parse(job(id)!.failure as string).message).toMatch(/while step "Run pnpm test" ran\. Diffinery doesn't start a job again by itself/); |
| 1121 | 1129 | } |
| 1130 | + // Right after a deploy this object can still have the previous RUNNER_IMAGE: the image of the | |
| 1131 | + // Worker that took the report decides whether a deploy replaced the runner. | |
| 1132 | + sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES ('r6', 'run1', 'r1', 'r6', 'queued'), ('r7', 'run1', 'r1', 'r7', 'queued')"); | |
| 1133 | + for (const [id, deployed, says] of [["r6", "diffinery-runner:next", "The platform stopped this job's container"], ["r7", undefined, "The platform replaced this job's runner with a new version"]] as const) { | |
| 1134 | + await launch(id); | |
| 1135 | + running(id, "diffinery-runner:next", [0, 1], install); | |
| 1136 | + await actions.runnerComplete(id, started.at(-1)!.token, replaced, deployed); | |
| 1137 | + expect(job(id)).toMatchObject({ status: "queued", restarts: 1 }); | |
| 1138 | + expect(logs(id)[0]!.text).toMatch(new RegExp(`^##\\[warning\\]${says} while step "Run pnpm install" ran`)); | |
| 1139 | + } | |
| 1122 | 1140 | |
| 1123 | 1141 | // A job that finished despite the replacement, or a runner on the deployed image that stops, completes as before. |
| 1124 | 1142 | sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES ('j3', 'run1', 'k3', 'k3', 'queued'), ('j4', 'run1', 'k4', 'k4', 'queued')"); |