diff --git a/weed/s3api/filer_multipart.go b/weed/s3api/filer_multipart.go index b02fcf501..1d26aba5f 100644 --- a/weed/s3api/filer_multipart.go +++ b/weed/s3api/filer_multipart.go @@ -205,6 +205,11 @@ type multipartPartBoundary struct { StartChunk int `json:"start"` EndChunk int `json:"end"` ETag string `json:"etag"` + // Byte offsets of the part within the object. Readers prefer these over + // the chunk indexes above, which stop matching the entry's chunk list + // once large completions fold it into manifest chunks. + StartOffset int64 `json:"startOffset,omitempty"` + EndOffset int64 `json:"endOffset,omitempty"` // exclusive } type multipartSSES3Info struct { @@ -492,7 +497,17 @@ func (s3a *S3ApiServer) prepareMultipartCompletionState(r *http.Request, input * continue } + // A part entry can itself carry manifest chunks (the filer folds an + // oversized part, e.g. a large UploadPartCopy range). Resolve them + // first: the rebase below shifts chunk offsets, which a manifest + // chunk cannot express. + if flattenErr := s3a.flattenManifestChunks(r.Context(), entry); flattenErr != nil { + glog.Errorf("completeMultipartUpload %s %s part %d resolve manifest chunks: %v", *input.Bucket, *input.UploadId, partNumber, flattenErr) + return nil, nil, s3err.ErrInternalError + } + partStartChunk := len(finalParts) + partStartOffset := offset partETag := getEtagFromEntry(entry) for _, chunk := range entry.GetChunks() { @@ -507,10 +522,12 @@ func (s3a *S3ApiServer) prepareMultipartCompletionState(r *http.Request, input * partEndChunk := len(finalParts) partBoundaries = append(partBoundaries, multipartPartBoundary{ - PartNumber: partNumber, - StartChunk: partStartChunk, - EndChunk: partEndChunk, - ETag: partETag, + PartNumber: partNumber, + StartChunk: partStartChunk, + EndChunk: partEndChunk, + ETag: partETag, + StartOffset: partStartOffset, + EndOffset: offset, }) found = true @@ -554,6 +571,11 @@ func (s3a *S3ApiServer) prepareMultipartCompletionState(r *http.Request, input * } } + // Fold huge completions (>filer.ManifestBatch chunks) into manifest chunks + // so the object entry stays small. Runs after the part boundaries above + // captured their byte offsets against the flat list. + finalParts = s3a.manifestizeChunks(dirName+"/"+entryName, *input.Bucket, 0, finalParts) + return &multipartCompletionState{ deleteEntries: deleteEntries, partEntries: partEntries, diff --git a/weed/s3api/s3api_chunk_manifest.go b/weed/s3api/s3api_chunk_manifest.go new file mode 100644 index 000000000..428250c2e --- /dev/null +++ b/weed/s3api/s3api_chunk_manifest.go @@ -0,0 +1,61 @@ +package s3api + +import ( + "context" + "io" + "math" + + "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" +) + +// saveManifestChunk returns the save function MaybeManifestize uses to store +// manifest blobs, assigning volumes against the object's real filer path so +// placement follows the bucket's storage rules. +func (s3a *S3ApiServer) saveManifestChunk(filePath string, bucket string, ttlSec int32) filer.SaveDataAsChunkFunctionType { + collection := "" + if s3a.option.FilerGroup != "" { + collection = s3a.getCollectionName(bucket) + } + return func(reader io.Reader, name string, offset int64, tsNs int64, expectedDataSize uint64) (*filer_pb.FileChunk, error) { + return filer.SaveGatewayDataAsChunk(filer.GatewayChunkUploadRequest{ + FilerClient: s3a, + Reader: reader, + FullPath: filePath, + Offset: offset, + TsNs: tsNs, + Collection: collection, + TtlSec: ttlSec, + DataCenter: s3a.option.DataCenter, + Cipher: s3a.cipher, + }) + } +} + +// manifestizeChunks folds a large flat chunk list into manifest chunks +// (filer.ManifestBatch data chunks per manifest). On failure the flat list is +// returned so the write can still proceed, matching the filer's soft-fail. +func (s3a *S3ApiServer) manifestizeChunks(filePath string, bucket string, ttlSec int32, chunks []*filer_pb.FileChunk) []*filer_pb.FileChunk { + manifested, err := filer.MaybeManifestize(s3a.saveManifestChunk(filePath, bucket, ttlSec), chunks) + if err != nil { + glog.V(0).Infof("MaybeManifestize %s: %v", filePath, err) + return chunks + } + return manifested +} + +// flattenManifestChunks resolves any manifest chunks on a copy source into the +// flat data-chunk list, so per-chunk copy logic reads real data chunks instead +// of raw manifest blobs. +func (s3a *S3ApiServer) flattenManifestChunks(ctx context.Context, entry *filer_pb.Entry) error { + if entry == nil || !filer.HasChunkManifest(entry.GetChunks()) { + return nil + } + dataChunks, _, err := filer.ResolveChunkManifest(ctx, s3a.createLookupFileIdFunction(), entry.GetChunks(), 0, math.MaxInt64) + if err != nil { + return err + } + entry.Chunks = dataChunks + return nil +} diff --git a/weed/s3api/s3api_object_handlers.go b/weed/s3api/s3api_object_handlers.go index 3c9427fa4..034d99e22 100644 --- a/weed/s3api/s3api_object_handlers.go +++ b/weed/s3api/s3api_object_handlers.go @@ -853,8 +853,18 @@ func (s3a *S3ApiServer) GetObjectHandler(w http.ResponseWriter, r *http.Request) // Note: ETag is NOT overridden - AWS S3 returns the complete object's ETag // even when requesting a specific part via PartNumber var startOffset, endOffset int64 - if partInfo != nil { - // Use part boundaries from metadata (accurate for multi-chunk parts) + if partInfo != nil && partInfo.EndOffset > partInfo.StartOffset { + // Byte-offset boundaries: valid regardless of how the entry's + // chunk list is laid out (flat or manifest chunks) + startOffset = partInfo.StartOffset + endOffset = partInfo.EndOffset - 1 + } else if partInfo != nil { + // Legacy boundaries carry chunk indexes into the flat chunk list + if partInfo.StartChunk < 0 || partInfo.EndChunk <= partInfo.StartChunk || partInfo.EndChunk > len(objectEntryForSSE.Chunks) { + glog.Errorf("GetObject: part %d boundary chunks [%d,%d) out of range (chunks: %d)", partNumber, partInfo.StartChunk, partInfo.EndChunk, len(objectEntryForSSE.Chunks)) + s3err.WriteErrorResponse(w, r, s3err.ErrInternalError) + return + } startOffset = objectEntryForSSE.Chunks[partInfo.StartChunk].Offset lastChunk := objectEntryForSSE.Chunks[partInfo.EndChunk-1] endOffset = lastChunk.Offset + int64(lastChunk.Size) - 1 @@ -3037,6 +3047,11 @@ type PartBoundaryInfo struct { StartChunk int `json:"start"` EndChunk int `json:"end"` // exclusive ETag string `json:"etag"` + // Byte offsets of the part within the object; preferred over the chunk + // indexes, which stop matching once the entry's chunk list is folded + // into manifest chunks. Zero EndOffset means a legacy boundary record. + StartOffset int64 `json:"startOffset,omitempty"` + EndOffset int64 `json:"endOffset,omitempty"` // exclusive } // rc is a helper type that wraps a Reader and Closer for proper resource cleanup diff --git a/weed/s3api/s3api_object_handlers_attributes.go b/weed/s3api/s3api_object_handlers_attributes.go index 8452b99fd..4f372e256 100644 --- a/weed/s3api/s3api_object_handlers_attributes.go +++ b/weed/s3api/s3api_object_handlers_attributes.go @@ -309,7 +309,9 @@ func (s3a *S3ApiServer) buildObjectAttributesParts(entry *filer_pb.Entry, maxPar } var partSize int64 - if b.StartChunk >= 0 && b.EndChunk >= 0 && b.StartChunk < len(chunks) && b.EndChunk <= len(chunks) && b.StartChunk < b.EndChunk { + if b.EndOffset > b.StartOffset { + partSize = b.EndOffset - b.StartOffset + } else if b.StartChunk >= 0 && b.EndChunk >= 0 && b.StartChunk < len(chunks) && b.EndChunk <= len(chunks) && b.StartChunk < b.EndChunk { for ci := b.StartChunk; ci < b.EndChunk; ci++ { partSize += int64(chunks[ci].Size) } diff --git a/weed/s3api/s3api_object_handlers_copy.go b/weed/s3api/s3api_object_handlers_copy.go index abec42af9..002d9fea9 100644 --- a/weed/s3api/s3api_object_handlers_copy.go +++ b/weed/s3api/s3api_object_handlers_copy.go @@ -423,7 +423,9 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request return } - dstEntry.Chunks = dstChunks + // Re-fold a large copied chunk list into manifest chunks, mirroring the + // PutObject path (no-op below filer.ManifestBatch or for SSE chunks). + dstEntry.Chunks = s3a.manifestizeChunks(fmt.Sprintf("%s/%s", s3a.bucketDir(dstBucket), dstObject), dstBucket, 0, dstChunks) // Apply destination-specific metadata (e.g., SSE-C IV and headers) if dstMetadata != nil { @@ -917,6 +919,14 @@ func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Req entry = cachedEntry } + // The part-copy paths below iterate entry.GetChunks() per chunk, so a + // manifested source must be resolved into its data chunks first. + if err := s3a.flattenManifestChunks(r.Context(), entry); err != nil { + glog.Errorf("CopyObjectPartHandler: resolve source manifest chunks %s/%s: %v", srcBucket, srcObject, err) + s3err.WriteErrorResponse(w, r, s3err.ErrInternalError) + return + } + // Validate conditional copy headers if err := s3a.validateConditionalCopyHeaders(r, entry); err != s3err.ErrNone { s3err.WriteErrorResponse(w, r, err) diff --git a/weed/s3api/s3api_object_handlers_copy_unified.go b/weed/s3api/s3api_object_handlers_copy_unified.go index 7146f5b5b..d41dd9ad4 100644 --- a/weed/s3api/s3api_object_handlers_copy_unified.go +++ b/weed/s3api/s3api_object_handlers_copy_unified.go @@ -14,6 +14,13 @@ import ( // executeUnifiedCopyStrategy executes the appropriate copy strategy based on encryption state // Returns chunks and destination metadata that should be applied to the destination entry func (s3a *S3ApiServer) executeUnifiedCopyStrategy(entry *filer_pb.Entry, r *http.Request, srcBucket, dstBucket, srcObject, dstObject string) ([]*filer_pb.FileChunk, map[string][]byte, error) { + // The per-chunk copy paths below must see real data chunks; copying a + // manifest chunk raw would store its blob as object data pointing at the + // source's chunks. + if err := s3a.flattenManifestChunks(r.Context(), entry); err != nil { + return nil, nil, fmt.Errorf("resolve source manifest chunks: %w", err) + } + // Detect encryption state (using entry-aware detection for multipart objects) srcPath := fmt.Sprintf("%s/%s", s3a.bucketDir(srcBucket), srcObject) dstPath := fmt.Sprintf("%s/%s", s3a.bucketDir(dstBucket), dstObject) diff --git a/weed/s3api/s3api_object_handlers_put.go b/weed/s3api/s3api_object_handlers_put.go index e83eb0761..d7c7dda22 100644 --- a/weed/s3api/s3api_object_handlers_put.go +++ b/weed/s3api/s3api_object_handlers_put.go @@ -807,6 +807,13 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader glog.V(3).Infof("putToFiler: storing SSE-S3 metadata - keyID=%s, raw len=%d", sseS3Key.KeyID, len(sseS3Metadata)) } + // Fold large flat chunk lists into manifest chunks before creating the + // entry. Part uploads (object == "") stay flat: completion rebases their + // offsets into the final object, which manifest chunks cannot express. + if object != "" { + entry.Chunks = s3a.manifestizeChunks(filePath, bucket, lifecycleTTLSec, entry.GetChunks()) + } + // Step 4: Save metadata to filer via gRPC // Use context.Background() to ensure metadata save completes even if HTTP request is cancelled // This matches the chunk upload behavior and prevents orphaned chunks @@ -892,9 +899,15 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader } // If the entry was never created, the uploaded chunks are orphaned and must be deleted. - if !entryCreated && len(chunkResult.FileChunks) > 0 { - glog.Warningf("putToFiler: finalization failed, attempting to cleanup %d orphaned chunks", len(chunkResult.FileChunks)) - s3a.deleteOrphanedChunks(chunkResult.FileChunks) + if !entryCreated { + orphaned := chunkResult.FileChunks + if manifestChunks, _ := filer.SeparateManifestChunks(entry.GetChunks()); len(manifestChunks) > 0 { + orphaned = append(manifestChunks, orphaned...) + } + if len(orphaned) > 0 { + glog.Warningf("putToFiler: finalization failed, attempting to cleanup %d orphaned chunks", len(orphaned)) + s3a.deleteOrphanedChunks(orphaned) + } } return "", createCode, SSEResponseMetadata{} diff --git a/weed/s3api/s3api_part_boundary_test.go b/weed/s3api/s3api_part_boundary_test.go new file mode 100644 index 000000000..d726372f3 --- /dev/null +++ b/weed/s3api/s3api_part_boundary_test.go @@ -0,0 +1,120 @@ +package s3api + +import ( + "encoding/json" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" +) + +// The writer (multipartPartBoundary) and reader (PartBoundaryInfo) structs are +// kept separate; their JSON keys must stay in sync, including the offset +// fields added for manifest-chunked entries. +func TestPartBoundaryJsonCompatibility(t *testing.T) { + written, err := json.Marshal([]multipartPartBoundary{ + {PartNumber: 1, StartChunk: 0, EndChunk: 2, ETag: "abc", StartOffset: 0, EndOffset: 16}, + {PartNumber: 2, StartChunk: 2, EndChunk: 3, ETag: "def", StartOffset: 16, EndOffset: 24}, + }) + if err != nil { + t.Fatal(err) + } + + var read []PartBoundaryInfo + if err := json.Unmarshal(written, &read); err != nil { + t.Fatal(err) + } + if len(read) != 2 { + t.Fatalf("expected 2 boundaries, got %d", len(read)) + } + if read[0].EndChunk != 2 || read[0].ETag != "abc" || read[0].StartOffset != 0 || read[0].EndOffset != 16 { + t.Errorf("boundary 1 mismatch: %+v", read[0]) + } + if read[1].StartChunk != 2 || read[1].StartOffset != 16 || read[1].EndOffset != 24 { + t.Errorf("boundary 2 mismatch: %+v", read[1]) + } + + // Legacy records carry no offset fields; they must unmarshal to zero so + // readers fall back to the chunk-index path. + var legacy []PartBoundaryInfo + if err := json.Unmarshal([]byte(`[{"part":1,"start":0,"end":2,"etag":"abc"}]`), &legacy); err != nil { + t.Fatal(err) + } + if legacy[0].StartOffset != 0 || legacy[0].EndOffset != 0 { + t.Errorf("legacy boundary should have zero offsets: %+v", legacy[0]) + } +} + +func partsEntry(t *testing.T, boundaries []multipartPartBoundary, chunks []*filer_pb.FileChunk) *filer_pb.Entry { + t.Helper() + boundariesJSON, err := json.Marshal(boundaries) + if err != nil { + t.Fatal(err) + } + return &filer_pb.Entry{ + Chunks: chunks, + Extended: map[string][]byte{ + s3_constants.SeaweedFSMultipartPartBoundaries: boundariesJSON, + }, + } +} + +func TestBuildObjectAttributesPartsPrefersOffsets(t *testing.T) { + s3a := &S3ApiServer{} + + // Entry whose chunk list was folded into a single manifest chunk: the + // stored chunk indexes no longer address the flat list, but the byte + // offsets still describe each part. + entry := partsEntry(t, []multipartPartBoundary{ + {PartNumber: 1, StartChunk: 0, EndChunk: 2, StartOffset: 0, EndOffset: 16}, + {PartNumber: 2, StartChunk: 2, EndChunk: 4, StartOffset: 16, EndOffset: 40}, + }, []*filer_pb.FileChunk{ + {FileId: "1,ab", Offset: 0, Size: 40, IsChunkManifest: true}, + }) + + parts := s3a.buildObjectAttributesParts(entry, 1000, 0) + if parts == nil || len(parts.Parts) != 2 { + t.Fatalf("expected 2 parts, got %+v", parts) + } + if parts.Parts[0].Size != 16 { + t.Errorf("part 1 size = %d, want 16", parts.Parts[0].Size) + } + if parts.Parts[1].Size != 24 { + t.Errorf("part 2 size = %d, want 24", parts.Parts[1].Size) + } +} + +func TestBuildObjectAttributesPartsLegacyChunkIndexes(t *testing.T) { + s3a := &S3ApiServer{} + + chunks := []*filer_pb.FileChunk{ + {FileId: "1,ab", Offset: 0, Size: 8}, + {FileId: "1,ac", Offset: 8, Size: 8}, + {FileId: "1,ad", Offset: 16, Size: 24}, + } + entry := partsEntry(t, []multipartPartBoundary{ + {PartNumber: 1, StartChunk: 0, EndChunk: 2}, + {PartNumber: 2, StartChunk: 2, EndChunk: 3}, + }, chunks) + + parts := s3a.buildObjectAttributesParts(entry, 1000, 0) + if parts == nil || len(parts.Parts) != 2 { + t.Fatalf("expected 2 parts, got %+v", parts) + } + if parts.Parts[0].Size != 16 { + t.Errorf("part 1 size = %d, want 16", parts.Parts[0].Size) + } + if parts.Parts[1].Size != 24 { + t.Errorf("part 2 size = %d, want 24", parts.Parts[1].Size) + } + + // Out-of-range legacy indexes (e.g. metadata from a differently shaped + // entry) must not panic; the part is reported with size 0. + badEntry := partsEntry(t, []multipartPartBoundary{ + {PartNumber: 1, StartChunk: 5, EndChunk: 9}, + }, chunks) + parts = s3a.buildObjectAttributesParts(badEntry, 1000, 0) + if parts == nil || len(parts.Parts) != 1 || parts.Parts[0].Size != 0 { + t.Fatalf("expected 1 part with size 0, got %+v", parts) + } +}