Skip to content
Sign up

flagon-io/g1t

public
0
241 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
Merge branch 'worktree-agent-af070177ded5baf00'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});
Merge branch 'worktree-agent-af070177ded5baf00'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 */
A broken-off body is resumed only with a rest that comes uncompressed, as its start did: the rest is joined on byte for byte, so a compressed one would have made a corrupt file.129function rangedUpstream(cuts: (number | null)[], options: { ranges?: boolean; encodedRest?: boolean } = {}) {
Merge branch 'worktree-agent-af070177ded5baf00'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,
A broken-off body is resumed only with a rest that comes uncompressed, as its start did: the rest is joined on byte for byte, so a compressed one would have made a corrupt file.140 headers: { "content-length": String(rest.length), "content-range": `bytes ${start}-${LARGE - 1}/${LARGE}`, "accept-ranges": "bytes", etag: ETAG, ...(options.encodedRest ? { "content-encoding": "gzip" } : {}) },
Merge branch 'worktree-agent-af070177ded5baf00'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
A broken-off body is resumed only with a rest that comes uncompressed, as its start did: the rest is joined on byte for byte, so a compressed one would have made a corrupt file.171test("a rest that comes compressed is never joined on to a body that came as it is", async () => {
172 const { requests, send } = rangedUpstream([3_000_000, null], { encodedRest: true });
173 const response = await forward(tarball(), send, sleep);
174 await assert.rejects(response.arrayBuffer());
175 assert.equal(requests.length, 2);
176});
177
Merge branch 'worktree-agent-af070177ded5baf00'178test("a large tarball sent whole streams as it is, asked for once", async () => {
179 const { requests, send } = rangedUpstream([null]);
180 const response = await forward(tarball(), send, sleep);
181 assert.deepEqual(new Uint8Array(await response.arrayBuffer()), file);
182 assert.equal(requests.length, 1);
183});
184
185test("a body that ends short of its length is resumed too", async () => {
186 const upstream = rangedUpstream([null]);
187 let calls = 0;
188 const send = async (request: Request) => {
189 calls += 1;
190 if (calls > 1) return upstream.send(request);
191 return new Response(breaking(file.slice(0, 1_000_000), null), {
192 headers: { "content-length": String(LARGE), "accept-ranges": "bytes", etag: ETAG },
193 });
194 };
195 const response = await forward(tarball(), send, sleep);
196 assert.deepEqual(new Uint8Array(await response.arrayBuffer()), file);
197 assert.equal(calls, 2);
198 assert.equal(upstream.requests[0].headers.get("range"), "bytes=1000000-");
199});
200
201test("a body that keeps breaking off errors after its resumptions, never handing on a short answer as whole", async () => {
202 const { requests, send } = rangedUpstream([1_000_000, 2_000_000, 3_000_000]);
203 const response = await forward(tarball(), send, sleep);
204 await assert.rejects(response.arrayBuffer());
205 assert.equal(requests.length, 1 + RETRY_WAITS_MS.length);
206});
207
208test("a resumption that fails to connect is tried again", async () => {
209 const upstream = rangedUpstream([2_000_000, null]);
210 let calls = 0;
211 const send = async (request: Request) => {
212 calls += 1;
213 if (calls === 2) throw new Error("Network connection lost.");
214 return upstream.send(request);
215 };
216 const response = await forward(tarball(), send, sleep);
217 assert.deepEqual(new Uint8Array(await response.arrayBuffer()), file);
218 assert.equal(calls, 3);
219});
220
221test("no rest is joined on when the host answers a range with the whole file", async () => {
222 const { requests, send } = rangedUpstream([1_000_000, null], { ranges: false });
223 const response = await forward(tarball(), send, sleep);
224 await assert.rejects(response.arrayBuffer());
225 assert.equal(requests.length, 2);
226});
227
228test("only a whole, plain answer to a GET, from a host that takes ranges, is resumable", () => {
229 const get = new Request("https://registry.npmjs.org/x.tgz");
230 const answer = (headers: Record<string, string>, status = 200) =>
231 new Response("x", { status, headers: { "content-length": "1", "accept-ranges": "bytes", etag: ETAG, ...headers } });
232 assert.deepEqual(resumeFrom(get, answer({})), { validator: ETAG, length: 1 });
233 const modified = "Thu, 16 Apr 2026 23:38:31 GMT";
234 assert.deepEqual(resumeFrom(get, answer({ etag: 'W/"weak"', "last-modified": modified })), { validator: modified, length: 1 });
235 assert.equal(resumeFrom(get, answer({ etag: 'W/"weak"' })), null);
236 assert.equal(resumeFrom(get, answer({ "accept-ranges": "none" })), null);
237 assert.equal(resumeFrom(get, answer({ "content-encoding": "gzip" })), null);
238 assert.equal(resumeFrom(get, answer({}, 206)), null);
239 assert.equal(resumeFrom(new Request(get.url, { headers: { range: "bytes=0-" } }), answer({})), null);
240 assert.equal(resumeFrom(new Request(get.url, { method: "HEAD" }), answer({})), null);
241});

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