Compare commits

...
7 changed files with 336 additions and 12 deletions

No files matched your search

+49 -1
View File
@@ -3,8 +3,56 @@ import os from 'node:os';
import path from 'node:path';
import fs from 'fs-extra';
import nock from 'nock';
import { computeFileMd5Base64Async } from '../uploads';
import { computeFileMd5Base64Async, uploadWithPresignedPostWithRetryAsync } from '../uploads';
describe(uploadWithPresignedPostWithRetryAsync, () => {
const presignedPost = {
url: 'https://storage.example.com/upload',
fields: { key: 'object-key' },
};
const contents = Array.from({ length: 50000 }, (_, i) => `line ${i}\n`).join('');
let tmpDir: string;
let file: string;
beforeAll(async () => {
tmpDir = await fs.mkdtemp(path.join(os.tmpdir(), 'eas-cli-presigned-post-test-'));
file = path.join(tmpDir, 'bundle.js');
await fs.writeFile(file, contents);
});
afterAll(async () => {
await fs.remove(tmpDir);
});
afterEach(() => {
nock.cleanAll();
});
it('sends the whole file and reports activity while it uploads', async () => {
let body = '';
nock('https://storage.example.com')
.post('/upload', requestBody => {
body = requestBody;
return true;
})
.reply(204);
const onAssetUploadBegin = jest.fn();
await uploadWithPresignedPostWithRetryAsync(file, presignedPost, onAssetUploadBegin);
expect(body).toContain('filename="bundle.js"');
expect(body).toContain(contents);
expect(onAssetUploadBegin.mock.calls.length).toBeGreaterThan(2);
expect(onAssetUploadBegin).toHaveBeenCalledWith(
expect.objectContaining({ file, fileSize: contents.length, uploadedBytes: 0 })
);
expect(onAssetUploadBegin).toHaveBeenLastCalledWith(
expect.objectContaining({ file, fileSize: contents.length, uploadedBytes: contents.length })
);
});
});
describe(computeFileMd5Base64Async, () => {
let tmpDir: string;
@@ -61,6 +61,7 @@ jest.mock('../../../graphql/queries/UpdateQuery');
jest.mock('../../../prompts');
jest.mock('../../../graphql/queries/EnvironmentVariablesQuery');
jest.mock('../../../ora', () => ({
isSpinnerEnabled: () => false,
ora: () => ({
start: () => ({ succeed: () => {}, fail: () => {}, stop: () => {} }),
}),
+46 -4
View File
@@ -3,6 +3,7 @@ import { EasJson, EasJsonAccessor, EasJsonUtils } from '@expo/eas-json';
import { Errors, Flags } from '@oclif/core';
import chalk from 'chalk';
import nullthrows from 'nullthrows';
import path from 'path';
import { ensureBranchExistsAsync } from '../../branch/queries';
import { ensureRepoIsCleanAsync } from '../../build/utils/repository';
@@ -28,7 +29,7 @@ import {
} from '../../graphql/generated';
import { PublishMutation } from '../../graphql/mutations/PublishMutation';
import Log, { learnMore, link } from '../../log';
import { ora } from '../../ora';
import { isSpinnerEnabled, ora } from '../../ora';
import { RequestedPlatform } from '../../platform';
import { maybeUploadAssetMapAsync } from '../../project/maybeUploadAssetMapAsync';
import {
@@ -56,6 +57,7 @@ import {
resolveInputDirectoryAsync,
uploadAssetsAsync,
} from '../../project/publish';
import { createUploadProgressReporter } from '../../project/uploadProgressReporter';
import { resolveWorkflowPerPlatformAsync } from '../../project/workflow';
import { resolveUpdateGroupsSupersedingActiveRolloutsAsync } from '../../update/active-rollout';
import { ensureEASUpdateIsConfiguredAsync } from '../../update/configure';
@@ -389,13 +391,41 @@ export default class UpdatePublish extends EasCommand {
// - Start with NO_ACTIVITY_TIMEOUT. 180 seconds is chosen because the cloud function that processes
// uploaded assets has a timeout of 60 seconds and uploading can take some time on a slow connection.
// - Each time one or more assets reports as ready, reset the timeout.
// - Each time an asset upload begins, reset the timeout. This includes retries.
// - Each time an asset upload begins or sends more bytes, reset the timeout. This includes retries.
// - Start upload. Internally, uploadAssetsAsync uploads them all first and then checks for successful
// processing every (5 + n) seconds with a linear backoff of n + 1 second.
// - At the same time as upload is started, start timeout checker which checks every 1 second to see
// if timeout has been reached. When timeout expires, send a cancellation signal to currently running
// upload function call to instruct it to stop uploading or checking for successful processing.
const NO_ACTIVITY_TIMEOUT = 180 * 1000; // 180 seconds
let uploadCountText = 'Uploading...';
let activeUploadLines: string[] = [];
let unfinishedAssetPaths: string[] = [];
const fullySentAssetPaths = new Set<string>();
const updateAssetSpinnerText = (): void => {
const processingAssetNames = unfinishedAssetPaths
.filter(assetPath => fullySentAssetPaths.has(assetPath))
.map(assetPath => path.basename(assetPath));
assetSpinner.text = [
uploadCountText,
...activeUploadLines,
...(processingAssetNames.length > 0
? [
`Waiting for the server to process ${processingAssetNames.slice(0, 3).join(', ')}` +
(processingAssetNames.length > 3
? ` and ${processingAssetNames.length - 3} more`
: ''),
]
: []),
].join('\n');
};
const reportUploadProgress = createUploadProgressReporter({
isInteractive: isSpinnerEnabled(),
onActiveUploadsChanged: lines => {
activeUploadLines = lines;
updateAssetSpinnerText();
},
});
let lastUploadedStorageKeys = new Set<string>();
let lastAssetUploadResults: {
asset: RawAsset & { storageKey: string };
@@ -422,11 +452,23 @@ export default class UpdatePublish extends EasCommand {
const totalAssets = assetUploadResults.length;
const missingAssetCount = assetUploadResults.filter(a => !a.finished).length;
assetSpinner.text = `Uploading (${totalAssets - missingAssetCount}/${totalAssets})`;
uploadCountText = `Uploading (${totalAssets - missingAssetCount}/${totalAssets})`;
unfinishedAssetPaths = assetUploadResults
.filter(r => !r.finished)
.map(r => r.asset.path);
updateAssetSpinnerText();
},
() => {
progress => {
// when an upload is retried, reset the timeout as we know this will now need more time
timeAtWhichToTimeout = Date.now() + NO_ACTIVITY_TIMEOUT; // reset timeout to NO_ACTIVITY_TIMEOUT
if (progress) {
if (progress.uploadedBytes >= progress.fileSize) {
fullySentAssetPaths.add(progress.file);
} else {
fullySentAssetPaths.delete(progress.file);
}
reportUploadProgress(progress);
}
}
),
(async () => {
@@ -0,0 +1,126 @@
import Log from '../../log';
import { UploadProgress } from '../../uploads';
import { createUploadProgressReporter } from '../uploadProgressReporter';
const MB = 1024 * 1024;
function makeProgress(file: string, uploadedBytes: number, elapsedMs: number): UploadProgress {
return { file, fileSize: 7 * MB, uploadedBytes, elapsedMs };
}
describe(createUploadProgressReporter, () => {
afterEach(() => {
jest.restoreAllMocks();
});
describe('interactive', () => {
it('shows a line for each active upload once it has run for 10 seconds', () => {
const onActiveUploadsChanged = jest.fn();
const report = createUploadProgressReporter({ isInteractive: true, onActiveUploadsChanged });
report(makeProgress('/dist/ios.hbc', MB, 5_000));
expect(onActiveUploadsChanged).toHaveBeenLastCalledWith([]);
report(makeProgress('/dist/ios.hbc', MB, 20_000));
report(makeProgress('/dist/android.hbc', 6 * MB, 10_000));
expect(onActiveUploadsChanged).toHaveBeenLastCalledWith([
'ios.hbc: 1.0 MB of 7.0 MB, 51.2 KB/s, about 2 minutes left',
'android.hbc: 6.0 MB of 7.0 MB, 614 KB/s',
]);
});
it('removes an upload once all of its bytes are sent', () => {
const onActiveUploadsChanged = jest.fn();
const report = createUploadProgressReporter({ isInteractive: true, onActiveUploadsChanged });
report(makeProgress('/dist/ios.hbc', MB, 20_000));
report(makeProgress('/dist/ios.hbc', 7 * MB, 30_000));
expect(onActiveUploadsChanged).toHaveBeenLastCalledWith([]);
});
it('shows at most 5 uploads', () => {
const onActiveUploadsChanged = jest.fn();
const report = createUploadProgressReporter({ isInteractive: true, onActiveUploadsChanged });
for (let i = 0; i < 7; i++) {
report(makeProgress(`/dist/asset-${i}.png`, MB, 20_000));
}
const lines = onActiveUploadsChanged.mock.calls.at(-1)?.[0];
expect(lines).toHaveLength(6);
expect(lines[5]).toBe('and 2 more uploads');
});
it('does not log', () => {
const warnSpy = jest.spyOn(Log, 'warn').mockImplementation(() => {});
const logSpy = jest.spyOn(Log, 'log').mockImplementation(() => {});
const report = createUploadProgressReporter({
isInteractive: true,
onActiveUploadsChanged: () => {},
});
report(makeProgress('/dist/ios.hbc', MB, 20_000));
expect(warnSpy).not.toHaveBeenCalled();
expect(logSpy).not.toHaveBeenCalled();
});
});
describe('non-interactive', () => {
it('warns when an upload is slow and then logs its progress every minute', () => {
let now = 0;
jest.spyOn(Date, 'now').mockImplementation(() => now);
const warnSpy = jest.spyOn(Log, 'warn').mockImplementation(() => {});
const logSpy = jest.spyOn(Log, 'log').mockImplementation(() => {});
const report = createUploadProgressReporter({
isInteractive: false,
onActiveUploadsChanged: () => {},
});
report(makeProgress('/dist/ios.hbc', MB, 20_000));
now = 30_000;
report(makeProgress('/dist/ios.hbc', 2 * MB, 50_000));
now = 60_000;
report(makeProgress('/dist/ios.hbc', 3 * MB, 80_000));
expect(warnSpy).toHaveBeenCalledTimes(1);
expect(warnSpy).toHaveBeenCalledWith(
'Uploading ios.hbc (7.0 MB) is slow (51.2 KB/s). It may take about 2 minutes.'
);
expect(logSpy).toHaveBeenCalledTimes(1);
expect(logSpy).toHaveBeenCalledWith(
'Still uploading ios.hbc: 3.0 MB of 7.0 MB, 38.4 KB/s, about 2 minutes left.'
);
});
it('warns again when a retried upload is still slow', () => {
const warnSpy = jest.spyOn(Log, 'warn').mockImplementation(() => {});
const report = createUploadProgressReporter({
isInteractive: false,
onActiveUploadsChanged: () => {},
});
report(makeProgress('/dist/ios.hbc', MB, 20_000));
report(makeProgress('/dist/ios.hbc', 0, 0));
report(makeProgress('/dist/ios.hbc', MB, 20_000));
expect(warnSpy).toHaveBeenCalledTimes(2);
});
it('does not log when an upload is fast', () => {
const warnSpy = jest.spyOn(Log, 'warn').mockImplementation(() => {});
const logSpy = jest.spyOn(Log, 'log').mockImplementation(() => {});
const report = createUploadProgressReporter({
isInteractive: false,
onActiveUploadsChanged: () => {},
});
report(makeProgress('/dist/ios.hbc', 6 * MB, 10_000));
expect(warnSpy).not.toHaveBeenCalled();
expect(logSpy).not.toHaveBeenCalled();
});
});
});
+3 -3
View File
@@ -44,7 +44,7 @@ import {
truncateString as truncateUpdateMessage,
updatePublishPlatformToAppPlatform,
} from '../update/utils';
import { PresignedPost, uploadWithPresignedPostWithRetryAsync } from '../uploads';
import { PresignedPost, UploadProgress, uploadWithPresignedPostWithRetryAsync } from '../uploads';
import {
expoCommandAsync,
shouldUseVersionedExpoCLI,
@@ -496,7 +496,7 @@ export async function uploadAssetsAsync(
onAssetUploadResultsChanged: (
assetUploadResults: { asset: RawAsset & { storageKey: string }; finished: boolean }[]
) => void,
onAssetUploadBegin: () => void
onAssetUploadBegin: (progress?: UploadProgress) => void
): Promise<AssetUploadResult> {
let assets: RawAsset[] = [];
let platform: keyof FilteredCollectedAssets;
@@ -552,7 +552,7 @@ export async function uploadAssetsAsync(
// Sign URLs in chunks (endpoint caps at 100), then upload all from one pool so it stays saturated.
const uploadBatchAsync = async (
assetsToUpload: AssetWithStorageKey[],
onBegin: () => void
onBegin: (progress?: UploadProgress) => void
): Promise<void> => {
const signedAssets: { asset: AssetWithStorageKey; specification: string }[] = [];
for (const assetChunk of chunk(assetsToUpload, 100)) {
@@ -0,0 +1,91 @@
import path from 'path';
import Log from '../log';
import { UploadProgress } from '../uploads';
import { formatBytes } from '../utils/files';
const MIN_UPLOAD_MS_BEFORE_REPORTING = 10_000;
const SLOW_UPLOAD_SECONDS_LEFT = 60;
const SLOW_UPLOAD_LOG_INTERVAL_MS = 60_000;
const MAX_ACTIVE_UPLOAD_LINES = 5;
function getBytesPerSecond(progress: UploadProgress): number {
return progress.uploadedBytes / (progress.elapsedMs / 1000);
}
function getSecondsLeft(progress: UploadProgress): number {
return (progress.fileSize - progress.uploadedBytes) / getBytesPerSecond(progress);
}
function formatMinutesLeft(progress: UploadProgress): string {
const minutes = Math.ceil(getSecondsLeft(progress) / 60);
return minutes === 1 ? '1 minute' : `${minutes} minutes`;
}
function formatUploadProgress(progress: UploadProgress): string {
const message =
`${path.basename(progress.file)}: ${formatBytes(progress.uploadedBytes)} of ${formatBytes(progress.fileSize)}, ` +
`${formatBytes(getBytesPerSecond(progress))}/s`;
return getSecondsLeft(progress) > SLOW_UPLOAD_SECONDS_LEFT
? `${message}, about ${formatMinutesLeft(progress)} left`
: message;
}
function isSlowUpload(progress: UploadProgress): boolean {
return (
progress.elapsedMs >= MIN_UPLOAD_MS_BEFORE_REPORTING &&
getSecondsLeft(progress) > SLOW_UPLOAD_SECONDS_LEFT
);
}
export function createUploadProgressReporter({
isInteractive,
onActiveUploadsChanged,
}: {
isInteractive: boolean;
onActiveUploadsChanged: (lines: string[]) => void;
}): (progress: UploadProgress) => void {
const activeUploads = new Map<string, UploadProgress>();
const lastSlowUploadLogTimes = new Map<string, number>();
return progress => {
if (progress.uploadedBytes === 0) {
lastSlowUploadLogTimes.delete(progress.file);
}
if (progress.uploadedBytes >= progress.fileSize) {
activeUploads.delete(progress.file);
} else {
activeUploads.set(progress.file, progress);
}
if (isInteractive) {
const lines = [...activeUploads.values()]
.filter(upload => upload.elapsedMs >= MIN_UPLOAD_MS_BEFORE_REPORTING)
.map(formatUploadProgress);
onActiveUploadsChanged(
lines.length > MAX_ACTIVE_UPLOAD_LINES
? [
...lines.slice(0, MAX_ACTIVE_UPLOAD_LINES),
`and ${lines.length - MAX_ACTIVE_UPLOAD_LINES} more uploads`,
]
: lines
);
return;
}
if (!isSlowUpload(progress)) {
return;
}
const lastLogTime = lastSlowUploadLogTimes.get(progress.file);
if (lastLogTime === undefined) {
Log.warn(
`Uploading ${path.basename(progress.file)} (${formatBytes(progress.fileSize)}) is slow ` +
`(${formatBytes(getBytesPerSecond(progress))}/s). It may take about ${formatMinutesLeft(progress)}.`
);
lastSlowUploadLogTimes.set(progress.file, Date.now());
} else if (Date.now() - lastLogTime >= SLOW_UPLOAD_LOG_INTERVAL_MS) {
Log.log(`Still uploading ${formatUploadProgress(progress)}.`);
lastSlowUploadLogTimes.set(progress.file, Date.now());
}
};
}
+20 -4
View File
@@ -20,6 +20,13 @@ export interface PresignedPost {
fields: Record<string, string>;
}
export interface UploadProgress {
file: string;
fileSize: number;
uploadedBytes: number;
elapsedMs: number;
}
export async function uploadFileAtPathToGCSAsync(
graphqlClient: ExpoGraphqlClient,
type: UploadSessionType,
@@ -88,7 +95,7 @@ export async function uploadAppScopedFileAtPathToGCSAsync(
export async function uploadWithPresignedPostWithRetryAsync(
file: string,
presignedPost: PresignedPost,
onAssetUploadBegin: () => void
onAssetUploadBegin: (progress?: UploadProgress) => void
): Promise<Response> {
return await promiseRetry(
async retry => {
@@ -96,7 +103,7 @@ export async function uploadWithPresignedPostWithRetryAsync(
let response: Response;
try {
onAssetUploadBegin();
response = await uploadWithPresignedPostAsync(file, presignedPost);
response = await uploadWithPresignedPostAsync(file, presignedPost, onAssetUploadBegin);
} catch (e: any) {
return retry(e);
}
@@ -127,7 +134,8 @@ export async function uploadWithPresignedPostWithRetryAsync(
async function uploadWithPresignedPostAsync(
file: string,
presignedPost: PresignedPost
presignedPost: PresignedPost,
onProgress: (progress: UploadProgress) => void
): Promise<Response> {
const fileStat = await fs.stat(file);
const fileSize = fileStat.size;
@@ -135,7 +143,15 @@ async function uploadWithPresignedPostAsync(
for (const [fieldKey, fieldValue] of Object.entries(presignedPost.fields)) {
form.append(fieldKey, fieldValue);
}
form.append('file', fs.createReadStream(file), { knownLength: fileSize });
const readStream = fs.createReadStream(file);
form.append('file', readStream, { knownLength: fileSize });
const startTime = Date.now();
let uploadedBytes = 0;
onProgress({ file, fileSize, uploadedBytes, elapsedMs: 0 });
readStream.addListener('data', (chunk: Buffer | string) => {
uploadedBytes += Buffer.byteLength(chunk);
onProgress({ file, fileSize, uploadedBytes, elapsedMs: Date.now() - startTime });
});
const formHeaders = form.getHeaders();
return await fetch(presignedPost.url, {
method: 'POST',