fix putObject never resolves when the input stream gets destroyed o…

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<void>

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<string, unknown> {
  */
 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<undefined | unknown> {
+  if (s.readable) {
+    return undefined
+  }
+  try {
+    await streamPromise.finished(s)
+    return undefined
+  } catch (error) {
+    return error
+  }
 }
 
 /**

tests/functional/functional-tests.js

@@ -36,6 +36,7 @@
 import * as minio from '../../src/minio.ts'
 
 const assert = chai.assert
+const expect = chai.expect
 
 const isWindowsPlatform = process.platform === 'win32'
 
@@ -599,32 +600,51 @@
       })
     })
 
-    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_`,
@@ -3751,13 +3771,13 @@
     const metadata = { 'X-Amz-Meta-Test': 'test-value' }
 
     before(() => {
       return client.makeBucket(bucketName, '').then((res) => {
         return client.putObject(bucketName, objectName, fdObject, fdObject.length, metadata).then((res) => {
           return client.setObjectTagging(bucketName, objectName, tags)
         })
       })
     })
     after(() => client.removeObject(bucketName, objectName).then((_) => client.removeBucket(bucketName)))
 
     step(
       `extensions.listObjectsV2WithMetadata(bucketName, prefix, recursive)_bucketName:${bucketName}, prefix:"", recursive:true`,

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')
         })
       })
     })