Github-Mirror/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-09-20 13:30:46 +02:00
Code Issues Packages Projects Releases Wiki Activity

review: cancel source GET when destination POST fails

Browse Source 
Address coderabbit review (line 165 / second pass on PR 9424): when
the POST leg fails or returns an error status, closing pipeReader
only fails the producer's *writes*. ReadUrlAsStream's own read loop
runs under the parent ctx, so it keeps draining the source body in
the background until EOF — wasting source-volume bandwidth and CPU
on a copy that's already failed.

Wrap streamCopyChunkRange in a child context cancelled on return.
ReadUrlAsStream checks ctx.Done() per 256 KiB tick, so the in-flight
read aborts on the next iteration once the function returns. The POST
also moves to streamCtx so the in-flight request can be cancelled the
same way if the producer fails first.

Defer-cancel runs after both legs return, so the success path still
sends EOF cleanly through pipeWriter.Close before cancellation.
This commit is contained in:
Chris Lu
2026-05-10 13:53:18 -07:00
parent f70f338e7f
commit 843e4b7cdb
1 changed files with 13 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
weed/s3api/s3api_object_handlers_copy_stream.go
+13 -3
View File
@@ -60,6 +60,15 @@ func (s3a *S3ApiServer) streamCopyChunkRange(
if size > int64(math.MaxInt32) {
return fmt.Errorf("chunk size %d exceeds maximum int32 size", size)
}
// Child context so a terminal error here unblocks the producer
// goroutine immediately. Without this, a failed POST closes
// pipeReader (which only fails the producer's writes), but
// ReadUrlAsStream can keep draining the source body in its read
// loop until EOF before noticing — wasting source-volume bandwidth
// and CPU. Cancelling streamCtx makes ReadUrlAsStream's per-tick
// ctx.Done() check return on the next iteration.
streamCtx, cancel := context.WithCancel(ctx)
defer cancel()
dstUrl := fmt.Sprintf("http://%s/%s", assignResult.Location.Url, assignResult.FileId)
dstJwt := security.EncodedJwt(assignResult.Auth)
srcJwt := filer.JwtForVolumeServer(srcFileId)
@@ -101,7 +110,7 @@ func (s3a *S3ApiServer) streamCopyChunkRange(
// HTTP transport picks it up — no per-chunk buffering on either
// side of the pipe.
var writeErr error
shouldRetry, readErr := util_http.ReadUrlAsStream(ctx, srcUrl, srcJwt, nil, false, false, offset, int(size), func(data []byte) {
shouldRetry, readErr := util_http.ReadUrlAsStream(streamCtx, srcUrl, srcJwt, nil, false, false, offset, int(size), func(data []byte) {
if writeErr != nil {
return
}
@@ -138,7 +147,7 @@ func (s3a *S3ApiServer) streamCopyChunkRange(
}
}()
req, err := http.NewRequestWithContext(ctx, http.MethodPost, dstUrl, pipeReader)
req, err := http.NewRequestWithContext(streamCtx, http.MethodPost, dstUrl, pipeReader)
if err != nil {
// Drain the pipe so the producer goroutine doesn't leak waiting
// on a never-read writer.
@@ -152,7 +161,8 @@ func (s3a *S3ApiServer) streamCopyChunkRange(
resp, err := util_http.GetGlobalHttpClient().Do(req)
if err != nil {
// Closing the reader unblocks the producer if it was still mid-write.
// Closing the reader unblocks the producer if it was still mid-write;
// the deferred cancel above also stops any in-flight source read.
pipeReader.CloseWithError(err)
return fmt.Errorf("POST: %w", err)
}