From 0716577ec8fdc16c9c1d66c8f369767fb5bb279e Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sun, 26 Apr 2026 02:17:55 -0700 Subject: [PATCH] fix(upload): rewind request body when retrying on connection reset (#9139) (#9222) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(upload): rewind request body when retrying on connection reset (#9139) When httpClient.Do() returned "connection reset by peer" or "use of closed network connection", upload_content retried with the same *http.Request. But the body is a *bytes.Reader the first attempt already consumed, so the retry sent 0 bytes and Go's transport surfaced "http: ContentLength=N with Body length 0". http.NewRequestWithContext populates req.GetBody for *bytes.Reader bodies; use it to attach a fresh body before retrying. Reproduces the issue with a unit test (asserts both attempts see the same payload bytes); the test fails without the fix. * upload: skip inner retry when body cannot be rewound Per review feedback: if req.GetBody is nil or returns an error, the inner retry would call Do(req) with an already-consumed body and the "connection reset" error would be replaced by the misleading "ContentLength=N with Body length 0" — the very symptom this PR set out to fix. Skip the inner retry on rewind failure and let the outer retriedUploadData loop reissue with a fresh request, and log when GetBody is unavailable for observability. * upload: log the actual transport error in the inner retry log line Per review feedback: the diagnostic glog at the top of the inner retry branch was logging postErr — the request-construction error from http.NewRequestWithContext, which is necessarily nil there because the function returns early at line 423 if it isn't. Operators were seeing "" instead of the transient transport error that triggered the rewind. Reference post_err so the connection-reset / closed-connection cause is actually visible. --- weed/operation/upload_content.go | 24 +++++++-- weed/operation/upload_content_test.go | 76 +++++++++++++++++++++++++++ 2 files changed, 97 insertions(+), 3 deletions(-) diff --git a/weed/operation/upload_content.go b/weed/operation/upload_content.go index 1312911ba..e16558cb7 100644 --- a/weed/operation/upload_content.go +++ b/weed/operation/upload_content.go @@ -439,10 +439,28 @@ func (uploader *Uploader) upload_content(ctx context.Context, fillBufferFunction if post_err != nil { if strings.Contains(post_err.Error(), "connection reset by peer") || strings.Contains(post_err.Error(), "use of closed network connection") { - glog.V(1).InfofCtx(ctx, "repeat error upload request %s: %v", option.UploadUrl, postErr) + glog.V(1).InfofCtx(ctx, "repeat error upload request %s: %v", option.UploadUrl, post_err) stats.FilerHandlerCounter.WithLabelValues(stats.RepeatErrorUploadContent).Inc() - resp, post_err = uploader.httpClient.Do(req) - defer util_http.CloseResponse(resp) + // The first attempt already consumed (or partially consumed) the + // body, so retrying with the same *http.Request would send 0 bytes + // and Go's transport would surface "ContentLength=N with Body + // length 0". http.NewRequestWithContext sets GetBody for + // *bytes.Reader bodies; use it to attach a fresh body for retry. + // If we can't rewind, skip the inner retry and let the outer + // retriedUploadData loop reissue the request with a fresh body — + // retrying here with a consumed body would mask the original + // "connection reset" error with a misleading "Body length 0". + if req.GetBody != nil { + if newBody, gbErr := req.GetBody(); gbErr == nil { + req.Body = newBody + resp, post_err = uploader.httpClient.Do(req) + defer util_http.CloseResponse(resp) + } else { + glog.V(1).InfofCtx(ctx, "skip inner retry for %s: GetBody returned %v", option.UploadUrl, gbErr) + } + } else { + glog.V(1).InfofCtx(ctx, "skip inner retry for %s: req.GetBody is nil", option.UploadUrl) + } } } if post_err != nil { diff --git a/weed/operation/upload_content_test.go b/weed/operation/upload_content_test.go index 501d6c318..a045eea03 100644 --- a/weed/operation/upload_content_test.go +++ b/weed/operation/upload_content_test.go @@ -1,6 +1,9 @@ package operation import ( + "bytes" + "context" + "errors" "fmt" "io" "net/http" @@ -122,3 +125,76 @@ func TestUploadWithRetryDataReassignsOnVolumeSizeExceeded(t *testing.T) { t.Fatalf("unexpected upload call sequence: %#v", httpClient.calls) } } + +// bodyCapturingHTTPClient drains req.Body on every Do, optionally failing the +// first attempt with a transport error so we can verify upload_content rewinds +// the body before retrying. +type bodyCapturingHTTPClient struct { + mu sync.Mutex + bodies [][]byte + failFirst string + successJSON string +} + +func (c *bodyCapturingHTTPClient) Do(req *http.Request) (*http.Response, error) { + c.mu.Lock() + defer c.mu.Unlock() + + var captured []byte + if req.Body != nil { + b, err := io.ReadAll(req.Body) + if err != nil { + return nil, err + } + captured = b + } + c.bodies = append(c.bodies, captured) + + if len(c.bodies) == 1 && c.failFirst != "" { + return nil, errors.New(c.failFirst) + } + return &http.Response{ + StatusCode: http.StatusCreated, + Header: make(http.Header), + Body: io.NopCloser(strings.NewReader(c.successJSON)), + }, nil +} + +// TestUploadRewindsBodyOnConnectionReset reproduces issue #9139 follow-up: +// when the inner Do retry fires on a "connection reset" / "closed network" +// error, the *bytes.Reader body has already been consumed, so without an +// explicit rewind the second attempt sends 0 bytes and Go's transport surfaces +// "ContentLength=N with Body length 0". +func TestUploadRewindsBodyOnConnectionReset(t *testing.T) { + for _, transient := range []string{ + "connection reset by peer", + "use of closed network connection", + } { + t.Run(transient, func(t *testing.T) { + client := &bodyCapturingHTTPClient{ + failFirst: transient, + successJSON: `{"name":"test.bin","size":42}`, + } + uploader := newUploader(client) + + payload := bytes.Repeat([]byte("payload-"), 256) // 2048 bytes + _, err := uploader.UploadData(context.Background(), payload, &UploadOption{ + UploadUrl: "http://volume/1,abc", + Filename: "test.bin", + }) + if err != nil { + t.Fatalf("upload should succeed after inner retry, got %v", err) + } + if len(client.bodies) != 2 { + t.Fatalf("expected 2 Do attempts, got %d", len(client.bodies)) + } + if len(client.bodies[0]) == 0 { + t.Fatalf("first attempt sent an empty body; test setup wrong") + } + if !bytes.Equal(client.bodies[0], client.bodies[1]) { + t.Fatalf("retry body length=%d differs from first attempt length=%d (body was not rewound)", + len(client.bodies[1]), len(client.bodies[0])) + } + }) + } +}