diff --git a/weed/s3api/filer_multipart.go b/weed/s3api/filer_multipart.go index 1d26aba5f..8af6a5dac 100644 --- a/weed/s3api/filer_multipart.go +++ b/weed/s3api/filer_multipart.go @@ -205,9 +205,8 @@ 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. + // Byte offsets of the part; the chunk indexes above stop matching once + // the chunk list is folded into manifest chunks. StartOffset int64 `json:"startOffset,omitempty"` EndOffset int64 `json:"endOffset,omitempty"` // exclusive } @@ -232,6 +231,11 @@ type multipartCompletionState struct { checksumHeaderName string // e.g. "X-Amz-Checksum-Crc32", empty if no checksum checksumValue string // multipart checksum: "base64-N" (COMPOSITE) or "base64" (FULL_OBJECT) checksumType string // s3_constants.ChecksumType* (COMPOSITE or FULL_OBJECT) + + newManifestChunks []*filer_pb.FileChunk // blobs folded for finalParts; orphans until the entry commits + manifestsReferenced bool // failed rollback left an entry holding newManifestChunks + supersededPartManifests []*filer_pb.FileChunk // part-entry blobs replaced by flattening; deleted after commit + metadataOnlyCleanup bool // deleteEntries share chunks with the live object; keep their data } func completeMultipartResult(r *http.Request, input *s3.CompleteMultipartUploadInput, etag string, entry *filer_pb.Entry) *CompleteMultipartUploadResult { @@ -378,7 +382,7 @@ func (s3a *S3ApiServer) prepareMultipartCompletionState(r *http.Request, input * if cleanupErr != nil && !errors.Is(cleanupErr, filer_pb.ErrNotFound) { glog.Warningf("completeMultipartUpload: failed to list stale upload directory %s for cleanup: %v", uploadDirectory, cleanupErr) } - return &multipartCompletionState{deleteEntries: cleanupEntries}, completeMultipartResult(r, input, getEtagFromEntry(entry), entry), s3err.ErrNone + return &multipartCompletionState{deleteEntries: cleanupEntries, metadataOnlyCleanup: true}, completeMultipartResult(r, input, getEtagFromEntry(entry), entry), s3err.ErrNone } } @@ -476,6 +480,7 @@ func (s3a *S3ApiServer) prepareMultipartCompletionState(r *http.Request, input * } finalParts := make([]*filer_pb.FileChunk, 0) partBoundaries := make([]multipartPartBoundary, 0, len(completedPartNumbers)) + var supersededPartManifests []*filer_pb.FileChunk var offset int64 for _, partNumber := range completedPartNumbers { @@ -497,14 +502,14 @@ 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 { + // a part may carry manifest chunks (the filer folds oversized part + // copies); the offset rebase below needs flat chunks + removedManifests, flattenErr := s3a.flattenManifestChunks(r.Context(), entry) + if flattenErr != nil { glog.Errorf("completeMultipartUpload %s %s part %d resolve manifest chunks: %v", *input.Bucket, *input.UploadId, partNumber, flattenErr) return nil, nil, s3err.ErrInternalError } + supersededPartManifests = append(supersededPartManifests, removedManifests...) partStartChunk := len(finalParts) partStartOffset := offset @@ -571,25 +576,27 @@ 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. + // Fold after the boundaries captured byte offsets; finalParts is flat + // here, so anything folded out is newly created. finalParts = s3a.manifestizeChunks(dirName+"/"+entryName, *input.Bucket, 0, finalParts) + newManifestChunks, _ := filer.SeparateManifestChunks(finalParts) return &multipartCompletionState{ - deleteEntries: deleteEntries, - partEntries: partEntries, - pentry: pentry, - sses3Info: sses3Info, - mime: mime, - finalParts: finalParts, - offset: offset, - partBoundaries: partBoundaries, - multipartETag: calculateMultipartETag(partEntries, completedPartNumbers), - entityWithTtl: entityWithTtl, - checksumHeaderName: checksumHeaderName, - checksumValue: checksumValue, - checksumType: checksumType, + deleteEntries: deleteEntries, + partEntries: partEntries, + pentry: pentry, + sses3Info: sses3Info, + mime: mime, + finalParts: finalParts, + offset: offset, + partBoundaries: partBoundaries, + multipartETag: calculateMultipartETag(partEntries, completedPartNumbers), + entityWithTtl: entityWithTtl, + checksumHeaderName: checksumHeaderName, + checksumValue: checksumValue, + checksumType: checksumType, + newManifestChunks: newManifestChunks, + supersededPartManifests: supersededPartManifests, }, nil, s3err.ErrNone } @@ -733,12 +740,14 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl if code := s3a.routedVersionedFinalize(owner, *input.Bucket, *input.Key, useInvertedFormat); code != s3err.ErrNone { if rollbackErr := s3a.rollbackMultipartVersion(versionDir, versionFileName); rollbackErr != nil { glog.Errorf("completeMultipartUpload: failed to rollback version %s for %s/%s after routed finalize error: %v", versionId, *input.Bucket, *input.Key, rollbackErr) + completionState.manifestsReferenced = true } return code } } else if err := s3a.updateLatestVersionInDirectory(*input.Bucket, *input.Key, versionId, versionFileName, versionEntryForCache); err != nil { if rollbackErr := s3a.rollbackMultipartVersion(versionDir, versionFileName); rollbackErr != nil { glog.Errorf("completeMultipartUpload: failed to rollback version %s for %s/%s after latest pointer update error: %v", versionId, *input.Bucket, *input.Key, rollbackErr) + completionState.manifestsReferenced = true } glog.Errorf("completeMultipartUpload: failed to update latest version in directory: %v", err) return s3err.ErrInternalError @@ -910,25 +919,34 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl }, completionBody) } if finalizeCode != s3err.ErrNone { + // this attempt's manifests are orphans unless a failed rollback left an entry holding them + if completionState != nil && !completionState.manifestsReferenced && len(completionState.newManifestChunks) > 0 { + s3a.deleteOrphanedChunks(completionState.newManifestChunks) + } return nil, finalizeCode } if completionState != nil { for _, deleteEntry := range completionState.deleteEntries { - if err := s3a.rm(uploadDirectory, deleteEntry.Name, true, true); err != nil { + if err := s3a.rm(uploadDirectory, deleteEntry.Name, !completionState.metadataOnlyCleanup, true); err != nil { glog.Warningf("completeMultipartUpload cleanup %s upload %s unused %s : %v", *input.Bucket, *input.UploadId, deleteEntry.Name, err) } } if err := s3a.rm(s3a.genUploadsFolder(*input.Bucket), *input.UploadId, false, true); err != nil { glog.V(1).Infof("completeMultipartUpload cleanup %s upload %s: %v", *input.Bucket, *input.UploadId, err) } + if len(completionState.supersededPartManifests) > 0 { + s3a.deleteOrphanedChunks(completionState.supersededPartManifests) + } } return } +// Metadata-only: the version file's chunks are the still-registered parts' +// chunks, which a retried completion needs. func (s3a *S3ApiServer) rollbackMultipartVersion(versionDir, versionFileName string) error { - return s3a.rmObject(versionDir, versionFileName, true, false) + return s3a.rmObject(versionDir, versionFileName, false, false) } func (s3a *S3ApiServer) getEntryNameAndDir(input *s3.CompleteMultipartUploadInput) (string, string) { diff --git a/weed/s3api/s3api_chunk_manifest.go b/weed/s3api/s3api_chunk_manifest.go index 428250c2e..c25a36ca2 100644 --- a/weed/s3api/s3api_chunk_manifest.go +++ b/weed/s3api/s3api_chunk_manifest.go @@ -10,9 +10,8 @@ import ( "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. +// saveManifestChunk stores manifest blobs, assigning volumes against the +// object's filer path so bucket placement rules apply. func (s3a *S3ApiServer) saveManifestChunk(filePath string, bucket string, ttlSec int32) filer.SaveDataAsChunkFunctionType { collection := "" if s3a.option.FilerGroup != "" { @@ -33,29 +32,43 @@ func (s3a *S3ApiServer) saveManifestChunk(filePath string, bucket string, ttlSec } } -// 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. +// manifestizeChunks folds a large flat chunk list into manifest chunks. A +// failed fold falls back to the flat list, deleting any blobs it saved. 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) + return manifestizeOrKeepFlat(s3a.saveManifestChunk(filePath, bucket, ttlSec), s3a.deleteOrphanedChunks, filePath, chunks) +} + +func manifestizeOrKeepFlat(save filer.SaveDataAsChunkFunctionType, deleteChunks func([]*filer_pb.FileChunk), filePath string, chunks []*filer_pb.FileChunk) []*filer_pb.FileChunk { + var saved []*filer_pb.FileChunk + record := func(reader io.Reader, name string, offset int64, tsNs int64, expectedDataSize uint64) (*filer_pb.FileChunk, error) { + chunk, err := save(reader, name, offset, tsNs, expectedDataSize) + if err == nil { + saved = append(saved, chunk) + } + return chunk, err + } + manifested, err := filer.MaybeManifestize(record, chunks) if err != nil { glog.V(0).Infof("MaybeManifestize %s: %v", filePath, err) + if len(saved) > 0 { + deleteChunks(saved) + } 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 { +// flattenManifestChunks resolves manifest chunks into flat data chunks, +// returning the resolved-away manifests for callers that must delete the +// superseded blobs once they own the data chunks. +func (s3a *S3ApiServer) flattenManifestChunks(ctx context.Context, entry *filer_pb.Entry) ([]*filer_pb.FileChunk, error) { if entry == nil || !filer.HasChunkManifest(entry.GetChunks()) { - return nil + return nil, nil } - dataChunks, _, err := filer.ResolveChunkManifest(ctx, s3a.createLookupFileIdFunction(), entry.GetChunks(), 0, math.MaxInt64) + dataChunks, manifestChunks, err := filer.ResolveChunkManifest(ctx, s3a.createLookupFileIdFunction(), entry.GetChunks(), 0, math.MaxInt64) if err != nil { - return err + return nil, err } entry.Chunks = dataChunks - return nil + return manifestChunks, nil } diff --git a/weed/s3api/s3api_chunk_manifest_test.go b/weed/s3api/s3api_chunk_manifest_test.go new file mode 100644 index 000000000..8c3f373f5 --- /dev/null +++ b/weed/s3api/s3api_chunk_manifest_test.go @@ -0,0 +1,134 @@ +package s3api + +import ( + "fmt" + "io" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" +) + +func flatChunks(n int) []*filer_pb.FileChunk { + chunks := make([]*filer_pb.FileChunk, n) + for i := range chunks { + chunks[i] = &filer_pb.FileChunk{ + FileId: fmt.Sprintf("1,%x", i+1), + Offset: int64(i) * 8, + Size: 8, + } + } + return chunks +} + +type fakeManifestStore struct { + saves int + failOn int // 1-based save call to fail on; 0 = never + deleted []*filer_pb.FileChunk +} + +func (s *fakeManifestStore) save(reader io.Reader, name string, offset int64, tsNs int64, expectedDataSize uint64) (*filer_pb.FileChunk, error) { + s.saves++ + if s.saves == s.failOn { + return nil, fmt.Errorf("save %d failed", s.saves) + } + if _, err := io.Copy(io.Discard, reader); err != nil { + return nil, err + } + return &filer_pb.FileChunk{FileId: fmt.Sprintf("2,%x", s.saves), Offset: offset}, nil +} + +func (s *fakeManifestStore) delete(chunks []*filer_pb.FileChunk) { + s.deleted = append(s.deleted, chunks...) +} + +func TestManifestizeOrKeepFlatFolds(t *testing.T) { + store := &fakeManifestStore{} + chunks := flatChunks(filer.ManifestBatch + 50) + + result := manifestizeOrKeepFlat(store.save, store.delete, "/buckets/b/o", chunks) + + manifests, data := filer.SeparateManifestChunks(result) + if len(manifests) != 1 || len(data) != 50 { + t.Fatalf("expected 1 manifest + 50 flat chunks, got %d + %d", len(manifests), len(data)) + } + if store.saves != 1 || len(store.deleted) != 0 { + t.Errorf("expected 1 save and no deletes, got %d saves, %d deleted", store.saves, len(store.deleted)) + } + if manifests[0].Offset != 0 || manifests[0].Size != uint64(filer.ManifestBatch*8) { + t.Errorf("manifest span [%d,%d) wrong", manifests[0].Offset, manifests[0].Size) + } +} + +func TestManifestizeOrKeepFlatBelowThreshold(t *testing.T) { + store := &fakeManifestStore{} + chunks := flatChunks(filer.ManifestBatch - 1) + + result := manifestizeOrKeepFlat(store.save, store.delete, "/buckets/b/o", chunks) + + if len(result) != len(chunks) || store.saves != 0 { + t.Fatalf("expected untouched flat list, got %d chunks, %d saves", len(result), store.saves) + } +} + +func TestManifestizeOrKeepFlatSkipsSse(t *testing.T) { + store := &fakeManifestStore{} + chunks := flatChunks(filer.ManifestBatch + 50) + chunks[0].SseType = filer_pb.SSEType_SSE_S3 + + result := manifestizeOrKeepFlat(store.save, store.delete, "/buckets/b/o", chunks) + + if len(result) != len(chunks) || store.saves != 0 { + t.Fatalf("SSE chunks must not be folded, got %d chunks, %d saves", len(result), store.saves) + } +} + +// A fold that fails midway must fall back to the flat list and delete the +// manifest blobs its earlier batches already uploaded. +func TestManifestizeOrKeepFlatRollsBackPartialFold(t *testing.T) { + store := &fakeManifestStore{failOn: 2} + chunks := flatChunks(2*filer.ManifestBatch + 50) + + result := manifestizeOrKeepFlat(store.save, store.delete, "/buckets/b/o", chunks) + + if len(result) != len(chunks) { + t.Fatalf("expected fallback to %d flat chunks, got %d", len(chunks), len(result)) + } + if filer.HasChunkManifest(result) { + t.Error("fallback list must not contain manifest chunks") + } + if len(store.deleted) != 1 || store.deleted[0].FileId != "2,1" { + t.Fatalf("expected the first saved blob deleted, got %+v", store.deleted) + } +} + +func TestPartRange(t *testing.T) { + chunks := []*filer_pb.FileChunk{ + {FileId: "1,a", Offset: 0, Size: 8}, + {FileId: "1,b", Offset: 8, Size: 8}, + {FileId: "1,c", Offset: 16, Size: 24}, + } + + // Byte offsets win even when the chunk indexes no longer match the list. + start, end, ok := partRange(&PartBoundaryInfo{StartChunk: 40, EndChunk: 80, StartOffset: 16, EndOffset: 40}, chunks) + if !ok || start != 16 || end != 39 { + t.Errorf("offset boundary: got [%d,%d] ok=%v, want [16,39]", start, end, ok) + } + + // Legacy record: range from chunk indexes. + start, end, ok = partRange(&PartBoundaryInfo{StartChunk: 1, EndChunk: 3}, chunks) + if !ok || start != 8 || end != 39 { + t.Errorf("legacy boundary: got [%d,%d] ok=%v, want [8,39]", start, end, ok) + } + + // Legacy record with indexes off the list must report, not panic. + for _, b := range []*PartBoundaryInfo{ + {StartChunk: 2, EndChunk: 9}, + {StartChunk: -1, EndChunk: 2}, + {StartChunk: 2, EndChunk: 2}, + } { + if _, _, ok := partRange(b, chunks); ok { + t.Errorf("boundary %+v should not resolve", b) + } + } +} diff --git a/weed/s3api/s3api_object_handlers.go b/weed/s3api/s3api_object_handlers.go index 034d99e22..8ac5db12d 100644 --- a/weed/s3api/s3api_object_handlers.go +++ b/weed/s3api/s3api_object_handlers.go @@ -853,21 +853,14 @@ 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 && 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) { + if partInfo != nil { + var ok bool + startOffset, endOffset, ok = partRange(partInfo, objectEntryForSSE.Chunks) + if !ok { 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 } else { // Fallback: assume 1:1 part-to-chunk mapping (backward compatibility) chunkIndex := partNumber - 1 @@ -3047,13 +3040,26 @@ 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. + // Byte offsets of the part; zero EndOffset means a legacy record carrying + // chunk indexes only. StartOffset int64 `json:"startOffset,omitempty"` EndOffset int64 `json:"endOffset,omitempty"` // exclusive } +// partRange returns the inclusive byte range of a part, preferring byte +// offsets; ok is false when a legacy record's chunk indexes do not address +// the entry's chunk list. +func partRange(partInfo *PartBoundaryInfo, chunks []*filer_pb.FileChunk) (startOffset, endOffset int64, ok bool) { + if partInfo.EndOffset > partInfo.StartOffset { + return partInfo.StartOffset, partInfo.EndOffset - 1, true + } + if partInfo.StartChunk < 0 || partInfo.EndChunk <= partInfo.StartChunk || partInfo.EndChunk > len(chunks) { + return 0, 0, false + } + lastChunk := chunks[partInfo.EndChunk-1] + return chunks[partInfo.StartChunk].Offset, lastChunk.Offset + int64(lastChunk.Size) - 1, true +} + // rc is a helper type that wraps a Reader and Closer for proper resource cleanup type rc struct { io.Reader diff --git a/weed/s3api/s3api_object_handlers_copy.go b/weed/s3api/s3api_object_handlers_copy.go index 002d9fea9..0881359c2 100644 --- a/weed/s3api/s3api_object_handlers_copy.go +++ b/weed/s3api/s3api_object_handlers_copy.go @@ -423,8 +423,7 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request return } - // Re-fold a large copied chunk list into manifest chunks, mirroring the - // PutObject path (no-op below filer.ManifestBatch or for SSE chunks). + // re-fold a large copied chunk list, mirroring the PutObject path 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) @@ -919,9 +918,8 @@ 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 { + // per-chunk part-copy needs a flat source; resolved manifests stay with the source + 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 diff --git a/weed/s3api/s3api_object_handlers_copy_unified.go b/weed/s3api/s3api_object_handlers_copy_unified.go index d41dd9ad4..3728ac7d9 100644 --- a/weed/s3api/s3api_object_handlers_copy_unified.go +++ b/weed/s3api/s3api_object_handlers_copy_unified.go @@ -14,10 +14,9 @@ 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 { + // Per-chunk copy must see data chunks: a manifest chunk copied raw becomes + // object data. Resolved manifests stay with the source. + if _, err := s3a.flattenManifestChunks(r.Context(), entry); err != nil { return nil, nil, fmt.Errorf("resolve source manifest chunks: %w", err) } diff --git a/weed/s3api/s3api_object_handlers_put.go b/weed/s3api/s3api_object_handlers_put.go index d7c7dda22..fa6fdcb96 100644 --- a/weed/s3api/s3api_object_handlers_put.go +++ b/weed/s3api/s3api_object_handlers_put.go @@ -807,9 +807,8 @@ 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. + // Parts (object == "") stay flat: completion rebases chunk offsets, which + // manifest chunks cannot express. if object != "" { entry.Chunks = s3a.manifestizeChunks(filePath, bucket, lifecycleTTLSec, entry.GetChunks()) }