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
234 linesCodeBlameRaw

Pick any line to see why it is the way it is: the commit, the pull request and issue it came from, and what the agent was thinking.

Merge brokered secrets: a secret a job uses only to call an API can be tied to the hosts it is for, and the job gets a stand-in in its place, which g1t fills in on the way out to those hosts and nowhere else, so the value never enters the sandbox; and a request to an allowed host that fails upstream is retried when it can be, or answered with a 502 that says why, rather than reset.1import assert from "node:assert/strict";
2import { test } from "node:test";
3
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.4import { BUFFERED_MAX_BYTES, RETRY_WAITS_MS, bufferable, forward, resumeFrom } from "./forward.ts";
Merge brokered secrets: a secret a job uses only to call an API can be tied to the hosts it is for, and the job gets a stand-in in its place, which g1t fills in on the way out to those hosts and nowhere else, so the value never enters the sandbox; and a request to an allowed host that fails upstream is retried when it can be, or answered with a 502 that says why, rather than reset.5
6/** A fetch that throws `failures` times, then answers 200; it counts its calls. */
7function flaky(failures: number) {
8 const calls: Request[] = [];
9 const send = async (request: Request) => {
10 calls.push(request);
11 if (calls.length <= failures) throw new Error("Network connection lost.");
12 return new Response("ok");
13 };
14 return { calls, send };
15}
16
17const waits: number[] = [];
18const sleep = async (ms: number) => {
19 waits.push(ms);
20};
21
22test("a GET that fails is tried again, and its answer comes back", async () => {
23 waits.length = 0;
24 const { calls, send } = flaky(2);
25 const response = await forward(new Request("https://registry.npmjs.org/react"), send, sleep);
26 assert.equal(response.status, 200);
27 assert.equal(await response.text(), "ok");
28 assert.equal(calls.length, 3);
29 assert.deepEqual(waits, [...RETRY_WAITS_MS]);
30});
31
32test("one that keeps failing gets a 502 naming the host and why, never a throw", async () => {
33 const { calls, send } = flaky(10);
34 const response = await forward(new Request("https://Registry.npmjs.org:443/react"), send, sleep);
35 assert.equal(calls.length, 3);
36 assert.equal(response.status, 502);
37 assert.equal(response.headers.get("x-g1t-egress"), "upstream-failed");
38 assert.match(response.headers.get("content-type") ?? "", /^text\/plain/);
39 const body = await response.text();
40 assert.match(body, /registry\.npmjs\.org could not be reached/);
41 assert.match(body, /Network connection lost\./);
42});
43
44test("a POST is never sent twice", async () => {
45 const { calls, send } = flaky(1);
46 const response = await forward(new Request("https://api.example.com/upload", { method: "POST", body: "data" }), send, sleep);
47 assert.equal(calls.length, 1);
48 assert.equal(response.status, 502);
49});
50
51test("an answer, even an error, is passed on as it is", async () => {
52 const response = await forward(new Request("https://example.com/"), async () => new Response("no", { status: 503 }), sleep);
53 assert.equal(response.status, 503);
54});
npm installs on g1t's cloud runners stop failing with "network aborted". The upstream sometimes broke off in the middle of a tarball's body, after its headers had already been handed to the sandbox, so the retry added for failed requests couldn't help: npm got a truncated file and gave up. Now an answer to a request that can be sent again (a GET or HEAD with no body), of known length up to 8 MB and sent uncompressed, as package tarballs are, is read whole by the runner Worker before the sandbox gets it; one whose body breaks off is tried again, twice, and still failing becomes a 502 saying why, never a truncated answer. Larger answers and ones of unknown length stream as before. And the repository's installs open at most 8 connections at once rather than npm's 15, since each passes through the Worker and bursts of parallel downloads were when it happened.55
56/** A tarball's answer whose body breaks off after its first chunk, as an upstream that drops mid-body does. */
57function brokenOff(length: number): Response {
58 const body = new ReadableStream<Uint8Array>({
59 start(controller) {
60 controller.enqueue(new Uint8Array(16));
61 controller.error(new Error("aborted"));
62 },
63 });
64 return new Response(body, { headers: { "content-length": String(length), "content-type": "application/octet-stream" } });
65}
66
67test("a body that breaks off midway is tried again, and the whole answer comes back", async () => {
68 let calls = 0;
69 const send = async () => {
70 calls += 1;
71 return calls === 1 ? brokenOff(1024) : new Response(new Uint8Array(1024), { headers: { "content-length": "1024" } });
72 };
73 const response = await forward(new Request("https://registry.npmjs.org/react/-/react-19.0.0.tgz"), send, sleep);
74 assert.equal(calls, 2);
75 assert.equal(response.status, 200);
76 assert.equal((await response.arrayBuffer()).byteLength, 1024);
77});
78
79test("a body that keeps breaking off ends as a 502, not a truncated answer", async () => {
80 let calls = 0;
81 const response = await forward(
82 new Request("https://registry.npmjs.org/x.tgz"),
83 async () => {
84 calls += 1;
85 return brokenOff(1024);
86 },
87 sleep,
88 );
89 assert.equal(calls, 3);
90 assert.equal(response.status, 502);
91});
92
93test("large, compressed or unsized answers stream as they come", () => {
94 const sized = (length: number, headers: Record<string, string> = {}) => new Response("x", { headers: { "content-length": String(length), ...headers } });
95 assert.equal(bufferable(sized(1024)), true);
96 assert.equal(bufferable(sized(BUFFERED_MAX_BYTES + 1)), false);
97 assert.equal(bufferable(sized(1024, { "content-encoding": "gzip" })), false);
98 assert.equal(bufferable(new Response(new ReadableStream())), false);
99 assert.equal(bufferable(new Response(null, { status: 304 })), false);
100});
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.101
102/** A file larger than is read whole, served by an upstream that takes byte ranges. */
103const LARGE = BUFFERED_MAX_BYTES + 4096;
104const file = Uint8Array.from({ length: LARGE }, (_, i) => (i * 7) % 251);
105const 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. */
108function 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 */
129function 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
150const tarball = () =>
151 new Request("https://registry.npmjs.org/@pagefind/linux-x64/-/linux-x64-1.5.2.tgz", { headers: { "user-agent": "npm/11" } });
152
153test("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
171test("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
178test("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
194test("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
201test("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
214test("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
221test("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});

This file's history is long; its oldest lines are credited to the oldest commit read.