diff --git a/src/internal/client.ts b/src/internal/client.ts index 823b6e7a..9272d971 100644 --- a/src/internal/client.ts +++ b/src/internal/client.ts @@ -35,6 +35,7 @@ import { calculateEvenSplits, extractMetadata, getContentLength, + getReadableStreamError, getScope, getSourceVersionId, getVersionId, @@ -1682,7 +1683,15 @@ export class TypedClient { // Adapts the non-stream interface into a stream. size = stream.length stream = readableStream(stream) - } else if (!isReadableStream(stream)) { + } else if (isReadableStream(stream)) { + const streamError = await getReadableStreamError(stream) + if (streamError) { + throw streamError + } + if (!stream.readable) { + throw new Error('stream is not readable') + } + } else { throw new TypeError('third argument should be of type "stream.Readable" or "Buffer" or "string"') } @@ -1793,58 +1802,61 @@ export class TypedClient { const chunkier = new BlockStream2({ size: partSize, zeroPadding: false }) - // eslint-disable-next-line @typescript-eslint/no-unused-vars - const [_, o] = await Promise.all([ - new Promise((resolve, reject) => { - body.pipe(chunkier).on('error', reject) - chunkier.on('end', resolve).on('error', reject) - }), - (async () => { - let partNumber = 1 - - for await (const chunk of chunkier) { - const md5 = crypto.createHash('md5').update(chunk).digest() - - const oldPart = oldParts[partNumber] - if (oldPart) { - if (oldPart.etag === md5.toString('hex')) { - eTags.push({ part: partNumber, etag: oldPart.etag }) - partNumber++ - continue - } - } + const upload = (async () => { + let partNumber = 1 - partNumber++ + for await (const chunk of chunkier) { + const md5 = crypto.createHash('md5').update(chunk).digest() - // now start to upload missing part - const options: RequestOption = { - method: 'PUT', - query: qs.stringify({ partNumber, uploadId }), - headers: { - 'Content-Length': chunk.length, - 'Content-MD5': md5.toString('base64'), - }, - bucketName, - objectName, + const oldPart = oldParts[partNumber] + if (oldPart) { + if (oldPart.etag === md5.toString('hex')) { + eTags.push({ part: partNumber, etag: oldPart.etag }) + partNumber++ + continue } + } - const response = await this.makeRequestAsyncOmit(options, chunk) + partNumber++ - let etag = response.headers.etag - if (etag) { - etag = etag.replace(/^"/, '').replace(/"$/, '') - } else { - etag = '' - } + // now start to upload missing part + const options: RequestOption = { + method: 'PUT', + query: qs.stringify({ partNumber, uploadId }), + headers: { + 'Content-Length': chunk.length, + 'Content-MD5': md5.toString('base64'), + }, + bucketName, + objectName, + } - eTags.push({ part: partNumber, etag }) + const response = await this.makeRequestAsyncOmit(options, chunk) + + let etag = response.headers.etag + if (etag) { + etag = etag.replace(/^"/, '').replace(/"$/, '') + } else { + etag = '' } - return await this.completeMultipartUpload(bucketName, objectName, uploadId, eTags) - })(), - ]) + eTags.push({ part: partNumber, etag }) + } + })().catch((err) => { + if (!chunkier.closed) { + chunkier.destroy(err) + } + throw err + }) + + try { + await Promise.all([streamPromise.pipeline(body, chunkier), upload]) + } catch (err) { + await this.abortMultipartUpload(bucketName, objectName, uploadId) + throw err + } - return o + return await this.completeMultipartUpload(bucketName, objectName, uploadId, eTags) } async removeBucketReplication(bucketName: string): Promise diff --git a/src/internal/helper.ts b/src/internal/helper.ts index 60ab11bb..9cc34866 100644 --- a/src/internal/helper.ts +++ b/src/internal/helper.ts @@ -16,6 +16,7 @@ import * as crypto from 'node:crypto' import * as stream from 'node:stream' +import * as streamPromise from 'node:stream/promises' import { XMLParser } from 'fast-xml-parser' import ipaddr from 'ipaddr.js' @@ -244,7 +245,23 @@ export function isPlainObject(arg: unknown): arg is Record { */ export function isReadableStream(arg: unknown): arg is stream.Readable { // eslint-disable-next-line @typescript-eslint/unbound-method - return isObject(arg) && isFunction((arg as stream.Readable)._read) && stream.isReadable(arg as stream.Readable) + return isObject(arg) && isFunction((arg as stream.Readable)._read) +} + +/** + * get the error of a readable stream, return undefined if the stream is still open + * or closed without error. + */ +export async function getReadableStreamError(s: stream.Readable): Promise { + if (s.readable) { + return undefined + } + try { + await streamPromise.finished(s) + return undefined + } catch (error) { + return error + } } /** diff --git a/tests/functional/functional-tests.js b/tests/functional/functional-tests.js index 8d9f2a75..3b0228c3 100644 --- a/tests/functional/functional-tests.js +++ b/tests/functional/functional-tests.js @@ -36,6 +36,7 @@ import { getVersionId } from '../../src/internal/helper.ts' import * as minio from '../../src/minio.ts' const assert = chai.assert +const expect = chai.expect const isWindowsPlatform = process.platform === 'win32' @@ -599,32 +600,51 @@ describe('functional tests', function () { }) }) - step(`putObject(bucketName, objectName, destroyedStream) should reject immediately`, function (done) { - this.timeout(10000) - var s = new stream.Readable({ read() {} }) - s.destroy() - client.putObject(bucketName, objectName, s, (e) => { - if (e && e instanceof TypeError && e.message.includes('stream.Readable')) { - return done() - } - done(new Error('expected TypeError for destroyed stream, got: ' + (e || 'no error'))) - }) - }) + step( + `putObject(bucketName, objectName, destroyedStream) should reject with the default premature close error`, + async function () { + this.timeout(10000) + var s = new stream.Readable({ read() {} }) + s.destroy() + await expect(client.putObject(bucketName, objectName, s)).to.be.rejectedWith('Premature close') + }, + ) - step(`putObject(bucketName, objectName, streamDestroyedDuringUpload) should reject`, function (done) { - this.timeout(10000) - // Create a stream large enough to trigger multipart upload (> partSize). - // Destroy it on the next event loop tick so it becomes unreadable during - // the async findUploadId/initiateNewMultipartUpload calls in uploadStream. - var s = new stream.Readable({ read() {} }) - setTimeout(() => s.destroy(), 0) - client.putObject(bucketName, objectName, s, _65mb.length, (e) => { - if (e) { - return done() - } - done(new Error('expected an error for stream destroyed during upload')) - }) - }) + step( + `putObject(bucketName, objectName, streamDestroyedWithError) should reject with the destroy error`, + async function () { + this.timeout(10000) + var s = new stream.Readable({ read() {} }).on('error', () => {}) + s.destroy(new Error('test stream error')) + await expect(client.putObject(bucketName, objectName, s)).to.be.rejectedWith('test stream error') + }, + ) + + step( + `putObject(bucketName, objectName, streamDestroyedDuringUpload) should reject with the default premature close error`, + async function () { + this.timeout(10000) + // Create a stream large enough to trigger multipart upload (> partSize). + // Destroy it after a short timeout so it becomes unreadable during + // the async findUploadId/initiateNewMultipartUpload calls in uploadStream. + var s = new stream.Readable({ read() {} }) + setTimeout(() => s.destroy(), 500) + await expect(client.putObject(bucketName, objectName, s, _65mb.length)).to.be.rejectedWith('Premature close') + }, + ) + + step( + `putObject(bucketName, objectName, streamDestroyedWithErrorDuringUpload) should reject with the destroy error`, + async function () { + this.timeout(10000) + // Create a stream large enough to trigger multipart upload (> partSize). + // Destroy it after a short timeout so it becomes unreadable during + // the async findUploadId/initiateNewMultipartUpload calls in uploadStream. + var s = new stream.Readable({ read() {} }).on('error', () => {}) + setTimeout(() => s.destroy(new Error('test stream error')), 500) + await expect(client.putObject(bucketName, objectName, s, _65mb.length)).to.be.rejectedWith('test stream error') + }, + ) step( `getPartialObject(bucketName, objectName, offset, length, cb)_bucketName:${bucketName}, objectName:${_65mbObjectName}, offset:0, length:100*1024_`, diff --git a/tests/unit/test.js b/tests/unit/test.js index 1b58d0b1..b91d97a2 100644 --- a/tests/unit/test.js +++ b/tests/unit/test.js @@ -665,7 +665,12 @@ describe('Client', function () { it('should fail when stream is destroyed', () => { const s = new Stream.Readable({ read() {} }) s.destroy() - return expect(client.putObject('bucket', 'object', s)).to.be.rejectedWith('stream.Readable') + return expect(client.putObject('bucket', 'object', s)).to.be.rejectedWith('Premature close') + }) + it('should fail when stream is destroyed with an error', () => { + const s = new Stream.Readable({ read() {} }).on('error', () => {}) + s.destroy(new Error('stream error')) + return expect(client.putObject('bucket', 'object', s)).to.be.rejectedWith('stream error') }) }) })