mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-10 13:58:24 +01:00
s3: evaluate multipart completion conditions under the object lock (#12209)
* filer_pb,filer: per-mutation preconditions on ObjectTransaction A transaction's request-level condition evaluates one entry before any mutation runs, so a commit guarding two paths — the object it publishes and the upload directory it assembles — could not be expressed. Give each ObjectMutation an optional condition (evaluated against condition_key, else the mutation's target, under the same lock just before the mutation applies) and report which mutation failed through failed_mutation on the response, so callers can tell a mutation guard's rejection apart from the request-level precondition's. * s3: evaluate multipart completion conditions under the object lock CompleteMultipartUpload checked conditional headers gateway-side before committing, and the routed commit transaction guarded only the upload directory — nothing made If-None-Match: * atomic with the publish, so concurrent conditional completions of the same absent key could all report success. The request's reducible write condition (If-None-Match/If-Match, star or single strong ETag) now rides the commit as the transaction's request condition on the owner filer: on the object path for unversioned buckets, falling back to the .versions directory for suspended and enabled versioning. The upload-exists guard moves onto the PUT mutation, keeping its check-before-publish ordering. Conditions the reducer cannot express stay on the distributed write lock, where check and commit already serialize. * filer,s3: resolve the latest version first for versioned conditional writes * filer: keep directory entries holding content out of the absent-slot rule * s3: keep a gateway-side upload-exists check for filers without mutation conditions * filer: honor null-version-is-latest resolving prefer-alt conditions * drop scratch multipart-race repro programs * s3: note the write lock does not cover routed writes on non-reducible conditions
This commit is contained in:
1 parent
3955e7218f
commit
2737315e9d
12 files changed
+670
-212
No files matched your search
@@ -689,6 +689,13 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl
|
||||
// retries. With no owner yet (no ring), keep the lock as the bootstrap path.
|
||||
owner := s3a.objectWriteOwner(*input.Bucket, *input.Key)
|
||||
routeKey := s3a.objectRouteKey(*input.Bucket, *input.Key)
|
||||
// The write condition rides the commit transaction so the owner filer
|
||||
// evaluates it under the object's path lock — the gateway's earlier
|
||||
// checkConditionalHeaders is only a fast-path. Conditions it cannot reduce
|
||||
// (time-based, combined, ETag lists, weak ETags) keep the pre-change
|
||||
// semantics: check and commit run under the distributed write lock, which
|
||||
// routed writes still bypass.
|
||||
cond, reducible := buildWriteCondition(r)
|
||||
completionBody := func() s3err.ErrorCode {
|
||||
var prepCode s3err.ErrorCode
|
||||
completionState, output, prepCode = s3a.prepareMultipartCompletionState(r, input, parts, uploadDirectory, entryName, dirName, completedPartNumbers, completedPartMap, maxPartNo)
|
||||
@@ -796,7 +803,7 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl
|
||||
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 := s3a.routedMultipartFinalize(owner, *input.Bucket, *input.Key, useInvertedFormat, versionDir, versionFileName, completionState.finalParts, decorateVersionEntry, *input.UploadId, cond); code != s3err.ErrNone {
|
||||
if code == s3err.ErrNoSuchUpload {
|
||||
return code
|
||||
}
|
||||
@@ -850,6 +857,9 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl
|
||||
glog.Errorf("completeMultipartUpload %s upload %s unused part cleanup: %v", *input.Bucket, *input.UploadId, err)
|
||||
return s3err.ErrInternalError
|
||||
}
|
||||
// The condition checks the null slot first, then the latest version,
|
||||
// same resolution the gateway's conditional check performs.
|
||||
versionsDir := dirName + "/" + entryName + s3_constants.VersionsFolder
|
||||
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)
|
||||
@@ -898,10 +908,13 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl
|
||||
entry.Attributes.Mime = completionState.mime
|
||||
}
|
||||
entry.Attributes.FileSize = uint64(completionState.offset)
|
||||
}, removal); err != nil {
|
||||
}, removal, cond, versionsDir); err != nil {
|
||||
if errors.Is(err, errUploadRemoved) {
|
||||
return s3err.ErrNoSuchUpload
|
||||
}
|
||||
if errors.Is(err, errObjectPrecondition) {
|
||||
return s3err.ErrPreconditionFailed
|
||||
}
|
||||
// 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 {
|
||||
@@ -991,10 +1004,13 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl
|
||||
if completionState.entityWithTtl {
|
||||
entry.Extended[s3_constants.SeaweedFSExpiresS3] = []byte("true")
|
||||
}
|
||||
}, removal); err != nil {
|
||||
}, removal, cond, ""); err != nil {
|
||||
if errors.Is(err, errUploadRemoved) {
|
||||
return s3err.ErrNoSuchUpload
|
||||
}
|
||||
if errors.Is(err, errObjectPrecondition) {
|
||||
return s3err.ErrPreconditionFailed
|
||||
}
|
||||
// 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 {
|
||||
@@ -1017,7 +1033,7 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl
|
||||
return s3err.ErrNone
|
||||
}
|
||||
var finalizeCode s3err.ErrorCode
|
||||
if owner != "" {
|
||||
if owner != "" && reducible {
|
||||
if code := s3a.checkConditionalHeaders(r, *input.Bucket, *input.Key); code != s3err.ErrNone {
|
||||
finalizeCode = code
|
||||
} else {
|
||||
|
||||
@@ -742,7 +742,7 @@ func (s3a *S3ApiServer) finalizeCopyDestination(dstBucket, dstObject, dstVersion
|
||||
if owner != "" {
|
||||
err = s3a.routedMkFile(owner, s3a.objectRouteKey(dstBucket, normalizedObject), bucketDir+"/"+normalizedObject+s3_constants.VersionsFolder, versionFileName, dstEntry.Chunks, func(entry *filer_pb.Entry) {
|
||||
copyEntryToTarget(entry, dstEntry)
|
||||
}, nil)
|
||||
}, nil, nil, "")
|
||||
} else {
|
||||
err = s3a.mkFile(bucketDir, versionObjectPath, dstEntry.Chunks, func(entry *filer_pb.Entry) {
|
||||
copyEntryToTarget(entry, dstEntry)
|
||||
@@ -779,7 +779,7 @@ func (s3a *S3ApiServer) finalizeCopyDestination(dstBucket, dstObject, dstVersion
|
||||
owner := s3a.objectWriteOwner(dstBucket, normalizedObject)
|
||||
if err = s3a.writeMultipartObject(owner, s3a.objectRouteKey(dstBucket, normalizedObject), dstDir, dstName, dstEntry.Chunks, func(entry *filer_pb.Entry) {
|
||||
copyEntryToTarget(entry, dstEntry)
|
||||
}, nil); err != nil {
|
||||
}, nil, nil, ""); err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
|
||||
@@ -799,7 +799,7 @@ func (s3a *S3ApiServer) finalizeCopyDestination(dstBucket, dstObject, dstVersion
|
||||
// default filer is invisible to the next copy until replication.
|
||||
if err = s3a.writeMultipartObject(s3a.objectWriteOwner(dstBucket, normalizedObject), s3a.objectRouteKey(dstBucket, normalizedObject), dstDir, dstName, dstEntry.Chunks, func(entry *filer_pb.Entry) {
|
||||
copyEntryToTarget(entry, dstEntry)
|
||||
}, nil); err != nil {
|
||||
}, nil, nil, ""); err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
|
||||
|
||||
@@ -1084,7 +1084,7 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader
|
||||
lockKey, finalizeMutations = finalize.lockKey, finalize.mutations
|
||||
}
|
||||
routeKey := s3a.objectRouteKey(bucket, object)
|
||||
resp, err := s3a.routedPut(owner, routeKey, lockKey, filePath, entry, cond, "", condAltKey, finalizeMutations)
|
||||
resp, err := s3a.routedPut(owner, routeKey, lockKey, filePath, entry, cond, "", condAltKey, false, nil, "", finalizeMutations)
|
||||
if err == nil && resp.ErrorCode == filer_pb.FilerError_PRECONDITION_FAILED && condAltKey != "" {
|
||||
// A filer that ignores condition_alt_key evaluated the empty
|
||||
// null slot alone; resend the transaction against the entry the
|
||||
@@ -1097,7 +1097,7 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader
|
||||
conditionKey = condAltKey + "/" + resolved.Name
|
||||
}
|
||||
glog.Warningf("putToFiler: routed PUT to %s precondition-failed for %s; resending against %s", owner, filePath, conditionKey)
|
||||
resp, err = s3a.routedPut(owner, routeKey, lockKey, filePath, entry, cond, conditionKey, condAltKey, finalizeMutations)
|
||||
resp, err = s3a.routedPut(owner, routeKey, lockKey, filePath, entry, cond, conditionKey, condAltKey, false, nil, "", finalizeMutations)
|
||||
}
|
||||
switch {
|
||||
case err != nil:
|
||||
|
||||
@@ -216,28 +216,37 @@ func (s3a *S3ApiServer) withWriteOwnerFilerClient(owner pb.ServerAddress, fn fun
|
||||
// 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, conditionKey, conditionAltKey string, finalize []*filer_pb.ObjectMutation) (*filer_pb.ObjectTransactionResponse, error) {
|
||||
// putCondition, when set, guards the PUT mutation itself against putConditionKey
|
||||
// (e.g. the upload directory a multipart commit must still find); the request
|
||||
// condition keeps guarding the object the transaction is about.
|
||||
func (s3a *S3ApiServer) routedPut(owner pb.ServerAddress, routeKey, lockKey, filePath string, entry *filer_pb.Entry, cond *filer_pb.WriteCondition, conditionKey, conditionAltKey string, conditionPreferAlt bool, putCondition *filer_pb.WriteCondition, putConditionKey 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,
|
||||
Directory: path.Dir(filePath),
|
||||
Entry: entry,
|
||||
Type: filer_pb.ObjectMutation_PUT,
|
||||
Directory: path.Dir(filePath),
|
||||
Entry: entry,
|
||||
Condition: putCondition,
|
||||
ConditionKey: putConditionKey,
|
||||
})
|
||||
mutations = append(mutations, finalize...)
|
||||
return s3a.objectTxnOnFiler(owner, &filer_pb.ObjectTransactionRequest{
|
||||
LockKey: lockKey,
|
||||
RouteKey: routeKey,
|
||||
Condition: cond,
|
||||
ConditionKey: conditionKey,
|
||||
ConditionAltKey: conditionAltKey,
|
||||
Mutations: mutations,
|
||||
LockKey: lockKey,
|
||||
RouteKey: routeKey,
|
||||
Condition: cond,
|
||||
ConditionKey: conditionKey,
|
||||
ConditionAltKey: conditionAltKey,
|
||||
ConditionPreferAlt: conditionPreferAlt,
|
||||
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), removal *uploadRemovalTxn) error {
|
||||
// default filer (e.g. multipart completion of a non-versioned object). cond is
|
||||
// the request's write condition, evaluated on the object under the transaction
|
||||
// lock; condAltKey falls back to the .versions directory when the object slot
|
||||
// is empty (suspended or versioned buckets).
|
||||
func (s3a *S3ApiServer) routedMkFile(owner pb.ServerAddress, routeKey, parentDir, name string, chunks []*filer_pb.FileChunk, fn func(*filer_pb.Entry), removal *uploadRemovalTxn, cond *filer_pb.WriteCondition, condAltKey string) error {
|
||||
now := time.Now().Unix()
|
||||
entry := &filer_pb.Entry{
|
||||
Name: name,
|
||||
@@ -254,14 +263,14 @@ func (s3a *S3ApiServer) routedMkFile(owner pb.ServerAddress, routeKey, parentDir
|
||||
fn(entry)
|
||||
}
|
||||
filePath := parentDir + "/" + name
|
||||
var cond *filer_pb.WriteCondition
|
||||
var conditionKey string
|
||||
var putCondition *filer_pb.WriteCondition
|
||||
var putConditionKey string
|
||||
var finalize []*filer_pb.ObjectMutation
|
||||
if removal != nil {
|
||||
cond, conditionKey = removal.condition, removal.conditionKey
|
||||
putCondition, putConditionKey = removal.putCondition, removal.putConditionKey
|
||||
finalize = []*filer_pb.ObjectMutation{removal.mutation}
|
||||
}
|
||||
resp, err := s3a.routedPut(owner, routeKey, filePath, filePath, entry, cond, conditionKey, "", finalize)
|
||||
resp, err := s3a.routedPut(owner, routeKey, filePath, filePath, entry, cond, "", condAltKey, false, putCondition, putConditionKey, finalize)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -269,12 +278,15 @@ func (s3a *S3ApiServer) routedMkFile(owner pb.ServerAddress, routeKey, parentDir
|
||||
// Other keys are nested under this one, so the object goes onto the
|
||||
// directory they live in — the same retry mkFile makes off the ring.
|
||||
entry.MarkPrefixObject()
|
||||
resp, err = s3a.routedPut(owner, routeKey, filePath, filePath, entry, cond, conditionKey, "", finalize)
|
||||
resp, err = s3a.routedPut(owner, routeKey, filePath, filePath, entry, cond, "", condAltKey, false, putCondition, putConditionKey, finalize)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if resp.ErrorCode == filer_pb.FilerError_PRECONDITION_FAILED {
|
||||
if resp.FailedMutation == nil {
|
||||
return errObjectPrecondition
|
||||
}
|
||||
return errUploadRemoved
|
||||
}
|
||||
if resp.Error != "" {
|
||||
@@ -283,32 +295,40 @@ func (s3a *S3ApiServer) routedMkFile(owner pb.ServerAddress, routeKey, parentDir
|
||||
return nil
|
||||
}
|
||||
|
||||
// errUploadRemoved is a routed commit rejected because its precondition found
|
||||
// the upload directory already deleted, mapped to NoSuchUpload by callers.
|
||||
// errUploadRemoved is a routed commit rejected because the upload directory's
|
||||
// precondition found it already deleted, mapped to NoSuchUpload by callers.
|
||||
var errUploadRemoved = errors.New("upload directory already removed")
|
||||
|
||||
// errObjectPrecondition is a routed commit rejected because the request's write
|
||||
// condition (e.g. If-None-Match: *) found the object slot already taken by a
|
||||
// racing commit, mapped to PreconditionFailed by callers.
|
||||
var errObjectPrecondition = errors.New("object precondition failed")
|
||||
|
||||
// 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. 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 {
|
||||
// selection and forwarding stay consistent. cond is the request's write
|
||||
// condition, enforced on the owner under the transaction lock; condAltKey is
|
||||
// the .versions directory a conditional check falls back to when the bucket
|
||||
// tracks versions.
|
||||
func (s3a *S3ApiServer) writeMultipartObject(owner pb.ServerAddress, routeKey, dir, name string, chunks []*filer_pb.FileChunk, fn func(*filer_pb.Entry), removal *uploadRemovalTxn, cond *filer_pb.WriteCondition, condAltKey string) error {
|
||||
if owner != "" {
|
||||
return s3a.routedMkFile(owner, routeKey, dir, name, chunks, fn, removal)
|
||||
return s3a.routedMkFile(owner, routeKey, dir, name, chunks, fn, removal, cond, condAltKey)
|
||||
}
|
||||
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.
|
||||
// uploadRemovalTxn carries the PUT 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
|
||||
putCondition *filer_pb.WriteCondition
|
||||
putConditionKey string
|
||||
mutation *filer_pb.ObjectMutation
|
||||
}
|
||||
|
||||
// routedUploadRemoval returns the transaction parts that remove a completed
|
||||
@@ -319,14 +339,20 @@ func (s3a *S3ApiServer) routedUploadRemoval(ctx context.Context, owner pb.Server
|
||||
if owner == "" {
|
||||
return nil, nil
|
||||
}
|
||||
// Older filers ignore the mutation's upload-exists condition, so during an
|
||||
// upgrade this gateway-side check keeps the same guard window the
|
||||
// pre-transaction code had.
|
||||
if exists, err := s3a.exists(s3a.genUploadsFolder(bucket), uploadId, true); err == nil && !exists {
|
||||
return nil, errUploadRemoved
|
||||
}
|
||||
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),
|
||||
putCondition: condition,
|
||||
putConditionKey: conditionKey,
|
||||
mutation: s3a.removeUploadDirMutation(bucket, uploadId),
|
||||
}, nil
|
||||
}
|
||||
|
||||
|
||||
@@ -514,7 +514,7 @@ func TestRoutedMultipartFinalize(t *testing.T) {
|
||||
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")
|
||||
}, "up1", clause(filer_pb.WriteCondition_IF_NOT_EXISTS))
|
||||
if code != s3err.ErrNone {
|
||||
t.Fatalf("routedMultipartFinalize = %v", code)
|
||||
}
|
||||
@@ -529,10 +529,15 @@ func TestRoutedMultipartFinalize(t *testing.T) {
|
||||
if req.RouteKey != "s3.object.write:/buckets/b/obj" {
|
||||
t.Fatalf("RouteKey = %q", req.RouteKey)
|
||||
}
|
||||
if req.ConditionKey != "/buckets/b/.uploads/up1" {
|
||||
// The request condition guards the object (at lock_key, falling back to
|
||||
// .versions); the upload directory rides the PUT as a mutation guard.
|
||||
if req.ConditionKey != "" {
|
||||
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 {
|
||||
if req.ConditionAltKey != "/buckets/b/obj.versions" {
|
||||
t.Fatalf("ConditionAltKey = %q", req.ConditionAltKey)
|
||||
}
|
||||
if req.Condition == nil || len(req.Condition.Clauses) != 1 || req.Condition.Clauses[0].Kind != filer_pb.WriteCondition_IF_NOT_EXISTS {
|
||||
t.Fatalf("Condition = %+v", req.Condition)
|
||||
}
|
||||
if len(req.Mutations) != 3 {
|
||||
@@ -548,6 +553,10 @@ func TestRoutedMultipartFinalize(t *testing.T) {
|
||||
string(put.Entry.Extended[s3_constants.SeaweedFSUploadId]) != "up1" {
|
||||
t.Fatalf("put mutation = %+v", put)
|
||||
}
|
||||
if put.ConditionKey != "/buckets/b/.uploads/up1" ||
|
||||
put.Condition == nil || len(put.Condition.Clauses) != 1 || put.Condition.Clauses[0].Kind != filer_pb.WriteCondition_IF_EXISTS {
|
||||
t.Fatalf("put condition = %+v key %q", put.Condition, put.ConditionKey)
|
||||
}
|
||||
removeUpload := req.Mutations[1]
|
||||
if removeUpload.Type != filer_pb.ObjectMutation_DELETE ||
|
||||
removeUpload.Directory != "/buckets/b/.uploads" ||
|
||||
@@ -580,7 +589,7 @@ func TestWriteMultipartObjectRemovesUploadDir(t *testing.T) {
|
||||
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 {
|
||||
if err := s3a.writeMultipartObject(owner, "s3.object.write:/buckets/b/o", "/buckets/b", "o", nil, nil, removal, nil, ""); err != nil {
|
||||
t.Fatalf("writeMultipartObject: %v", err)
|
||||
}
|
||||
|
||||
@@ -591,12 +600,6 @@ func TestWriteMultipartObjectRemovesUploadDir(t *testing.T) {
|
||||
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))
|
||||
}
|
||||
@@ -604,6 +607,10 @@ func TestWriteMultipartObjectRemovesUploadDir(t *testing.T) {
|
||||
if put.Type != filer_pb.ObjectMutation_PUT || put.Directory != "/buckets/b" || put.Entry == nil || put.Entry.Name != "o" {
|
||||
t.Fatalf("put mutation = %+v", put)
|
||||
}
|
||||
if put.ConditionKey != "/buckets/b/.uploads/up1" ||
|
||||
put.Condition == nil || len(put.Condition.Clauses) != 1 || put.Condition.Clauses[0].Kind != filer_pb.WriteCondition_IF_EXISTS {
|
||||
t.Fatalf("put condition = %+v key %q", put.Condition, put.ConditionKey)
|
||||
}
|
||||
removeUpload := req.Mutations[1]
|
||||
if removeUpload.Type != filer_pb.ObjectMutation_DELETE ||
|
||||
removeUpload.Directory != "/buckets/b/.uploads" ||
|
||||
|
||||
@@ -116,8 +116,10 @@ func uploadExistsCondition(uploadDirectory string) (string, *filer_pb.WriteCondi
|
||||
// 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 {
|
||||
// with data. cond is the request's write condition, checked against the object
|
||||
// (its latest version when the slot at the object path is empty) under the same
|
||||
// lock; the upload directory rides the PUT as a mutation precondition.
|
||||
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, cond *filer_pb.WriteCondition) s3err.ErrorCode {
|
||||
now := time.Now().Unix()
|
||||
versionEntry := &filer_pb.Entry{
|
||||
Name: versionFileName,
|
||||
@@ -138,8 +140,14 @@ func (s3a *S3ApiServer) routedMultipartFinalize(owner pb.ServerAddress, bucket,
|
||||
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{
|
||||
// A filer older than the mutation-condition support silently ignores
|
||||
// putCondition, so on upgrade the guard degrades to this gateway-side
|
||||
// check — the same window the pre-transaction code had.
|
||||
if exists, err := s3a.exists(s3a.genUploadsFolder(bucket), uploadID, true); err == nil && !exists {
|
||||
return s3err.ErrNoSuchUpload
|
||||
}
|
||||
putConditionKey, putCondition := uploadExistsCondition(s3a.genUploadsFolder(bucket) + "/" + uploadID)
|
||||
resp, err := s3a.routedPut(owner, routeKey, s3a.toFilerPath(bucket, object), versionDir+"/"+versionFileName, versionEntry, cond, "", versionDir, true, putCondition, putConditionKey, []*filer_pb.ObjectMutation{
|
||||
s3a.removeUploadDirMutation(bucket, uploadID),
|
||||
s3a.latestPointerRecompute(bucket, object, useInvertedFormat, "", true),
|
||||
})
|
||||
@@ -148,6 +156,9 @@ func (s3a *S3ApiServer) routedMultipartFinalize(owner pb.ServerAddress, bucket,
|
||||
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:
|
||||
if resp.FailedMutation == nil {
|
||||
return s3err.ErrPreconditionFailed
|
||||
}
|
||||
return s3err.ErrNoSuchUpload
|
||||
case resp.Error != "":
|
||||
glog.Errorf("routedMultipartFinalize: %s/%s upload %s: %s", bucket, object, uploadID, resp.Error)
|
||||
|
||||
Reference in new issue
Block a user