fix: stop multipart producer after early response without race/hang

Create multipart cancel before the writer goroutine starts, and on
transport return cancel + close pipe/readers so execute does not race
on multipartCancelFunc or hang on multipartErrChan when a source
Reader is blocked inside Read. Fixes #1186.

Author: Solaris-star

client.go

@@ -2485,9 +2485,12 @@ func (c *Client) execute(req *Request) (*Response, error) {
 
 	req.StartTime = time.Now()
 	resp, err := c.Client().Do(req.withTimeout())
-	// Cancel multipart context for io.Copy to stop reading/writing further
-	if req.isMultiPart && req.multipartCancelFunc != nil {
-		req.multipartCancelFunc()
+	// Stop multipart production once the transport has returned. Cancel alone
+	// is not enough when a source Reader is already blocked inside Read; also
+	// close closable field readers so the writer goroutine can finish and
+	// multipartErrChan can close (see #1186).
+	if req.isMultiPart {
+		req.stopMultipart()
 	}
 
 	// Take ownership of the per-attempt timeout cancel func set by
@@ -2509,8 +2512,14 @@ func (c *Client) execute(req *Request) (*Response, error) {
 	}
 	if req.isMultiPart && req.multipartErrChan != nil {
 		// read all multipart errors from channel
-		for err = range req.multipartErrChan {
-			response.CascadeError = wrapErrors(err, response.CascadeError)
+		for merr := range req.multipartErrChan {
+			// stopMultipart intentionally closes the pipe/readers after the
+			// transport returns. Those shutdown signals are not request failures
+			// when an HTTP response was already obtained.
+			if resp != nil && isMultipartStopError(merr) {
+				continue
+			}
+			response.CascadeError = wrapErrors(merr, response.CascadeError)
 		}
 	}
 

middleware.go

@@ -373,6 +373,7 @@ func handleMultipart(c *Client, r *Request) error {
 	br, bw := io.Pipe()
 	mw := multipart.NewWriter(bw)
 	r.Body = br
+	r.multipartPipeWriter = bw
 
 	// set custom multipart boundary if exists
 	if err := multipartSetBoundary(mw, r); err != nil {
@@ -382,6 +383,10 @@ func handleMultipart(c *Client, r *Request) error {
 
 	r.Header.Set(hdrContentTypeKey, mw.FormDataContentType())
 
+	// Create cancel before the writer goroutine so Client.execute can stop
+	// production without racing on multipartCancelFunc publication.
+	ctx, cancel := context.WithCancel(r.Context())
+	r.multipartCancelFunc = cancel
 	r.multipartErrChan = make(chan error, 1)
 	go func() {
 		defer close(r.multipartErrChan)
@@ -399,8 +404,6 @@ func handleMultipart(c *Client, r *Request) error {
 			return
 		}
 
-		ctx, cancel := context.WithCancel(r.Context())
-		r.multipartCancelFunc = cancel
 		for _, mf := range r.multipartFields {
 			if mf.isValues() {
 				for _, v := range mf.Values {

middleware_test.go

@@ -21,6 +21,7 @@ import (
 	"strings"
 	"sync"
 	"testing"
+	"time"
 )
 
 func Test_parseRequestURL(t *testing.T) {
@@ -1096,6 +1097,7 @@ func TestMiddleware_multipartWriteFormData(t *testing.T) {
 	})
 
 	req := &Request{
+		mu:          new(sync.Mutex),
 		Header:      http.Header{},
 		isMultiPart: true,
 		multipartFields: []*MultipartField{
@@ -1262,3 +1264,105 @@ func TestMiddlewareCoverage(t *testing.T) {
 	err1 := createRawRequest(c, req1)
 	assertTrue(t, strings.Contains(err1.Error(), "invalid character"), "invalid URL error expected")
 }
+
+func TestMultipartEarlyResponseRace(t *testing.T) {
+	client := NewWithClient(&http.Client{
+		Transport: earlyResponseTransport{},
+	})
+
+	for range 200 {
+		_, _ = client.R().
+			SetMultipartFields(&MultipartField{
+				Name:        "audio",
+				FileName:    "audio.wav",
+				ContentType: "audio/wav",
+				Reader:      bytes.NewReader(make([]byte, 1<<20)),
+			}).
+			Post("http://resty.test/upload")
+	}
+}
+
+type earlyResponseTransport struct{}
+
+func (earlyResponseTransport) RoundTrip(req *http.Request) (*http.Response, error) {
+	// net/http allows a RoundTripper to consume and close the request body
+	// asynchronously after RoundTrip returns.
+	go func() {
+		_, _ = io.Copy(io.Discard, req.Body)
+		_ = req.Body.Close()
+	}()
+
+	return &http.Response{
+		StatusCode: http.StatusUnprocessableEntity,
+		Status:     "422 Unprocessable Entity",
+		Header:     make(http.Header),
+		Body:       io.NopCloser(strings.NewReader("rejected")),
+		Request:    req,
+	}, nil
+}
+
+type blockingReadCloser struct {
+	started chan struct{}
+	release chan struct{}
+	once    sync.Once
+}
+
+func newBlockingReadCloser() *blockingReadCloser {
+	return &blockingReadCloser{
+		started: make(chan struct{}),
+		release: make(chan struct{}),
+	}
+}
+
+func (r *blockingReadCloser) Read([]byte) (int, error) {
+	select {
+	case <-r.started:
+	default:
+		close(r.started)
+	}
+
+	<-r.release
+
+	return 0, io.EOF
+}
+
+func (r *blockingReadCloser) Close() error {
+	r.once.Do(func() {
+		close(r.release)
+	})
+
+	return nil
+}
+
+func TestMultipartReturnsAfterEarlyResponse(t *testing.T) {
+	// RoundTrip returns before the request body is fully written, leaving the
+	// multipart producer blocked inside Reader.Read. stopMultipart must cancel
+	// production and close closable field readers so execute does not hang on
+	// multipartErrChan (see #1186).
+	reader := newBlockingReadCloser()
+	client := NewWithClient(&http.Client{
+		Transport: earlyResponseTransport{},
+	})
+
+	done := make(chan error, 1)
+	go func() {
+		_, err := client.R().
+			SetMultipartFields(&MultipartField{
+				Name:        "audio",
+				FileName:    "audio.wav",
+				ContentType: "audio/wav",
+				Reader:      reader,
+			}).
+			Post("http://resty.test/upload")
+		done <- err
+	}()
+
+	select {
+	case <-done:
+		// returned after early response without caller unblocking the reader
+	case <-time.After(2 * time.Second):
+		_ = reader.Close()
+		<-done
+		t.Fatal("multipart request did not return after the transport responded")
+	}
+}

request.go

@@ -108,6 +108,7 @@ type Request struct {
 	unescapeQueryParams  bool
 	multipartErrChan     chan error
 	multipartCancelFunc  context.CancelFunc
+	multipartPipeWriter  *io.PipeWriter
 }
 
 // SetCorrelationID method is used to set the correlation ID for the request
@@ -1597,6 +1598,26 @@ func (r *Request) Execute(method, url string) (res *Response, err error) {
 	return
 }
 
+// stopMultipart cancels multipart streaming, closes the pipe writer, and closes
+// field readers so a writer blocked in io.Reader.Read can unblock once the HTTP
+// round-trip ends (or needs to be aborted).
+func (r *Request) stopMultipart() {
+	if r == nil {
+		return
+	}
+	if r.multipartCancelFunc != nil {
+		r.multipartCancelFunc()
+	}
+	if r.multipartPipeWriter != nil {
+		// Unblock Client.Do / RoundTrip waiting on the request body pipe.
+		_ = r.multipartPipeWriter.CloseWithError(io.ErrClosedPipe)
+		r.multipartPipeWriter = nil
+	}
+	for _, mf := range r.multipartFields {
+		mf.close()
+	}
+}
+
 // Clone returns a deep copy of r with its context changed to ctx.
 // It does clone appropriate fields, reset, and reinitialize, so
 // [Request] can be used again.
@@ -1662,6 +1683,8 @@ func (r *Request) Clone(ctx context.Context) *Request {
 	rr.initTraceIfEnabled()
 	rr.values = make(map[string]any)
 	rr.multipartErrChan = nil
+	rr.multipartCancelFunc = nil
+	rr.multipartPipeWriter = nil
 	rr.ctxCancelFunc = nil
 
 	// copy bodyBuf

util.go

@@ -7,6 +7,7 @@ package resty
 
 import (
 	"bytes"
+	"context"
 	"crypto/rand"
 	"crypto/sha256"
 	"encoding/binary"
@@ -545,3 +546,17 @@ func readMachineID() []byte {
 	// This panic would happen at program startup, so no worries at runtime panic.
 	panic(errors.New("resty - guid: unable to get hostname and random bytes"))
 }
+
+func isMultipartStopError(err error) bool {
+	if err == nil {
+		return false
+	}
+	if errors.Is(err, io.ErrClosedPipe) || errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+		return true
+	}
+	// multipart.Writer.Close can surface closed-pipe as a plain error string
+	// depending on Go version / close path.
+	msg := err.Error()
+	return strings.Contains(msg, "io: read/write on closed pipe") ||
+		strings.Contains(msg, "context canceled")
+}