s3: return IncompleteBody instead of 500 for truncated PUT bodies (#10186)

* s3: add IncompleteBody error code

* operation: tag a truncated source read as ErrTruncatedBody

The chunked uploader wraps both source-read failures and volume-upload failures,
and a mid-write volume server drop also carries io.ErrUnexpectedEOF. Tag only the
source read so callers can tell a truncated input apart from a server-side fault.

* s3: return IncompleteBody for a truncated PUT body

A client abort or reverse-proxy timeout truncates the request body mid-upload.
putToFiler mapped every streaming-upload failure to InternalError (500), which a
reverse proxy relays as a 502. Classify a source-read truncation as IncompleteBody
(400) so the response matches AWS and passes through. All S3 write paths share
putToFiler, so they all benefit.

weed/operation/upload_chunked.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"crypto/md5"
 	"encoding/base64"
+	"errors"
 	"fmt"
 	"hash"
 	"io"
@@ -18,6 +19,11 @@ import (
 	util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
 )
 
+// ErrTruncatedBody tags a source read that ended before the expected bytes
+// arrived, so callers can tell a truncated input (client abort, reverse-proxy
+// timeout) apart from a volume-server upload fault.
+var ErrTruncatedBody = errors.New("truncated request body")
+
 // ChunkedUploadResult contains the result of a chunked upload
 type ChunkedUploadResult struct {
 	FileChunks   []*filer_pb.FileChunk
@@ -104,6 +110,9 @@ uploadLoop:
 			// before any data (offset=0,got=0) from mid-stream truncation.
 			// A bare io.ErrUnexpectedEOF is not actionable on its own (see #9149).
 			wrapped := fmt.Errorf("read chunk at offset %d (got %d bytes): %w", chunkOffset, dataSize, err)
+			if errors.Is(err, io.ErrUnexpectedEOF) {
+				wrapped = fmt.Errorf("%w: %w", ErrTruncatedBody, wrapped)
+			}
 			glog.V(2).Infof("UploadReaderInChunks: %v", wrapped)
 			chunkBufferPool.Put(bytesBuffer)
 			<-bytesBufferLimitChan

weed/operation/upload_chunked_test.go

@@ -372,3 +372,46 @@ func TestUploadReaderInChunksReaderFailure(t *testing.T) {
 	t.Logf("✓ Got partial result on read failure: chunks=%d, totalSize=%d",
 		len(result.FileChunks), result.TotalSize)
 }
+
+// truncatedReader yields some bytes then reports io.ErrUnexpectedEOF, mimicking a
+// request body cut short by a client abort or reverse-proxy timeout.
+type truncatedReader struct {
+	data []byte
+	pos  int
+}
+
+func (r *truncatedReader) Read(p []byte) (int, error) {
+	if r.pos >= len(r.data) {
+		return 0, io.ErrUnexpectedEOF
+	}
+	n := copy(p, r.data[r.pos:])
+	r.pos += n
+	return n, nil
+}
+
+func TestUploadReaderInChunksTagsTruncatedBody(t *testing.T) {
+	reader := &truncatedReader{data: bytes.Repeat([]byte("x"), 10000)}
+
+	assignFunc := func(ctx context.Context, count int, expectedDataSize uint64) (*VolumeAssignRequest, *AssignResult, error) {
+		return nil, &AssignResult{Fid: "test-fid,1234", Url: "http://test-volume:8080", Count: 1}, nil
+	}
+	uploadFunc := func(ctx context.Context, data []byte, option *UploadOption) (*UploadResult, error) {
+		return &UploadResult{Size: uint32(len(data))}, nil
+	}
+
+	_, err := UploadReaderInChunks(context.Background(), reader, &ChunkedUploadOption{
+		ChunkSize:  8 * 1024,
+		Collection: "test",
+		AssignFunc: assignFunc,
+		UploadFunc: uploadFunc,
+	})
+	if err == nil {
+		t.Fatal("expected an error for a truncated body")
+	}
+	if !errors.Is(err, ErrTruncatedBody) {
+		t.Errorf("expected ErrTruncatedBody, got %v", err)
+	}
+	if !errors.Is(err, io.ErrUnexpectedEOF) {
+		t.Errorf("expected io.ErrUnexpectedEOF to remain in the chain, got %v", err)
+	}
+}