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.

Faster cache restores (parallel ranges streamed into tar), Docker started early, faster pnpm installs (#221)

* Cache restores download in parallel ranges and extract as they arrive; Docker starts early for jobs that use it
test-bot authored 1 parent 86c0486commit 3acc94948309ec7d40bdda8fc744c81bae5c23f0Browse files
16 files changed+353−34
Files (16)
modifiedapps/diffinery/docs/actions/caching-and-artifacts.md+3−1View file
@@ -157,6 +157,8 @@
157157
158158 The built-in creates the same archive as `@actions/cache`: a POSIX tar file of the matched paths, stored relative to the workspace, compressed with zstd when `zstd` is installed (it is on hosted runners) and with gzip otherwise. Restoring extracts it to the same paths relative to the workspace.
159159
160+Restoring is built for speed. The runner downloads the archive in 8 MiB ranges, 16 at a time (4 on the 1 GiB `tiny` runner), and pipes them straight into `tar`, so the archive extracts while it is still arriving. On hosted runners a 216 MB pnpm store restores in about 3 to 5 seconds. Every range must come from the same stored entry; if one doesn't, or a range keeps failing after its retries, the log says `Streaming the cache entry failed (<reason>); downloading it in full and extracting it again.` and the runner downloads the whole archive to a file and extracts it again, which replaces anything the stream left half done. If that fails too, the restore removes what it had extracted and behaves like a miss, so a job never runs with half a cache. Artifact downloads use the same ranged download.
161+
160162 - The compression is part of the version, so an entry saved on a runner with zstd doesn't restore on one without it.
161163 - A path outside the workspace, such as `~/.npm`, is stored relative to the workspace too. A runner whose workspace sits at a different depth restores it into a different directory. Include `runner.environment` or a label in the key if you mix runner types.
162164 - The step log shows the archive size as `Cache Size: ~<n> MB (<bytes> B)` on both save and restore.
@@ -233,7 +235,7 @@
233235
234236 Cache service problems are reported as warnings and the job carries on, as on GitHub:
235237
236-- `Failed to restore: <reason>` on restore. The step behaves like a miss.
238+- `Failed to restore: <reason>` on restore. The step behaves like a miss, and nothing the restore extracted before it failed is left behind.
237239 - `Failed to save: <reason>` on save.
238240 - `The cache service is not available to this job; skipping.` when the job has no runtime services.
239241
modifiedapps/diffinery/docs/actions/docker.md+1−1View file
@@ -14,7 +14,7 @@
1414
1515 On hosted runners:
1616
17-- The `docker` command starts the daemon on first use and prints `Starting the Docker daemon`. Jobs that never use Docker don't start it.
17+- The `docker` command starts the daemon on first use and prints `Starting the Docker daemon`. Jobs that never use Docker don't start it. A job that will use Docker (a job `container:`, `services:`, a `docker://` action, or a step whose `uses:` or `run:` names `docker`) gets the daemon started in the background as the job begins, so it is usually ready by the time checkout and setup are done.
1818 - Docker Hub images are pulled through the registry mirror `https://mirror.gcr.io`, which avoids Docker Hub's anonymous pull limits for official images. Images the mirror doesn't have come from Docker Hub directly.
1919 - Containers use the DNS servers `1.1.1.1` and `8.8.8.8`.
2020 - Bridge networks, service names, port publishing and image builds all work, as on GitHub's hosted runners.
modifiedapps/diffinery/docs/actions/runners.md+1−1View file
@@ -155,7 +155,7 @@
155155
156156 Docker works on hosted runners the same way as on GitHub's: job containers, service containers, Docker container actions, and `docker build`, `docker run` and `docker compose` in `run:` steps. Details:
157157
158-- **Started on first use.** The `docker` command is a wrapper that starts the Docker daemon the first time anything in the job needs it (a `docker` command in a step, a job `container:`, `services:`, or a Docker action). The first use prints `Starting the Docker daemon` and takes a few seconds; jobs that never touch Docker don't pay for it. If the daemon can't start, the step fails with `The Docker daemon did not start:` followed by the end of the daemon's log.
158+- **Started on first use.** The `docker` command is a wrapper that starts the Docker daemon the first time anything in the job needs it (a `docker` command in a step, a job `container:`, `services:`, or a Docker action). The first use prints `Starting the Docker daemon` and takes a few seconds; jobs that never touch Docker don't pay for it. A job that will use Docker (a job `container:`, `services:`, a `docker://` action, or a step whose `uses:` or `run:` names `docker`) gets the daemon started in the background as the job begins, so those seconds overlap checkout and setup. If the daemon can't start, the step fails with `The Docker daemon did not start:` followed by the end of the daemon's log.
159159 - **Docker Hub through `mirror.gcr.io`.** The daemon uses `https://mirror.gcr.io` as its registry mirror, so pulls of Docker Hub images (`redis:7-alpine`, `postgres:16`) usually come from Google's public mirror and avoid Docker Hub's anonymous rate limits. Images the mirror doesn't have are pulled from Docker Hub as usual.
160160 - **DNS.** The daemon gives containers the DNS servers `1.1.1.1` and `8.8.8.8`.
161161 - **Pushing to the Diffinery registry.** The daemon and BuildKit builders reach `diffinery.com` through a local relay that splits image layer uploads into chunks, so layers larger than Cloudflare's 100 MB request limit push from hosted workflows. See [Container registry](/docs/packages/container-registry).
modifiedcontainers/runner/src/agent.ts+3−1View file
@@ -26,6 +26,8 @@
2626 * container is the job's own machine; on a shared self-hosted machine the numbers would be the host's).
2727 */
2828 measureUsage?: boolean | UsageSampler;
29+ /** Hosted runners: start the Docker daemon as the job begins when the job will use Docker (docker.ts prewarm). */
30+ prewarmDocker?: boolean;
2931 }
3032
3133 export function apiBaseFor(apiUrl: string | undefined | null, jobId: string): string | null {
@@ -42,7 +44,7 @@
4244 onError: (e) => process.stderr.write(`[diffinery-runner] log upload failed: ${errMsg(e)}\n`),
4345 })
4446 : null;
45- const runner = new JobRunner(spec, api, null, { workRoot: o.workRoot, toolCache: o.toolCache, echoLogs: o.echoLogs ?? !api.online, baseEnv: o.baseEnv }, batcher);
47+ const runner = new JobRunner(spec, api, null, { workRoot: o.workRoot, toolCache: o.toolCache, echoLogs: o.echoLogs ?? !api.online, baseEnv: o.baseEnv, prewarmDocker: o.prewarmDocker }, batcher);
4648 o.onRunner?.(runner);
4749 const onSignal = (sig: string) => {
4850 runner.logger.line(-1, `##[error]Received ${sig}; cancelling the job.`);
modifiedcontainers/runner/src/builtins/cache.ts+55−6View file
@@ -1,6 +1,7 @@
11 import { createHash, randomUUID } from "node:crypto";
2-import { mkdir, rm, stat, writeFile } from "node:fs/promises";
2+import { lstat, mkdir, readFile, rm, rmdir, stat, writeFile } from "node:fs/promises";
33 import path from "node:path";
4+import type { Readable } from "node:stream";
45 import { errMsg, failure, type JobRunner } from "../job.js";
56 import { RuntimeError } from "../runtime.js";
67 import { globPaths, inputList, truthyInput } from "../util.js";
@@ -75,16 +76,37 @@
7576 const exact = isExactKeyMatch(key, hit.matched_key);
7677 if (lookupOnly) return { matchedKey: hit.matched_key, exact };
7778 const dir = path.join(c.job.tempDir, `cache-${randomUUID()}`);
78- const file = path.join(dir, archiveName(compression));
79+ const ws = c.job.workspace;
80+ const indexes = [path.join(dir, "extracted-1"), path.join(dir, "extracted-2")];
81+ const since = Date.now();
82+ const extract = (from: string, index: string, input?: Readable) =>
83+ c.exec("tar", ["-xvf", from, "--index-file", index, "-P", "-C", ws, ...compressArgs(compression, "extract")], { silent: true, capture: true, input });
84+ const tarFailed = (r: { exitCode: number; stderr: string }) => new Error(`tar failed with exit code ${r.exitCode}: ${r.stderr.trim().split("\n").slice(-3).join(" ")}`);
7985 try {
8086 await mkdir(dir, { recursive: true });
87+ await mkdir(ws, { recursive: true });
88+ // Downloaded in ranges, several at once, straight into tar: the archive extracts while it arrives.
89+ let streamError: unknown;
90+ try {
91+ const dl = rt.downloadStream(hit.signed_download_url, { onSize: (size) => c.log(`Cache Size: ~${Math.round(size / 1024 / 1024)} MB (${size} B)`) });
92+ dl.once("error", (e) => (streamError = e));
93+ const r = await extract("-", indexes[0]!, dl);
94+ if (streamError) throw streamError;
95+ if (r.exitCode !== 0) throw tarFailed(r);
96+ return { matchedKey: hit.matched_key, exact };
97+ } catch (e) {
98+ if (c.job.isStopping) throw e;
99+ c.log(`Streaming the cache entry failed (${errMsg(streamError ?? e)}); downloading it in full and extracting it again.`);
100+ }
101+ // Once more from a file, like @actions/cache: extracting the whole archive again replaces anything the stream left half done.
102+ const file = path.join(dir, archiveName(compression));
81103 await rt.download(hit.signed_download_url, file);
82- const size = (await stat(file)).size;
83- c.log(`Cache Size: ~${Math.round(size / 1024 / 1024)} MB (${size} B)`);
84- await mkdir(c.job.workspace, { recursive: true });
85- await c.run("tar", ["-xf", file, "-P", "-C", c.job.workspace, ...compressArgs(compression, "extract")], { silent: true });
104+ const r = await extract(file, indexes[1]!);
105+ if (r.exitCode !== 0) throw tarFailed(r);
86106 return { matchedKey: hit.matched_key, exact };
87107 } catch (e) {
108+ // Nothing half restored stays behind: a job that misses the cache installs from scratch, one with half a cache may not notice.
109+ await undoExtraction(ws, indexes, since).catch((u: unknown) => c.warn(`Could not remove a partly restored cache: ${errMsg(u)}`));
88110 c.warn(`Failed to restore: ${errMsg(e)}`);
89111 return { exact: false };
90112 } finally {
@@ -92,6 +114,33 @@
92114 }
93115 }
94116
117+/**
118+ * Removes what a failed restore extracted: the entries tar listed (its --index-file) that were
119+ * created or replaced since the restore began (their ctime, which tar can't set). Directories go
120+ * only when empty, deepest first; the workspace itself stays.
121+ */
122+export async function undoExtraction(ws: string, indexes: string[], since: number): Promise<void> {
123+ const dirs = new Set<string>();
124+ for (const index of indexes) {
125+ let text: string;
126+ try {
127+ text = await readFile(index, "utf8");
128+ } catch {
129+ continue;
130+ }
131+ for (const name of text.split("\n")) {
132+ if (!name) continue;
133+ const p = path.resolve(ws, name);
134+ if (p === ws || p === path.parse(p).root) continue;
135+ const st = await lstat(p).catch(() => null);
136+ if (!st || st.ctimeMs < since - 1) continue;
137+ if (st.isDirectory()) dirs.add(p);
138+ else await rm(p, { force: true });
139+ }
140+ }
141+ for (const d of [...dirs].sort((a, b) => b.length - a.length)) await rmdir(d).catch(() => {});
142+}
143+
95144 /** Archives `paths` and saves it under `key`. Never throws for service errors. */
96145 export async function saveCache(c: BuiltinContext, paths: string[], key: string): Promise<boolean> {
97146 const rt = runtimeOf(c);
modifiedcontainers/runner/src/builtins/index.ts+3−2View file
@@ -1,4 +1,5 @@
11 import { failure, type JobRunner, type Scope, type StepIO } from "../job.js";
2+import type { Readable } from "node:stream";
23 import type { StepExec } from "../steps/run.js";
34 import type { ExecResult } from "../util.js";
45 import { artifactDownload, artifactUpload } from "./artifacts.js";
@@ -26,9 +27,9 @@
2627 addPath(dir: string): void;
2728 /** Env for child tools (job env + GITHUB_ENV + PATH additions). */
2829 toolEnv(extra?: Record<string, string>): NodeJS.ProcessEnv;
29- exec(cmd: string, args: string[], o?: { cwd?: string; env?: NodeJS.ProcessEnv; capture?: boolean; silent?: boolean; input?: string }): Promise<ExecResult>;
30+ exec(cmd: string, args: string[], o?: { cwd?: string; env?: NodeJS.ProcessEnv; capture?: boolean; silent?: boolean; input?: string | Readable }): Promise<ExecResult>;
3031 /** Like exec but throws on non-zero exit. */
31- run(cmd: string, args: string[], o?: { cwd?: string; env?: NodeJS.ProcessEnv; capture?: boolean; silent?: boolean; input?: string }): Promise<ExecResult>;
32+ run(cmd: string, args: string[], o?: { cwd?: string; env?: NodeJS.ProcessEnv; capture?: boolean; silent?: boolean; input?: string | Readable }): Promise<ExecResult>;
3233 registerPost(name: string, condition: string, fn: (c: BuiltinContext) => Promise<void>): void;
3334 }
3435
modifiedcontainers/runner/src/docker.ts+20−0View file
@@ -1,3 +1,4 @@
1+import { spawn } from "node:child_process";
12 import { createHash } from "node:crypto";
23 import { existsSync } from "node:fs";
34 import { mkdir } from "node:fs/promises";
@@ -57,6 +58,25 @@
5758
5859 // ------------------------------------------------------------ CLI
5960
61+ /**
62+ * Hosted runners start the Docker daemon the first time anything runs `docker` (image/docker), which
63+ * takes 2-5 s. A job that will use Docker starts it as the job begins instead, while checkout and
64+ * setup run; the first real `docker` call then waits only for what is left (the wrapper's lock).
65+ * If this start fails, that first call starts the daemon again and reports why it can't.
66+ */
67+ prewarm(): void {
68+ const env = this.job.hostEnv();
69+ const bin = which("docker", env.PATH);
70+ if (!bin) return;
71+ try {
72+ const child = spawn(bin, ["version"], { env, stdio: "ignore", detached: true });
73+ child.on("error", () => {});
74+ child.unref();
75+ } catch {
76+ /* the first real docker call starts it */
77+ }
78+ }
79+
6080 /** Fails with a clear message when the runner has no usable Docker. */
6181 async ensure(io: StepIO, what: string): Promise<string> {
6282 if (this.bin) return this.bin;
modifiedcontainers/runner/src/job.ts+14−1View file
@@ -3,6 +3,7 @@
33 import { existsSync } from "node:fs";
44 import { mkdir, readFile, writeFile } from "node:fs/promises";
55 import path from "node:path";
6+import type { Readable } from "node:stream";
67 import type { Conclusion, ContainerSpec, JobCompletion, JobSpec, StepReport, StepSpec } from "@diffinery/shared";
78 import { evaluate, evaluateIf, interpolate, type StatusFns } from "@diffinery/workflow-yaml";
89 import { cpuTimes, runnerInfo, type RunnerApi } from "./api.js";
@@ -61,10 +62,21 @@
6162 echoLogs?: boolean;
6263 /** Base env for child processes (defaults to process.env). */
6364 baseEnv?: NodeJS.ProcessEnv;
65+ /** Start the Docker daemon as the job begins when the job will use Docker (hosted runners). */
66+ prewarmDocker?: boolean;
6467 }
6568
6669 export class StepFailure extends Error {}
6770
71+/**
72+ * Whether a job will use Docker: job or service containers, `docker://` actions, or a step that names
73+ * docker (a `run:` script, or an action such as docker/build-push-action). A guess, so only for a head start.
74+ */
75+export function jobUsesDocker(spec: JobSpec): boolean {
76+ if (spec.container?.image || Object.keys(spec.services ?? {}).length) return true;
77+ return spec.steps.some((s) => /\bdocker\b|testcontainers/i.test(`${s.uses ?? ""}\n${s.run ?? ""}`));
78+}
79+
6880 /** The runner's own credentials: never in a step's environment. */
6981 const SENSITIVE_ENV = ["DIFFINERY_JOB_TOKEN", "DIFFINERY_SANDBOX_SECRET", "DIFFINERY_GITHUB_TOKEN"];
7082
@@ -165,6 +177,7 @@
165177 jobTimer.unref();
166178 }
167179 const top: Scope = { steps: {}, inputs: spec.inputs ?? {}, failed: false, inheritedEnv: {}, composite: false, depth: 0 };
180+ if (this.opts.prewarmDocker && jobUsesDocker(spec)) this.docker.prewarm();
168181 try {
169182 await this.setup();
170183 await this.initContainers(top);
@@ -723,7 +736,7 @@
723736 }
724737
725738 /** Runs an internal tool (git, tar, ...) streaming its output into the step log. */
726- async tool(cmd: string, args: string[], o: { io: StepIO; cwd?: string; env?: NodeJS.ProcessEnv; capture?: boolean; silent?: boolean; input?: string }): Promise<ExecResult> {
739+ async tool(cmd: string, args: string[], o: { io: StepIO; cwd?: string; env?: NodeJS.ProcessEnv; capture?: boolean; silent?: boolean; input?: string | Readable }): Promise<ExecResult> {
727740 return exec(cmd, args, {
728741 cwd: o.cwd ?? this.workspace,
729742 env: o.env ?? this.processEnv(this.baseEnvContext({ steps: {}, inputs: {}, failed: false, inheritedEnv: {}, composite: false, depth: 0 })),
modifiedcontainers/runner/src/main.ts+1−0View file
@@ -150,6 +150,7 @@
150150 // two are for cancelling the steps and reporting the result before SIGKILL.
151151 drainOnSigtermMs: args.spec ? undefined : 13 * 60_000,
152152 // A hosted runner is its own machine (Cloudflare sets the Durable Object id in its container).
153+ prewarmDocker: !!env.CLOUDFLARE_DURABLE_OBJECT_ID,
153154 measureUsage: !!env.CLOUDFLARE_DURABLE_OBJECT_ID || env.DIFFINERY_MEASURE_USAGE === "1" ? new UsageSampler({ cores: Number(env.DIFFINERY_RUNNER_VCPU) || undefined }) : false,
154155 });
155156 process.stdout.write(`[diffinery-runner] job finished: ${completion.conclusion}\n`);
modifiedcontainers/runner/src/runtime.ts+111−18View file
@@ -1,5 +1,6 @@
11 import { createWriteStream } from "node:fs";
22 import { open, stat } from "node:fs/promises";
3+import { totalmem } from "node:os";
34 import { Readable } from "node:stream";
45 import { pipeline } from "node:stream/promises";
56 import type { ReadableStream as WebReadableStream } from "node:stream/web";
@@ -29,6 +30,10 @@
2930 const SINGLE_SHOT = 32 * 1024 * 1024;
3031 const BLOCK_SIZE = 32 * 1024 * 1024;
3132 const UPLOAD_CONCURRENCY = 4;
33+/** Downloads come in ranges of this size, this many at a time (measured on hosted runners: 16 of 8 MiB ran 2-5x faster than one stream). */
34+const DOWNLOAD_CHUNK = 8 * 1024 * 1024;
35+// Fewer on the smallest runner (1 GiB), so ranges waiting their turn hold at most 32 MiB there.
36+const DOWNLOAD_CONCURRENCY = totalmem() < 2 * 1024 ** 3 ? 4 : 16;
3237
3338 export class RuntimeClient {
3439 constructor(
@@ -41,7 +46,7 @@
4146 return spec.runtime ? new RuntimeClient(spec.runtime.token, spec.runtime.resultsUrl) : null;
4247 }
4348
44- private async withRetries<T>(what: string, fn: () => Promise<T>): Promise<T> {
49+ async withRetries<T>(what: string, fn: () => Promise<T>): Promise<T> {
4550 const retries = this.opts.retries ?? 5;
4651 let last: unknown;
4752 for (let attempt = 0; attempt <= retries; attempt++) {
@@ -127,25 +132,113 @@
127132 }
128133 }
129134
130- /** Downloads a signed URL to a file. Returns the response's content type and file name. */
135+ /**
136+ * Downloads a signed URL to a file. Large blobs come down in ranges, several at once, which is
137+ * what makes a cache or artifact of a few hundred MB restore in seconds (one stream from R2 through
138+ * the Worker often runs at 20-30 MB/s). Returns the response's content type and file name.
139+ */
131140 async download(url: string, file: string): Promise<{ contentType: string; fileName: string | null }> {
132- return this.withRetries("download", async () => {
133- const res = await fetch(url, { headers: { "user-agent": "diffinery-runner" } });
134- if (!res.ok || !res.body) {
135- const text = await res.text().catch(() => "");
136- throw new RuntimeError(res.status, "download", `download failed (${res.status}): ${text.slice(0, 200)}`);
141+ const out = createWriteStream(file);
142+ const dl = this.downloadStream(url);
143+ await pipeline(dl, out);
144+ return dl.meta!;
145+ }
146+
147+ /**
148+ * The blob at a signed URL as a stream, in order. It asks for the first range; a server that
149+ * answers with ranges (as Diffinery's does) gets the rest DOWNLOAD_CONCURRENCY at a time, each
150+ * range retried on its own, and at most that many ranges are held in memory. A server that
151+ * ignores ranges sends the whole body in one response. Every range must come from the same
152+ * object (size and ETag), or the stream fails rather than mix two versions.
153+ */
154+ downloadStream(url: string, o: { onSize?: (size: number) => void } = {}): Readable & { meta?: { contentType: string; fileName: string | null }; size?: number } {
155+ const abort = new AbortController();
156+ const client = this;
157+ const get = (range: { start: number; end: number } | null) =>
158+ fetch(url, { headers: { "user-agent": "diffinery-runner", ...(range ? { range: `bytes=${range.start}-${range.end}` } : {}) }, signal: abort.signal });
159+ const fail = async (res: Response) => {
160+ const text = await res.text().catch(() => "");
161+ return new RuntimeError(res.status, "download", `download failed (${res.status}): ${text.slice(0, 200)}`);
162+ };
163+ async function* chunks(this: void): AsyncGenerator<Buffer> {
164+ // The first range, or the whole body from a server that doesn't do ranges (or an empty blob: 416).
165+ const first = await client.withRetries("download", async () => {
166+ let res = await get({ start: 0, end: DOWNLOAD_CHUNK - 1 });
167+ if (res.status === 416) {
168+ await res.body?.cancel();
169+ res = await get(null);
170+ }
171+ if (!res.ok || !res.body) throw await fail(res);
172+ return res;
173+ });
174+ stream.meta = { contentType: (first.headers.get("content-type") ?? "").split(";")[0]!.trim().toLowerCase(), fileName: fileNameOf(first.headers.get("content-disposition") ?? "") };
175+ const total = first.status === 206 ? Number(/\/(\d+)$/.exec(first.headers.get("content-range") ?? "")?.[1]) : NaN;
176+ if (first.status !== 206 || !Number.isSafeInteger(total)) {
177+ const declared = first.headers.get("content-length");
178+ // fetch decodes a compressed body, so its length only counts when the body isn't encoded.
179+ const length = declared === null || first.headers.get("content-encoding") ? NaN : Number(declared);
180+ if (Number.isSafeInteger(length)) o.onSize?.(length);
181+ let n = 0;
182+ for await (const c of Readable.fromWeb(first.body as unknown as WebReadableStream)) {
183+ n += (c as Buffer).length;
184+ yield c as Buffer;
185+ }
186+ if (Number.isSafeInteger(length) && n !== length) throw new Error(`download: got ${n} of ${length} bytes`);
187+ stream.size = n;
188+ return;
137189 }
138- await pipeline(Readable.fromWeb(res.body as unknown as WebReadableStream), createWriteStream(file));
139- const cd = res.headers.get("content-disposition") ?? "";
140- const star = /filename\*\s*=\s*UTF-8''([^;\r\n]*)/i.exec(cd)?.[1];
141- const plain = /(?<!\*)filename\s*=\s*"?([^;\r\n"]*)"?/i.exec(cd)?.[1];
142- let fileName: string | null = null;
143- try {
144- fileName = star ? decodeURIComponent(star) : (plain ?? null);
145- } catch {
146- fileName = plain ?? null;
190+ stream.size = total;
191+ o.onSize?.(total);
192+ const etag = first.headers.get("etag");
193+ const ranges: Array<{ start: number; end: number }> = [];
194+ for (let start = 0; start < total; start += DOWNLOAD_CHUNK) ranges.push({ start, end: Math.min(start + DOWNLOAD_CHUNK, total) - 1 });
195+ const body = async (res: Response, r: { start: number; end: number }) => {
196+ const buf = Buffer.from(await res.arrayBuffer());
197+ if (buf.length !== r.end - r.start + 1) throw new Error(`download: got ${buf.length} bytes for bytes ${r.start}-${r.end}`);
198+ return buf;
199+ };
200+ const fetchRange = (r: { start: number; end: number }) =>
201+ client.withRetries("download", async () => {
202+ const res = await get(r);
203+ if (res.status !== 206) {
204+ await res.body?.cancel();
205+ throw res.ok ? new RuntimeError(412, "download", `download failed: the server stopped answering with ranges (${res.status})`) : await fail(res);
206+ }
207+ const m = /^bytes (\d+)-(\d+)\/(\d+)$/.exec(res.headers.get("content-range") ?? "");
208+ if (!m || Number(m[1]) !== r.start || Number(m[2]) !== r.end || Number(m[3]) !== total || (etag && res.headers.get("etag") !== etag)) {
209+ await res.body?.cancel();
210+ throw new RuntimeError(412, "download", "download failed: the blob changed while it was being downloaded");
211+ }
212+ return body(res, r);
213+ });
214+ const pending: Array<Promise<Buffer>> = [body(first, ranges[0]!).catch(() => fetchRange(ranges[0]!))];
215+ let next = 1;
216+ const fill = () => {
217+ while (next < ranges.length && pending.length < DOWNLOAD_CONCURRENCY) {
218+ const p = fetchRange(ranges[next++]!);
219+ p.catch(() => {}); // awaited in order below; this keeps an early failure from going unhandled meanwhile
220+ pending.push(p);
221+ }
222+ };
223+ fill();
224+ while (pending.length) {
225+ const buf = await pending.shift()!;
226+ fill();
227+ yield buf;
147228 }
148- return { contentType: (res.headers.get("content-type") ?? "").split(";")[0]!.trim().toLowerCase(), fileName };
149- });
229+ }
230+ const stream: Readable & { meta?: { contentType: string; fileName: string | null }; size?: number } = Readable.from(chunks(), { objectMode: false });
231+ stream.once("close", () => abort.abort());
232+ return stream;
150233 }
151234 }
235+
236+function fileNameOf(cd: string): string | null {
237+ const star = /filename\*\s*=\s*UTF-8''([^;\r\n]*)/i.exec(cd)?.[1];
238+ const plain = /(?<!\*)filename\s*=\s*"?([^;\r\n"]*)"?/i.exec(cd)?.[1];
239+ try {
240+ return star ? decodeURIComponent(star) : (plain ?? null);
241+ } catch {
242+ return plain ?? null;
243+ }
244+}
modifiedcontainers/runner/src/util.ts+8−2View file
@@ -3,6 +3,8 @@
33 import { existsSync, statSync, accessSync, readdirSync, readFileSync, constants as fsc } from "node:fs";
44 import path from "node:path";
55 import { constants } from "node:os";
6+import type { Readable } from "node:stream";
7+import { pipeline } from "node:stream/promises";
68
79 export const sleep = (ms: number) => new Promise<void>((r) => setTimeout(r, ms));
810
@@ -60,7 +62,8 @@
6062 cwd?: string;
6163 env?: NodeJS.ProcessEnv;
6264 onLine?: (line: string, stream: "stdout" | "stderr") => void;
63- input?: string;
65+ /** Written to the process's stdin: a string, or a stream piped in (destroying the stream ends stdin). */
66+ input?: string | Readable;
6467 timeoutMs?: number;
6568 /** Called with the child so callers can kill it on cancellation. */
6669 onSpawn?: (child: ChildProcess) => void;
@@ -162,9 +165,12 @@
162165 }
163166 se.write(d);
164167 });
165- if (o.input !== undefined) {
168+ if (typeof o.input === "string") {
166169 child.stdin!.on("error", () => {});
167170 child.stdin!.end(o.input);
171+ } else if (o.input) {
172+ // Either side failing stops the other: a failed source ends stdin, an exited process stops the source.
173+ pipeline(o.input, child.stdin!).catch(() => {});
168174 }
169175 let timer: NodeJS.Timeout | undefined;
170176 if (o.timeoutMs && o.timeoutMs > 0) {
modifiedcontainers/runner/test/helpers.ts+25−1View file
@@ -16,6 +16,10 @@
1616 /** Cache entries by key (version in `cacheVersions`). */
1717 cache: Map<string, Buffer>;
1818 cacheVersions: Map<string, string>;
19+ /** The Range header of each blob GET ("" for none). */
20+ blobGets: string[];
21+ /** How many ranged blob GETs past the first range still answer as if the blob had changed (another ETag). */
22+ blobChanges: number;
1923 /** Finalized artifacts of the run, by name. */
2024 artifacts: Map<string, { id: number; body: Buffer; mimeType: string; hash?: string }>;
2125 requests: Array<{ method: string; path: string; auth: string | undefined; runner?: string }>;
@@ -37,6 +41,8 @@
3741 runtimeToken: fakeRuntimeToken(),
3842 cache: new Map(),
3943 cacheVersions: new Map(),
44+ blobGets: [],
45+ blobChanges: 0,
4046 artifacts: new Map(),
4147 requests: [],
4248 failLogs: 0,
@@ -69,7 +75,25 @@
6975 const art = [...api.artifacts.values()].find((a) => String(a.id) === id);
7076 const b = art?.body ?? api.cache.get(id);
7177 if (!b) return json(404, {});
72- res.writeHead(200, { "content-type": art?.mimeType ?? "application/octet-stream" });
78+ api.blobGets.push(String(headers.range ?? ""));
79+ // Ranged reads, like Diffinery's blob store.
80+ const range = /^bytes=(\d+)-(\d+)$/.exec(String(headers.range ?? ""));
81+ const type = { "content-type": art?.mimeType ?? "application/octet-stream", etag: '"e1"', "accept-ranges": "bytes" };
82+ if (range && b.length === 0) {
83+ res.writeHead(416, { "content-range": "bytes */0" });
84+ return res.end();
85+ }
86+ if (range) {
87+ const start = Number(range[1]);
88+ const end = Math.min(Number(range[2]), b.length - 1);
89+ if (start > 0 && api.blobChanges > 0) {
90+ api.blobChanges--;
91+ type.etag = '"e2"';
92+ }
93+ res.writeHead(206, { ...type, "content-range": `bytes ${start}-${end}/${b.length}` });
94+ return res.end(b.subarray(start, end + 1));
95+ }
96+ res.writeHead(200, type);
7397 return res.end(b);
7498 }
7599 if (headers.authorization !== `Bearer ${api.runtimeToken}`) return json(401, { code: "unauthenticated", msg: "bad token" });
modifiedcontainers/runner/test/job.test.ts+45−0View file
@@ -1,4 +1,5 @@
11 import { execFileSync, spawn } from "node:child_process";
2+import { createHash, randomBytes } from "node:crypto";
23 import { createServer } from "node:http";
34 import type { AddressInfo } from "node:net";
45 import { existsSync, mkdirSync, readFileSync, rmSync, writeFileSync } from "node:fs";
@@ -538,6 +539,50 @@
538539 expect([...api.cache.keys()]).toEqual([key]);
539540 });
540541
542+ it("streams a large cache entry in ranges, falls back to a full download, and leaves nothing half restored", async () => {
543+ const steps: StepSpec[] = [
544+ { id: "cache", uses: "actions/cache@v4", with: { path: "deps", key: "big-v1" } },
545+ { run: 'echo "hit=${{ steps.cache.outputs.cache-hit }}"; cat deps/marker; sha256sum deps/big.bin | cut -c1-16' },
546+ ];
547+ // Random data doesn't compress, so the archive spans several 8 MiB ranges.
548+ const big = randomBytes(20 * 1024 * 1024);
549+ const sum = createHash("sha256").update(big).digest("hex").slice(0, 16);
550+ const fill = () => {
551+ mkdirSync(path.join(workspace(), "deps"), { recursive: true });
552+ writeFileSync(path.join(workspace(), "deps", "marker"), "cached");
553+ writeFileSync(path.join(workspace(), "deps", "big.bin"), big);
554+ };
555+ fill();
556+ expect((await run(makeSpec(api, steps, { runId: "save" }))).conclusion, api.logText()).toBe("success");
557+ expect(api.cache.get("big-v1")!.length).toBeGreaterThan(16 * 1024 * 1024);
558+
559+ const again = async (changes: number) => {
560+ rmSync(path.join(workspace(), "deps"), { recursive: true, force: true });
561+ api.logs = [];
562+ api.steps = [];
563+ api.blobGets = [];
564+ api.blobChanges = changes;
565+ return run(makeSpec(api, [steps[0]!, { run: 'set +e; echo "hit=${{ steps.cache.outputs.cache-hit }}"; ls deps 2>&1; cat deps/marker; sha256sum deps/big.bin | cut -c1-16; true' }], { runId: `r${changes}` }));
566+ };
567+ // Streamed: ranges, more than one, all of the same blob.
568+ expect((await again(0)).conclusion, api.logText()).toBe("success");
569+ expect(api.blobGets.length).toBeGreaterThan(2);
570+ expect(api.blobGets.every((r) => r.startsWith("bytes="))).toBe(true);
571+ expect(api.logText(1)).toContain("hit=true");
572+ expect(api.logText(1)).toContain(sum);
573+
574+ // The blob changes under the stream once: the restore starts over from a full download.
575+ expect((await again(1)).conclusion, api.logText()).toBe("success");
576+ expect(api.logText(0)).toMatch(/Streaming the cache entry failed \(download failed: the blob changed while it was being downloaded\); downloading it in full/);
577+ expect(api.logText(1)).toContain(sum);
578+
579+ // It keeps changing: the restore fails, and what the stream extracted is gone again.
580+ expect((await again(1000)).conclusion, api.logText()).toBe("success");
581+ expect(api.logText(0)).toContain("Failed to restore");
582+ expect(api.logText(1)).toMatch(/hit=\n/);
583+ expect(existsSync(path.join(workspace(), "deps"))).toBe(false);
584+ });
585+
541586 it("checks out a repository with actions/checkout", async () => {
542587 const origin = tmp("origin-");
543588 const git = (...args: string[]) => execFileSync("git", args, { cwd: origin, env: { ...process.env, GIT_AUTHOR_NAME: "t", GIT_AUTHOR_EMAIL: "t@t", GIT_COMMITTER_NAME: "t", GIT_COMMITTER_EMAIL: "t@t" } }).toString().trim();
modifiedcontainers/runner/test/runtime.test.ts+43−0View file
@@ -188,4 +188,47 @@
188188 await new Promise<void>((r) => server.close(() => r()));
189189 }
190190 });
191+
192+ it("downloads in ranges, several at once, retrying a range on its own", async () => {
193+ const blob = Buffer.alloc(8 * 1024 * 1024 * 3 + 12345);
194+ for (let i = 0; i < blob.length; i += 997) blob[i] = (i / 997) % 256;
195+ const seen: string[] = [];
196+ let flaky = 1;
197+ let inFlight = 0;
198+ let most = 0;
199+ const server = createServer(async (req, res) => {
200+ const b = req.url!.startsWith("/empty") ? Buffer.alloc(0) : blob;
201+ const range = /^bytes=(\d+)-(\d+)$/.exec(req.headers.range ?? "");
202+ seen.push(req.headers.range ?? "");
203+ inFlight++;
204+ most = Math.max(most, inFlight);
205+ await new Promise((r) => setTimeout(r, 20));
206+ inFlight--;
207+ if (!range) return res.writeHead(200, { "content-length": b.length }).end(b);
208+ if (!b.length) return res.writeHead(416).end();
209+ const start = Number(range[1]);
210+ const end = Math.min(Number(range[2]), b.length - 1);
211+ if (start > 0 && flaky-- > 0) return res.writeHead(503).end("busy");
212+ res.writeHead(206, { etag: '"x"', "content-range": `bytes ${start}-${end}/${b.length}` }).end(b.subarray(start, end + 1));
213+ });
214+ await new Promise<void>((r) => server.listen(0, "127.0.0.1", () => r()));
215+ const base = `http://127.0.0.1:${(server.address() as AddressInfo).port}`;
216+ const d = scratch("ranges-");
217+ try {
218+ const rt = new RuntimeClient("t", `${base}/`, { retries: 2, minBackoffMs: 1 });
219+ await rt.download(`${base}/blob?sig=1`, path.join(d, "back"));
220+ expect(readFileSync(path.join(d, "back")).equals(blob)).toBe(true);
221+ // The first range, three more at once, and the one that failed again.
222+ expect(seen[0]).toBe(`bytes=0-${8 * 1024 * 1024 - 1}`);
223+ expect(seen).toHaveLength(5);
224+ expect(most).toBeGreaterThanOrEqual(3);
225+ // An empty blob: the range is refused (416), so it is fetched whole.
226+ seen.length = 0;
227+ await rt.download(`${base}/empty?sig=1`, path.join(d, "empty"));
228+ expect(readFileSync(path.join(d, "empty")).length).toBe(0);
229+ expect(seen).toEqual([`bytes=0-${8 * 1024 * 1024 - 1}`, ""]);
230+ } finally {
231+ await new Promise<void>((r) => server.close(() => r()));
232+ }
233+ });
191234 });
modifiedcontainers/runner/test/unit.test.ts+15−0View file
@@ -11,6 +11,8 @@
1111 import { resolveNodeVersion } from "../src/builtins/setup-node.js";
1212 import { cloneUrlFor, basicAuthHeader } from "../src/builtins/checkout.js";
1313 import { tmp } from "./helpers.js";
14+import { jobUsesDocker } from "../src/job.js";
15+import type { JobSpec } from "@diffinery/shared";
1416
1517 describe("yaml", () => {
1618 it("parses a typical action.yml", () => {
@@ -238,3 +240,16 @@
238240 expect(cpuTimes("intr 1 2 3\n")).toBeNull();
239241 });
240242 });
243+
244+describe("docker head start", () => {
245+ const spec = (steps: Array<{ run?: string; uses?: string }>, more: Partial<JobSpec> = {}) => ({ steps, ...more }) as JobSpec;
246+ it("starts the daemon early only for jobs that will use Docker", () => {
247+ expect(jobUsesDocker(spec([{ uses: "actions/checkout@v4" }, { run: "docker build -t app ." }]))).toBe(true);
248+ expect(jobUsesDocker(spec([{ uses: "docker/build-push-action@v6" }]))).toBe(true);
249+ expect(jobUsesDocker(spec([{ uses: "docker://alpine:3.20" }]))).toBe(true);
250+ expect(jobUsesDocker(spec([{ run: "npm test" }], { services: { db: { image: "postgres:16" } } as JobSpec["services"] }))).toBe(true);
251+ expect(jobUsesDocker(spec([{ run: "npm test" }], { container: { image: "node:22" } as JobSpec["container"] }))).toBe(true);
252+ expect(jobUsesDocker(spec([{ uses: "actions/checkout@v4" }, { run: "pnpm install && pnpm test" }]))).toBe(false);
253+ expect(jobUsesDocker(spec([{ run: "echo dockerfile-lint" }]))).toBe(false);
254+ });
255+});
modifiedpnpm-workspace.yaml+5−0View file
@@ -2,3 +2,8 @@
22 - apps/*
33 - packages/*
44 - containers/*
5+
6+# pnpm's side-effects cache made each install 8-10 s slower on hosted runners (12 s against 3.4 s
7+# for the same restored store): reusing the cached build of esbuild and workerd cost more than
8+# running their postinstall again. Measured on the speed-bench branch, 2026-10-16.
9+sideEffectsCache: false