diff --git a/weed/operation/upload_chunked.go b/weed/operation/upload_chunked.go index 84aba9f82..ec40df623 100644 --- a/weed/operation/upload_chunked.go +++ b/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 diff --git a/weed/operation/upload_chunked_test.go b/weed/operation/upload_chunked_test.go index 0f09baa49..3c457200c 100644 --- a/weed/operation/upload_chunked_test.go +++ b/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) + } +} diff --git a/weed/s3api/s3api_object_handlers_put.go b/weed/s3api/s3api_object_handlers_put.go index 9e7736b44..e83eb0761 100644 --- a/weed/s3api/s3api_object_handlers_put.go +++ b/weed/s3api/s3api_object_handlers_put.go @@ -545,10 +545,7 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader s3a.deleteOrphanedChunks(chunkResult.FileChunks) } - if strings.Contains(err.Error(), s3err.ErrMsgPayloadChecksumMismatch) { - return "", s3err.ErrInvalidDigest, SSEResponseMetadata{} - } - return "", s3err.ErrInternalError, SSEResponseMetadata{} + return "", mapChunkedUploadErrorToS3Error(err), SSEResponseMetadata{} } // Step 3: Calculate MD5 hash and add SSE metadata to chunks @@ -1170,6 +1167,22 @@ func filerErrorToS3Error(err error) s3err.ErrorCode { } } +// mapChunkedUploadErrorToS3Error classifies a failed streaming upload. A truncated +// request body (client abort or reverse-proxy timeout) is a request error, so report +// IncompleteBody (400) rather than a 500 a reverse proxy would relay as a confusing +// 502. Only the source read is tagged, so a volume-server upload fault still maps to +// InternalError. +func mapChunkedUploadErrorToS3Error(err error) s3err.ErrorCode { + switch { + case strings.Contains(err.Error(), s3err.ErrMsgPayloadChecksumMismatch): + return s3err.ErrInvalidDigest + case errors.Is(err, operation.ErrTruncatedBody): + return s3err.ErrIncompleteBody + default: + return s3err.ErrInternalError + } +} + // setObjectOwnerFromRequest sets the object owner metadata based on the bucket ownership policy. // When BucketOwnerEnforced (the modern AWS default), the bucket owner owns all objects. // Otherwise, the uploader's account ID is used (ObjectWriter mode). diff --git a/weed/s3api/s3api_object_handlers_put_error_test.go b/weed/s3api/s3api_object_handlers_put_error_test.go new file mode 100644 index 000000000..ff70b12fd --- /dev/null +++ b/weed/s3api/s3api_object_handlers_put_error_test.go @@ -0,0 +1,51 @@ +package s3api + +import ( + "errors" + "fmt" + "io" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/operation" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3err" +) + +func TestMapChunkedUploadErrorToS3Error(t *testing.T) { + tests := []struct { + name string + err error + want s3err.ErrorCode + }{ + { + // A truncated body (client abort or reverse-proxy timeout) reaches + // putToFiler tagged exactly like UploadReaderInChunks reports it. + name: "truncated source read maps to IncompleteBody", + err: fmt.Errorf("%w: read chunk at offset %d (got %d bytes): %w", operation.ErrTruncatedBody, 0, 8056500, io.ErrUnexpectedEOF), + want: s3err.ErrIncompleteBody, + }, + { + // A volume-server upload dropping mid-write is a server fault, not a + // client truncation, even though it also carries io.ErrUnexpectedEOF. + name: "volume upload unexpected EOF maps to InternalError", + err: fmt.Errorf("upload chunk: %w", io.ErrUnexpectedEOF), + want: s3err.ErrInternalError, + }, + { + name: "payload checksum mismatch maps to InvalidDigest", + err: errors.New(s3err.ErrMsgPayloadChecksumMismatch), + want: s3err.ErrInvalidDigest, + }, + { + name: "other errors map to InternalError", + err: errors.New("assign volume: no free volumes"), + want: s3err.ErrInternalError, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := mapChunkedUploadErrorToS3Error(tt.err); got != tt.want { + t.Errorf("mapChunkedUploadErrorToS3Error(%v) = %v, want %v", tt.err, got, tt.want) + } + }) + } +} diff --git a/weed/s3api/s3err/s3api_errors.go b/weed/s3api/s3err/s3api_errors.go index 76cd81075..e4281feab 100644 --- a/weed/s3api/s3err/s3api_errors.go +++ b/weed/s3api/s3err/s3api_errors.go @@ -155,6 +155,9 @@ const ( ErrKeyTooLongError ErrNoSuchConfiguration + + // Truncated request body (fewer bytes than Content-Length) + ErrIncompleteBody ) // Error message constants for checksum validation @@ -297,6 +300,11 @@ var errorCodeResponse = map[ErrorCode]APIError{ Description: "We encountered an internal error, please try again.", HTTPStatusCode: http.StatusInternalServerError, }, + ErrIncompleteBody: { + Code: "IncompleteBody", + Description: "You did not provide the number of bytes specified by the Content-Length HTTP header.", + HTTPStatusCode: http.StatusBadRequest, + }, ErrInvalidPart: { Code: "InvalidPart",