Skip to content

fix: drain intermediate upload responses #1551

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
IgorGanapolsky wants to merge 1 commit into
base: main
Choose a base branch
from
+62 −12
Open

fix: drain intermediate upload responses #1551

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions src/cross/_cross_uploader.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,16 @@ export const INITIAL_RETRY_DELAY_MS = 1000;
export const DELAY_MULTIPLIER = 2;
export const X_GOOG_UPLOAD_STATUS_HEADER_FIELD = 'x-goog-upload-status';

export async function discardResponseBody(
response: HttpResponse | undefined,
): Promise<void> {
const body = response?.responseInternal?.body;
if (!body || response?.responseInternal?.bodyUsed) {
return;
}
await body.cancel();
}

export class CrossUploader implements Uploader {
async upload(
file: string | Blob,
Expand Down Expand Up @@ -160,6 +170,7 @@ async function uploadBlobInternal(
if (response?.headers?.[X_GOOG_UPLOAD_STATUS_HEADER_FIELD]) {
break;
}
await discardResponseBody(response);
retryCount++;
await sleep(currentDelayMs);
currentDelayMs = currentDelayMs * DELAY_MULTIPLIER;
Expand All @@ -177,6 +188,7 @@ async function uploadBlobInternal(
'All content has been uploaded, but the upload status is not finalized.',
);
}
await discardResponseBody(response);
}

return response;
Expand Down
3 changes: 3 additions & 0 deletions src/node/_node_uploader.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@ import {
MAX_CHUNK_SIZE,
MAX_RETRY_COUNT,
X_GOOG_UPLOAD_STATUS_HEADER_FIELD,
discardResponseBody,
getBlobStat,
sleep,
uploadBlob,
Expand Down Expand Up @@ -301,6 +302,7 @@ export class NodeUploader implements Uploader {
if (response?.headers?.[X_GOOG_UPLOAD_STATUS_HEADER_FIELD]) {
break;
}
await discardResponseBody(response);
retryCount++;
await sleep(currentDelayMs);
currentDelayMs = currentDelayMs * DELAY_MULTIPLIER;
Expand All @@ -318,6 +320,7 @@ export class NodeUploader implements Uploader {
'All content has been uploaded, but the upload status is not finalized.',
);
}
await discardResponseBody(response);
}
return response;
} finally {
Expand Down
59 changes: 47 additions & 12 deletions test/unit/file_test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -100,12 +100,13 @@ describe('File', () => {
},
url: 'some-url',
};
const mockResponse = new Response(
JSON.stringify({
data: 'data1',
}),
uploadOkOptions,
);
const createUploadActiveResponse = () =>
new Response(
JSON.stringify({
data: 'data1',
}),
uploadOkOptions,
);
const fileSize = TEST_FILE_SIZE;
describe('Node_client', () => {
it('It should upload the file from a string path.', async () => {
Expand All @@ -117,9 +118,12 @@ describe('File', () => {
const mockResponses = [
Promise.resolve(new Response('', createUrlOkoptions)),
];
const activeResponses = [];

for (let i = 0; i < numRequests - 1; i++) {
mockResponses.push(Promise.resolve(mockResponse));
const activeResponse = createUploadActiveResponse();
activeResponses.push(activeResponse);
mockResponses.push(Promise.resolve(activeResponse));
}
mockResponses.push(
Promise.resolve(
Expand All @@ -138,6 +142,9 @@ describe('File', () => {

await client.files.upload({file: filePath});
expect(fetchSpy).toHaveBeenCalledTimes(numRequests + 1);
expect(
activeResponses.every((response) => response.bodyUsed),
).toBeTrue();

const allArgs = fetchSpy.calls.allArgs();

Expand Down Expand Up @@ -180,11 +187,18 @@ describe('File', () => {
// of the requests to upload the file.
const mockResponses = [
Promise.resolve(new Response('', createUrlOkoptions)),
Promise.resolve(new Response('', uploadMissingStatusOptions)),
];
const missingStatusResponse = new Response(
'',
uploadMissingStatusOptions,
);
mockResponses.push(Promise.resolve(missingStatusResponse));
const activeResponses = [];

for (let i = 0; i < numRequests - 1; i++) {
mockResponses.push(Promise.resolve(mockResponse));
const activeResponse = createUploadActiveResponse();
activeResponses.push(activeResponse);
mockResponses.push(Promise.resolve(activeResponse));
}
mockResponses.push(
Promise.resolve(
Expand All @@ -205,6 +219,10 @@ describe('File', () => {
// 1 initial request to get the upload url, 1 response missing x-goog-upload-status header and then the rest
// of the requests to upload the file.
expect(fetchSpy).toHaveBeenCalledTimes(numRequests + 1 + 1);
expect(missingStatusResponse.bodyUsed).toBeTrue();
expect(
activeResponses.every((response) => response.bodyUsed),
).toBeTrue();

const allArgs = fetchSpy.calls.allArgs();

Expand Down Expand Up @@ -241,9 +259,12 @@ describe('File', () => {
const mockResponses = [
Promise.resolve(new Response('', createUrlOkoptions)),
];
const activeResponses = [];

for (let i = 0; i < numRequests - 1; i++) {
mockResponses.push(Promise.resolve(mockResponse));
const activeResponse = createUploadActiveResponse();
activeResponses.push(activeResponse);
mockResponses.push(Promise.resolve(activeResponse));
}
mockResponses.push(
Promise.resolve(
Expand All @@ -263,6 +284,9 @@ describe('File', () => {
await client.files.upload({file: testBlob});

expect(fetchSpy).toHaveBeenCalledTimes(numRequests + 1);
expect(
activeResponses.every((response) => response.bodyUsed),
).toBeTrue();
const allArgs = fetchSpy.calls.allArgs();

// make sure we get the correct create url. mimeType and fileSize in the
Expand Down Expand Up @@ -304,11 +328,18 @@ describe('File', () => {
// of the requests to upload the file.
const mockResponses = [
Promise.resolve(new Response('', createUrlOkoptions)),
Promise.resolve(new Response('', uploadMissingStatusOptions)),
];
const missingStatusResponse = new Response(
'',
uploadMissingStatusOptions,
);
mockResponses.push(Promise.resolve(missingStatusResponse));
const activeResponses = [];

for (let i = 0; i < numRequests - 1; i++) {
mockResponses.push(Promise.resolve(mockResponse));
const activeResponse = createUploadActiveResponse();
activeResponses.push(activeResponse);
mockResponses.push(Promise.resolve(activeResponse));
}
mockResponses.push(
Promise.resolve(
Expand All @@ -329,6 +360,10 @@ describe('File', () => {

// 1 initial request to get the upload url, 1 response missing x-goog-upload-status header and then the rest
expect(fetchSpy).toHaveBeenCalledTimes(numRequests + 1 + 1);
expect(missingStatusResponse.bodyUsed).toBeTrue();
expect(
activeResponses.every((response) => response.bodyUsed),
).toBeTrue();
const allArgs = fetchSpy.calls.allArgs();

// make sure we get the correct create url. mimeType and fileSize in the
Expand Down