From 843e4b7cdb12d14e6c3237392d06d196ce103df9 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sun, 10 May 2026 13:53:18 -0700 Subject: [PATCH] review: cancel source GET when destination POST fails MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- weed/s3api/s3api_object_handlers_copy_stream.go | 16 +++++++++++++--- 1 file changed, 13 insertions(+), 3 deletions(-) diff --git a/weed/s3api/s3api_object_handlers_copy_stream.go b/weed/s3api/s3api_object_handlers_copy_stream.go index 76cec9da1..19adef011 100644 --- a/weed/s3api/s3api_object_handlers_copy_stream.go +++ b/weed/s3api/s3api_object_handlers_copy_stream.go @@ -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) }