fix: improve error handling in the multipart request failure #1030

Author: jeevatkm

client.go

@@ -2380,7 +2380,7 @@ func (c *Client) execute(req *Request) (*Response, error) {
 	}
 	if req.multipartErrChan != nil {
 		if err = <-req.multipartErrChan; err != nil {
-			return response, err
+			response.CascadeError = wrapErrors(err, response.CascadeError)
 		}
 	}
 	if resp != nil {
@@ -2390,18 +2390,18 @@ func (c *Client) execute(req *Request) (*Response, error) {
 
 		response.Body = resp.Body
 		if err = response.wrapContentDecompresser(); err != nil {
-			return response, err
+			return response, response.wrapError(err, false)
 		}
 
 		response.wrapLimitReadCloser()
-	}
 
-	if !req.DoNotParseResponse {
-		if req.ResponseBodyUnlimitedReads || req.Debug {
-			response.wrapCopyReadCloser()
+		if !req.DoNotParseResponse {
+			if req.ResponseBodyUnlimitedReads || req.Debug {
+				response.wrapCopyReadCloser()
 
-			if err = response.readAll(); err != nil {
-				return response, err
+				if err = response.readAll(); err != nil {
+					return response, response.wrapError(err, false)
+				}
 			}
 		}
 	}
@@ -2415,8 +2415,7 @@ func (c *Client) execute(req *Request) (*Response, error) {
 		}
 	}
 
-	err = response.CascadeError
-	return response, err
+	return response, response.wrapError(nil, false)
 }
 
 // getting TLS client config if not exists then create one

middleware.go

@@ -314,16 +314,38 @@ func handleMultipart(c *Client, r *Request) error {
 		}
 
 		r.Header.Set(hdrContentTypeKey, mw.FormDataContentType())
-		closeq(mw)
 
-		return nil
+		return mw.Close()
+	}
+
+	// pre-process multipart fields to catch possible errors
+	for _, mf := range r.multipartFields {
+		if len(mf.Values) > 0 {
+			continue
+		}
+		if err := mf.openFileIfRequired(); err != nil {
+			return err
+		}
+
+		// probe the file to catch possible errors
+		// and also detect content type if empty
+		p := make([]byte, 512)
+		size, err := mf.Reader.Read(p)
+		if err != nil && err != io.EOF {
+			return err
+		}
+		mf.tempBuf = p[:size]
+
+		// auto detect content type if empty
+		if isStringEmpty(mf.ContentType) {
+			mf.ContentType = http.DetectContentType(mf.tempBuf)
+		}
 	}
 
 	// multipart streaming
 	bodyReader, bodyWriter := io.Pipe()
 	mw := multipart.NewWriter(bodyWriter)
 	r.Body = bodyReader
-	r.multipartErrChan = make(chan error, 1)
 
 	// set boundary if it is provided by the user
 	if !isStringEmpty(r.multipartBoundary) {
@@ -332,13 +354,21 @@ func handleMultipart(c *Client, r *Request) error {
 		}
 	}
 
+	r.multipartErrChan = make(chan error, 1)
 	go func() {
-		defer close(r.multipartErrChan)
+		defer func() {
+			if err := mw.Close(); err != nil {
+				r.multipartErrChan <- err
+			}
+			if err := bodyWriter.Close(); err != nil {
+				r.multipartErrChan <- err
+			}
+			close(r.multipartErrChan)
+		}()
+
 		if err := createMultipart(mw, r); err != nil {
 			r.multipartErrChan <- err
 		}
-		closeq(mw)
-		closeq(bodyWriter)
 	}()
 
 	r.Header.Set(hdrContentTypeKey, mw.FormDataContentType())
@@ -362,27 +392,15 @@ func createMultipart(w *multipart.Writer, r *Request) error {
 			continue
 		}
 
-		if err := mf.openFileIfRequired(); err != nil {
-			return err
-		}
-
-		p := make([]byte, 512)
-		size, err := mf.Reader.Read(p)
-		if err != nil && err != io.EOF {
-			return err
-		}
-		// auto detect content type if empty
-		if isStringEmpty(mf.ContentType) {
-			mf.ContentType = http.DetectContentType(p[:size])
-		}
-
 		partWriter, err := mpCreatePart(w, mf.createHeader())
 		if err != nil {
 			return err
 		}
 
 		partWriter = mf.wrapProgressCallbackIfPresent(partWriter)
-		partWriter.Write(p[:size])
+		if len(mf.tempBuf) > 0 {
+			partWriter.Write(mf.tempBuf)
+		}
 
 		if _, err = ioCopy(partWriter, mf.Reader); err != nil {
 			return err
@@ -482,7 +500,8 @@ func handleRequestBody(c *Client, r *Request) error {
 // based on registered HTTP response `Content-Type` decoder, see [Client.AddContentTypeDecoder];
 // if [Request.SetResult], [Request.SetResultError], or [Client.SetResultError] is used
 func AutoParseResponseMiddleware(c *Client, res *Response) (err error) {
-	if res.CascadeError != nil || res.Request.DoNotParseResponse {
+	if (res.CascadeError != nil && (res.Request.isMultiPart && res.StatusCode() == 0)) ||
+		res.Request.DoNotParseResponse {
 		return // move on
 	}
 

multipart.go

@@ -58,6 +58,10 @@ type MultipartField struct {
 	//
 	// It is primarily added for ordered multipart form-data field use cases
 	Values []string
+
+	// tempBuf is used to preserve the byte(s) read from the file to detect the content type.
+	// Or any possible read error early.
+	tempBuf []byte
 }
 
 // Clone method returns the deep copy of m except [io.Reader].

multipart_test.go

@@ -19,6 +19,7 @@ import (
 	"path/filepath"
 	"strconv"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 )
@@ -88,8 +89,8 @@ func TestMultipartUploadError(t *testing.T) {
 		Post(ts.URL + "/upload")
 
 	assertNotNil(t, err)
-	assertNotNil(t, resp)
-	assertTrue(t, errors.Is(err, fs.ErrNotExist))
+	assertNil(t, resp)
+	assertEqual(t, true, errors.Is(err, fs.ErrNotExist))
 }
 
 func TestMultipartUploadFiles(t *testing.T) {
@@ -572,8 +573,7 @@ func TestMultipartReaderErrors(t *testing.T) {
 
 		assertNotNil(t, err)
 		assertEqual(t, errTestErrorReader, err)
-		assertNotNil(t, resp)
-		assertEqual(t, nil, resp.Body)
+		assertNil(t, resp)
 	})
 
 	t.Run("multipart files with errorReader", func(t *testing.T) {
@@ -583,8 +583,7 @@ func TestMultipartReaderErrors(t *testing.T) {
 
 		assertNotNil(t, err)
 		assertEqual(t, errTestErrorReader, err)
-		assertNotNil(t, resp)
-		assertEqual(t, nil, resp.Body)
+		assertNil(t, resp)
 	})
 
 	t.Run("multipart with file not found", func(t *testing.T) {
@@ -593,9 +592,8 @@ func TestMultipartReaderErrors(t *testing.T) {
 			Post("/upload")
 
 		assertNotNil(t, err)
-		assertTrue(t, errors.Is(err, fs.ErrNotExist))
-		assertNotNil(t, resp)
-		assertEqual(t, nil, resp.Body)
+		assertEqual(t, true, errors.Is(err, fs.ErrNotExist))
+		assertNil(t, resp)
 	})
 }
 
@@ -668,6 +666,89 @@ func TestMultipartRequest_createMultipart(t *testing.T) {
 	})
 }
 
+func TestMultipartUploadFailAutoErrorParse(t *testing.T) {
+	type ErrorResponse struct {
+		Code    int    `json:"code"`
+		Message string `json:"message"`
+	}
+
+	ts := createTestServer(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set(hdrContentTypeKey, "application/json")
+		w.WriteHeader(http.StatusForbidden)
+		_, _ = w.Write([]byte(`{ "code": 403, "message": "forbidden error message" }`))
+	})
+	defer ts.Close()
+
+	c := dcnl()
+
+	t.Run("single request", func(t *testing.T) {
+		res, err := c.R().
+			SetFile("profile_img", filepath.Join(getTestDataPath(), "test-img.png")).
+			SetResultError(&ErrorResponse{}).
+			Post(ts.URL)
+
+		assertErrorIs(t, io.ErrClosedPipe, err)
+		assertEqual(t, http.StatusForbidden, res.StatusCode())
+
+		er := res.ResultError().(*ErrorResponse)
+		assertEqual(t, 403, er.Code)
+		assertEqual(t, "forbidden error message", er.Message)
+	})
+
+	t.Run("concurrent requests", func(t *testing.T) {
+		concurrencyCount := 100
+		wg := sync.WaitGroup{}
+		for i := 0; i < concurrencyCount; i++ {
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				res, _ := c.R().
+					SetFile("profile_img", filepath.Join(getTestDataPath(), "test-img.png")).
+					SetResultError(&ErrorResponse{}).
+					Post(ts.URL)
+
+				er := res.ResultError().(*ErrorResponse)
+				assertEqual(t, http.StatusForbidden, res.StatusCode())
+				assertEqual(t, 403, er.Code)
+				assertEqual(t, "forbidden error message", er.Message)
+			}()
+		}
+		wg.Wait()
+	})
+
+}
+
+func TestMultipartConcurrentRequests(t *testing.T) {
+	ts := createFormPostServer(t)
+	defer ts.Close()
+	defer cleanupFiles(".testdata/upload")
+
+	c := dcnl()
+	c.SetFormData(map[string]string{"zip_code": "00001", "city": "Los Angeles"})
+
+	concurrencyCount := 100
+	wg := sync.WaitGroup{}
+	for i := 0; i < concurrencyCount; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			res, err := c.R().
+				SetFormData(map[string]string{
+					"welcome1": "welcome value 1",
+					"welcome2": "welcome value 2",
+					"welcome3": "welcome value 3",
+				}).
+				SetFile("profile_img", filepath.Join(getTestDataPath(), "test-img.png")).
+				Post(ts.URL + "/upload")
+
+			assertError(t, err)
+			assertEqual(t, http.StatusOK, res.StatusCode())
+			assertEqual(t, true, strings.Contains(res.String(), "test-img.png"))
+		}()
+	}
+	wg.Wait()
+}
+
 type returnValueTestWriter struct {
 }
 

response.go

@@ -335,3 +335,13 @@ func (r *Response) wrapContentDecompresser() error {
 
 	return nil
 }
+
+func (r *Response) wrapError(err error, preserve bool) error {
+	r.CascadeError = wrapErrors(err, r.CascadeError)
+	if preserve {
+		return nil
+	}
+	e := r.CascadeError
+	r.CascadeError = nil
+	return e
+}

resty_test.go

@@ -355,6 +355,8 @@ func createPostServer(t *testing.T) *httptest.Server {
 
 func createFormPostServer(t *testing.T) *httptest.Server {
 	ts := createTestServer(func(w http.ResponseWriter, r *http.Request) {
+		t.Logf("Content-Type: %v", r.Header.Get(hdrConnectionKey))
+
 		if r.Method == MethodPost {
 			_ = r.ParseMultipartForm(10e6)
 
@@ -406,9 +408,9 @@ func createFormPostServer(t *testing.T) *httptest.Server {
 						defer func() {
 							_ = f.Close()
 						}()
-						_, _ = io.Copy(f, infile)
+						size, _ := io.Copy(f, infile)
 
-						_, _ = w.Write([]byte(fmt.Sprintf("File: %v, uploaded as: %v\n", hdr.Filename, fname)))
+						_, _ = w.Write([]byte(fmt.Sprintf("File: %v, uploaded as: %v, size: %v\n", hdr.Filename, fname, size)))
 					}
 				}