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.

Hosted jobs that get no container retry; the Live test imports with a GitHub token (#117)

A hosted job whose container gets no machine (its start fails for lack of an instance, or the container stops before the runner reports in) goes back in line and the run's workflow starts it again after 30 s, doubling to 5 minutes, up to 8 times before it fails. The Live test imports with IMPORT_GITHUB_TOKEN, which the sync now copies to Diffinery.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01AizKXSsengXZRegJxRATkL
Canberk Demirci authored · GitHub committed signed1 parent 1d56710commit 3c0691183244dcf9de2421e5ff13026eb7b2d285Browse files
8 files changed+141−12
Files (8)
modified.github/workflows/live-test.yml+8−3View file
@@ -531,16 +531,21 @@
531531 env:
532532 # A small public repository with issues and pull requests; a repository variable can change it.
533533 SOURCE: ${{ vars.LIVE_IMPORT_SOURCE || 'octocat/git-consortium' }}
534+ # A read-only GitHub token (copied here by the sync), so GitHub's rate limit doesn't skip the check. Never printed.
535+ GITHUB_TOKEN_FOR_IMPORT: ${{ secrets.IMPORT_GITHUB_TOKEN }}
534536 run: |
535537 set -euo pipefail
536538 api() { curl -fsS -X "$1" "$DIFFINERY/api/v3$2" -H "Authorization: Bearer $TOKEN" -H "content-type: application/json" ${3:+--data "$3"}; }
537539 name="$REPO-import"
538- # Token-free, so GitHub allows 60 requests an hour per address: a rate limit skips this check with a warning.
540+ body=$(jq -nc --arg s "$SOURCE" --arg o "$OWNER" --arg n "$name" --arg t "${GITHUB_TOKEN_FOR_IMPORT:-}" \
541+ '{source: $s, owner: $o, name: $n, visibility: "private"} + (if $t == "" then {} else {github_token: $t} end)')
542+ if [ -n "${GITHUB_TOKEN_FOR_IMPORT:-}" ]; then echo "importing with IMPORT_GITHUB_TOKEN"; else echo "::warning::IMPORT_GITHUB_TOKEN is not set, so the import runs token-free"; fi
543+ # Token-free, GitHub allows 60 requests an hour per address: a rate limit skips this check with a warning.
539544 # Starting the import reads the repository from GitHub; a GitHub outage (502 "could not reach GitHub")
540545 # is retried, and skips this check with a warning if it outlasts the retries.
541546 for try in 1 2 3 4; do
542547 code=$(curl -sS -o import.json -w '%{http_code}' -X POST "$DIFFINERY/api/v3/diffinery/imports" -H "Authorization: Bearer $TOKEN" -H "content-type: application/json" \
543- --data "{\"source\":\"$SOURCE\",\"owner\":\"$OWNER\",\"name\":\"$name\",\"visibility\":\"private\"}" || true)
548+ --data-binary @- <<<"$body" || true)
544549 unreachable=$(jq -r '.message // "" | startswith("could not reach GitHub")' import.json 2>/dev/null || echo false)
545550 { [ "$code" = 502 ] && [ "$unreachable" = true ]; } || [ "$code" = 000 ] || break
546551 echo "starting the import returned $code ($(jq -r .message import.json 2>/dev/null)); trying again in 30 seconds"
@@ -569,7 +574,7 @@
569574 # The same refs, issue and pull request numbers as on GitHub (when GitHub still answers this runner).
570575 R="/repos/$OWNER/$name"
571576 [ "$(api GET "$R" | jq -r .private)" = true ] || { echo "::error::the imported repository is not private"; exit 1; }
572- gh() { curl -fsS "https://api.github.com/repos/$SOURCE$1" -H "Accept: application/vnd.github+json"; }
577+ gh() { curl -fsS "https://api.github.com/repos/$SOURCE$1" -H "Accept: application/vnd.github+json" ${GITHUB_TOKEN_FOR_IMPORT:+-H "Authorization: Bearer $GITHUB_TOKEN_FOR_IMPORT"}; }
573578 if src=$(gh "") && branches=$(gh "/branches?per_page=100") && items=$(gh "/issues?state=all&per_page=30&sort=created&direction=asc"); then
574579 def=$(jq -r .default_branch <<<"$src")
575580 want=$(jq -r --arg b "$def" '.[] | select(.name == $b) | .commit.sha' <<<"$branches")
modified.github/workflows/sync-to-diffinery.yml+2−1View file
@@ -70,6 +70,7 @@
7070 S_CLOUDFLARE_API_TOKEN: ${{ secrets.CLOUDFLARE_API_TOKEN }}
7171 S_CLOUDFLARE_ACCOUNT_ID: ${{ secrets.CLOUDFLARE_ACCOUNT_ID }}
7272 S_DIFFINERY_LIVE_TOKEN: ${{ secrets.DIFFINERY_LIVE_TOKEN }}
73+ S_IMPORT_GITHUB_TOKEN: ${{ secrets.IMPORT_GITHUB_TOKEN }}
7374 V_ACCESS_TEAM_DOMAIN: ${{ vars.ACCESS_TEAM_DOMAIN }}
7475 V_DIFFINERY_DOMAIN: ${{ vars.DIFFINERY_DOMAIN }}
7576 V_DIFFINERY_ZONE: ${{ vars.DIFFINERY_ZONE }}
@@ -77,7 +78,7 @@
7778 run: |
7879 set -euo pipefail
7980 put() { curl -fsS -o /dev/null -X PUT "$DIFFINERY/api/v3/repos/$ORG/$NAME/actions/$1/$2" -H "Authorization: Bearer $TOKEN" -H "content-type: application/json" --data-binary @-; }
80- for name in CLOUDFLARE_API_TOKEN CLOUDFLARE_ACCOUNT_ID DIFFINERY_LIVE_TOKEN; do
81+ for name in CLOUDFLARE_API_TOKEN CLOUDFLARE_ACCOUNT_ID DIFFINERY_LIVE_TOKEN IMPORT_GITHUB_TOKEN; do
8182 v="S_$name"
8283 if [ -n "${!v}" ]; then jq -n --arg v "${!v}" '{value: $v}' | put secrets "$name"; echo "secret $name copied"; else echo "::warning::secret $name is empty here, so it was not copied"; fi
8384 done
modifiedapps/diffinery/docs/reference/limits.md+1−0View file
@@ -182,6 +182,7 @@
182182 | Hosted runner size | Cloudflare Containers `standard-2` | Each `runs-on: ubuntu-latest` job runs in its own container of this instance type. |
183183 | Hosted job containers at once | 100 | `max_instances` of the runner container class. |
184184 | Hosted jobs at once per repository | 20 | Further jobs stay queued and start, oldest first, as the repository's own jobs finish, so one repository can't take every container. Self-hosted jobs don't count. |
185+| No free container | Wait, then retry, 8 times | When the platform has no container free for a job (many jobs starting at once, or just after a deploy), the job goes back to queued and starts again after 30 seconds, then 1, 2, 4 and 5 minutes. After 8 tries, about half an hour, it fails with `No hosted runner could be started`. Only jobs whose runner never started are retried, so no step runs twice. |
185186 | Unfinished runs per repository | 500 | Past it, a `workflow_dispatch` or `repository_dispatch` is refused with `429 this repository has 500 unfinished workflow runs, the most allowed; wait for some to finish`, and other events start no run. |
186187 | Self-hosted runners per repository | 100, each with at most 20 labels | Registering one more gets `429 this repository has 100 self-hosted runners, the most allowed; remove one first`. A registration token works for an hour. |
187188 | Purging unreachable objects | Repositories of up to 300,000 objects | A larger repository gets `422`; ask a site administrator. See [Removing a secret that was pushed](/docs/security/push-protection#removing-a-secret-that-was-pushed). |
modifiedapps/diffinery/src/repo/RepoDO.ts+3−0View file
@@ -1023,6 +1023,9 @@
10231023 containerStopped(...a: Parameters<Actions["containerStopped"]>) {
10241024 return this.actions.containerStopped(...a);
10251025 }
1026+ relaunchJob(...a: Parameters<Actions["relaunchJob"]>) {
1027+ return this.actions.relaunchJob(...a);
1028+ }
10261029 setSecret(...a: Parameters<Actions["setSecret"]>) {
10271030 return this.actions.setSecret(...a);
10281031 }
modifiedapps/diffinery/src/repo/actions.ts+44−1View file
@@ -46,6 +46,15 @@
4646 export const MAX_WORKFLOW_BYTES = 512 * 1024;
4747 /** Hosted jobs one repository runs at once (GitHub's free plan allows 20 per account); more wait their turn. */
4848 export const MAX_HOSTED_JOBS = 20;
49+/**
50+ * A hosted job whose container never got a machine (the platform had none free, usually when many
51+ * jobs start at once or right after a deploy) goes back in line and starts again later, this many
52+ * times, before it fails. Only jobs whose runner never reported in: nothing of theirs has run.
53+ */
54+export const MAX_PLACEMENT_TRIES = 8;
55+/** Seconds before the next start of a job whose container got no machine: 30 s, doubling to 5 minutes. */
56+export const placementDelay = (tries: number): number => Math.min(30 * 2 ** (tries - 1), 300);
57+const NO_INSTANCE = "no container instance";
4958 /** Runs one repository may have unfinished at once; past it, new ones aren't started. */
5059 export const MAX_ACTIVE_RUNS = 500;
5160 /** Self-hosted runners one repository may have registered. */
@@ -122,6 +131,7 @@
122131 log_chars: number;
123132 fleet_held: number;
124133 runner_id: number | null;
134+ placement_tries: number;
125135 };
126136
127137 export type RunInfo = {
@@ -961,11 +971,42 @@
961971 const stub = this.env.RUNNER.get(this.env.RUNNER.idFromName(container));
962972 await stub.startJob({ repoId: m.repoId, jobId, token, apiUrl: this.env.PUBLIC_URL });
963973 } catch (e) {
964- await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: `Could not start a runner: ${(e as Error).message}` } });
974+ const message = (e as Error).message;
975+ if (message.includes(NO_INSTANCE)) return this.placementFailed(jobId, message);
976+ await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: `Could not start a runner: ${message}` } });
965977 }
966978 }
967979
968980 /**
981+ * The job's container got no machine (its start failed, or it stopped before the runner reported
982+ * in). Put the job back in line and have the run's workflow start it again after a pause, or fail it
983+ * once it has waited MAX_PLACEMENT_TRIES times.
984+ */
985+ private async placementFailed(jobId: string, why: string): Promise<void> {
986+ const job = this.jobRow(jobId);
987+ if (job.status === "completed") return;
988+ const tries = job.placement_tries + 1;
989+ if (tries > MAX_PLACEMENT_TRIES) {
990+ await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: `No hosted runner could be started: Diffinery tried ${MAX_PLACEMENT_TRIES} times over about half an hour (${why}). Re-run the job to try again.` } });
991+ return;
992+ }
993+ this.sql.exec("UPDATE jobs SET status = 'queued', container = NULL, token_hash = NULL, placement_tries = ? WHERE id = ?", tries, jobId);
994+ this.repo.broadcast(`run:${job.run_id}`, { action: "job", jobId });
995+ try {
996+ const instance = await this.env.RUN_WORKFLOW.get(job.run_id);
997+ await instance.sendEvent({ type: `job-${jobId}`, payload: { retryInSeconds: placementDelay(tries) } });
998+ } catch (e) {
999+ console.error("could not ask the run's workflow to start the job again", e);
1000+ await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: `Could not start a runner: ${why}` } });
1001+ }
1002+ }
1003+
1004+ /** The run's workflow starts a job again after placementFailed put it back in line. */
1005+ async relaunchJob(jobId: string): Promise<void> {
1006+ if (this.jobRow(jobId).status === "queued") await this.launch(jobId);
1007+ }
1008+
1009+ /**
9691010 * Job-level `concurrency`, like GitHub: at most one job of a group runs at a time in
9701011 * the repository. With cancel-in-progress the newcomer cancels the group's other jobs;
9711012 * without it the newcomer waits as `pending` and replaces any older pending job.
@@ -1002,6 +1043,8 @@
10021043 async containerStopped(jobId: string, exitCode: number | null): Promise<void> {
10031044 const j = this.sql.exec<JobRow>("SELECT * FROM jobs WHERE id = ?", jobId).toArray()[0];
10041045 if (!j || j.status === "completed") return;
1046+ // Stopped while still waiting for its runner, without an exit code: the container never got a machine.
1047+ if (j.status === "waiting" && !j.labels && !exitCode) return this.placementFailed(jobId, "the container stopped before the runner started");
10051048 await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: exitCode ? `The runner stopped unexpectedly (exit code ${exitCode}).` : "The runner's container stopped before the job finished, without an exit code (the platform restarted or replaced it)." } });
10061049 }
10071050
modifiedapps/diffinery/src/repo/schema.ts+2−0View file
@@ -185,6 +185,8 @@
185185 ["jobs", "fleet_held", "INTEGER NOT NULL DEFAULT 0"],
186186 // The registered self-hosted runner that claimed the job (null for hosted jobs and personal-token claims).
187187 ["jobs", "runner_id", "INTEGER"],
188+ // Times a hosted job's container couldn't get a machine and the job went back in line (repo/actions.ts MAX_PLACEMENT_TRIES).
189+ ["jobs", "placement_tries", "INTEGER NOT NULL DEFAULT 0"],
188190 // The GitHub import that wrote this row (null for everything else); a retry may only overwrite its own rows.
189191 ["issues", "imported", "TEXT"],
190192 ["pulls", "imported", "TEXT"],
modifiedapps/diffinery/src/runner/workflow.ts+18−6View file
@@ -1,6 +1,7 @@
11 import { WorkflowEntrypoint, type WorkflowEvent, type WorkflowStep } from "cloudflare:workers";
22 import type { Env } from "../env.js";
33 import { graphFromLevels, runGraph, type JobNode } from "../repo/job-graph.js";
4+import { MAX_PLACEMENT_TRIES } from "../repo/actions.js";
45
56 export interface RunParams {
67 repoId: string;
@@ -31,12 +32,23 @@
3132 }
3233
3334 const waitFor = async (jobId: string) => {
34- const status = await step.do(`job ${jobId} info`, () => repo().jobStatus(jobId));
35- if (status === "completed") return;
36- try {
37- await step.waitForEvent(`wait ${jobId}`, { type: `job-${jobId}`, timeout: "6 hours" });
38- } catch {
39- await step.do(`timeout ${jobId}`, () => repo().jobTimedOut(jobId));
35+ // A job whose container got no machine comes back with `retryInSeconds`: pause, start it again
36+ // and keep waiting. The first round keeps its step names, so runs already in flight replay.
37+ for (let i = 0; ; i++) {
38+ const n = i ? ` ${i}` : "";
39+ const status = await step.do(`job ${jobId} info${n}`, () => repo().jobStatus(jobId));
40+ if (status === "completed") return;
41+ let event: { payload?: { retryInSeconds?: number } };
42+ try {
43+ event = await step.waitForEvent(`wait ${jobId}${n}`, { type: `job-${jobId}`, timeout: "6 hours" });
44+ } catch {
45+ await step.do(`timeout ${jobId}${n}`, () => repo().jobTimedOut(jobId));
46+ return;
47+ }
48+ const pause = event.payload?.retryInSeconds;
49+ if (!pause || i >= MAX_PLACEMENT_TRIES) return;
50+ await step.sleep(`pause ${jobId} ${i + 1}`, `${pause} seconds`);
51+ await step.do(`relaunch ${jobId} ${i + 1}`, () => repo().relaunchJob(jobId));
4052 }
4153 };
4254 // Runs planned before plans carried a graph replay with their recorded levels.
modifiedapps/diffinery/test/unit/actions-runtime.test.ts+63−1View file
@@ -1,7 +1,7 @@
11 import { describe, expect, it } from "vitest";
22 import { BodyTooLarge, STEP_OPS, batchParts, blockKey, chunkKey, compose, finalKey, listStaged, parseBlockList, parseRange, planCompose, putStream, runComposeStep, stagingPrefix, type BlobBucket, type ComposeLimits, type ComposeStep } from "../../src/lib/blob-store.js";
33 import { DatabaseSync } from "node:sqlite";
4-import { Actions, MAX_ACTIVE_RUNS, MAX_HOSTED_JOBS } from "../../src/repo/actions.js";
4+import { Actions, MAX_ACTIVE_RUNS, MAX_HOSTED_JOBS, MAX_PLACEMENT_TRIES, placementDelay } from "../../src/repo/actions.js";
55 import { DEFAULT_ARTIFACT_QUOTA, Runtime, artifactFileName, artifactObjectKeys } from "../../src/repo/runtime.js";
66 import { REPO_SCHEMA, migrateRepoSchema } from "../../src/repo/schema.js";
77 import type { RepoDO } from "../../src/repo/RepoDO.js";
@@ -679,6 +679,68 @@
679679 expect(started.at(-1)).toBe(`job-j${MAX_HOSTED_JOBS}`);
680680 expect(held()).toEqual([`j${MAX_HOSTED_JOBS + 1}`]);
681681 });
682+
683+ it("a job whose container gets no machine waits and starts again, and fails only after MAX_PLACEMENT_TRIES", async () => {
684+ const sql = sqlStorage();
685+ sql.exec(REPO_SCHEMA);
686+ migrateRepoSchema(sql as unknown as SqlStorage);
687+ let full = true;
688+ const events: { type: string; payload: Record<string, unknown> }[] = [];
689+ const repo = {
690+ sql,
691+ meta: { repoId: "r", owner: "acme", name: "app", defaultBranch: "main" },
692+ upsertCheck: () => {},
693+ broadcast: () => {},
694+ deploys: { jobCompleted: () => {} },
695+ runtime: { sweepJob: async () => {} },
696+ } as unknown as RepoDO;
697+ const env = {
698+ PUBLIC_URL: "https://d.test",
699+ BUCKET: memoryBucket().bucket,
700+ RUNNER: {
701+ idFromName: (n: string) => n,
702+ get: () => ({
703+ startJob: async () => {
704+ if (full) throw new Error("there is no container instance that can be provided to this durable object");
705+ },
706+ stopJob: async () => {},
707+ }),
708+ },
709+ RUN_WORKFLOW: { get: async () => ({ sendEvent: async (e: { type: string; payload: Record<string, unknown> }) => void events.push(e) }) },
710+ } as unknown as Env;
711+ const actions = new Actions(repo, env);
712+ const t = new Date().toISOString();
713+ 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);
714+ sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status) VALUES ('j1', 'run1', 'k', 'k', 'queued'), ('j2', 'run1', 'k2', 'k2', 'queued')");
715+ const launch = (id: string) => (actions as unknown as { launch(id: string): Promise<void> }).launch(id);
716+ const job = (id: string) => sql.exec("SELECT status, conclusion, container, placement_tries FROM jobs WHERE id = ?", id).toArray()[0];
717+
718+ // No machine: the job goes back in line and the run's workflow is asked to start it again after a pause.
719+ await launch("j1");
720+ expect(job("j1")).toMatchObject({ status: "queued", conclusion: null, container: null, placement_tries: 1 });
721+ expect(events).toEqual([{ type: "job-j1", payload: { retryInSeconds: placementDelay(1) } }]);
722+
723+ // Its container stopping before the runner reported in counts the same way; a started job's loss doesn't.
724+ full = false;
725+ await actions.relaunchJob("j1");
726+ expect(job("j1")).toMatchObject({ status: "waiting", container: "job-j1" });
727+ await actions.containerStopped("j1", null);
728+ expect(job("j1")).toMatchObject({ status: "queued", placement_tries: 2 });
729+ expect(events.at(-1)).toEqual({ type: "job-j1", payload: { retryInSeconds: placementDelay(2) } });
730+ await launch("j2");
731+ sql.exec("UPDATE jobs SET status = 'in_progress' WHERE id = 'j2'");
732+ await actions.containerStopped("j2", null);
733+ expect(job("j2")).toMatchObject({ status: "completed", conclusion: "failure" });
734+
735+ // Past the limit it fails, and says why.
736+ full = true;
737+ for (let i = 2; i < MAX_PLACEMENT_TRIES; i++) await actions.relaunchJob("j1");
738+ expect(job("j1")).toMatchObject({ status: "queued", placement_tries: MAX_PLACEMENT_TRIES });
739+ await actions.relaunchJob("j1");
740+ expect(job("j1")).toMatchObject({ status: "completed", conclusion: "failure" });
741+ expect(sql.exec("SELECT failure FROM jobs WHERE id = 'j1'").one().failure).toMatch(/No hosted runner could be started/);
742+ expect([1, 2, 3, 4, 5, 6].map(placementDelay)).toEqual([30, 60, 120, 240, 300, 300]);
743+ });
682744 });
683745
684746 describe("unfinished runs", () => {