fix(api): use multipart for uploads (#912)

* perf: reduce sleep timer duration to 3 minutes * feat: download vm to multipart upload * feat: add abort method * feat: update worker to use multipart * Create grumpy-years-judge.md * fix: use old logic for files smaller than 50MB * fix: adjust sleep counter to 2 minutes but reset per part * fix: status code 201 for put * fix: do not convert body to blob for single uploads * feat: use bun deflate instead * fix: wait for idle * refactor: handle bucket errors in a separate function * perf: use fflate instead * fix: zipping large directories * fix: invalid part format * fix: add blobs to formdata properly * perf: limit multipart to 80mb zips * perf: stream responses instead of buffering * fix: header content disposition * perf: use set to optimise manifest pruning * Create large-tomatoes-cover.md * Update grumpy-years-judge.md
fontsource · Dec 15, 2023 · 247c3af · 247c3af
1 parent 423390e
commit 247c3af
Show file tree

Hide file tree

Showing 11 changed files with 477 additions and 109 deletions.
diff --git a/.changeset/grumpy-years-judge.md b/.changeset/grumpy-years-judge.md
@@ -0,0 +1,6 @@
+---
+"download": minor
+"upload": minor
+---
+
+feat(api): use multipart for uploads
diff --git a/.changeset/large-tomatoes-cover.md b/.changeset/large-tomatoes-cover.md
@@ -0,0 +1,6 @@
+---
+"cdn": patch
+"upload": patch
+---
+
+perf(api): stream responses instead of buffering
diff --git a/api/cdn/src/router.ts b/api/cdn/src/router.ts
@@ -65,20 +65,22 @@ router.get('/fonts/:tag/:file', withParams, async (request, env, ctx) => {
 			? IMMUTABLE_CACHE
 			: STALE_CACHE;
 
-	const headers = {
-		'Cache-Control': cacheControl,
-		'Content-Type': isZip ? 'application/zip' : `font/${extension}`,
-		'Content-Disposition': `attachment; filename="${
+	const headers = new Headers();
+	headers.set('Cache-Control', cacheControl);
+	headers.set('Content-Type', isZip ? 'application/zip' : `font/${extension}`);
+	headers.set(
+		'Content-Disposition',
+		`attachment; filename="${
 			isZip ? `${id}_${version}.zip` : `${id}_${version}_${file}`
 		}"`,
-	};
+	);
 
 	// Check R2 bucket for file
 	const key = isVariable ? `${fullTag}/variable/${file}` : `${fullTag}/${file}`;
 	let item = await env.FONTS.get(key);
 	if (item !== null) {
-		const blob = await item.arrayBuffer();
-		const response = new Response(blob, {
+		headers.set('ETag', item.etag);
+		const response = new Response(item.body, {
 			status: 200,
 			headers,
 		});
@@ -115,8 +117,8 @@ router.get('/fonts/:tag/:file', withParams, async (request, env, ctx) => {
 		item = await updateFile(fullTag, file, env);
 	}
 	if (item !== null) {
-		const blob = await item.arrayBuffer();
-		const response = new Response(blob, {
+		headers.set('ETag', item.etag);
+		const response = new Response(item.body, {
 			status: 200,
 			headers,
 		});
@@ -151,10 +153,9 @@ router.get('/css/:tag/:file', withParams, async (request, env, ctx) => {
 			? IMMUTABLE_CACHE
 			: STALE_CACHE;
 
-	const headers = {
-		'Cache-Control': cacheControl,
-		'Content-Type': 'text/css',
-	};
+	const headers = new Headers();
+	headers.set('Cache-Control', cacheControl);
+	headers.set('Content-Type', 'text/css');
 
 	// Check KV for file
 	const key = isVariable ? `variable:${fullTag}:${file}` : `${fullTag}:${file}`;

diff --git a/api/download/package.json b/api/download/package.json
@@ -20,8 +20,8 @@
 	},
 	"dependencies": {
 		"diary": "^0.4.4",
+		"fflate": "^0.8.1",
 		"itty-router": "^4.0.23",
-		"jszip": "^3.10.1",
 		"p-queue": "^7.4.1",
 		"woff2sfnt-sfnt2woff": "^1.0.0"
 	}

diff --git a/api/download/src/bucket.ts b/api/download/src/bucket.ts
@@ -1,3 +1,4 @@
+import { info } from 'diary';
 import { StatusError } from 'itty-router';
 
 import { type Manifest, type ManifestVariable } from './manifest';
@@ -10,6 +11,11 @@ interface ListBucket {
 	objects: R2Object[];
 }
 
+interface R2UploadedPart {
+	etag: string;
+	partNumber: number;
+}
+
 type BucketPath = Pick<
 	Manifest,
 	'id' | 'subset' | 'weight' | 'style' | 'extension' | 'version'
@@ -38,6 +44,17 @@ export const bucketPathVariable = ({
 }: BucketPathVariable) =>
 	`${id}@${version}/variable/${subset}-${axes}-${style}.woff2`;
 
+const handleBucketError = (resp: Response, msg: string) => {
+	if (resp.status === 401) {
+		throw new StatusError(
+			401,
+			'Unauthorized. Please check your UPLOAD_KEY environment variable.',
+		);
+	}
+
+	throw new StatusError(500, `Internal Server Error. ${msg}`);
+};
+
 export const listBucket = async (prefix: string) => {
 	keepAwake(SLEEP_MINUTES);
 
@@ -48,36 +65,196 @@ export const listBucket = async (prefix: string) => {
 		},
 	});
 	if (!resp.ok) {
+		handleBucketError(resp, 'Unable to list bucket.');
+	}
+
+	return await resp.json<ListBucket>();
+};
+
+const abortMultiPartUpload = async (
+	bucketPath: string,
+	uploadId: string,
+	msg?: string,
+) => {
+	const resp = await fetch(
+		`https://upload.fontsource.org/multipart/${bucketPath}?action=mpu-abort`,
+		{
+			method: 'DELETE',
+			headers: {
+				// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+				Authorization: `Bearer ${process.env.UPLOAD_KEY!}`,
+			},
+			body: JSON.stringify({
+				uploadId,
+			}),
+		},
+	);
+
+	if (!resp.ok) {
+		const error = await resp.text();
+		const errorMsg = `Unable to abort multipart upload. ${error}`;
+		handleBucketError(resp, msg ? `${msg} ${errorMsg}` : errorMsg);
+	}
+};
+
+const initiateMultipartUpload = async (bucketPath: string): Promise<string> => {
+	const resp = await fetch(
+		`https://upload.fontsource.org/multipart/${bucketPath}?action=mpu-create`,
+		{
+			method: 'POST',
+			headers: {
+				// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+				Authorization: `Bearer ${process.env.UPLOAD_KEY!}`,
+			},
+		},
+	);
+
+	if (!resp.ok) {
+		const error = await resp.text();
+		handleBucketError(resp, `Unable to initiate multipart upload. ${error}`);
+	}
+
+	const data = await resp.json();
+	if (!data.uploadId) {
 		throw new StatusError(
 			500,
-			'Internal Server Error. Unable to fetch bucket.',
+			`Internal Server Error. Upload ID is missing. ${JSON.stringify(data)}`,
 		);
 	}
-
-	return await resp.json<ListBucket>();
+	return data.uploadId;
 };
 
-export const putBucket = async (
+const uploadPart = async (
 	bucketPath: string,
-	body: Uint8Array | ArrayBuffer,
+	uploadId: string,
+	partNumber: number,
+	partData: Uint8Array | ArrayBuffer,
 ) => {
 	keepAwake(SLEEP_MINUTES);
+	const formData = new FormData();
+	formData.append('partNumber', String(partNumber));
+	formData.append('uploadId', uploadId);
 
-	const resp = await fetch(`https://upload.fontsource.org/put/${bucketPath}`, {
-		method: 'PUT',
-		headers: {
-			// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
-			Authorization: `Bearer ${process.env.UPLOAD_KEY!}`,
+	const blob = new Blob([partData]);
+	const filename = `${bucketPath}-${partNumber}`;
+	formData.append('file', blob, filename);
+
+	const resp = await fetch(
+		`https://upload.fontsource.org/multipart/${bucketPath}?action=mpu-uploadpart`,
+		{
+			method: 'PUT',
+			headers: {
+				// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+				Authorization: `Bearer ${process.env.UPLOAD_KEY!}`,
+			},
+			body: formData,
 		},
-		body,
-	});
+	);
+
 	if (!resp.ok) {
 		const error = await resp.text();
+		const msg = `Unable to upload part. ${error}`;
+
+		await abortMultiPartUpload(bucketPath, uploadId, msg);
+		handleBucketError(resp, msg);
+	}
+
+	const data = await resp.json();
+	if (!data.etag) {
 		throw new StatusError(
 			500,
-			`Internal Server Error. Unable to upload to bucket. ${error}`,
+			`Internal Server Error. ETag is missing. ${JSON.stringify(data)}`,
+		);
+	}
+	return data.etag;
+};
+
+const completeMultipartUpload = async (
+	bucketPath: string,
+	uploadId: string,
+	parts: R2UploadedPart[],
+) => {
+	const resp = await fetch(
+		`https://upload.fontsource.org/multipart/${bucketPath}?action=mpu-complete`,
+		{
+			method: 'POST',
+			headers: {
+				// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+				Authorization: `Bearer ${process.env.UPLOAD_KEY!}`,
+				'Content-Type': 'application/json',
+			},
+			body: JSON.stringify({
+				uploadId,
+				parts,
+			}),
+		},
+	);
+
+	if (!resp.ok) {
+		const error = await resp.text();
+		const msg = `Unable to complete multipart upload. ${error}`;
+
+		await abortMultiPartUpload(bucketPath, uploadId, msg);
+		handleBucketError(resp, msg);
+	}
+};
+
+export const putBucket = async (
+	bucketPath: string,
+	body: Uint8Array | ArrayBuffer,
+) => {
+	// We only use multipart uploads for files larger than 80MB since Cloudflare
+	// limits the maximum request size to 100MB
+	const partSize = 80 * 1024 * 1024;
+	if (body.byteLength < partSize) {
+		keepAwake(SLEEP_MINUTES);
+		const resp = await fetch(
+			`https://upload.fontsource.org/put/${bucketPath}`,
+			{
+				method: 'PUT',
+				headers: {
+					// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+					Authorization: `Bearer ${process.env.UPLOAD_KEY!}`,
+				},
+				body,
+			},
 		);
+
+		if (!resp.ok) {
+			const error = await resp.text();
+			handleBucketError(resp, `Unable to upload file ${bucketPath}. ${error}`);
+		}
+
+		return;
+	}
+
+	info(`Uploading ${bucketPath} in parts with size ${body.byteLength}`);
+
+	const uploadId = await initiateMultipartUpload(bucketPath);
+
+	const parts: R2UploadedPart[] = [];
+	let offset = 0;
+	let partNumber = 1;
+
+	// Upload buffers in parts
+	while (offset < body.byteLength) {
+		const end = Math.min(offset + partSize, body.byteLength);
+		const partData = body.slice(offset, end);
+		info(`Uploading part ${partNumber} with size ${partData.byteLength}`);
+
+		const etag = await uploadPart(bucketPath, uploadId, partNumber, partData);
+
+		parts.push({
+			etag,
+			partNumber,
+		});
+
+		offset = end;
+		partNumber++;
 	}
+
+	// Complete multipart upload
+	await completeMultipartUpload(bucketPath, uploadId, parts);
 };
 
 export const getBucket = async (bucketPath: string) => {
@@ -90,10 +267,7 @@ export const getBucket = async (bucketPath: string) => {
 		},
 	});
 	if (!resp.ok) {
-		throw new StatusError(
-			500,
-			'Internal Server Error. Unable to fetch bucket.',
-		);
+		handleBucketError(resp, `Unable to fetch ${bucketPath}.`);
 	}
 
 	return resp;