| 1 | /** |
| 2 | * How the outbound handler sends a sandbox's request on, for a workflow |
| 3 | * job whose requests pass the Worker, and how a self-hosted job's brokered |
| 4 | * request is relayed. An upstream that resets the connection or fails outright would |
| 5 | * otherwise throw out of the handler, and the sandbox would see its own |
| 6 | * connection reset, which tools such as npm count as a failure rather |
| 7 | * than something to retry. So a request that can be sent again (a GET or |
| 8 | * HEAD with no body) is tried twice more after a short wait, and what |
| 9 | * still fails comes back as a `502` saying which host and why, which |
| 10 | * those tools do retry. It never throws. |
| 11 | * |
| 12 | * An upstream can also break off in the middle of a body, after its |
| 13 | * headers have been handed on: npm then fails with `ECONNRESET` and |
| 14 | * `network aborted`, and never retries it. So an answer of known, modest |
| 15 | * size to such a request is read whole before it is handed on, and tried |
| 16 | * again when it breaks off (`BUFFERED_MAX_BYTES`). A larger one, such as a |
| 17 | * 50 MB package tarball, streams as it comes, and when it breaks off the |
| 18 | * rest is asked for with a `Range` request from where it stopped and |
| 19 | * joined on, so the sandbox sees one unbroken answer (`resuming`). |
| 20 | */ |
| 21 | |
| 22 | import { normalizeHost } from "./egress.ts"; |
| 23 | |
| 24 | /** The waits before the second and third tries, in milliseconds; also before each resumption of a body. */ |
| 25 | export const RETRY_WAITS_MS: readonly number[] = [250, 1000]; |
| 26 | |
| 27 | /** |
| 28 | * The largest answer to a replayable request that is read whole before the |
| 29 | * sandbox is answered. An answer read whole first is tried again like any |
| 30 | * other failure. Package tarballs and most API answers are well under this; |
| 31 | * larger ones stream, and resume when they break off (`resuming`). |
| 32 | */ |
| 33 | export const BUFFERED_MAX_BYTES = 8 * 1024 * 1024; |
| 34 | |
| 35 | /** The `content-length` of an answer sent as it is (no `content-encoding`), or null. */ |
| 36 | function plainLength(response: Response): number | null { |
| 37 | const encoding = (response.headers.get("content-encoding") ?? "identity").toLowerCase(); |
| 38 | if (encoding !== "identity") return null; |
| 39 | const header = response.headers.get("content-length"); |
| 40 | const length = header === null ? NaN : Number(header); |
| 41 | return Number.isSafeInteger(length) && length >= 0 ? length : null; |
| 42 | } |
| 43 | |
| 44 | /** Whether an answer is to be read whole: a known length within the limit, sent as it is. */ |
| 45 | export function bufferable(response: Response): boolean { |
| 46 | if (response.body === null) return false; |
| 47 | const length = plainLength(response); |
| 48 | return length !== null && length <= BUFFERED_MAX_BYTES; |
| 49 | } |
| 50 | |
| 51 | /** An answer whose body is read whole; throws when the body breaks off. */ |
| 52 | async function whole(response: Response): Promise<Response> { |
| 53 | const body = await response.arrayBuffer(); |
| 54 | const expected = Number(response.headers.get("content-length")); |
| 55 | if (body.byteLength !== expected) throw new Error(`the answer broke off after ${body.byteLength} of ${expected} bytes`); |
| 56 | return new Response(body, { status: response.status, statusText: response.statusText, headers: response.headers }); |
| 57 | } |
| 58 | |
| 59 | /** Whether a request can be sent again as it was: no body, nothing changed. */ |
| 60 | export function replayable(request: Request): boolean { |
| 61 | return (request.method === "GET" || request.method === "HEAD") && request.body === null; |
| 62 | } |
| 63 | |
| 64 | /** |
| 65 | * What a broken-off body can be resumed against: the validator that makes |
| 66 | * sure the rest comes from the same file (a strong `ETag`, else |
| 67 | * `Last-Modified`), and its whole length. Null when it cannot be: only a |
| 68 | * whole `200` to a GET that asked for no range, of known length, sent as it |
| 69 | * is, from a host that takes byte ranges. |
| 70 | */ |
| 71 | export function resumeFrom(request: Request, response: Response): { validator: string; length: number } | null { |
| 72 | if (request.method !== "GET" || request.headers.has("range") || response.status !== 200 || response.body === null) return null; |
| 73 | if (!(response.headers.get("accept-ranges") ?? "").toLowerCase().split(/[,\s]+/).includes("bytes")) return null; |
| 74 | const length = plainLength(response); |
| 75 | if (length === null || length === 0) return null; |
| 76 | const etag = response.headers.get("etag"); |
| 77 | const validator = etag && !etag.startsWith("W/") ? etag : response.headers.get("last-modified"); |
| 78 | return validator ? { validator, length } : null; |
| 79 | } |
| 80 | |
| 81 | /** Whether a `206` carries exactly the rest of the file from `offset`. */ |
| 82 | function restFrom(response: Response, offset: number, length: number): boolean { |
| 83 | // The rest is joined on byte for byte, so it must come as it is, like the start did. |
| 84 | const encoding = (response.headers.get("content-encoding") ?? "identity").toLowerCase(); |
| 85 | if (response.status !== 206 || response.body === null || encoding !== "identity") return false; |
| 86 | const range = /^bytes (\d+)-(\d+)\/(\d+)$/.exec((response.headers.get("content-range") ?? "").trim()); |
| 87 | return range !== null && Number(range[1]) === offset && Number(range[2]) === length - 1 && Number(range[3]) === length; |
| 88 | } |
| 89 | |
| 90 | /** |
| 91 | * `response`'s body, joined across breaks: when it breaks off (an error, or |
| 92 | * an end short of its length), the rest is asked for with |
| 93 | * `Range: bytes=<received>-` and `If-Range`, after the waits of |
| 94 | * `RETRY_WAITS_MS`, and read on from there. When the rest cannot be had |
| 95 | * (no `206` for exactly that range, or every try failed), the body errors |
| 96 | * as it would have. Each break is one line of the Worker's log. |
| 97 | */ |
| 98 | export function resuming( |
| 99 | request: Request, |
| 100 | response: Response, |
| 101 | from: { validator: string; length: number }, |
| 102 | send: (request: Request) => Promise<Response>, |
| 103 | sleep: (ms: number) => Promise<void>, |
| 104 | ): Response { |
| 105 | const host = normalizeHost(new URL(request.url).host); |
| 106 | const { validator, length } = from; |
| 107 | let reader = (response.body as ReadableStream<Uint8Array>).getReader(); |
| 108 | let received = 0; |
| 109 | let resumes = 0; |
| 110 | |
| 111 | /** A reader for the rest from `received`, or the reason there is none. */ |
| 112 | const reopen = async (broke: unknown): Promise<ReadableStreamDefaultReader<Uint8Array>> => { |
| 113 | let failure = broke; |
| 114 | while (resumes < RETRY_WAITS_MS.length) { |
| 115 | const wait = RETRY_WAITS_MS[resumes++]; |
| 116 | console.log("egress body broke off", host, `${received} of ${length} bytes`, `resuming, try ${resumes} of ${RETRY_WAITS_MS.length}`, String(failure)); |
| 117 | await sleep(wait); |
| 118 | const headers = new Headers(request.headers); |
| 119 | headers.set("range", `bytes=${received}-`); |
| 120 | headers.set("if-range", validator); |
| 121 | let rest: Response; |
| 122 | try { |
| 123 | rest = await send(new Request(request.url, { method: "GET", headers, redirect: request.redirect })); |
| 124 | } catch (error) { |
| 125 | failure = error; |
| 126 | continue; |
| 127 | } |
| 128 | if (restFrom(rest, received, length)) return (rest.body as ReadableStream<Uint8Array>).getReader(); |
| 129 | await rest.body?.cancel().catch(() => {}); |
| 130 | // The file changed, or the host would not send that range: there is no rest to join on. |
| 131 | const why = `the rest could not be had (${rest.status} ${rest.headers.get("content-range") ?? "without a range"})`; |
| 132 | console.log("egress body not resumed", host, `${received} of ${length} bytes`, why); |
| 133 | throw new Error(`the answer broke off after ${received} of ${length} bytes, and ${why}`); |
| 134 | } |
| 135 | console.log("egress body not resumed", host, `${received} of ${length} bytes`, String(failure)); |
| 136 | throw failure instanceof Error ? failure : new Error(String(failure)); |
| 137 | }; |
| 138 | |
| 139 | const body = new ReadableStream<Uint8Array>({ |
| 140 | async pull(controller) { |
| 141 | for (;;) { |
| 142 | let chunk: ReadableStreamReadResult<Uint8Array>; |
| 143 | try { |
| 144 | chunk = await reader.read(); |
| 145 | if (chunk.done && received < length) throw new Error(`the answer ended after ${received} of ${length} bytes`); |
| 146 | } catch (error) { |
| 147 | try { |
| 148 | reader = await reopen(error); |
| 149 | } catch (final) { |
| 150 | controller.error(final); |
| 151 | return; |
| 152 | } |
| 153 | continue; |
| 154 | } |
| 155 | if (chunk.done) { |
| 156 | controller.close(); |
| 157 | return; |
| 158 | } |
| 159 | if (received + chunk.value.byteLength > length) { |
| 160 | controller.error(new Error(`the answer ran past its ${length} bytes`)); |
| 161 | return; |
| 162 | } |
| 163 | received += chunk.value.byteLength; |
| 164 | controller.enqueue(chunk.value); |
| 165 | return; |
| 166 | } |
| 167 | }, |
| 168 | cancel(reason) { |
| 169 | return reader.cancel(reason); |
| 170 | }, |
| 171 | }); |
| 172 | // In the Worker, a stream of fixed length keeps the answer's |
| 173 | // `content-length`, so the sandbox can tell a whole answer from a short one. |
| 174 | const Fixed = (globalThis as { FixedLengthStream?: new (length: number) => TransformStream<Uint8Array, Uint8Array> }).FixedLengthStream; |
| 175 | const sent = Fixed ? body.pipeThrough(new Fixed(length)) : body; |
| 176 | return new Response(sent, { status: response.status, statusText: response.statusText, headers: response.headers }); |
| 177 | } |
| 178 | |
| 179 | /** What a request that could not be sent gets back. */ |
| 180 | export function upstreamFailed(host: string, error: unknown): Response { |
| 181 | const why = error instanceof Error ? error.message : String(error); |
| 182 | return new Response(`g1t egress: ${normalizeHost(host)} could not be reached: ${why}\n`, { |
| 183 | status: 502, |
| 184 | headers: { "content-type": "text/plain; charset=utf-8", "x-g1t-egress": "upstream-failed" }, |
| 185 | }); |
| 186 | } |
| 187 | |
| 188 | const wait = (ms: number) => new Promise<void>((resolve) => setTimeout(resolve, ms)); |
| 189 | |
| 190 | /** |
| 191 | * Sends `request` with `send` (`fetch`), trying a replayable one again on |
| 192 | * a thrown error, and answering `502` when it cannot. Each failure is one |
| 193 | * line of the Worker's log. |
| 194 | */ |
| 195 | export async function forward( |
| 196 | request: Request, |
| 197 | send: (request: Request) => Promise<Response> = fetch, |
| 198 | sleep: (ms: number) => Promise<void> = wait, |
| 199 | ): Promise<Response> { |
| 200 | const host = new URL(request.url).host; |
| 201 | const tries = replayable(request) ? RETRY_WAITS_MS.length + 1 : 1; |
| 202 | let failure: unknown = null; |
| 203 | for (let attempt = 1; attempt <= tries; attempt++) { |
| 204 | if (attempt > 1) await sleep(RETRY_WAITS_MS[attempt - 2]); |
| 205 | try { |
| 206 | const response = await send(request); |
| 207 | if (tries === 1) return response; |
| 208 | // Read whole first when it can be sent again, so a body that breaks off is retried too. |
| 209 | if (bufferable(response)) return await whole(response); |
| 210 | // Too large to hold: streamed, and resumed where it breaks off. |
| 211 | const from = resumeFrom(request, response); |
| 212 | return from ? resuming(request, response, from, send, sleep) : response; |
| 213 | } catch (error) { |
| 214 | failure = error; |
| 215 | console.log("egress upstream failed", normalizeHost(host), request.method, `try ${attempt} of ${tries}`, String(error)); |
| 216 | } |
| 217 | } |
| 218 | return upstreamFailed(host, failure); |
| 219 | } |