Allow for multipart uploads

Change-type: minor

config/confd/templates/env.tmpl

@@ -49,6 +49,7 @@ NUM_WORKERS={{if ne (getenv "PRODUCTION_MODE") "true"}}1{{end}}
 PG_SCHEMA=public
 PINEJS_QUEUE_CONCURRENCY={{getenv "PINEJS_QUEUE_CONCURRENCY"}}
 PINEJS_QUEUE_INTERVAL_MS={{getenv "PINEJS_QUEUE_INTERVAL_MS"}}
+PINEJS_WEBRESOURCE_MULTIPART_ENABLED={{getenv "PINEJS_WEBRESOURCE_MULTIPART_ENABLED"}}
 PORT=80
 PRODUCTION_MODE={{getenv "PRODUCTION_MODE"}}
 PROFILING_MODE=false

package.json

@@ -50,9 +50,9 @@
     "@balena/env-parsing": "^1.2.0",
     "@balena/es-version": "^1.0.3",
     "@balena/node-metrics-gatherer": "^6.0.3",
-    "@balena/pinejs": "^20.0.3",
+    "@balena/pinejs": "20.1.0-build-large-file-uploads-2-d7ceca84533fbdb4b7a775e3e94f5de8dcec6fb1-1",
     "@balena/pinejs-webresource-cloudfront": "^1.0.2",
-    "@balena/pinejs-webresource-s3": "^1.0.2",
+    "@balena/pinejs-webresource-s3": "2.0.0-build-new-multiparthandle-interface-89b8ac06df331a867a569ccd702dc90e0ad12959-1",
     "@sentry/node": "^8.48.0",
     "@swc-node/register": "^1.10.9",
     "@types/basic-auth": "^1.1.8",

src/index.ts

@@ -56,6 +56,7 @@ import {
 	ASYNC_TASKS_ENABLED,
 	PINEJS_QUEUE_INTERVAL_MS,
 	PINEJS_QUEUE_CONCURRENCY,
+	PINEJS_WEBRESOURCE_MULTIPART_ENABLED,
 } from './lib/config.js';
 
 import {
@@ -323,6 +324,9 @@ export async function setup(app: Application, options: SetupOptions) {
 		pine.env.db.queryTimeout = DB_QUERY_TIMEOUT;
 	}
 
+	pine.env.webResource.multipartUploadEnabled =
+		PINEJS_WEBRESOURCE_MULTIPART_ENABLED;
+
 	app.disable('x-powered-by');
 
 	app.use('/connectivity-check', (_req, res) => {

src/lib/config.ts

@@ -418,6 +418,11 @@ export const LOGS_WRITE_BUFFER_LIMIT = intVar('LOGS_WRITE_BUFFER_LIMIT', 50);
 export const PINEJS_QUEUE_CONCURRENCY = intVar('PINEJS_QUEUE_CONCURRENCY', 1);
 export let PINEJS_QUEUE_INTERVAL_MS = intVar('PINEJS_QUEUE_INTERVAL_MS', 1000);
 
+export let PINEJS_WEBRESOURCE_MULTIPART_ENABLED = boolVar(
+	'PINEJS_WEBRESOURCE_MULTIPART_ENABLED',
+	true,
+);
+
 export const ASYNC_TASK_ATTEMPT_LIMIT = intVar(
 	'ASYNC_TASK_ATTEMPT_LIMIT',
 	2 ** 31 - 1,
@@ -681,4 +686,10 @@ export const TEST_MOCK_ONLY = {
 		guardTestMockOnly();
 		PINEJS_QUEUE_INTERVAL_MS = v;
 	},
+	set PINEJS_WEBRESOURCE_MULTIPART_ENABLED(
+		v: typeof PINEJS_WEBRESOURCE_MULTIPART_ENABLED,
+	) {
+		guardTestMockOnly();
+		PINEJS_WEBRESOURCE_MULTIPART_ENABLED = v;
+	},
 };

test/23_release_asset.ts

@@ -4,6 +4,7 @@ import * as fixtures from './test-lib/fixtures.js';
 import { supertest } from './test-lib/supertest.js';
 import {
 	checkFileExists,
+	createTempFile,
 	expectEqualBlobs,
 } from './test-lib/fileupload-helper.js';
 import * as versions from './test-lib/versions.js';
@@ -134,6 +135,112 @@ export default () => {
 						})
 						.expect(500); // This should ideally be 4xx
 				});
+
+				it('should be able to upload using multipart uploads', async function () {
+					const res = await supertest(this.user)
+						.post(`/${version}/release_asset`)
+						.field('release', this.release1.id)
+						.field('asset_key', 'unique_key_3')
+						.expect(201);
+
+					expect(res.body).to.have.property('id').that.is.a('number');
+					expect(res.body).to.have.property('asset').that.is.null;
+					expect(res.body)
+						.to.have.nested.property('release.__id')
+						.that.equals(this.release1.id);
+					expect(res.body.asset_key).to.equal('unique_key_3');
+
+					const releaseAssetId = res.body.id;
+					const testFileSize = 6291456;
+					const { tempFilePath, content } = await createTempFile(testFileSize);
+
+					const { body } = await supertest(this.user)
+						.post(`/${version}/release_asset(${releaseAssetId})/beginUpload`)
+						.send({
+							asset: {
+								filename: 'sample.txt',
+								content_type: 'text/plain',
+								size: testFileSize,
+								chunk_size: 6000000,
+							},
+						});
+
+					expect(body).to.have.property('asset').that.is.an('object');
+					expect(body.asset).to.have.property('uuid').that.is.a('string');
+					expect(body.asset.uploadParts).to.be.an('array').that.has.length(2);
+					expect(body.asset.uploadParts[0].chunkSize).to.be.eq(6000000);
+					expect(body.asset.uploadParts[0].partNumber).to.be.eq(1);
+					expect(body.asset.uploadParts[1].chunkSize).to.be.eq(291456);
+					expect(body.asset.uploadParts[1].partNumber).to.be.eq(2);
+
+					const uploadRes = await Promise.all([
+						fetch(body.asset.uploadParts[0].url, {
+							method: 'PUT',
+							body: Uint8Array.prototype.slice.call(content, 0, 6000000),
+						}),
+						fetch(body.asset.uploadParts[1].url, {
+							method: 'PUT',
+							body: Uint8Array.prototype.slice.call(content, 6000000),
+						}),
+					]);
+
+					expect(uploadRes[0].status).to.be.eq(200);
+					expect(uploadRes[0].headers.get('Etag')).to.be.a('string');
+
+					expect(uploadRes[1].status).to.be.eq(200);
+					expect(uploadRes[1].headers.get('Etag')).to.be.a('string');
+
+					const { body: commitResponse } = await supertest(this.user)
+						.post(`/${version}/release_asset(${releaseAssetId})/commitUpload`)
+						.send({
+							uuid: body.asset.uuid,
+							providerCommitData: {
+								Parts: [
+									{
+										PartNumber: 1,
+										ETag: uploadRes[0].headers.get('Etag'),
+									},
+									{
+										PartNumber: 2,
+										ETag: uploadRes[1].headers.get('Etag'),
+									},
+								],
+							},
+						})
+						.expect(200);
+
+					expect(commitResponse).to.have.property('href').that.is.a('string');
+					expect(commitResponse)
+						.to.have.property('filename')
+						.that.is.eq('sample.txt');
+					expect(commitResponse)
+						.to.have.property('size')
+						.that.is.eq(testFileSize);
+
+					const { body: releaseAsset } = await supertest(this.user)
+						.get(
+							`/${version}/release_asset(${releaseAssetId})?$select=id,release,asset&$orderby=release asc`,
+						)
+						.expect(200);
+
+					expect(releaseAsset)
+						.to.have.property('d')
+						.that.is.an('array')
+						.and.has.length(1);
+					expect(releaseAsset)
+						.to.have.nested.property('d[0].id')
+						.that.is.a('number');
+					expect(releaseAsset)
+						.to.have.nested.property('d[0].release.__id')
+						.that.is.a('number');
+					expect(releaseAsset)
+						.to.have.nested.property('d[0].asset.href')
+						.that.is.a('string');
+
+					const href = releaseAsset.d[0].asset.href;
+					expect(await checkFileExists(href, 450)).to.be.true;
+					await expectEqualBlobs(href, tempFilePath);
+				});
 			});
 
 			describe('retrieve release assets', function () {

test/test-lib/fileupload-helper.ts

@@ -1,4 +1,7 @@
-import * as fs from 'fs/promises';
+import fs from 'fs/promises';
+import path from 'path';
+import os from 'os';
+import crypto from 'node:crypto';
 import { requestAsync } from '../../src/infra/request-promise/index.js';
 import { expect } from 'chai';
 
@@ -35,3 +38,10 @@ export async function expectEqualBlobs(url: string, localBlobPath: string) {
 	const diff = originalFile.compare(fileRes);
 	expect(diff).to.equal(0);
 }
+
+export const createTempFile = async (sizeInBytes: number) => {
+	const content = crypto.randomBytes(sizeInBytes);
+	const tempFilePath = path.join(os.tmpdir(), `temp-${Date.now()}.tmp`);
+	await fs.writeFile(tempFilePath, content, 'utf8');
+	return { tempFilePath, content };
+};

test/test-lib/init-tests.ts

@@ -23,6 +23,8 @@ export const preInit = async () => {
 	config.TEST_MOCK_ONLY.ASYNC_TASK_CREATE_SERVICE_INSTALLS_ENABLED = true;
 	config.TEST_MOCK_ONLY.PINEJS_QUEUE_INTERVAL_MS = 100;
 
+	config.TEST_MOCK_ONLY.PINEJS_WEBRESOURCE_MULTIPART_ENABLED = true;
+
 	// override the interval used to emit the queue stats event...
 	const { DeviceOnlineStateManager } = await import(
 		'../../src/features/device-heartbeat/index.js'