Skip to content
Sign up

flagon-io/g1t

public
00

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

g1t.sh
gitagentscloudflarerustforgemcp

Commit

A large download through a cloud runner's egress that breaks off partway is resumed instead of failing the job. npm ci kept failing with ECONNRESET and "network aborted" (17 install steps over three days, one in thirty on g1t's machines, all cloud sandboxes): an answer broke off after its headers had reached the sandbox, and npm never retries a body that stops. Answers up to 8 MB were already read whole and fetched again, but the tarballs that take longest to arrive are larger and streamed as they came: @pagefind/linux-x64 is 50 MB, workerd 38 MB, mermaid 25 MB. Now a GET whose whole 200 answer is sent uncompressed with a known length, from a host that takes byte ranges and names a strong ETag or a Last-Modified date, streams through a body that counts what it has passed on; when the upstream breaks off or ends short, the rest is asked for with Range: bytes=<received>- and If-Range, after the same 250 ms and 1 s waits, and joined on only when the answer is a 206 for exactly that rest, so the sandbox sees one unbroken answer with its Content-Length kept. A host that sends the whole file again, or a file that changed, ends the body as before. Each break and each resumption is one line of the Worker's log (egress body broke off, egress body not resumed), so a reset that the Worker never saw is known to be between it and the sandbox. The Actions guide says how requests to allowed hosts are retried and resumed.

syntaqxcommitted Parent85c083aBrowse files
3 files+286−140/3 viewed
+9−0
588588 allowed domains under **Settings → Guardrails**; a project whose guardrails
589589 turn the network restriction off runs its jobs with an open network.
590590
591+A request to an allowed host that fails on the way is handled for the job.
592+A `GET` or `HEAD` with no body is sent again, twice, after a short wait; one
593+that still fails gets `502` with `x-g1t-egress: upstream-failed` and a body
594+naming the host and the error, which npm, pip and most tools retry. A
595+request with a body is never sent twice. An answer up to 8 MB that breaks
596+off partway is fetched again whole, and a larger one, such as a 50 MB
597+package tarball, is resumed from where it stopped with a `Range` request
598+when its host takes ranges, so the job receives one unbroken answer.
599+
591600 A host only workflows should reach, such as the API a deploy uploads to,
592601 goes in **Workflow-only domains** instead, limited to the workflows and
593602 environments that need it: `api.cloudflare.com | deploy.yml | production`
+135−1
11 import assert from "node:assert/strict";
22 import { test } from "node:test";
33
4−import { BUFFERED_MAX_BYTES, RETRY_WAITS_MS, bufferable, forward } from "./forward.ts";
4+import { BUFFERED_MAX_BYTES, RETRY_WAITS_MS, bufferable, forward, resumeFrom } from "./forward.ts";
55
66 /** A fetch that throws `failures` times, then answers 200; it counts its calls. */
77 function flaky(failures: number) {
9898 assert.equal(bufferable(new Response(new ReadableStream())), false);
9999 assert.equal(bufferable(new Response(null, { status: 304 })), false);
100100 });
101+
102+/** A file larger than is read whole, served by an upstream that takes byte ranges. */
103+const LARGE = BUFFERED_MAX_BYTES + 4096;
104+const file = Uint8Array.from({ length: LARGE }, (_, i) => (i * 7) % 251);
105+const ETAG = '"0136d6b610a8ef0251f65aecb4c9fc24"';
106+
107+/** A body of `bytes` that breaks off after `cut` of them (null: sent whole), in chunks as a network body comes. */
108+function breaking(bytes: Uint8Array, cut: number | null): ReadableStream<Uint8Array> {
109+ const end = cut ?? bytes.length;
110+ let at = 0;
111+ // Pulled a chunk at a time: an error in `start` would drop what was queued before it.
112+ return new ReadableStream<Uint8Array>({
113+ pull(controller) {
114+ if (at < end) {
115+ controller.enqueue(bytes.slice(at, Math.min(at + 65536, end)));
116+ at += 65536;
117+ } else if (cut === null) controller.close();
118+ else controller.error(new Error("aborted"));
119+ },
120+ });
121+}
122+
123+/**
124+ * An upstream for `file`: each answer in turn breaks off after the byte
125+ * of the file given in `cuts` (null: sent whole). A range request with the
126+ * file's ETag gets a `206` with the rest, as registry.npmjs.org answers
127+ * one, unless `ranges` is false.
128+ */
129+function rangedUpstream(cuts: (number | null)[], options: { ranges?: boolean } = {}) {
130+ const requests: Request[] = [];
131+ const send = async (request: Request) => {
132+ requests.push(request);
133+ const cut = cuts[requests.length - 1] ?? null;
134+ const range = /^bytes=(\d+)-$/.exec(request.headers.get("range") ?? "");
135+ if (range && options.ranges !== false && request.headers.get("if-range") === ETAG) {
136+ const start = Number(range[1]);
137+ const rest = file.slice(start);
138+ return new Response(breaking(rest, cut === null ? null : cut - start), {
139+ status: 206,
140+ headers: { "content-length": String(rest.length), "content-range": `bytes ${start}-${LARGE - 1}/${LARGE}`, "accept-ranges": "bytes", etag: ETAG },
141+ });
142+ }
143+ return new Response(breaking(file, cut), {
144+ headers: { "content-length": String(LARGE), "content-type": "application/octet-stream", "accept-ranges": "bytes", etag: ETAG },
145+ });
146+ };
147+ return { requests, send };
148+}
149+
150+const tarball = () =>
151+ new Request("https://registry.npmjs.org/@pagefind/linux-x64/-/linux-x64-1.5.2.tgz", { headers: { "user-agent": "npm/11" } });
152+
153+test("a large tarball that breaks off midway is resumed from where it stopped, and arrives whole", async () => {
154+ waits.length = 0;
155+ const { requests, send } = rangedUpstream([3_000_000, 6_000_000, null]);
156+ const response = await forward(tarball(), send, sleep);
157+ assert.equal(response.status, 200);
158+ assert.equal(response.headers.get("content-length"), String(LARGE));
159+ const body = new Uint8Array(await response.arrayBuffer());
160+ assert.equal(body.length, LARGE);
161+ assert.deepEqual(body, file);
162+ assert.equal(requests.length, 3);
163+ assert.equal(requests[0].headers.get("range"), null);
164+ assert.equal(requests[1].headers.get("range"), "bytes=3000000-");
165+ assert.equal(requests[2].headers.get("range"), "bytes=6000000-");
166+ assert.equal(requests[1].headers.get("if-range"), ETAG);
167+ assert.equal(requests[2].headers.get("user-agent"), "npm/11");
168+ assert.deepEqual(waits, [...RETRY_WAITS_MS]);
169+});
170+
171+test("a large tarball sent whole streams as it is, asked for once", async () => {
172+ const { requests, send } = rangedUpstream([null]);
173+ const response = await forward(tarball(), send, sleep);
174+ assert.deepEqual(new Uint8Array(await response.arrayBuffer()), file);
175+ assert.equal(requests.length, 1);
176+});
177+
178+test("a body that ends short of its length is resumed too", async () => {
179+ const upstream = rangedUpstream([null]);
180+ let calls = 0;
181+ const send = async (request: Request) => {
182+ calls += 1;
183+ if (calls > 1) return upstream.send(request);
184+ return new Response(breaking(file.slice(0, 1_000_000), null), {
185+ headers: { "content-length": String(LARGE), "accept-ranges": "bytes", etag: ETAG },
186+ });
187+ };
188+ const response = await forward(tarball(), send, sleep);
189+ assert.deepEqual(new Uint8Array(await response.arrayBuffer()), file);
190+ assert.equal(calls, 2);
191+ assert.equal(upstream.requests[0].headers.get("range"), "bytes=1000000-");
192+});
193+
194+test("a body that keeps breaking off errors after its resumptions, never handing on a short answer as whole", async () => {
195+ const { requests, send } = rangedUpstream([1_000_000, 2_000_000, 3_000_000]);
196+ const response = await forward(tarball(), send, sleep);
197+ await assert.rejects(response.arrayBuffer());
198+ assert.equal(requests.length, 1 + RETRY_WAITS_MS.length);
199+});
200+
201+test("a resumption that fails to connect is tried again", async () => {
202+ const upstream = rangedUpstream([2_000_000, null]);
203+ let calls = 0;
204+ const send = async (request: Request) => {
205+ calls += 1;
206+ if (calls === 2) throw new Error("Network connection lost.");
207+ return upstream.send(request);
208+ };
209+ const response = await forward(tarball(), send, sleep);
210+ assert.deepEqual(new Uint8Array(await response.arrayBuffer()), file);
211+ assert.equal(calls, 3);
212+});
213+
214+test("no rest is joined on when the host answers a range with the whole file", async () => {
215+ const { requests, send } = rangedUpstream([1_000_000, null], { ranges: false });
216+ const response = await forward(tarball(), send, sleep);
217+ await assert.rejects(response.arrayBuffer());
218+ assert.equal(requests.length, 2);
219+});
220+
221+test("only a whole, plain answer to a GET, from a host that takes ranges, is resumable", () => {
222+ const get = new Request("https://registry.npmjs.org/x.tgz");
223+ const answer = (headers: Record<string, string>, status = 200) =>
224+ new Response("x", { status, headers: { "content-length": "1", "accept-ranges": "bytes", etag: ETAG, ...headers } });
225+ assert.deepEqual(resumeFrom(get, answer({})), { validator: ETAG, length: 1 });
226+ const modified = "Thu, 16 Apr 2026 23:38:31 GMT";
227+ assert.deepEqual(resumeFrom(get, answer({ etag: 'W/"weak"', "last-modified": modified })), { validator: modified, length: 1 });
228+ assert.equal(resumeFrom(get, answer({ etag: 'W/"weak"' })), null);
229+ assert.equal(resumeFrom(get, answer({ "accept-ranges": "none" })), null);
230+ assert.equal(resumeFrom(get, answer({ "content-encoding": "gzip" })), null);
231+ assert.equal(resumeFrom(get, answer({}, 206)), null);
232+ assert.equal(resumeFrom(new Request(get.url, { headers: { range: "bytes=0-" } }), answer({})), null);
233+ assert.equal(resumeFrom(new Request(get.url, { method: "HEAD" }), answer({})), null);
234+});
+142−13
66 * than something to retry. So a request that can be sent again (a GET or
77 * HEAD with no body) is tried twice more after a short wait, and what
88 * still fails comes back as a `502` saying which host and why, which
9− * those tools do retry. It never throws. An answer of known, modest size
10− * to such a request is read whole before it is handed on, so one whose
11− * body breaks off midway is tried again too (`BUFFERED_MAX_BYTES`).
9+ * those tools do retry. It never throws.
10+ *
11+ * An upstream can also break off in the middle of a body, after its
12+ * headers have been handed on: npm then fails with `ECONNRESET` and
13+ * `network aborted`, and never retries it. So an answer of known, modest
14+ * size to such a request is read whole before it is handed on, and tried
15+ * again when it breaks off (`BUFFERED_MAX_BYTES`). A larger one, such as a
16+ * 50 MB package tarball, streams as it comes, and when it breaks off the
17+ * rest is asked for with a `Range` request from where it stopped and
18+ * joined on, so the sandbox sees one unbroken answer (`resuming`).
1219 */
1320
1421 import { normalizeHost } from "./egress.ts";
1522
16−/** The waits before the second and third tries, in milliseconds. */
23+/** The waits before the second and third tries, in milliseconds; also before each resumption of a body. */
1724 export const RETRY_WAITS_MS: readonly number[] = [250, 1000];
1825
1926 /**
2027 * The largest answer to a replayable request that is read whole before the
21− * sandbox is answered. An upstream can also break off in the middle of a
22− * body, after its headers: npm then reports `network aborted` with a
23− * truncated tarball, and by then the answer was already the sandbox's, too
24− * late to try again. An answer read whole first is tried again like any
28+ * sandbox is answered. An answer read whole first is tried again like any
2529 * other failure. Package tarballs and most API answers are well under this;
26− * larger ones, and ones of unknown length, stream as they come.
30+ * larger ones stream, and resume when they break off (`resuming`).
2731 */
2832 export const BUFFERED_MAX_BYTES = 8 * 1024 * 1024;
2933
34+/** The `content-length` of an answer sent as it is (no `content-encoding`), or null. */
35+function plainLength(response: Response): number | null {
36+ const encoding = (response.headers.get("content-encoding") ?? "identity").toLowerCase();
37+ if (encoding !== "identity") return null;
38+ const header = response.headers.get("content-length");
39+ const length = header === null ? NaN : Number(header);
40+ return Number.isSafeInteger(length) && length >= 0 ? length : null;
41+}
42+
3043 /** Whether an answer is to be read whole: a known length within the limit, sent as it is. */
3144 export function bufferable(response: Response): boolean {
3245 if (response.body === null) return false;
33− const length = Number(response.headers.get("content-length") ?? NaN);
34− const encoding = (response.headers.get("content-encoding") ?? "identity").toLowerCase();
35− return Number.isFinite(length) && length <= BUFFERED_MAX_BYTES && encoding === "identity";
46+ const length = plainLength(response);
47+ return length !== null && length <= BUFFERED_MAX_BYTES;
3648 }
3749
3850 /** An answer whose body is read whole; throws when the body breaks off. */
4860 return (request.method === "GET" || request.method === "HEAD") && request.body === null;
4961 }
5062
63+/**
64+ * What a broken-off body can be resumed against: the validator that makes
65+ * sure the rest comes from the same file (a strong `ETag`, else
66+ * `Last-Modified`), and its whole length. Null when it cannot be: only a
67+ * whole `200` to a GET that asked for no range, of known length, sent as it
68+ * is, from a host that takes byte ranges.
69+ */
70+export function resumeFrom(request: Request, response: Response): { validator: string; length: number } | null {
71+ if (request.method !== "GET" || request.headers.has("range") || response.status !== 200 || response.body === null) return null;
72+ if (!(response.headers.get("accept-ranges") ?? "").toLowerCase().split(/[,\s]+/).includes("bytes")) return null;
73+ const length = plainLength(response);
74+ if (length === null || length === 0) return null;
75+ const etag = response.headers.get("etag");
76+ const validator = etag && !etag.startsWith("W/") ? etag : response.headers.get("last-modified");
77+ return validator ? { validator, length } : null;
78+}
79+
80+/** Whether a `206` carries exactly the rest of the file from `offset`. */
81+function restFrom(response: Response, offset: number, length: number): boolean {
82+ if (response.status !== 206 || response.body === null) return false;
83+ const range = /^bytes (\d+)-(\d+)\/(\d+)$/.exec((response.headers.get("content-range") ?? "").trim());
84+ return range !== null && Number(range[1]) === offset && Number(range[2]) === length - 1 && Number(range[3]) === length;
85+}
86+
87+/**
88+ * `response`'s body, joined across breaks: when it breaks off (an error, or
89+ * an end short of its length), the rest is asked for with
90+ * `Range: bytes=<received>-` and `If-Range`, after the waits of
91+ * `RETRY_WAITS_MS`, and read on from there. When the rest cannot be had
92+ * (no `206` for exactly that range, or every try failed), the body errors
93+ * as it would have. Each break is one line of the Worker's log.
94+ */
95+export function resuming(
96+ request: Request,
97+ response: Response,
98+ from: { validator: string; length: number },
99+ send: (request: Request) => Promise<Response>,
100+ sleep: (ms: number) => Promise<void>,
101+): Response {
102+ const host = normalizeHost(new URL(request.url).host);
103+ const { validator, length } = from;
104+ let reader = (response.body as ReadableStream<Uint8Array>).getReader();
105+ let received = 0;
106+ let resumes = 0;
107+
108+ /** A reader for the rest from `received`, or the reason there is none. */
109+ const reopen = async (broke: unknown): Promise<ReadableStreamDefaultReader<Uint8Array>> => {
110+ let failure = broke;
111+ while (resumes < RETRY_WAITS_MS.length) {
112+ const wait = RETRY_WAITS_MS[resumes++];
113+ console.log("egress body broke off", host, `${received} of ${length} bytes`, `resuming, try ${resumes} of ${RETRY_WAITS_MS.length}`, String(failure));
114+ await sleep(wait);
115+ const headers = new Headers(request.headers);
116+ headers.set("range", `bytes=${received}-`);
117+ headers.set("if-range", validator);
118+ let rest: Response;
119+ try {
120+ rest = await send(new Request(request.url, { method: "GET", headers, redirect: request.redirect }));
121+ } catch (error) {
122+ failure = error;
123+ continue;
124+ }
125+ if (restFrom(rest, received, length)) return (rest.body as ReadableStream<Uint8Array>).getReader();
126+ await rest.body?.cancel().catch(() => {});
127+ // The file changed, or the host would not send that range: there is no rest to join on.
128+ const why = `the rest could not be had (${rest.status} ${rest.headers.get("content-range") ?? "without a range"})`;
129+ console.log("egress body not resumed", host, `${received} of ${length} bytes`, why);
130+ throw new Error(`the answer broke off after ${received} of ${length} bytes, and ${why}`);
131+ }
132+ console.log("egress body not resumed", host, `${received} of ${length} bytes`, String(failure));
133+ throw failure instanceof Error ? failure : new Error(String(failure));
134+ };
135+
136+ const body = new ReadableStream<Uint8Array>({
137+ async pull(controller) {
138+ for (;;) {
139+ let chunk: ReadableStreamReadResult<Uint8Array>;
140+ try {
141+ chunk = await reader.read();
142+ if (chunk.done && received < length) throw new Error(`the answer ended after ${received} of ${length} bytes`);
143+ } catch (error) {
144+ try {
145+ reader = await reopen(error);
146+ } catch (final) {
147+ controller.error(final);
148+ return;
149+ }
150+ continue;
151+ }
152+ if (chunk.done) {
153+ controller.close();
154+ return;
155+ }
156+ if (received + chunk.value.byteLength > length) {
157+ controller.error(new Error(`the answer ran past its ${length} bytes`));
158+ return;
159+ }
160+ received += chunk.value.byteLength;
161+ controller.enqueue(chunk.value);
162+ return;
163+ }
164+ },
165+ cancel(reason) {
166+ return reader.cancel(reason);
167+ },
168+ });
169+ // In the Worker, a stream of fixed length keeps the answer's
170+ // `content-length`, so the sandbox can tell a whole answer from a short one.
171+ const Fixed = (globalThis as { FixedLengthStream?: new (length: number) => TransformStream<Uint8Array, Uint8Array> }).FixedLengthStream;
172+ const sent = Fixed ? body.pipeThrough(new Fixed(length)) : body;
173+ return new Response(sent, { status: response.status, statusText: response.statusText, headers: response.headers });
174+}
175+
51176 /** What a request that could not be sent gets back. */
52177 export function upstreamFailed(host: string, error: unknown): Response {
53178 const why = error instanceof Error ? error.message : String(error);
76201 if (attempt > 1) await sleep(RETRY_WAITS_MS[attempt - 2]);
77202 try {
78203 const response = await send(request);
204+ if (tries === 1) return response;
79205 // Read whole first when it can be sent again, so a body that breaks off is retried too.
80− return tries > 1 && bufferable(response) ? await whole(response) : response;
206+ if (bufferable(response)) return await whole(response);
207+ // Too large to hold: streamed, and resumed where it breaks off.
208+ const from = resumeFrom(request, response);
209+ return from ? resuming(request, response, from, send, sleep) : response;
81210 } catch (error) {
82211 failure = error;
83212 console.log("egress upstream failed", normalizeHost(host), request.method, `try ${attempt} of ${tries}`, String(error));

Comments

Sign in to comment.