Diffinery

codician/diffinerypublic

Open-source, agent-first GitHub and GitHub Actions alternative that runs on your own Cloudflare account: git hosting, pull requests, code review and drop-in GitHub Actions YAML on Workers, Durable Objects, R2 and Containers.

Restart read-only jobs whose runner a platform rollout replaced (#209)

* Restart read-only jobs whose runner a platform rollout replaced
test-bot authored 1 parent 1eb39eacommit b4c01c57c9d38fa429705b6201ba1a1e6d601454Browse files
4 files changed+118−23
Files (4)
modifiedapps/diffinery/docs/actions/runners.md+1−1View file
@@ -165,7 +165,7 @@
165165 | Matrix size | 256 jobs per matrix | The job fails with "A matrix may produce at most 256 jobs." |
166166 | Log line length | 16,384 characters | Longer lines are cut and end with `…` |
167167 | Log size | 16 MiB of text per job | The log ends with "The log is cut off here: a job's log is limited to 16 MiB of text." Later lines are dropped; the job keeps running. |
168-| Platform updates | A deploy is set up not to replace the container of a running job, so running jobs normally finish on the version they started with. The platform doesn't guarantee it, though. If it replaces a container anyway, its stop signal reaches the running step too. If no step but `actions/checkout` had started, Diffinery starts the job again from the beginning on a new runner, up to twice. Once another step has started, the job fails instead, because running that step twice could repeat what it did (a deploy, a publish). Re-run it from the run page | The log shows "This runner is being replaced by a new version" and, on the step, "The platform stopped this step's processes while replacing the runner with a new version." A restarted job's log starts over with "Diffinery started the job again from the beginning on a new runner (restart N of 2)". A failed one says "The platform replaced this job's runner with a new version while step … ran" |
168+| Platform updates | A deploy is set up not to replace the container of a running job, so running jobs normally finish on the version they started with. The platform doesn't guarantee it, though. If it replaces a container anyway, its stop signal reaches the running step too. Diffinery then starts the job again from the beginning on a new runner, up to twice, if no step but `actions/checkout` had started or if the job can only read: it uses no secrets (other than `GITHUB_TOKEN`), no deployment environment, and its token can't write. What the stopped attempt uploaded (artifacts, caches) is dropped first, as a re-run would. A job that could change something outside the run fails instead once another step has started, because running that step twice could repeat what it did (a deploy, a publish). Re-run it from the run page | The log shows "This runner is being replaced by a new version" and, on the step, "The platform stopped this step's processes while replacing the runner with a new version." A restarted job's log starts over with "Diffinery started the job again from the beginning on a new runner (restart N of 2)". A failed one says "The platform replaced this job's runner with a new version while step … ran" (or, after the third replacement, "… 3 times") |
169169
170170 If the container stops unexpectedly (for example it runs out of memory), the job fails with `The runner stopped unexpectedly (exit code N).` A container that stops before its runner fetched the job is replaced instead, since no step ran.
171171
modifiedapps/diffinery/src/repo/actions.ts+42−13View file
@@ -2187,26 +2187,32 @@
21872187 }
21882188
21892189 /**
2190- * The platform replaced a hosted job's runner while the job ran (a deploy's container rollout).
2191- * Before any step but checkout ran, the job starts again from the beginning on a new runner, as
2192- * if it had just been queued, up to MAX_RUNNER_RESTARTS times. After that, or once a step ran
2193- * whose effects running it again could repeat, the job fails and says why.
2190+ * The platform replaced a hosted job's runner while the job ran (a deploy's container rollout, or
2191+ * the platform moving the container). Its stop signal reaches the running step too, so the job
2192+ * can't finish there. It starts again from the beginning on a new runner, as if it had just been
2193+ * queued, up to MAX_RUNNER_RESTARTS times, when running its steps again can't repeat anything
2194+ * outside the run: no step but checkout had started, or the job could only read (see `readOnly`).
2195+ * Otherwise, or past the limit, the job fails and says why.
21942196 */
21952197 private async runnerReplaced(job: JobRow): Promise<void> {
21962198 const run = this.runRow(job.run_id);
21972199 const steps = JSON.parse(job.steps) as Array<StepReport & { index: number }>;
2198- let defs: Job["steps"] = [];
2200+ let exp: Expanded | null = null;
21992201 try {
2200- defs = (await this.expanded(run)).workflow.jobs[job.job_key]?.steps ?? [];
2202+ exp = await this.expanded(run);
22012203 } catch (e) {
22022204 console.error("could not read the job's steps", e);
22032205 }
2204- const ran = steps.find((s) => s.startedAt && !/^actions\/checkout@/.test(defs[s.index]?.uses ?? ""));
2206+ const defs = exp?.workflow.jobs[job.job_key]?.steps ?? [];
2207+ const ran = steps.filter((s) => s.startedAt && !/^actions\/checkout@/.test(defs[s.index]?.uses ?? ""));
2208+ // The step the platform stopped: the last one that started.
2209+ const stopped = ran.at(-1);
22052210 const restarts = (job.restarts ?? 0) + 1;
2206- if (ran || restarts > MAX_RUNNER_RESTARTS) {
2207- const why = ran
2208- ? `while step "${ran.name}" ran. Diffinery doesn't start a job again by itself once its steps have run, since running them twice could repeat what they did`
2209- : `before the job's steps ran, ${MAX_RUNNER_RESTARTS + 1} times`;
2211+ const repeatable = !stopped || (exp !== null && this.readOnly(run, job, exp));
2212+ if (!repeatable || restarts > MAX_RUNNER_RESTARTS) {
2213+ const why = !repeatable
2214+ ? `while step "${stopped!.name}" ran. Diffinery doesn't start a job again by itself once its steps have run if it could change something outside the run (it uses secrets, a deployment environment, or a token that can write), since running them twice could repeat what they did`
2215+ : `${stopped ? "after its steps had started" : "before the job's steps ran"}, ${MAX_RUNNER_RESTARTS + 1} times`;
22102216 await this.completeJob(job.id, { conclusion: "failure", outputs: {}, failure: { message: `The platform replaced this job's runner with a new version ${why}. Re-run the job.` } });
22112217 return;
22122218 }
@@ -2217,12 +2223,14 @@
22172223 job.id,
22182224 );
22192225 // The log starts over, with a line that says why (its seq keeps counting up, for live readers).
2220- const text = `##[warning]The platform replaced this job's runner with a new version before the job's steps ran. Diffinery started the job again from the beginning on a new runner (restart ${restarts} of ${MAX_RUNNER_RESTARTS}).`;
2226+ const when = stopped ? `while step "${stopped.name}" ran. The job can only read (no secrets, deployment environment or token that can write), so` : "before the job's steps ran.";
2227+ const text = `##[warning]The platform replaced this job's runner with a new version ${when} Diffinery started the job again from the beginning on a new runner (restart ${restarts} of ${MAX_RUNNER_RESTARTS}).`;
22212228 const seq = (this.sql.exec<{ n: number | null }>("SELECT max(seq) AS n FROM logs WHERE job_id = ?", job.id).one().n ?? -1) + 1;
22222229 this.sql.exec("DELETE FROM logs WHERE job_id = ?", job.id);
22232230 this.sql.exec("INSERT INTO logs (job_id, seq, t, step, text) VALUES (?, ?, ?, -1, ?)", job.id, seq, t, text);
22242231 this.sql.exec("UPDATE jobs SET log_chars = ? WHERE id = ?", text.length, job.id);
2225- await this.repo.runtime.sweepJob(job.id).catch((e) => console.error("runtime sweep failed", e));
2232+ // What the stopped attempt uploaded goes too: the new attempt uploads its own, under the same names.
2233+ await this.repo.runtime.sweepJob(job.id, { restart: true }).catch((e) => console.error("runtime sweep failed", e));
22262234 await endCompute(this.env, job.id).catch((e: unknown) => console.error("compute usage failed", e));
22272235 this.repo.upsertCheck({ sha: run.head_sha ?? run.sha, name: this.checkName(run, job), status: "queued", runId: run.id, jobId: job.id, detailsUrl: this.jobUrl(run, job.id) });
22282236 this.repo.broadcast(`job:${job.id}`, { action: "restarted" });
@@ -2236,6 +2244,27 @@
22362244 }
22372245 }
22382246
2247+ /**
2248+ * Whether the job could only read, so running it again from the beginning can't repeat anything
2249+ * outside the run: its steps name no secrets (its token aside), it deploys to no environment, and
2250+ * its token's effective `permissions:` write nothing (an OIDC token counts as a write). What it
2251+ * left in the run itself (log, check, uploads, caches) is reset when it starts again.
2252+ */
2253+ private readOnly(run: RunRow, job: JobRow, exp: Expanded): boolean {
2254+ try {
2255+ const def = exp.workflow.jobs[job.job_key];
2256+ const wf = exp.workflowOf.get(job.job_key);
2257+ if (!def || !wf || def.environment) return false;
2258+ const used = referencedSecrets(def, wf.env);
2259+ if (used === "all" || [...used].some((name) => name !== "GITHUB_TOKEN" && !RUNNER_SECRETS.includes(name))) return false;
2260+ const permissions = permissionsForJob(exp, job.job_key, { untrusted: this.untrusted(run), repoDefault: this.repo.policy.settings().defaultPermissions });
2261+ return Object.values(permissions).every((access) => access !== "write");
2262+ } catch (e) {
2263+ console.error("could not tell whether the job only reads", e);
2264+ return false;
2265+ }
2266+ }
2267+
22392268 private async completeJob(jobId: string, completion: JobCompletion, override?: "timed_out"): Promise<void> {
22402269 const job = this.jobRow(jobId);
22412270 if (job.status === "completed") return;
modifiedapps/diffinery/src/repo/runtime.ts+4−3View file
@@ -304,9 +304,10 @@
304304 * When a job ends, its unfinished uploads go: their rows, so they stop counting
305305 * against the budgets, and their objects.
306306 */
307- async sweepJob(jobId: string): Promise<void> {
308- const caches = this.sql.exec<{ id: number }>("SELECT id FROM rt_caches WHERE job_id = ? AND status = 'reserved'", jobId).toArray().map((r) => r.id);
309- const artifacts = this.sql.exec<{ id: number }>("SELECT id FROM rt_artifacts WHERE job_id = ? AND status = 'uploading'", jobId).toArray().map((r) => r.id);
307+ async sweepJob(jobId: string, o: { restart?: boolean } = {}): Promise<void> {
308+ // A job starting again from the beginning (the platform replaced its runner) drops what it finished uploading too.
309+ const caches = this.sql.exec<{ id: number }>(`SELECT id FROM rt_caches WHERE job_id = ?${o.restart ? "" : " AND status = 'reserved'"}`, jobId).toArray().map((r) => r.id);
310+ const artifacts = this.sql.exec<{ id: number }>(`SELECT id FROM rt_artifacts WHERE job_id = ?${o.restart ? "" : " AND status = 'uploading'"}`, jobId).toArray().map((r) => r.id);
310311 for (const id of caches) this.sql.exec("DELETE FROM rt_caches WHERE id = ?", id);
311312 for (const id of artifacts) this.sql.exec("DELETE FROM rt_artifacts WHERE id = ?", id);
312313 if (caches.length) await this.dropObjects("cache", caches);
modifiedapps/diffinery/test/unit/actions-runtime.test.ts+71−6View file
@@ -532,6 +532,27 @@
532532 await f.rt.expireArtifacts();
533533 expect(f.rt.artifactGet(keep.id, { includeExpired: true })).toBeNull();
534534 });
535+
536+ it("a job starting again from the beginning drops what its stopped attempt uploaded, and nothing else", async () => {
537+ const f = runtimeFixture();
538+ const j = f.job("refs/heads/main");
539+ const other = f.job("refs/heads/main");
540+ const mine = await upload(f, j, "report");
541+ const theirs = await upload(f, other, "report");
542+ await f.save(j, "deps-1");
543+ await f.save(other, "deps-2");
544+ // A job's end keeps what it finished uploading.
545+ await f.rt.sweepJob(j);
546+ expect(f.rt.artifactGet(mine.id)).not.toBeNull();
547+ // Its restart doesn't: the new attempt uploads its own, under the same names.
548+ await f.rt.sweepJob(j, { restart: true });
549+ expect(f.rt.artifactGet(mine.id, { includeExpired: true })).toBeNull();
550+ expect(f.m.objects.has(finalKey({ repo: "r", kind: "artifact", id: mine.id }))).toBe(false);
551+ expect(f.restore(j, "deps-1")).toBeNull();
552+ expect(f.rt.artifactGet(theirs.id)).not.toBeNull();
553+ expect(f.restore(other, "deps-2")).not.toBeNull();
554+ expect((await upload(f, j, "report")).expired).toBe(false);
555+ });
535556 });
536557
537558 describe("artifacts from older runners", () => {
@@ -877,18 +898,20 @@
877898 await expect(refuse("s1", "my-image")).resolves.toBeUndefined();
878899 });
879900
880- it("a job whose runner a rollout replaced starts again on a new runner before its steps ran, and fails after", async () => {
901+ it("a job whose runner a rollout replaced starts again on a new runner before its steps ran or when it can only read, and fails otherwise", async () => {
881902 const sql = sqlStorage();
882903 sql.exec(REPO_SCHEMA);
883904 migrateRepoSchema(sql as unknown as SqlStorage);
884905 const events: { type: string; payload: Record<string, unknown> }[] = [];
906+ const swept: [string, boolean][] = [];
885907 const repo = {
886908 sql,
887909 meta: { repoId: "r", owner: "acme", ownerId: "acme", name: "app", defaultBranch: "main" },
888910 upsertCheck: () => {},
889911 broadcast: () => {},
890912 deploys: { jobCompleted: () => {} },
891- runtime: { sweepJob: async () => {}, grantFor: () => ({ role: "read", permissions: {} }) },
913+ runtime: { sweepJob: async (id: string, o?: { restart?: boolean }) => void swept.push([id, !!o?.restart]), grantFor: () => ({ role: "read", permissions: {} }) },
914+ policy: { settings: () => ({ defaultPermissions: "write" }), isForkPull: () => false },
892915 } as unknown as RepoDO;
893916 const DB = testD1();
894917 DB.sqlite.prepare("INSERT INTO accounts (id, login, kind, status, created_at) VALUES ('acme', 'acme', 'org', 'active', ?)").run(new Date().toISOString());
@@ -903,9 +926,24 @@
903926 } as unknown as Env;
904927 const actions = new Actions(repo, env);
905928 // The job's steps: checkout, then one whose effects running it twice would repeat.
929+ // k to k4 get the repository's default token (read and write); the others only read (r1 hands its read-only token to `gh`), unless they name a secret, deploy to an environment or ask for an OIDC token.
906930 (actions as unknown as { expanded: () => Promise<unknown> }).expanded = async () => {
907931 const steps = [{ uses: "actions/checkout@v4" }, { run: "make deploy" }];
908- return { workflow: { jobs: { k: { steps }, k2: { steps }, k3: { steps } } } };
932+ const read = { contents: "read" };
933+ const install = [{ uses: "actions/checkout@v4" }, { run: "pnpm install" }, { run: "pnpm test" }];
934+ const jobs: Record<string, unknown> = {
935+ k: { steps },
936+ k2: { steps },
937+ k3: { steps },
938+ k4: { steps },
939+ r1: { steps: install, permissions: read, env: { GH_TOKEN: "${{ secrets.GITHUB_TOKEN }}" } },
940+ r2: { steps: [...install, { run: "npm publish", env: { NODE_AUTH_TOKEN: "${{ secrets.NPM_TOKEN }}" } }], permissions: read },
941+ r3: { steps: install, permissions: read, environment: { name: "production" } },
942+ r4: { steps: install, permissions: { contents: "read", "id-token": "write" } },
943+ r5: { steps: [...install, { run: "echo ${{ toJSON(secrets) }}" }], permissions: read },
944+ };
945+ const workflow = { env: {}, jobs };
946+ return { workflow, calls: new Map(), parentOf: new Map(), needs: new Map(), workflowOf: new Map(Object.keys(jobs).map((key) => [key, workflow])) };
909947 };
910948 const t = new Date().toISOString();
911949 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);
@@ -914,8 +952,8 @@
914952 const job = (id: string) => sql.exec("SELECT status, conclusion, container, token_hash, steps, restarts, failure FROM jobs WHERE id = ?", id).toArray()[0];
915953 const logs = (id: string) => sql.exec("SELECT seq, step, text FROM logs WHERE job_id = ? ORDER BY seq", id).toArray();
916954 // The runner fetched the job on `image` and reported these steps as started.
917- const running = (id: string, image: string, steps: number[]) => {
918- const reports = steps.map((index) => ({ index, name: index ? "Deploy" : "Run actions/checkout@v4", status: "in_progress", startedAt: t }));
955+ const running = (id: string, image: string, steps: number[], names = ["Run actions/checkout@v4", "Deploy"]) => {
956+ const reports = steps.map((index) => ({ index, name: names[index] ?? `Step ${index}`, status: "in_progress", startedAt: t }));
919957 sql.exec("UPDATE jobs SET status = 'in_progress', started_at = ?, runner_info = ?, steps = ? WHERE id = ?", t, JSON.stringify({ image }), JSON.stringify(reports), id);
920958 sql.exec("INSERT INTO logs (job_id, seq, t, step, text) VALUES (?, 0, ?, -1, 'set up'), (?, 1, ?, 0, 'checking out')", id, t, id, t);
921959 };
@@ -959,8 +997,35 @@
959997 running("j2", "diffinery-runner:old", [0, 1]);
960998 await actions.runnerComplete("j2", started.at(-1)!.token, replaced);
961999 expect(job("j2")).toMatchObject({ status: "completed", conclusion: "failure", restarts: 0 });
962- expect(JSON.parse(job("j2")!.failure as string).message).toMatch(/while step "Deploy" ran\. Diffinery doesn't start a job again by itself/);
1000+ expect(JSON.parse(job("j2")!.failure as string).message).toMatch(/while step "Deploy" ran\. Diffinery doesn't start a job again by itself once its steps have run if it could change something outside the run/);
9631001
1002+ // A job that can only read starts again even after its steps ran (here the platform stopped
1003+ // `pnpm install`), and what its stopped attempt uploaded goes. Each restart is counted.
1004+ sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES ('r1', 'run1', 'r1', 'r1', 'queued'), ('r2', 'run1', 'r2', 'r2', 'queued'), ('r3', 'run1', 'r3', 'r3', 'queued'), ('r4', 'run1', 'r4', 'r4', 'queued'), ('r5', 'run1', 'r5', 'r5', 'queued')");
1005+ const install = ["Run actions/checkout@v4", "Run pnpm install", "Run pnpm test"];
1006+ await launch("r1");
1007+ running("r1", "diffinery-runner:new", [0, 1], install);
1008+ await actions.runnerComplete("r1", started.at(-1)!.token, replaced);
1009+ expect(job("r1")).toMatchObject({ status: "queued", conclusion: null, steps: "[]", restarts: 1 });
1010+ expect(swept.at(-1)).toEqual(["r1", true]);
1011+ expect(logs("r1")).toEqual([{ seq: 2, step: -1, text: expect.stringMatching(/while step "Run pnpm install" ran\. The job can only read .* Diffinery started the job again from the beginning on a new runner \(restart 1 of 2\)/) }]);
1012+ // It still stops after MAX_RUNNER_RESTARTS, wherever the platform stopped it.
1013+ for (let i = 1; i <= MAX_RUNNER_RESTARTS; i++) {
1014+ await actions.relaunchJob("r1");
1015+ running("r1", "diffinery-runner:new", [0, 1, 2], install);
1016+ await actions.runnerComplete("r1", started.at(-1)!.token, replaced);
1017+ }
1018+ expect(job("r1")).toMatchObject({ status: "completed", conclusion: "failure", restarts: MAX_RUNNER_RESTARTS });
1019+ expect(JSON.parse(job("r1")!.failure as string).message).toMatch(/replaced this job's runner with a new version after its steps had started, 3 times\. Re-run the job\./);
1020+ // One that names a secret, deploys to an environment, can get an OIDC token, or reads every secret fails as before, naming the step that was stopped.
1021+ for (const id of ["r2", "r3", "r4", "r5"]) {
1022+ await launch(id);
1023+ running(id, "diffinery-runner:new", [0, 1, 2], install);
1024+ await actions.runnerComplete(id, started.at(-1)!.token, replaced);
1025+ expect(job(id)).toMatchObject({ status: "completed", conclusion: "failure", restarts: 0 });
1026+ 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/);
1027+ }
1028+
9641029 // A job that finished despite the replacement, or a runner on the deployed image that stops, completes as before.
9651030 sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES ('j3', 'run1', 'k3', 'k3', 'queued'), ('j4', 'run1', 'k4', 'k4', 'queued')");
9661031 await launch("j3");