diff --git a/weed/pb/filer_pb/filer_client.go b/weed/pb/filer_pb/filer_client.go index 1fd86de7e..4052ba9f1 100644 --- a/weed/pb/filer_pb/filer_client.go +++ b/weed/pb/filer_pb/filer_client.go @@ -202,7 +202,7 @@ func DoSeaweedListWithSnapshot(ctx context.Context, client SeaweedFilerClient, f } prevEntry = resp.Entry count++ - if count > int(limit) && limit != 0 { + if limit != 0 && uint64(count) > uint64(limit) { prevEntry = nil } } diff --git a/weed/s3api/filer_multipart.go b/weed/s3api/filer_multipart.go index 33792dc2c..046e0aab0 100644 --- a/weed/s3api/filer_multipart.go +++ b/weed/s3api/filer_multipart.go @@ -241,6 +241,7 @@ type multipartCompletionState struct { 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 + uploadDirRemoved bool // the finalize transaction already removed .uploads/ } func completeMultipartResult(r *http.Request, input *s3.CompleteMultipartUploadInput, etag string, entry *filer_pb.Entry) *CompleteMultipartUploadResult { @@ -397,6 +398,9 @@ func (s3a *S3ApiServer) prepareMultipartCompletionState(r *http.Request, input * if err != nil { glog.Errorf("completeMultipartUpload %s %s error: %v", *input.Bucket, *input.UploadId, err) if isFilerNotFound(err) { + if output, code := s3a.resumeCommittedObject(r, input, dirName, entryName); code != s3err.ErrNoSuchUpload { + return &multipartCompletionState{metadataOnlyCleanup: true}, output, code + } stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedNoSuchUpload).Inc() return nil, nil, s3err.ErrNoSuchUpload } @@ -404,6 +408,9 @@ func (s3a *S3ApiServer) prepareMultipartCompletionState(r *http.Request, input * return nil, nil, s3err.ErrInternalError } if len(entries) == 0 { + if output, code := s3a.resumeCommittedObject(r, input, dirName, entryName); code != s3err.ErrNoSuchUpload { + return &multipartCompletionState{metadataOnlyCleanup: true}, output, code + } stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedNoSuchUpload).Inc() return nil, nil, s3err.ErrNoSuchUpload } @@ -674,8 +681,9 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl versionMtime := time.Now().Unix() amzAccountId := r.Header.Get(s3_constants.AmzAccountId) - // Create the version file in the .versions directory - if err := s3a.mkFile(versionDir, versionFileName, completionState.finalParts, func(versionEntry *filer_pb.Entry) { + // Fills in the version entry; on the routed path this runs inside the + // finalize transaction, on the fallback inside mkFile. + decorateVersionEntry := func(versionEntry *filer_pb.Entry) { if versionEntry.Extended == nil { versionEntry.Extended = make(map[string][]byte) } @@ -724,9 +732,6 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl } versionEntry.Attributes.FileSize = uint64(completionState.offset) versionEntry.Attributes.Mtime = versionMtime - }); err != nil { - glog.Errorf("completeMultipartUpload: failed to create version %s: %v", versionId, err) - return s3err.ErrInternalError } // Construct entry with metadata for caching in .versions directory @@ -745,25 +750,48 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl versionEntryForCache.Extended[s3_constants.ExtAmzOwnerKey] = []byte(amzAccountId) } - // Update the .versions directory metadata to indicate this is the latest version - // Pass entry to cache its metadata for single-scan list efficiency - // Route the pointer flip to the owner (off the lock) via - // RECOMPUTE_LATEST; the just-written version file is the newest. - if owner != "" { - 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) + // objectTxnOnFiler needs an owner to route to or a filerClient to + // pick a filer from; only the bootstrap fallback keeps the + // mkFile + pointer-update sequence. + if owner != "" || s3a.filerClient != nil { + // The transaction removes the upload directory metadata-only, + // so the part entries the object does not reference are freed + // first or their chunks leak. + if err := s3a.deleteUnusedPartEntries(r.Context(), uploadDirectory, *input.Bucket, *input.UploadId, completionState); err != nil { + glog.Errorf("completeMultipartUpload %s upload %s unused part cleanup: %v", *input.Bucket, *input.UploadId, err) + return s3err.ErrInternalError + } + if code := s3a.routedMultipartFinalize(owner, *input.Bucket, *input.Key, useInvertedFormat, versionDir, versionFileName, completionState.finalParts, decorateVersionEntry, *input.UploadId); code != s3err.ErrNone { + if code == s3err.ErrNoSuchUpload { + return code + } + // Roll back only while the upload directory survives: once the + // transaction removed it, the version file is the only record + // left and deleting it would make a retry impossible. + if exists, _ := s3a.exists(s3a.genUploadsFolder(*input.Bucket), *input.UploadId, true); exists { + 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 + } + } else { 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 + completionState.uploadDirRemoved = true + } else { + if err := s3a.mkFile(versionDir, versionFileName, completionState.finalParts, decorateVersionEntry); err != nil { + glog.Errorf("completeMultipartUpload: failed to create version %s: %v", versionId, err) + return s3err.ErrInternalError + } + 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 } - glog.Errorf("completeMultipartUpload: failed to update latest version in directory: %v", err) - return s3err.ErrInternalError } // For versioned buckets, all content is stored in .versions directory @@ -782,6 +810,11 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl if versioningState == s3_constants.VersioningSuspended { // For suspended versioning, add "null" version ID metadata and return "null" version ID + removal, err := s3a.routedUploadRemoval(r.Context(), owner, uploadDirectory, *input.Bucket, *input.UploadId, completionState) + if err != nil { + glog.Errorf("completeMultipartUpload %s upload %s unused part cleanup: %v", *input.Bucket, *input.UploadId, err) + return s3err.ErrInternalError + } if err := s3a.writeMultipartObject(owner, routeKey, dirName, entryName, completionState.finalParts, func(entry *filer_pb.Entry) { if entry.Extended == nil { entry.Extended = make(map[string][]byte) @@ -830,15 +863,25 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl entry.Attributes.Mime = completionState.mime } entry.Attributes.FileSize = uint64(completionState.offset) - }); err != nil { + }, removal); err != nil { + if errors.Is(err, errUploadRemoved) { + return s3err.ErrNoSuchUpload + } + // The transaction may have committed the object before failing on + // the upload removal; a surviving entry references the manifests. + if exists, err := s3a.exists(dirName, entryName, false); err != nil || exists { + completionState.manifestsReferenced = true + } glog.Errorf("completeMultipartUpload: failed to create suspended versioning object: %v", err) return s3err.ErrInternalError } + completionState.uploadDirRemoved = removal != nil // A failed finalize leaves the key reading as deleted, so fail rather than // return 200 — a non-ErrNone finalize keeps the upload directory, so the // caller's retry replays. if err := s3a.finalizeSuspendedNullWrite(owner, *input.Bucket, normalizedKey, s3_constants.SeaweedFSUploadId, *input.UploadId); err != nil { + completionState.manifestsReferenced = true glog.Errorf("completeMultipartUpload: failed to retire the null delete marker for %s/%s: %v", *input.Bucket, normalizedKey, err) return s3err.ErrInternalError } @@ -857,6 +900,11 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl } // For non-versioned buckets, create main object file + removal, err := s3a.routedUploadRemoval(r.Context(), owner, uploadDirectory, *input.Bucket, *input.UploadId, completionState) + if err != nil { + glog.Errorf("completeMultipartUpload %s upload %s unused part cleanup: %v", *input.Bucket, *input.UploadId, err) + return s3err.ErrInternalError + } if err := s3a.writeMultipartObject(owner, routeKey, dirName, entryName, completionState.finalParts, func(entry *filer_pb.Entry) { if entry.Extended == nil { entry.Extended = make(map[string][]byte) @@ -908,10 +956,19 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl if completionState.entityWithTtl { entry.Extended[s3_constants.SeaweedFSExpiresS3] = []byte("true") } - }); err != nil { + }, removal); err != nil { + if errors.Is(err, errUploadRemoved) { + return s3err.ErrNoSuchUpload + } + // The transaction may have committed the object before failing on + // the upload removal; a surviving entry references the manifests. + if exists, err := s3a.exists(dirName, entryName, false); err != nil || exists { + completionState.manifestsReferenced = true + } glog.Errorf("completeMultipartUpload %s/%s error: %v", dirName, entryName, err) return s3err.ErrInternalError } + completionState.uploadDirRemoved = removal != nil // For non-versioned buckets, return response without VersionId output = &CompleteMultipartUploadResult{ @@ -945,18 +1002,19 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl } if completionState != nil { - // The object is already committed and the client is still waiting, so the - // cleanup below runs on its own context but spends one allowance between - // all of it rather than a retry backoff per unused entry. - cleanupCtx := withFilerRetryBudget(context.Background(), filerRetryRequestBudget) - for _, deleteEntry := range completionState.deleteEntries { - if err := s3a.rm(cleanupCtx, 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 !completionState.uploadDirRemoved { + // The object is already committed and the client is still waiting, so the + // cleanup below runs on its own context but spends one allowance between + // all of it rather than a retry backoff per unused entry. + cleanupCtx := withFilerRetryBudget(context.Background(), filerRetryRequestBudget) + // Keep the directory when an entry delete failed so its metadata + // still references the chunks; removing it metadata-only orphans them. + if err := s3a.deleteUnusedPartEntries(cleanupCtx, uploadDirectory, *input.Bucket, *input.UploadId, completionState); err != nil { + glog.V(1).Infof("completeMultipartUpload cleanup %s upload %s: %v", *input.Bucket, *input.UploadId, err) + } else if err := s3a.rm(cleanupCtx, 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 err := s3a.rm(cleanupCtx, 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) } @@ -965,6 +1023,52 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl return } +// deleteUnusedPartEntries frees the part entries the completed object does not +// reference. A finalize that removes the whole upload directory metadata-only +// frees them first, or their chunks leak; a failure here must abort the +// completion, since the surviving entries keep the upload retriable. +func (s3a *S3ApiServer) deleteUnusedPartEntries(ctx context.Context, uploadDirectory, bucket, uploadId string, completionState *multipartCompletionState) error { + var lastErr error + for _, deleteEntry := range completionState.deleteEntries { + if err := s3a.rm(ctx, uploadDirectory, deleteEntry.Name, !completionState.metadataOnlyCleanup, true); err != nil { + glog.Warningf("completeMultipartUpload cleanup %s upload %s unused %s : %v", bucket, uploadId, deleteEntry.Name, err) + lastErr = err + } + } + return lastErr +} + +// 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 +// retire, so re-running it repairs the key and lets the retry succeed. Any +// other versioning state means a newer write owns the key and the marker must +// not be demoted. +func (s3a *S3ApiServer) resumeCommittedObject(r *http.Request, input *s3.CompleteMultipartUploadInput, dirName, entryName string) (*CompleteMultipartUploadResult, s3err.ErrorCode) { + entry, err := s3a.getEntry(dirName, entryName) + if err != nil { + if isFilerNotFound(err) { + return nil, s3err.ErrNoSuchUpload + } + return nil, s3err.ErrInternalError + } + if entry == nil || string(entry.Extended[s3_constants.SeaweedFSUploadId]) != *input.UploadId { + return nil, s3err.ErrNoSuchUpload + } + state, err := s3a.getVersioningState(*input.Bucket) + if err != nil { + return nil, s3err.ErrInternalError + } + if state != s3_constants.VersioningSuspended { + return nil, s3err.ErrNoSuchUpload + } + if err := s3a.finalizeSuspendedNullWrite(s3a.objectWriteOwner(*input.Bucket, *input.Key), *input.Bucket, s3_constants.NormalizeObjectKey(*input.Key), s3_constants.SeaweedFSUploadId, *input.UploadId); err != nil { + glog.Errorf("completeMultipartUpload: failed to retire the null delete marker for %s/%s: %v", *input.Bucket, *input.Key, err) + return nil, s3err.ErrInternalError + } + return completeMultipartResult(r, input, getEtagFromEntry(entry), entry), s3err.ErrNone +} + // 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 { diff --git a/weed/s3api/s3api_object_handlers_put.go b/weed/s3api/s3api_object_handlers_put.go index 9cbad8c83..dd49e442f 100644 --- a/weed/s3api/s3api_object_handlers_put.go +++ b/weed/s3api/s3api_object_handlers_put.go @@ -966,7 +966,7 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader if finalize != nil && len(finalize.mutations) > 0 { lockKey, finalizeMutations = finalize.lockKey, finalize.mutations } - resp, err := s3a.routedPut(owner, s3a.objectRouteKey(bucket, object), lockKey, filePath, entry, cond, finalizeMutations) + resp, err := s3a.routedPut(owner, s3a.objectRouteKey(bucket, object), lockKey, filePath, entry, cond, "", finalizeMutations) switch { case err != nil: glog.Warningf("putToFiler: routed PUT to %s failed for %s, falling back to lock: %v", owner, filePath, err) diff --git a/weed/s3api/s3api_object_routed_write.go b/weed/s3api/s3api_object_routed_write.go index b2a602722..d1865b58c 100644 --- a/weed/s3api/s3api_object_routed_write.go +++ b/weed/s3api/s3api_object_routed_write.go @@ -2,6 +2,7 @@ package s3api import ( "context" + "errors" "fmt" "net/http" "path" @@ -180,7 +181,7 @@ func (s3a *S3ApiServer) objectTxnOnFiler(owner pb.ServerAddress, req *filer_pb.O // object path plus a RECOMPUTE_LATEST finalize, so the version's PUT and its // .versions pointer flip commit atomically (the recompute scans .versions/ after // the PUT and sees the new version). -func (s3a *S3ApiServer) routedPut(owner pb.ServerAddress, routeKey, lockKey, filePath string, entry *filer_pb.Entry, cond *filer_pb.WriteCondition, finalize []*filer_pb.ObjectMutation) (*filer_pb.ObjectTransactionResponse, error) { +func (s3a *S3ApiServer) routedPut(owner pb.ServerAddress, routeKey, lockKey, filePath string, entry *filer_pb.Entry, cond *filer_pb.WriteCondition, conditionKey string, finalize []*filer_pb.ObjectMutation) (*filer_pb.ObjectTransactionResponse, error) { mutations := make([]*filer_pb.ObjectMutation, 0, 1+len(finalize)) mutations = append(mutations, &filer_pb.ObjectMutation{ Type: filer_pb.ObjectMutation_PUT, @@ -189,17 +190,18 @@ func (s3a *S3ApiServer) routedPut(owner pb.ServerAddress, routeKey, lockKey, fil }) mutations = append(mutations, finalize...) return s3a.objectTxnOnFiler(owner, &filer_pb.ObjectTransactionRequest{ - LockKey: lockKey, - RouteKey: routeKey, - Condition: cond, - Mutations: mutations, + LockKey: lockKey, + RouteKey: routeKey, + Condition: cond, + ConditionKey: conditionKey, + Mutations: mutations, }) } // routedMkFile builds an entry like filer_pb.MkFile and writes it through a // routed PUT on the owner filer, for callers that would otherwise mkFile to the // default filer (e.g. multipart completion of a non-versioned object). -func (s3a *S3ApiServer) routedMkFile(owner pb.ServerAddress, routeKey, parentDir, name string, chunks []*filer_pb.FileChunk, fn func(*filer_pb.Entry)) error { +func (s3a *S3ApiServer) routedMkFile(owner pb.ServerAddress, routeKey, parentDir, name string, chunks []*filer_pb.FileChunk, fn func(*filer_pb.Entry), removal *uploadRemovalTxn) error { now := time.Now().Unix() entry := &filer_pb.Entry{ Name: name, @@ -216,27 +218,73 @@ func (s3a *S3ApiServer) routedMkFile(owner pb.ServerAddress, routeKey, parentDir fn(entry) } filePath := parentDir + "/" + name - resp, err := s3a.routedPut(owner, routeKey, filePath, filePath, entry, nil, nil) + var cond *filer_pb.WriteCondition + var conditionKey string + var finalize []*filer_pb.ObjectMutation + if removal != nil { + cond, conditionKey = removal.condition, removal.conditionKey + finalize = []*filer_pb.ObjectMutation{removal.mutation} + } + resp, err := s3a.routedPut(owner, routeKey, filePath, filePath, entry, cond, conditionKey, finalize) if err != nil { return err } + if resp.ErrorCode == filer_pb.FilerError_PRECONDITION_FAILED { + return errUploadRemoved + } if resp.Error != "" { return fmt.Errorf("routed mkfile %s/%s: %s", parentDir, name, resp.Error) } return nil } +// errUploadRemoved is a routed commit rejected because its precondition found +// the upload directory already deleted, mapped to NoSuchUpload by callers. +var errUploadRemoved = errors.New("upload directory already removed") + // writeMultipartObject writes a completed multipart object entry, routed to the // owner when known (so it serializes with concurrent writes to the same key) -// and falling back to a plain mkFile otherwise. routeKey must be the same key the -// caller used to resolve owner, so owner selection and forwarding stay consistent. -func (s3a *S3ApiServer) writeMultipartObject(owner pb.ServerAddress, routeKey, dir, name string, chunks []*filer_pb.FileChunk, fn func(*filer_pb.Entry)) error { +// and falling back to a plain mkFile otherwise. removal carries the upload +// precondition and deletion applied in the same transaction as the routed PUT. +// routeKey must be the same key the caller used to resolve owner, so owner +// selection and forwarding stay consistent. +func (s3a *S3ApiServer) writeMultipartObject(owner pb.ServerAddress, routeKey, dir, name string, chunks []*filer_pb.FileChunk, fn func(*filer_pb.Entry), removal *uploadRemovalTxn) error { if owner != "" { - return s3a.routedMkFile(owner, routeKey, dir, name, chunks, fn) + return s3a.routedMkFile(owner, routeKey, dir, name, chunks, fn, removal) } return s3a.mkFile(dir, name, chunks, fn) } +// uploadRemovalTxn carries the precondition and mutation that drop a completed +// upload's directory inside the object's commit transaction. The precondition +// fails the commit when a delete that does not take the object lock (abort, +// lifecycle, s3.clean.uploads) removed the directory first, instead of +// publishing the object over freed chunks. +type uploadRemovalTxn struct { + condition *filer_pb.WriteCondition + conditionKey string + mutation *filer_pb.ObjectMutation +} + +// routedUploadRemoval returns the transaction parts that remove a completed +// upload's directory inside the object's commit, after freeing the part +// entries the object does not reference. It is nil when the write is not +// routed and the upload directory still needs post-commit cleanup. +func (s3a *S3ApiServer) routedUploadRemoval(ctx context.Context, owner pb.ServerAddress, uploadDirectory, bucket, uploadId string, completionState *multipartCompletionState) (*uploadRemovalTxn, error) { + if owner == "" { + return nil, nil + } + if err := s3a.deleteUnusedPartEntries(ctx, uploadDirectory, bucket, uploadId, completionState); err != nil { + return nil, err + } + conditionKey, condition := uploadExistsCondition(uploadDirectory) + return &uploadRemovalTxn{ + condition: condition, + conditionKey: conditionKey, + mutation: s3a.removeUploadDirMutation(bucket, uploadId), + }, nil +} + func (s3a *S3ApiServer) routedDelete(owner pb.ServerAddress, bucket, object string, cond *filer_pb.WriteCondition) (*filer_pb.ObjectTransactionResponse, error) { // NewFullPath normalizes a trailing-slash directory-marker key (e.g. "dir/") // to the entry name "dir", matching deleteUnversionedObjectWithClient. diff --git a/weed/s3api/s3api_object_routed_write_test.go b/weed/s3api/s3api_object_routed_write_test.go index 5ffc63f08..9ebaa4686 100644 --- a/weed/s3api/s3api_object_routed_write_test.go +++ b/weed/s3api/s3api_object_routed_write_test.go @@ -421,6 +421,122 @@ func TestObjectTxnFailsOverStaleOwner(t *testing.T) { } } +// A completed multipart upload commits in one transaction: the version file's +// PUT, the metadata-only removal of .uploads/ (its chunks are the object's +// chunks), and the latest-pointer recompute, in that order. +func TestRoutedMultipartFinalize(t *testing.T) { + filer := &fakeTxnFiler{} + owner := startFakeFiler(t, filer) + s3a := &S3ApiServer{ + option: &S3ApiServerOption{ + BucketsPath: "/buckets", + GrpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + }, + } + + chunks := []*filer_pb.FileChunk{{FileId: "1,01637037d6"}} + code := s3a.routedMultipartFinalize(owner, "b", "obj", false, "/buckets/b/obj.versions", "v_1", chunks, func(entry *filer_pb.Entry) { + entry.Extended = map[string][]byte{s3_constants.SeaweedFSUploadId: []byte("up1")} + }, "up1") + if code != s3err.ErrNone { + t.Fatalf("routedMultipartFinalize = %v", code) + } + + req := filer.lastReq + if req == nil { + t.Fatal("expected an ObjectTransaction request") + } + if req.LockKey != "/buckets/b/obj" { + t.Fatalf("LockKey = %q", req.LockKey) + } + if req.RouteKey != "s3.object.write:/buckets/b/obj" { + t.Fatalf("RouteKey = %q", req.RouteKey) + } + if req.ConditionKey != "/buckets/b/.uploads/up1" { + t.Fatalf("ConditionKey = %q", req.ConditionKey) + } + if req.Condition == nil || len(req.Condition.Clauses) != 1 || req.Condition.Clauses[0].Kind != filer_pb.WriteCondition_IF_EXISTS { + t.Fatalf("Condition = %+v", req.Condition) + } + if len(req.Mutations) != 3 { + t.Fatalf("mutations = %d, want 3", len(req.Mutations)) + } + + put := req.Mutations[0] + if put.Type != filer_pb.ObjectMutation_PUT || + put.Directory != "/buckets/b/obj.versions" || + put.Entry == nil || + put.Entry.Name != "v_1" || + len(put.Entry.Chunks) != 1 || + string(put.Entry.Extended[s3_constants.SeaweedFSUploadId]) != "up1" { + t.Fatalf("put mutation = %+v", put) + } + removeUpload := req.Mutations[1] + if removeUpload.Type != filer_pb.ObjectMutation_DELETE || + removeUpload.Directory != "/buckets/b/.uploads" || + removeUpload.Name != "up1" || + !removeUpload.IsRecursive || + removeUpload.IsDeleteData { + t.Fatalf("remove upload mutation = %+v", removeUpload) + } + if req.Mutations[2].Type != filer_pb.ObjectMutation_RECOMPUTE_LATEST { + t.Fatalf("recompute mutation = %+v", req.Mutations[2]) + } +} + +// A non-versioned multipart completion removes .uploads/ metadata-only in +// the same transaction as the object's PUT. +func TestWriteMultipartObjectRemovesUploadDir(t *testing.T) { + filer := &fakeTxnFiler{} + owner := startFakeFiler(t, filer) + s3a := &S3ApiServer{ + option: &S3ApiServerOption{ + BucketsPath: "/buckets", + GrpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + }, + } + + if removal, err := s3a.routedUploadRemoval(context.Background(), "", "/buckets/b/.uploads/up1", "b", "up1", &multipartCompletionState{}); removal != nil || err != nil { + t.Fatalf("unrouted write should not carry the removal, got %+v, %v", removal, err) + } + removal, err := s3a.routedUploadRemoval(context.Background(), owner, "/buckets/b/.uploads/up1", "b", "up1", &multipartCompletionState{}) + if err != nil { + t.Fatalf("routedUploadRemoval: %v", err) + } + if err := s3a.writeMultipartObject(owner, "s3.object.write:/buckets/b/o", "/buckets/b", "o", nil, nil, removal); err != nil { + t.Fatalf("writeMultipartObject: %v", err) + } + + req := filer.lastReq + if req == nil { + t.Fatal("expected an ObjectTransaction request") + } + if req.LockKey != "/buckets/b/o" { + t.Fatalf("LockKey = %q", req.LockKey) + } + if req.ConditionKey != "/buckets/b/.uploads/up1" { + t.Fatalf("ConditionKey = %q", req.ConditionKey) + } + if req.Condition == nil || len(req.Condition.Clauses) != 1 || req.Condition.Clauses[0].Kind != filer_pb.WriteCondition_IF_EXISTS { + t.Fatalf("Condition = %+v", req.Condition) + } + if len(req.Mutations) != 2 { + t.Fatalf("mutations = %d, want 2", len(req.Mutations)) + } + put := req.Mutations[0] + if put.Type != filer_pb.ObjectMutation_PUT || put.Directory != "/buckets/b" || put.Entry == nil || put.Entry.Name != "o" { + t.Fatalf("put mutation = %+v", put) + } + removeUpload := req.Mutations[1] + if removeUpload.Type != filer_pb.ObjectMutation_DELETE || + removeUpload.Directory != "/buckets/b/.uploads" || + removeUpload.Name != "up1" || + !removeUpload.IsRecursive || + removeUpload.IsDeleteData { + t.Fatalf("remove upload mutation = %+v", removeUpload) + } +} + func TestRouteWriteCondition(t *testing.T) { // Unconditional routes either way. if c, ok := routeWriteCondition(reqWith(nil), false); !ok || c != nil { diff --git a/weed/s3api/s3api_object_versioned_finalize.go b/weed/s3api/s3api_object_versioned_finalize.go index e74790bd7..5f4726331 100644 --- a/weed/s3api/s3api_object_versioned_finalize.go +++ b/weed/s3api/s3api_object_versioned_finalize.go @@ -87,6 +87,76 @@ func (s3a *S3ApiServer) routedVersionedFinalize(owner pb.ServerAddress, bucket, } } +// removeUploadDirMutation deletes a completed upload's directory metadata-only: +// the finished object's chunks are the part chunks, so freeing their data would +// destroy the object. +func (s3a *S3ApiServer) removeUploadDirMutation(bucket, uploadID string) *filer_pb.ObjectMutation { + return &filer_pb.ObjectMutation{ + Type: filer_pb.ObjectMutation_DELETE, + Directory: s3a.genUploadsFolder(bucket), + Name: uploadID, + IsRecursive: true, + } +} + +// uploadExistsCondition requires the upload directory to still exist when the +// commit transaction runs, so a delete that does not take the object lock +// (abort, lifecycle, s3.clean.uploads) fails the commit instead of letting it +// publish the object over freed chunks. +func uploadExistsCondition(uploadDirectory string) (string, *filer_pb.WriteCondition) { + return uploadDirectory, &filer_pb.WriteCondition{ + Clauses: []*filer_pb.WriteCondition_Clause{{Kind: filer_pb.WriteCondition_IF_EXISTS}}, + } +} + +// routedMultipartFinalize commits a completed multipart upload in one +// ObjectTransaction on the owner filer: PUT the version file, remove the upload +// directory, recompute the latest pointer. The transaction applies mutations in +// order with no rollback, so the order picks which partial states are +// reachable: PUT first keeps the chunks referenced at all times, and removing +// the upload directory before the recompute means a published object never +// coexists with a leftover .uploads directory that s3.clean.uploads would purge +// with data. +func (s3a *S3ApiServer) routedMultipartFinalize(owner pb.ServerAddress, bucket, object string, useInvertedFormat bool, versionDir, versionFileName string, chunks []*filer_pb.FileChunk, decorate func(*filer_pb.Entry), uploadID string) s3err.ErrorCode { + now := time.Now().Unix() + versionEntry := &filer_pb.Entry{ + Name: versionFileName, + Attributes: &filer_pb.FuseAttributes{ + Mtime: now, + Crtime: now, + FileMode: uint32(0770), + Uid: filer_pb.OS_UID, + Gid: filer_pb.OS_GID, + }, + Chunks: chunks, + } + if decorate != nil { + decorate(versionEntry) + } + + routeKey := "" + if owner != "" { + routeKey = s3a.objectRouteKey(bucket, object) + } + conditionKey, condition := uploadExistsCondition(s3a.genUploadsFolder(bucket) + "/" + uploadID) + resp, err := s3a.routedPut(owner, routeKey, s3a.toFilerPath(bucket, object), versionDir+"/"+versionFileName, versionEntry, condition, conditionKey, []*filer_pb.ObjectMutation{ + s3a.removeUploadDirMutation(bucket, uploadID), + s3a.latestPointerRecompute(bucket, object, useInvertedFormat, "", true), + }) + switch { + case err != nil: + glog.Errorf("routedMultipartFinalize: %s/%s upload %s on %s: %v", bucket, object, uploadID, owner, err) + return s3err.ErrInternalError + case resp.ErrorCode == filer_pb.FilerError_PRECONDITION_FAILED: + return s3err.ErrNoSuchUpload + case resp.Error != "": + glog.Errorf("routedMultipartFinalize: %s/%s upload %s: %s", bucket, object, uploadID, resp.Error) + return s3err.ErrInternalError + default: + return s3err.ErrNone + } +} + // wormDeleteCondition returns the object-lock guards for a delete, or nil when // the bucket has no object lock. Governance bypass gates the retention check to // COMPLIANCE mode so the filer still protects compliance versions under lock. diff --git a/weed/shell/command_s3_clean_uploads.go b/weed/shell/command_s3_clean_uploads.go index b9d9bea5e..1da7a02d4 100644 --- a/weed/shell/command_s3_clean_uploads.go +++ b/weed/shell/command_s3_clean_uploads.go @@ -2,10 +2,13 @@ package shell import ( "context" + "errors" "flag" "fmt" "io" "math" + "path" + "strings" "time" "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" @@ -74,12 +77,12 @@ func (c *commandS3CleanUploads) Do(args []string, commandEnv *CommandEnv, writer func (c *commandS3CleanUploads) cleanupUploads(commandEnv *CommandEnv, writer io.Writer, filerBucketsPath string, bucket string, timeAgo time.Duration, signingKey string) error { uploadsDir := filerBucketsPath + "/" + bucket + "/" + s3_constants.MultipartUploadsFolder - var staleUploads []string + var staleUploads []*filer_pb.Entry now := time.Now() err := filer_pb.List(context.Background(), commandEnv, uploadsDir, "", func(entry *filer_pb.Entry, isLast bool) error { ctime := time.Unix(entry.Attributes.Crtime, 0) if ctime.Add(timeAgo).Before(now) { - staleUploads = append(staleUploads, entry.Name) + staleUploads = append(staleUploads, entry) } return nil }, "", false, math.MaxUint32) @@ -93,14 +96,75 @@ func (c *commandS3CleanUploads) cleanupUploads(commandEnv *CommandEnv, writer io } for _, staleUpload := range staleUploads { - deleteUrl := fmt.Sprintf("http://%s%s/%s?recursive=true&ignoreRecursiveError=true", commandEnv.option.FilerAddress.ToHttpAddress(), uploadsDir, staleUpload) + // A completed upload's part entries share chunks with the finished + // object, so purging their data corrupts it. Completion normally + // removes this directory itself; a survivor means that cleanup failed + // and only the metadata should go. An undecidable lookup is left for + // the next run rather than risk live chunks. + completed, checkErr := c.uploadCompleted(commandEnv, filerBucketsPath+"/"+bucket, staleUpload) + if checkErr != nil { + fmt.Fprintf(writer, "skip %s: %v\n", staleUpload.Name, checkErr) + continue + } + deleteUrl := fmt.Sprintf("http://%s%s/%s?recursive=true&ignoreRecursiveError=true", commandEnv.option.FilerAddress.ToHttpAddress(), uploadsDir, staleUpload.Name) + if completed { + deleteUrl += "&skipChunkDeletion=true" + } fmt.Fprintf(writer, "purge %s\n", deleteUrl) err = util_http.Delete(deleteUrl, string(encodedJwt)) if err != nil && err.Error() != "" { - return fmt.Errorf("purge %s/%s: %v", uploadsDir, staleUpload, err) + return fmt.Errorf("purge %s/%s: %v", uploadsDir, staleUpload.Name, err) } } return nil } + +// uploadCompleted reports whether the upload assembled into an object: the +// object entry, or any version file under .versions, still carries the +// upload id completion stamps on it. +func (c *commandS3CleanUploads) uploadCompleted(filerClient filer_pb.FilerClient, bucketDir string, upload *filer_pb.Entry) (bool, error) { + objectKey := string(upload.Extended[s3_constants.ExtMultipartObjectKey]) + if objectKey == "" { + return false, nil + } + // Derive the object location the same way completion's getEntryNameAndDir + // does: a trailing-slash key stores the object inside the directory it + // names, so FullPath+DirAndName would look one level too high. + name := path.Base(objectKey) + dir := path.Dir(objectKey) + if dir == "." { + dir = "" + } + objectDir := util.FullPath(bucketDir + "/" + dir) + + completed := false + err := filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + resp, err := filer_pb.LookupEntry(context.Background(), client, &filer_pb.LookupDirectoryEntryRequest{Directory: string(objectDir), Name: name}) + if errors.Is(err, filer_pb.ErrNotFound) { + return nil + } + if err != nil { + return err + } + if resp.Entry != nil && string(resp.Entry.Extended[s3_constants.SeaweedFSUploadId]) == upload.Name { + completed = true + } + return nil + }) + if err != nil || completed { + return completed, err + } + + err = filer_pb.List(context.Background(), filerClient, string(objectDir)+"/"+name+s3_constants.VersionsFolder, "", func(entry *filer_pb.Entry, isLast bool) error { + if string(entry.Extended[s3_constants.SeaweedFSUploadId]) == upload.Name { + completed = true + } + return nil + }, "", false, math.MaxUint32) + if err != nil && (errors.Is(err, filer_pb.ErrNotFound) || strings.Contains(err.Error(), filer_pb.ErrNotFound.Error())) { + return false, nil + } + return completed, err +} diff --git a/weed/shell/command_s3_clean_uploads_test.go b/weed/shell/command_s3_clean_uploads_test.go new file mode 100644 index 000000000..6da4ff1e7 --- /dev/null +++ b/weed/shell/command_s3_clean_uploads_test.go @@ -0,0 +1,171 @@ +package shell + +import ( + "context" + "errors" + "io" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "google.golang.org/grpc" + "google.golang.org/grpc/metadata" +) + +type stubFilerClient struct { + client filer_pb.SeaweedFilerClient +} + +func (s stubFilerClient) WithFilerClient(_ bool, fn func(filer_pb.SeaweedFilerClient) error) error { + return fn(s.client) +} + +func (s stubFilerClient) AdjustedUrl(location *filer_pb.Location) string { return "" } + +func (s stubFilerClient) GetDataCenter() string { return "" } + +type stubFilerSvc struct { + filer_pb.SeaweedFilerClient + lookupResp *filer_pb.LookupDirectoryEntryResponse + lookupErr error + listResp []*filer_pb.Entry + listErr error + lookupReq *filer_pb.LookupDirectoryEntryRequest + listReq *filer_pb.ListEntriesRequest +} + +func (s *stubFilerSvc) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest, opts ...grpc.CallOption) (*filer_pb.LookupDirectoryEntryResponse, error) { + s.lookupReq = req + return s.lookupResp, s.lookupErr +} + +func (s *stubFilerSvc) ListEntries(ctx context.Context, req *filer_pb.ListEntriesRequest, opts ...grpc.CallOption) (filer_pb.SeaweedFiler_ListEntriesClient, error) { + s.listReq = req + if s.listErr != nil { + return nil, s.listErr + } + return &stubListStream{entries: s.listResp}, nil +} + +type stubListStream struct { + entries []*filer_pb.Entry + idx int +} + +func (s *stubListStream) Header() (metadata.MD, error) { return nil, nil } +func (s *stubListStream) Trailer() metadata.MD { return nil } +func (s *stubListStream) CloseSend() error { return nil } +func (s *stubListStream) Context() context.Context { return context.Background() } +func (s *stubListStream) SendMsg(m any) error { return nil } +func (s *stubListStream) RecvMsg(m any) error { return nil } + +func (s *stubListStream) Recv() (*filer_pb.ListEntriesResponse, error) { + if s.idx >= len(s.entries) { + return nil, io.EOF + } + entry := s.entries[s.idx] + s.idx++ + return &filer_pb.ListEntriesResponse{Entry: entry}, nil +} + +func uploadEntry(uploadId, objectKey string) *filer_pb.Entry { + return &filer_pb.Entry{ + Name: uploadId, + Extended: map[string][]byte{s3_constants.ExtMultipartObjectKey: []byte(objectKey)}, + } +} + +func completedEntry(uploadId string) *filer_pb.Entry { + return &filer_pb.Entry{ + Extended: map[string][]byte{s3_constants.SeaweedFSUploadId: []byte(uploadId)}, + } +} + +func TestUploadCompleted(t *testing.T) { + c := &commandS3CleanUploads{} + + t.Run("object entry carries the upload id", func(t *testing.T) { + fc := stubFilerClient{client: &stubFilerSvc{ + lookupResp: &filer_pb.LookupDirectoryEntryResponse{Entry: completedEntry("up1")}, + }} + completed, err := c.uploadCompleted(fc, "/buckets/b", uploadEntry("up1", "obj")) + if err != nil || !completed { + t.Fatalf("got (%v, %v), want (true, nil)", completed, err) + } + }) + + t.Run("object entry has a different upload id", func(t *testing.T) { + fc := stubFilerClient{client: &stubFilerSvc{ + lookupResp: &filer_pb.LookupDirectoryEntryResponse{Entry: completedEntry("up2")}, + }} + completed, err := c.uploadCompleted(fc, "/buckets/b", uploadEntry("up1", "obj")) + if err != nil || completed { + t.Fatalf("got (%v, %v), want (false, nil)", completed, err) + } + }) + + t.Run("a version file carries the upload id", func(t *testing.T) { + fc := stubFilerClient{client: &stubFilerSvc{ + lookupErr: filer_pb.ErrNotFound, + listResp: []*filer_pb.Entry{ + completedEntry("up0"), + completedEntry("up1"), + }, + }} + completed, err := c.uploadCompleted(fc, "/buckets/b", uploadEntry("up1", "obj")) + if err != nil || !completed { + t.Fatalf("got (%v, %v), want (true, nil)", completed, err) + } + }) + + t.Run("no object and no versions", func(t *testing.T) { + fc := stubFilerClient{client: &stubFilerSvc{ + lookupErr: filer_pb.ErrNotFound, + }} + completed, err := c.uploadCompleted(fc, "/buckets/b", uploadEntry("up1", "obj")) + if err != nil || completed { + t.Fatalf("got (%v, %v), want (false, nil)", completed, err) + } + }) + + t.Run("upload dir without an object key", func(t *testing.T) { + completed, err := c.uploadCompleted(stubFilerClient{}, "/buckets/b", &filer_pb.Entry{Name: "up1"}) + if err != nil || completed { + t.Fatalf("got (%v, %v), want (false, nil)", completed, err) + } + }) + + t.Run("lookup error propagates", func(t *testing.T) { + fc := stubFilerClient{client: &stubFilerSvc{ + lookupErr: errors.New("store unavailable"), + }} + completed, err := c.uploadCompleted(fc, "/buckets/b", uploadEntry("up1", "obj")) + if err == nil || completed { + t.Fatalf("got (%v, %v), want (false, error)", completed, err) + } + }) + + t.Run("versions list not found means no completed object", func(t *testing.T) { + fc := stubFilerClient{client: &stubFilerSvc{ + lookupErr: filer_pb.ErrNotFound, + listErr: filer_pb.ErrNotFound, + }} + completed, err := c.uploadCompleted(fc, "/buckets/b", uploadEntry("up1", "obj")) + if err != nil || completed { + t.Fatalf("got (%v, %v), want (false, nil)", completed, err) + } + }) + + t.Run("trailing slash key resolves inside its directory", func(t *testing.T) { + svc := &stubFilerSvc{ + lookupResp: &filer_pb.LookupDirectoryEntryResponse{Entry: completedEntry("up1")}, + } + completed, err := c.uploadCompleted(stubFilerClient{client: svc}, "/buckets/b", uploadEntry("up1", "dir/")) + if err != nil || !completed { + t.Fatalf("got (%v, %v), want (true, nil)", completed, err) + } + if svc.lookupReq.Directory != "/buckets/b/dir" || svc.lookupReq.Name != "dir" { + t.Fatalf("lookup = %s/%s, want /buckets/b/dir/dir", svc.lookupReq.Directory, svc.lookupReq.Name) + } + }) +}