fix(upload): rewind request body when retrying on connection reset (#9139) (#9222)

* 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 "<nil>" 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

@@ -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 {

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]))
+			}
+		})
+	}
+}