fix putObject never resolves when the input stream gets destroyed o…#1480
Open
yucao2521 wants to merge 1 commit into
Open
fix putObject never resolves when the input stream gets destroyed o…#1480yucao2521 wants to merge 1 commit into
putObject never resolves when the input stream gets destroyed o…#1480yucao2521 wants to merge 1 commit into
Conversation
📝 WalkthroughWalkthroughThis PR fixes a critical issue where ChangesStream Error Handling for putObject
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/internal/client.ts`:
- Around line 1802-1849: The async uploader awaits the pipeline and the consumer
separately which can deadlock if makeRequestAsyncOmit rejects inside the
consumer (o); change the try block to await both the pipeline and the consumer
together (e.g., await Promise.all([streamPromise.pipeline(body, chunkier), o]))
so the pipeline and the async iterator (o) are observed as one unit and ensure
abortMultipartUpload is still called on any rejection from either side; keep
references to chunkier, o, makeRequestAsyncOmit, streamPromise.pipeline and
abortMultipartUpload when making the change.
- Around line 1808-1840: The bug is that partNumber is incremented before
uploading the current chunk, causing chunk N to be uploaded as part N+1; fix by
keeping the current chunk on the current partNumber: do not increment partNumber
until after you have either skipped a matching oldPart (the existing branch
where you compare oldPart.etag === md5.toString('hex') should increment and
continue) or after a successful upload and pushing to eTags; specifically,
remove the premature partNumber++ that appears immediately before the upload
RequestOption, and instead increment partNumber only after eTags.push({ part:
partNumber, etag }) (and keep the existing increment where you skip matched
oldPart).
In `@tests/unit/test.js`:
- Around line 670-674: The test creates a Readable stream `s` and calls
`s.destroy(new Error('stream error'))` without an 'error' listener which can
cause an unhandled exception; before calling `s.destroy(...)` add an empty error
handler like `s.on('error', () => {})` (i.e., update the test around the
`it('should fail when stream is destroyed with an error', ...)` block so the
`Stream.Readable` instance `s` has `s.on('error', () => {})` attached prior to
calling `client.putObject('bucket', 'object', s)` and `s.destroy(...)`).
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: cbae5fd8-c249-4ba8-910e-9df28e9a463a
📒 Files selected for processing (4)
src/internal/client.tssrc/internal/helper.tstests/functional/functional-tests.jstests/unit/test.js
Comment on lines
+1808
to
+1840
| 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, | ||
| } | ||
|
|
||
| const response = await this.makeRequestAsyncOmit(options, chunk) | ||
|
|
||
| eTags.push({ part: partNumber, etag }) | ||
| 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 }) |
There was a problem hiding this comment.
Keep the current chunk on the current part number.
partNumber is incremented before uploading the current chunk. After the first mismatch, chunk N is sent as part N+1, which can overwrite a later existing part and produce duplicate/out-of-order entries in eTags during resume.
💡 Suggested fix
const o = (async () => {
let partNumber = 1
for await (const chunk of chunkier) {
+ const currentPartNumber = partNumber
const md5 = crypto.createHash('md5').update(chunk).digest()
- const oldPart = oldParts[partNumber]
+ const oldPart = oldParts[currentPartNumber]
if (oldPart) {
if (oldPart.etag === md5.toString('hex')) {
- eTags.push({ part: partNumber, etag: oldPart.etag })
+ eTags.push({ part: currentPartNumber, etag: oldPart.etag })
partNumber++
continue
}
}
- partNumber--
-
// now start to upload missing part
const options: RequestOption = {
method: 'PUT',
- query: qs.stringify({ partNumber, uploadId }),
+ query: qs.stringify({ partNumber: currentPartNumber, uploadId }),
headers: {
'Content-Length': chunk.length,
'Content-MD5': md5.toString('base64'),
},
bucketName,
objectName,
}
const response = await this.makeRequestAsyncOmit(options, chunk)
@@
- eTags.push({ part: partNumber, etag })
+ eTags.push({ part: currentPartNumber, etag })
+ partNumber++
}
})()🧰 Tools
🪛 ast-grep (0.43.0)
[warning] 1809-1809: Do not use weak hash functions
Context: md5.toString('hex')
Note: [CWE-328].
(insecure-hash-typescript)
[warning] 1824-1824: Do not use weak hash functions
Context: 'Content-MD5'
Note: [CWE-328].
(insecure-hash-typescript)
[warning] 1824-1824: Do not use weak hash functions
Context: md5.toString('base64')
Note: [CWE-328].
(insecure-hash-typescript)
[warning] 1809-1809: Avoid weak hash algorithm from CryptoJS
Context: md5.toString('hex')
Note: Security best practice.
(crypto-avoid-weak-hash-typescript)
[warning] 1824-1824: Avoid weak hash algorithm from CryptoJS
Context: 'Content-MD5'
Note: Security best practice.
(crypto-avoid-weak-hash-typescript)
[warning] 1824-1824: Avoid weak hash algorithm from CryptoJS
Context: md5.toString('base64')
Note: Security best practice.
(crypto-avoid-weak-hash-typescript)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/internal/client.ts` around lines 1808 - 1840, The bug is that partNumber
is incremented before uploading the current chunk, causing chunk N to be
uploaded as part N+1; fix by keeping the current chunk on the current
partNumber: do not increment partNumber until after you have either skipped a
matching oldPart (the existing branch where you compare oldPart.etag ===
md5.toString('hex') should increment and continue) or after a successful upload
and pushing to eTags; specifically, remove the premature partNumber++ that
appears immediately before the upload RequestOption, and instead increment
partNumber only after eTags.push({ part: partNumber, etag }) (and keep the
existing increment where you skip matched oldPart).
yucao2521
force-pushed
the
fix/put-stream
branch
from
9f01f3e to
55172c4
Compare
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/internal/client.ts`:
- Around line 1688-1692: The code currently only checks for destroy errors via
getReadableStreamError(stream) but accepts streams that have already been
consumed without error. Since getReadableStreamError() returns undefined for
both "still open" and "finished cleanly" states, you need to add an additional
check after the error probe to reject exhausted streams. After checking if an
error exists from getReadableStreamError(), also verify that the stream is not
already finished or exhausted. If the stream is in a finished/exhausted state
(meaning it was already consumed and returned no error), throw an error to
prevent uploading empty data when the caller provided a positive size. This
ensures that only viable streams proceed past this validation point.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: e4f2da02-ff3a-4313-a8fe-d8fdc0d8fb5a
📒 Files selected for processing (4)
src/internal/client.tssrc/internal/helper.tstests/functional/functional-tests.jstests/unit/test.js
🚧 Files skipped from review as they are similar to previous changes (2)
- tests/unit/test.js
- tests/functional/functional-tests.js
Comment on lines
+1688
to
+1692
| } else { | ||
| const error = await getReadableStreamError(stream) | ||
| if (error) { | ||
| throw error | ||
| } |
There was a problem hiding this comment.
Reject exhausted streams after probing for a destroy error.
getReadableStreamError() returns undefined for both “still open” and “finished cleanly”. In this branch that means an already-consumed stream is accepted and can be uploaded as empty data on the single-part path, even when the caller provided a positive size.
💡 Suggested fix
} else {
const error = await getReadableStreamError(stream)
if (error) {
throw error
}
+ if (!stream.readable) {
+ throw new TypeError('third argument should be of type "stream.Readable" or "Buffer" or "string"')
+ }
}🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/internal/client.ts` around lines 1688 - 1692, The code currently only
checks for destroy errors via getReadableStreamError(stream) but accepts streams
that have already been consumed without error. Since getReadableStreamError()
returns undefined for both "still open" and "finished cleanly" states, you need
to add an additional check after the error probe to reject exhausted streams.
After checking if an error exists from getReadableStreamError(), also verify
that the stream is not already finished or exhausted. If the stream is in a
finished/exhausted state (meaning it was already consumed and returned no
error), throw an error to prevent uploading empty data when the caller provided
a positive size. This ensures that only viable streams proceed past this
validation point.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
fixes: #1479
Summary by CodeRabbit
Bug Fixes
Tests
Refactor