diff --git a/weed/s3api/filer_multipart.go b/weed/s3api/filer_multipart.go index 9a66b6f879..8d8acc670e 100644 --- a/weed/s3api/filer_multipart.go +++ b/weed/s3api/filer_multipart.go @@ -387,14 +387,12 @@ func applyMultipartSSES3HeadersFromUploadEntry(dst *filer_pb.Entry, sses3Info *m } func (s3a *S3ApiServer) prepareMultipartCompletionState(r *http.Request, input *s3.CompleteMultipartUploadInput, parts *CompleteMultipartUpload, uploadDirectory, entryName, dirName string, completedPartNumbers []int, completedPartMap map[int][]string, maxPartNo int) (*multipartCompletionState, *CompleteMultipartUploadResult, s3err.ErrorCode) { - if entry, err := s3a.resolveObjectEntry(*input.Bucket, *input.Key, ""); err == nil && entry != nil && entry.Extended != nil { - if uploadId, ok := entry.Extended[s3_constants.SeaweedFSUploadId]; ok && *input.UploadId == string(uploadId) { - cleanupEntries, _, cleanupErr := s3a.list(uploadDirectory, "", "", false, s3_constants.MaxS3MultipartParts+1) - 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, metadataOnlyCleanup: true}, completeMultipartResult(r, input, getEtagFromEntry(entry), entry), s3err.ErrNone + if output, _ := s3a.committedCompletion(r, input); output != nil { + cleanupEntries, _, cleanupErr := s3a.list(uploadDirectory, "", "", false, s3_constants.MaxS3MultipartParts+1) + 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, metadataOnlyCleanup: true}, output, s3err.ErrNone } entries, _, err := s3a.list(uploadDirectory, "", "", false, s3_constants.MaxS3MultipartParts+1) @@ -1044,11 +1042,27 @@ 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) + // A finalize can fail after its commit. When the key now holds this + // upload's object, answer as a retry would. + var committed *CompleteMultipartUploadResult + var committedEntry *filer_pb.Entry + if completionState != nil && completionState.manifestsReferenced { + committed, committedEntry = s3a.committedCompletion(r, input) } - return nil, finalizeCode + if committed == nil { + // 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 + } + glog.Warningf("completeMultipartUpload %s/%s upload %s: committed despite %s", *input.Bucket, *input.Key, *input.UploadId, s3err.GetAPIError(finalizeCode).Code) + // Another attempt of this upload may own the commit, and its parts can + // include ones this attempt counts as unused. + if !sameFileChunks(committedEntry.GetChunks(), completionState.finalParts) { + completionState.metadataOnlyCleanup = true + } + output = committed } if completionState != nil { @@ -1088,6 +1102,19 @@ func (s3a *S3ApiServer) deleteUnusedPartEntries(ctx context.Context, uploadDirec return lastErr } +// committedCompletion returns this upload's completion result, and the object +// entry, when the key's current object is the one the upload committed. +func (s3a *S3ApiServer) committedCompletion(r *http.Request, input *s3.CompleteMultipartUploadInput) (*CompleteMultipartUploadResult, *filer_pb.Entry) { + entry, err := s3a.resolveObjectEntry(*input.Bucket, *input.Key, "") + if err != nil || entry == nil || entry.Extended == nil { + return nil, nil + } + if uploadId, ok := entry.Extended[s3_constants.SeaweedFSUploadId]; !ok || string(uploadId) != *input.UploadId { + return nil, nil + } + return completeMultipartResult(r, input, getEtagFromEntry(entry), entry), entry +} + // resumeCommittedObject finds an entry this upload already committed at the // regular path when the upload directory is gone. Suspended versioning can // leave such an entry hidden behind a delete marker the finalize failed to diff --git a/weed/s3api/filer_multipart_commit_test.go b/weed/s3api/filer_multipart_commit_test.go new file mode 100644 index 0000000000..220043f3d3 --- /dev/null +++ b/weed/s3api/filer_multipart_commit_test.go @@ -0,0 +1,316 @@ +package s3api + +import ( + "context" + "net/http/httptest" + "sync" + "testing" + "time" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/s3" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3err" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +// memCompleteFiler keeps entries in memory. A failing CreateEntry still +// stores what storedOnFailure returns, and a failing ObjectTransaction still +// applies its mutations, as a filer whose commit landed before the call failed. +type memCompleteFiler struct { + filer_pb.UnimplementedSeaweedFilerServer + mu sync.Mutex + entries map[string]map[string]*filer_pb.Entry + createFailure error + storedOnFailure func(*filer_pb.Entry) *filer_pb.Entry + deletedData map[string]bool +} + +func (f *memCompleteFiler) put(dir string, entry *filer_pb.Entry) { + if f.entries[dir] == nil { + f.entries[dir] = map[string]*filer_pb.Entry{} + } + f.entries[dir][entry.Name] = entry +} + +func (f *memCompleteFiler) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + if entry := f.entries[req.Directory][req.Name]; entry != nil { + return &filer_pb.LookupDirectoryEntryResponse{Entry: entry}, nil + } + return nil, status.Error(codes.NotFound, filer_pb.ErrNotFound.Error()) +} + +func (f *memCompleteFiler) ListEntries(req *filer_pb.ListEntriesRequest, stream filer_pb.SeaweedFiler_ListEntriesServer) error { + f.mu.Lock() + var entries []*filer_pb.Entry + for _, entry := range f.entries[req.Directory] { + entries = append(entries, entry) + } + f.mu.Unlock() + for _, entry := range entries { + if err := stream.Send(&filer_pb.ListEntriesResponse{Entry: entry}); err != nil { + return err + } + } + return nil +} + +func (f *memCompleteFiler) CreateEntry(ctx context.Context, req *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + if f.createFailure == nil { + f.put(req.Directory, req.Entry) + return &filer_pb.CreateEntryResponse{}, nil + } + if f.storedOnFailure != nil { + if entry := f.storedOnFailure(req.Entry); entry != nil { + f.put(req.Directory, entry) + } + } + return nil, f.createFailure +} + +func (f *memCompleteFiler) DeleteEntry(ctx context.Context, req *filer_pb.DeleteEntryRequest) (*filer_pb.DeleteEntryResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.remove(req.Directory, req.Name, req.IsDeleteData) + return &filer_pb.DeleteEntryResponse{}, nil +} + +func (f *memCompleteFiler) remove(dir, name string, isDeleteData bool) { + delete(f.entries[dir], name) + if f.deletedData == nil { + f.deletedData = map[string]bool{} + } + f.deletedData[name] = isDeleteData +} + +// ObjectTransaction applies PUT and DELETE as given and points a recompute at +// the transaction's PUT, the newest version. +func (f *memCompleteFiler) ObjectTransaction(ctx context.Context, req *filer_pb.ObjectTransactionRequest) (*filer_pb.ObjectTransactionResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + var put *filer_pb.Entry + for _, m := range req.Mutations { + switch m.Type { + case filer_pb.ObjectMutation_PUT: + put = m.Entry + f.put(m.Directory, m.Entry) + case filer_pb.ObjectMutation_DELETE: + f.remove(m.Directory, m.Name, m.IsDeleteData) + case filer_pb.ObjectMutation_RECOMPUTE_LATEST: + f.put(m.Directory, &filer_pb.Entry{Name: m.Name, IsDirectory: true, Extended: map[string][]byte{ + s3_constants.ExtLatestVersionIdKey: put.Extended[s3_constants.ExtVersionIdKey], + s3_constants.ExtLatestVersionFileNameKey: []byte(put.Name), + }}) + } + } + if f.createFailure != nil { + return nil, f.createFailure + } + return &filer_pb.ObjectTransactionResponse{}, nil +} + +func newCompleteTestServer(t *testing.T, f *memCompleteFiler) *S3ApiServer { + t.Helper() + s3a := newFailoverTestServer(t, startFakeFiler(t, f)) + s3a.bucketConfigCache = NewBucketConfigCache(time.Minute) + s3a.bucketConfigCache.Set("b", &BucketConfig{Name: "b"}) + uploadsDir := s3a.genUploadsFolder("b") + f.put(uploadsDir, &filer_pb.Entry{ + Name: "u1", + IsDirectory: true, + Extended: map[string][]byte{s3_constants.ExtMultipartObjectKey: []byte("a.bin")}, + }) + f.put(uploadsDir+"/u1", &filer_pb.Entry{ + Name: "0001.part", + Attributes: &filer_pb.FuseAttributes{FileSize: 11}, + Chunks: []*filer_pb.FileChunk{{FileId: "3,01637037d6", Size: 11}}, + Extended: map[string][]byte{s3_constants.ExtETagKey: []byte("5eb63bbbe01eeed093cb22bb8f5acdc3")}, + }) + // An earlier upload of part 1 the completion does not use. + f.put(uploadsDir+"/u1", &filer_pb.Entry{ + Name: "0001_old.part", + Attributes: &filer_pb.FuseAttributes{FileSize: 11}, + Chunks: []*filer_pb.FileChunk{{FileId: "3,02637037d6", Size: 11}}, + Extended: map[string][]byte{s3_constants.ExtETagKey: []byte("0f343b0931126a20f133d67c2b018a3b")}, + }) + return s3a +} + +func completeU1(s3a *S3ApiServer) (*CompleteMultipartUploadResult, s3err.ErrorCode) { + input := &s3.CompleteMultipartUploadInput{Bucket: aws.String("b"), Key: aws.String("a.bin"), UploadId: aws.String("u1")} + parts := &CompleteMultipartUpload{Parts: []CompletedPart{{PartNumber: 1, ETag: `"5eb63bbbe01eeed093cb22bb8f5acdc3"`}}} + return s3a.completeMultipartUpload(httptest.NewRequest("POST", "/b/a.bin?uploadId=u1", nil), input, parts) +} + +// A completion whose commit landed but whose call failed answers success, as +// its retry would, instead of a 500 the client cannot interpret. +func TestCompleteMultipartUploadCommittedDespiteFailure(t *testing.T) { + f := &memCompleteFiler{entries: map[string]map[string]*filer_pb.Entry{}, createFailure: status.Error(codes.Internal, "response lost after commit"), + storedOnFailure: func(entry *filer_pb.Entry) *filer_pb.Entry { return entry }} + s3a := newCompleteTestServer(t, f) + + output, code := completeU1(s3a) + if code != s3err.ErrNone || output == nil { + t.Fatalf("code = %v, want success for the committed object", s3err.GetAPIError(code).Code) + } + entry := f.entries[s3a.bucketDir("b")]["a.bin"] + if entry == nil || string(entry.Extended[s3_constants.SeaweedFSUploadId]) != "u1" { + t.Fatalf("object entry = %v, want the committed upload u1", entry) + } + if got, want := aws.StringValue(output.ETag), getEtagFromEntry(entry); got != want { + t.Errorf("ETag = %s, want the committed object's %s", got, want) + } + if _, ok := f.entries[s3a.genUploadsFolder("b")]["u1"]; ok { + t.Error("the upload directory survived the completed upload") + } + if deleteData, ok := f.deletedData["0001_old.part"]; !ok || !deleteData { + t.Errorf("unused part deleted = %v, with data = %v; want its data freed as on success", ok, deleteData) + } + if deleteData, ok := f.deletedData["u1"]; !ok || deleteData { + t.Errorf("upload directory deleted = %v, with data = %v; want metadata only, its parts are the object's", ok, deleteData) + } +} + +func newVersionedCommitTestServer(t *testing.T, versioning string) (*S3ApiServer, *memCompleteFiler) { + t.Helper() + f := &memCompleteFiler{entries: map[string]map[string]*filer_pb.Entry{}} + s3a := newFailoverTestServer(t, startFakeFiler(t, f)) + s3a.bucketConfigCache = NewBucketConfigCache(time.Minute) + s3a.bucketConfigCache.Set("b", &BucketConfig{Name: "b", Versioning: versioning}) + return s3a, f +} + +func pointLatestVersion(f *memCompleteFiler, s3a *S3ApiServer, versionId string, version *filer_pb.Entry) { + f.put(s3a.bucketDir("b"), &filer_pb.Entry{Name: "a.bin" + s3_constants.VersionsFolder, IsDirectory: true, Extended: map[string][]byte{ + s3_constants.ExtLatestVersionIdKey: []byte(versionId), + s3_constants.ExtLatestVersionFileNameKey: []byte(version.Name), + }}) + f.put(s3a.bucketDir("b")+"/a.bin"+s3_constants.VersionsFolder, version) +} + +// On a versioned bucket the committed version is read through the latest +// pointer, and the answer carries its version id and checksum. +func TestCommittedCompletionReadsTheLatestVersion(t *testing.T) { + s3a, f := newVersionedCommitTestServer(t, s3_constants.VersioningEnabled) + pointLatestVersion(f, s3a, "3a1b", &filer_pb.Entry{Name: "v_3a1b", Attributes: &filer_pb.FuseAttributes{FileSize: 11}, Extended: map[string][]byte{ + s3_constants.SeaweedFSUploadId: []byte("u1"), + s3_constants.ExtVersionIdKey: []byte("3a1b"), + s3_constants.ExtETagKey: []byte("8a6d5f0e5c8f7b2f4e1d0c9b8a7f6e5d-1"), + s3_constants.ExtChecksumAlgorithm: []byte(s3_constants.AmzChecksumCRC32C), + s3_constants.ExtChecksumValue: []byte("fx4FQw==-1"), + s3_constants.ExtChecksumType: []byte(s3_constants.ChecksumTypeComposite), + }}) + + input := &s3.CompleteMultipartUploadInput{Bucket: aws.String("b"), Key: aws.String("a.bin"), UploadId: aws.String("u1")} + output, entry := s3a.committedCompletion(httptest.NewRequest("POST", "/b/a.bin?uploadId=u1", nil), input) + if output == nil || entry == nil { + t.Fatal("did not find the committed version") + } + if got := aws.StringValue(output.VersionId); got != "3a1b" { + t.Errorf("VersionId = %q, want 3a1b", got) + } + if output.ChecksumCRC32C != "fx4FQw==-1" || output.ChecksumType != s3_constants.ChecksumTypeComposite { + t.Errorf("checksum = %q %q, want the committed version's", output.ChecksumCRC32C, output.ChecksumType) + } +} + +// A versioned finalize whose transaction landed but whose call failed answers +// with the committed version's id and checksum. +func TestCompleteMultipartUploadVersionedCommittedDespiteFailure(t *testing.T) { + f := &memCompleteFiler{entries: map[string]map[string]*filer_pb.Entry{}, createFailure: status.Error(codes.Internal, "response lost after commit")} + s3a := newCompleteTestServer(t, f) + s3a.bucketConfigCache.Set("b", &BucketConfig{Name: "b", Versioning: s3_constants.VersioningEnabled}) + uploadsDir := s3a.genUploadsFolder("b") + f.entries[uploadsDir]["u1"].Extended[s3_constants.ExtChecksumAlgorithm] = []byte(s3_constants.AmzChecksumCRC32C) + part := f.entries[uploadsDir+"/u1"]["0001.part"] + part.Extended[s3_constants.ExtChecksumAlgorithm] = []byte(s3_constants.AmzChecksumCRC32C) + part.Extended[s3_constants.ExtChecksumValue] = []byte("yZRlqg==") + + input := &s3.CompleteMultipartUploadInput{Bucket: aws.String("b"), Key: aws.String("a.bin"), UploadId: aws.String("u1")} + parts := &CompleteMultipartUpload{Parts: []CompletedPart{{PartNumber: 1, ETag: `"5eb63bbbe01eeed093cb22bb8f5acdc3"`, ChecksumResult: ChecksumResult{ChecksumCRC32C: "yZRlqg=="}}}} + output, code := s3a.completeMultipartUpload(httptest.NewRequest("POST", "/b/a.bin?uploadId=u1", nil), input, parts) + if code != s3err.ErrNone || output == nil { + t.Fatalf("code = %v, want success for the committed version", s3err.GetAPIError(code).Code) + } + pointer := f.entries[s3a.bucketDir("b")]["a.bin"+s3_constants.VersionsFolder] + version := f.entries[s3a.bucketDir("b")+"/a.bin"+s3_constants.VersionsFolder][string(pointer.Extended[s3_constants.ExtLatestVersionFileNameKey])] + if version == nil { + t.Fatal("the transaction did not store the version") + } + if got, want := aws.StringValue(output.VersionId), string(version.Extended[s3_constants.ExtVersionIdKey]); got == "" || got != want { + t.Errorf("VersionId = %q, want the committed version's %q", got, want) + } + if got, want := output.ChecksumCRC32C, string(version.Extended[s3_constants.ExtChecksumValue]); got == "" || got != want { + t.Errorf("ChecksumCRC32C = %q, want the committed version's %q", got, want) + } + if output.ChecksumType != s3_constants.ChecksumTypeComposite { + t.Errorf("ChecksumType = %q, want %q", output.ChecksumType, s3_constants.ChecksumTypeComposite) + } +} + +// Suspended versioning whose null write left the delete marker current reads +// as deleted, so the upload's object behind it is not a completed answer. +func TestCommittedCompletionBehindADeleteMarker(t *testing.T) { + s3a, f := newVersionedCommitTestServer(t, s3_constants.VersioningSuspended) + f.put(s3a.bucketDir("b"), &filer_pb.Entry{Name: "a.bin", Attributes: &filer_pb.FuseAttributes{FileSize: 11}, Extended: map[string][]byte{ + s3_constants.SeaweedFSUploadId: []byte("u1"), + s3_constants.ExtVersionIdKey: []byte("null"), + }}) + pointLatestVersion(f, s3a, "7c2d", &filer_pb.Entry{Name: "v_7c2d", Attributes: &filer_pb.FuseAttributes{}, Extended: map[string][]byte{ + s3_constants.ExtVersionIdKey: []byte("7c2d"), + s3_constants.ExtDeleteMarkerKey: []byte("true"), + }}) + + input := &s3.CompleteMultipartUploadInput{Bucket: aws.String("b"), Key: aws.String("a.bin"), UploadId: aws.String("u1")} + if output, _ := s3a.committedCompletion(httptest.NewRequest("POST", "/b/a.bin?uploadId=u1", nil), input); output != nil { + t.Fatalf("answered %+v for a key that reads as deleted", output) + } +} + +// The object found may be another attempt's commit of the same upload, whose +// parts this attempt can count as unused: their data must stay. +func TestCompleteMultipartUploadKeepsPartDataUnderAnotherAttemptsCommit(t *testing.T) { + other := &filer_pb.Entry{ + Name: "a.bin", + Attributes: &filer_pb.FuseAttributes{FileSize: 11}, + Chunks: []*filer_pb.FileChunk{{FileId: "3,02637037d6", Size: 11}}, + Extended: map[string][]byte{ + s3_constants.SeaweedFSUploadId: []byte("u1"), + s3_constants.ExtETagKey: []byte("8a6d5f0e5c8f7b2f4e1d0c9b8a7f6e5d-1"), + }, + } + f := &memCompleteFiler{entries: map[string]map[string]*filer_pb.Entry{}, createFailure: status.Error(codes.Internal, "lost the race"), + storedOnFailure: func(*filer_pb.Entry) *filer_pb.Entry { return other }} + s3a := newCompleteTestServer(t, f) + + output, code := completeU1(s3a) + if code != s3err.ErrNone || output == nil { + t.Fatalf("code = %v, want success for the upload's committed object", s3err.GetAPIError(code).Code) + } + if got, want := aws.StringValue(output.ETag), getEtagFromEntry(other); got != want { + t.Errorf("ETag = %s, want the committed object's %s", got, want) + } + if deleteData, ok := f.deletedData["0001_old.part"]; ok && deleteData { + t.Error("freed the data of a part the committed object references") + } +} + +// A failure that left another object at the key still reports the failure. +func TestCompleteMultipartUploadFailureOverAnotherObject(t *testing.T) { + f := &memCompleteFiler{entries: map[string]map[string]*filer_pb.Entry{}, createFailure: status.Error(codes.Internal, "store failed")} + s3a := newCompleteTestServer(t, f) + f.put(s3a.bucketDir("b"), &filer_pb.Entry{Name: "a.bin", Attributes: &filer_pb.FuseAttributes{}, Extended: map[string][]byte{s3_constants.SeaweedFSUploadId: []byte("u0")}}) + + if output, code := completeU1(s3a); code != s3err.ErrInternalError || output != nil { + t.Fatalf("code = %v, output = %v, want InternalError", s3err.GetAPIError(code).Code, output) + } + if _, ok := f.entries[s3a.genUploadsFolder("b")]["u1"]; !ok { + t.Error("the failed completion removed its upload directory") + } +}