fix: resume pinned Drive downloads after transient failures

Author: richiemcilroy

apps/media-server/src/__tests__/lib/media-transfer.test.ts

@@ -115,6 +115,42 @@ describe("bounded revision downloads", () => {
 		expect(received).toBe(content.length);
 		expect(calls).toBe(2);
 	});
+	test.each(["connection", "unavailable"])(
+		"retains downloaded bytes across a %s failure before response headers",
+		async (fault) => {
+			const path = await destination();
+			let calls = 0;
+			let received = 0;
+			await downloadDriveRevision(target(), path, {
+				fetcher: fetcher((_input, init) => {
+					calls++;
+					if (calls === 1)
+						return new Response(content.subarray(0, 12), {
+							headers: { "Content-Length": String(content.length) },
+						});
+					expect(new Headers(init?.headers).get("range")).toBe("bytes=12-");
+					if (calls === 2) {
+						if (fault === "connection")
+							throw new TypeError("Connection failed");
+						return new Response(null, { status: 503 });
+					}
+					return new Response(content.subarray(12), {
+						status: 206,
+						headers: {
+							"Content-Length": String(content.length - 12),
+							"Content-Range": `bytes 12-${content.length - 1}/${content.length}`,
+						},
+					});
+				}),
+				onBytes: (bytes) => {
+					received += bytes;
+				},
+			});
+			expect(await readFile(path)).toEqual(content);
+			expect(received).toBe(content.length);
+			expect(calls).toBe(3);
+		},
+	);
 	test("rejects corruption and removes the incomplete file", async () => {
 		const path = await destination();
 		await expect(

apps/media-server/src/lib/media-transfer.ts

@@ -219,15 +219,32 @@ export async function downloadDriveRevision(
 		for (let attempt = 0; attempt < 3 && bytes < target.size; attempt++) {
 			options.signal?.throwIfAborted();
 			const start = bytes;
-			const response = await fetcher(target.url, {
-				headers: {
-					Authorization: target.authorization,
-					"Accept-Encoding": "identity",
-					...(start ? { Range: `bytes=${start}-` } : {}),
-				},
-				signal: options.signal,
-				redirect: "error",
-			});
+			let response: Response;
+			try {
+				response = await fetcher(target.url, {
+					headers: {
+						Authorization: target.authorization,
+						"Accept-Encoding": "identity",
+						...(start ? { Range: `bytes=${start}-` } : {}),
+					},
+					signal: options.signal,
+					redirect: "error",
+				});
+			} catch (error) {
+				options.signal?.throwIfAborted();
+				if (error instanceof MediaTransferBudgetError || attempt === 2)
+					throw error;
+				await Bun.sleep(250 * 2 ** attempt);
+				continue;
+			}
+			if (
+				[408, 429, 500, 502, 503, 504].includes(response.status) &&
+				attempt < 2
+			) {
+				await response.body?.cancel();
+				await Bun.sleep(250 * 2 ** attempt);
+				continue;
+			}
 			const valid = start
 				? response.status === 206 &&
 					response.headers.get("content-range") ===