diff --git a/weed/s3api/s3api_object_handlers_copy_part_sse.go b/weed/s3api/s3api_object_handlers_copy_part_sse.go index 097dcc26c..66442a6d7 100644 --- a/weed/s3api/s3api_object_handlers_copy_part_sse.go +++ b/weed/s3api/s3api_object_handlers_copy_part_sse.go @@ -397,7 +397,7 @@ func (s3a *S3ApiServer) copyObjectPartViaReencryption( filePath := s3a.genPartUploadPath(dstBucket, uploadID, partID) // Copy-part is an MPU part write under .uploads//; lifecycle // TTL only applies to the eventual completed object. Pass 0. - tag, code, putSSE := s3a.putToFiler(cloned, filePath, srcReader, dstBucket, "", partID, 0, nil) + tag, code, putSSE := s3a.putToFiler(cloned, filePath, srcReader, dstBucket, "", partID, 0, nil, false) if code != s3err.ErrNone { return "", SSEResponseMetadata{}, code } diff --git a/weed/s3api/s3api_object_handlers_delete.go b/weed/s3api/s3api_object_handlers_delete.go index 304a27825..67bf2c2d9 100644 --- a/weed/s3api/s3api_object_handlers_delete.go +++ b/weed/s3api/s3api_object_handlers_delete.go @@ -214,33 +214,59 @@ func (s3a *S3ApiServer) DeleteObjectHandler(w http.ResponseWriter, r *http.Reque } var deleteResult deleteMutationResult - deleteCode := s3a.withObjectWriteLock(bucket, object, func() s3err.ErrorCode { - return s3a.checkDeleteIfMatch(bucket, object, versionId, versioningState, r.Header.Get(s3_constants.IfMatch), s3err.ErrPreconditionFailed) - }, func() s3err.ErrorCode { - if versioningConfigured { - result, errCode := s3a.deleteVersionedObject(r, bucket, object, versionId, versioningState) - if errCode != s3err.ErrNone { - return errCode + var deleteCode s3err.ErrorCode + + // Fast path: route the delete to the owner filer under its per-path lock; + // routedObjectOwner excludes versioned/object-lock buckets. + deleteHandled := false + if !versioningConfigured { + if cond, condOk := buildDeleteCondition(r); condOk { + if owner, ownerOk := s3a.routedObjectOwner(bucket, object); ownerOk { + resp, err := s3a.routedDelete(owner, bucket, object, cond) + switch { + case err != nil: + glog.Warningf("DeleteObjectHandler: routed delete to %s failed for %s/%s, falling back to lock: %v", owner, bucket, object, err) + case resp.ErrorCode == filer_pb.FilerError_PRECONDITION_FAILED: + deleteCode, deleteHandled = s3err.ErrPreconditionFailed, true + case resp.Error != "": + // Non-precondition error: fall back (the lock path handles cases + // the raw delete cannot, e.g. a non-empty directory marker). + glog.Warningf("DeleteObjectHandler: routed delete to %s returned %q for %s/%s, falling back to lock", owner, resp.Error, bucket, object) + default: + deleteCode, deleteHandled = s3err.ErrNone, true + } } - deleteResult = result + } + } + if !deleteHandled { + deleteCode = s3a.withObjectWriteLock(bucket, object, func() s3err.ErrorCode { + return s3a.checkDeleteIfMatch(bucket, object, versionId, versioningState, r.Header.Get(s3_constants.IfMatch), s3err.ErrPreconditionFailed) + }, func() s3err.ErrorCode { + if versioningConfigured { + result, errCode := s3a.deleteVersionedObject(r, bucket, object, versionId, versioningState) + if errCode != s3err.ErrNone { + return errCode + } + deleteResult = result + return s3err.ErrNone + } + + governanceBypassAllowed := s3a.evaluateGovernanceBypassRequest(r, bucket, object) + if err := s3a.enforceObjectLockProtections(r, bucket, object, "", governanceBypassAllowed); err != nil { + glog.V(2).Infof("DeleteObjectHandler: object lock check failed for %s/%s: %v", bucket, object, err) + return s3err.ErrAccessDenied + } + + if err := s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + return s3a.deleteUnversionedObjectWithClient(client, bucket, object, false) + }); err != nil { + glog.Errorf("DeleteObjectHandler: failed to delete %s/%s: %v", bucket, object, err) + return s3err.ErrInternalError + } + return s3err.ErrNone - } - - governanceBypassAllowed := s3a.evaluateGovernanceBypassRequest(r, bucket, object) - if err := s3a.enforceObjectLockProtections(r, bucket, object, "", governanceBypassAllowed); err != nil { - glog.V(2).Infof("DeleteObjectHandler: object lock check failed for %s/%s: %v", bucket, object, err) - return s3err.ErrAccessDenied - } - - if err := s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - return s3a.deleteUnversionedObjectWithClient(client, bucket, object, false) - }); err != nil { - glog.Errorf("DeleteObjectHandler: failed to delete %s/%s: %v", bucket, object, err) - return s3err.ErrInternalError - } - - return s3err.ErrNone - }) + }) + } if deleteCode != s3err.ErrNone { s3err.WriteErrorResponse(w, r, deleteCode) return diff --git a/weed/s3api/s3api_object_handlers_multipart.go b/weed/s3api/s3api_object_handlers_multipart.go index cb53d51a3..43084be69 100644 --- a/weed/s3api/s3api_object_handlers_multipart.go +++ b/weed/s3api/s3api_object_handlers_multipart.go @@ -457,7 +457,7 @@ func (s3a *S3ApiServer) PutObjectPartHandler(w http.ResponseWriter, r *http.Requ // transient .uploads// path, and a part write would otherwise // start the TTL clock before CompleteMultipartUpload ever assembled // the object. - etag, errCode, sseMetadata := s3a.putToFiler(r, filePath, dataReader, bucket, "", partID, 0, nil) + etag, errCode, sseMetadata := s3a.putToFiler(r, filePath, dataReader, bucket, "", partID, 0, nil, false) if errCode != s3err.ErrNone { glog.Errorf("PutObjectPart: putToFiler failed with error code %v for bucket=%s, object=%s, partNumber=%d", errCode, bucket, object, partID) diff --git a/weed/s3api/s3api_object_handlers_postpolicy.go b/weed/s3api/s3api_object_handlers_postpolicy.go index 94a461b6b..b3cab13b4 100644 --- a/weed/s3api/s3api_object_handlers_postpolicy.go +++ b/weed/s3api/s3api_object_handlers_postpolicy.go @@ -135,7 +135,7 @@ func (s3a *S3ApiServer) PostPolicyBucketHandler(w http.ResponseWriter, r *http.R // fields and boundaries inflates ContentLength relative to the // object body, which would mis-evaluate any size-filtered rule. ttlSec := s3a.lifecycleTTLForObjectWrite(bucket, object, fileSize) - etag, errCode, sseMetadata := s3a.putToFiler(r, filePath, fileBody, bucket, object, 1, ttlSec, nil) + etag, errCode, sseMetadata := s3a.putToFiler(r, filePath, fileBody, bucket, object, 1, ttlSec, nil, false) if errCode != s3err.ErrNone { s3err.WriteErrorResponse(w, r, errCode) diff --git a/weed/s3api/s3api_object_handlers_put.go b/weed/s3api/s3api_object_handlers_put.go index 01ef148d7..7ad2b0f11 100644 --- a/weed/s3api/s3api_object_handlers_put.go +++ b/weed/s3api/s3api_object_handlers_put.go @@ -297,7 +297,7 @@ func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request) } ttlSec := s3a.lifecycleTTLForObjectWrite(bucket, object, r.ContentLength) - etag, errCode, sseMetadata := s3a.putToFiler(r, filePath, dataReader, bucket, object, 1, ttlSec, nil) + etag, errCode, sseMetadata := s3a.putToFiler(r, filePath, dataReader, bucket, object, 1, ttlSec, nil, false) if errCode != s3err.ErrNone { s3err.WriteErrorResponse(w, r, errCode) @@ -359,7 +359,7 @@ func (s3a *S3ApiServer) withObjectWriteLock(bucket, object string, preconditionF // pass 0 because their own keys aren't the user-visible object the rule // targets and a part write would otherwise bind a TTL clock starting // before CompleteMultipartUpload. -func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader io.Reader, bucket string, object string, partNumber int, lifecycleTTLSec int32, afterCreate func(entry *filer_pb.Entry) s3err.ErrorCode) (etag string, code s3err.ErrorCode, sseMetadata SSEResponseMetadata) { +func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader io.Reader, bucket string, object string, partNumber int, lifecycleTTLSec int32, afterCreate func(entry *filer_pb.Entry) s3err.ErrorCode, uniqueWritePath bool) (etag string, code s3err.ErrorCode, sseMetadata SSEResponseMetadata) { // NEW OPTIMIZATION: Write directly to volume servers, bypassing filer proxy // This eliminates the filer proxy overhead for PUT operations // Note: filePath is now passed directly instead of URL (no parsing needed) @@ -802,7 +802,7 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader } return s3a.checkConditionalHeaders(r, bucket, object) } - createCode := s3a.withObjectWriteLock(bucket, object, preconditionFn, func() s3err.ErrorCode { + createUnderLock := func() s3err.ErrorCode { createErr = s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { req := &filer_pb.CreateEntryRequest{ Directory: path.Dir(filePath), @@ -831,7 +831,36 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader } } return s3err.ErrNone - }) + } + + // Route the create to the object's owner filer, whose per-path lock + // serializes it, then run afterCreate (e.g. a versioned finalize that routes + // itself). Conditional/object-lock/non-reducible cases fall back to the + // distributed lock. + var createCode s3err.ErrorCode + routed := false + if owner := s3a.routableWriteOwner(bucket, object); owner != "" { + if cond, ok := routeWriteCondition(r, uniqueWritePath); ok { + resp, err := s3a.routedPut(owner, filePath, entry, cond) + switch { + case err != nil: + glog.Warningf("putToFiler: routed PUT to %s failed for %s, falling back to lock: %v", owner, filePath, err) + case resp.ErrorCode == filer_pb.FilerError_PRECONDITION_FAILED: + createCode, routed = s3err.ErrPreconditionFailed, true + case resp.Error != "": + // Non-precondition mutation error: fall back so the lock path maps it. + glog.Warningf("putToFiler: routed PUT to %s returned %q for %s, falling back to lock", owner, resp.Error, filePath) + default: + entryCreated, routed, createCode = true, true, s3err.ErrNone + if afterCreate != nil { + createCode = afterCreate(entry) + } + } + } + } + if !routed { + createCode = s3a.withObjectWriteLock(bucket, object, preconditionFn, createUnderLock) + } if createCode != s3err.ErrNone { if createErr != nil { glog.Errorf("putToFiler: failed to create entry for %s: %v", filePath, createErr) @@ -1287,7 +1316,7 @@ func (s3a *S3ApiServer) putSuspendedVersioningObject(r *http.Request, bucket, ob glog.Warningf("putSuspendedVersioningObject: failed to update IsLatest flags: %v", err) } return s3err.ErrNone - }) + }, false) if errCode != s3err.ErrNone { glog.Errorf("putSuspendedVersioningObject: failed to upload object: %v", errCode) return "", errCode, SSEResponseMetadata{} @@ -1459,7 +1488,7 @@ func (s3a *S3ApiServer) putVersionedObject(r *http.Request, bucket, object strin return s3err.ErrInternalError } return s3err.ErrNone - }) + }, false) if errCode != s3err.ErrNone { glog.Errorf("putVersionedObject: failed to upload version: %v", errCode) return "", "", errCode, SSEResponseMetadata{} diff --git a/weed/s3api/s3api_object_routed_write.go b/weed/s3api/s3api_object_routed_write.go new file mode 100644 index 000000000..947e4f1b9 --- /dev/null +++ b/weed/s3api/s3api_object_routed_write.go @@ -0,0 +1,171 @@ +package s3api + +import ( + "context" + "fmt" + "net/http" + "path" + "strings" + + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3err" + "github.com/seaweedfs/seaweedfs/weed/util" +) + +// routableWriteOwner returns the owner filer for an object's writes, or "" to +// keep them on the distributed lock. Versioned and object-lock buckets stay on +// the lock (handled by later routing PRs); any lookup error also falls back. +func (s3a *S3ApiServer) routableWriteOwner(bucket, object string) pb.ServerAddress { + if object == "" || s3a.objectWriteLockClient == nil { + return "" + } + if configured, err := s3a.isVersioningConfigured(bucket); err != nil || configured { + return "" + } + if locked, err := s3a.isObjectLockEnabled(bucket); err != nil || locked { + return "" + } + return s3a.objectWriteLockClient.PrimaryForKey(fmt.Sprintf("s3.object.write:%s", s3a.toFilerPath(bucket, object))) +} + +// routedObjectOwner resolves the owner for the unversioned DELETE fast path. +func (s3a *S3ApiServer) routedObjectOwner(bucket, object string) (pb.ServerAddress, bool) { + owner := s3a.routableWriteOwner(bucket, object) + return owner, owner != "" +} + +// routeWriteCondition reduces the request's conditional headers for a routed +// create. A unique version path carries no precondition and only routes when the +// request is unconditional (a conditional versioned write must check the latest, +// which the lock path does); an overwrite carries the reduced condition. +func routeWriteCondition(r *http.Request, uniqueWritePath bool) (*filer_pb.WriteCondition, bool) { + cond, ok := buildWriteCondition(r) + if !ok { + return nil, false + } + if uniqueWritePath && cond != nil { + return nil, false + } + return cond, true +} + +// buildWriteCondition reduces the request's conditional headers to a +// WriteCondition. ok=false (combined headers, time conditions, ETag lists, weak +// ETags) keeps gateway-side evaluation under the lock; a nil condition with +// ok=true means unconditional. +func buildWriteCondition(r *http.Request) (*filer_pb.WriteCondition, bool) { + headers, errCode := parseConditionalHeaders(r) + if errCode != s3err.ErrNone { + return nil, false + } + if !headers.isSet { + return nil, true + } + if !headers.ifModifiedSince.IsZero() || !headers.ifUnmodifiedSince.IsZero() { + return nil, false + } + hasMatch := headers.ifMatch != "" + hasNoneMatch := headers.ifNoneMatch != "" + switch { + case hasMatch && !hasNoneMatch: + if headers.ifMatch == "*" { + return clause(filer_pb.WriteCondition_IF_EXISTS), true + } + if etag, single := singleStrongETag(headers.ifMatch); single { + return etagClause(filer_pb.WriteCondition_IF_ETAG_MATCH, etag), true + } + return nil, false + case hasNoneMatch && !hasMatch: + if headers.ifNoneMatch == "*" { + return clause(filer_pb.WriteCondition_IF_NOT_EXISTS), true + } + if etag, single := singleStrongETag(headers.ifNoneMatch); single { + return etagClause(filer_pb.WriteCondition_IF_ETAG_NOT_MATCH, etag), true + } + return nil, false + default: + return nil, false + } +} + +// buildDeleteCondition reduces a DeleteObject's If-Match header to a condition; +// DeleteObject honors only If-Match, matching checkDeleteIfMatch. +func buildDeleteCondition(r *http.Request) (*filer_pb.WriteCondition, bool) { + ifMatch := strings.TrimSpace(r.Header.Get(s3_constants.IfMatch)) + switch { + case ifMatch == "": + return nil, true + case ifMatch == "*": + return clause(filer_pb.WriteCondition_IF_EXISTS), true + default: + if etag, single := singleStrongETag(ifMatch); single { + return etagClause(filer_pb.WriteCondition_IF_ETAG_MATCH, etag), true + } + return nil, false + } +} + +func clause(kind filer_pb.WriteCondition_Kind) *filer_pb.WriteCondition { + return &filer_pb.WriteCondition{Clauses: []*filer_pb.WriteCondition_Clause{{Kind: kind}}} +} + +func etagClause(kind filer_pb.WriteCondition_Kind, etag string) *filer_pb.WriteCondition { + return &filer_pb.WriteCondition{Clauses: []*filer_pb.WriteCondition_Clause{{Kind: kind, Etags: []string{etag}}}} +} + +// singleStrongETag returns the normalized ETag when v carries exactly one strong +// ETag, and false for ETag lists or weak ("W/") ETags. +func singleStrongETag(v string) (string, bool) { + v = strings.TrimSpace(v) + if strings.Contains(v, ",") { + return "", false + } + if strings.HasPrefix(v, "W/") || strings.HasPrefix(v, "w/") { + return "", false + } + return strings.Trim(v, `"`), true +} + +func (s3a *S3ApiServer) objectTxnOnFiler(owner pb.ServerAddress, req *filer_pb.ObjectTransactionRequest) (*filer_pb.ObjectTransactionResponse, error) { + var resp *filer_pb.ObjectTransactionResponse + err := pb.WithFilerClient(false, 0, owner, s3a.option.GrpcDialOption, func(client filer_pb.SeaweedFilerClient) error { + var e error + resp, e = client.ObjectTransaction(context.Background(), req) + return e + }) + return resp, err +} + +// routedPut writes an object entry as a one-mutation ObjectTransaction on the +// owner filer. lock_key is the object's full path so the transaction shares the +// per-path lock with a concurrent create or delete of the same key. +func (s3a *S3ApiServer) routedPut(owner pb.ServerAddress, filePath string, entry *filer_pb.Entry, cond *filer_pb.WriteCondition) (*filer_pb.ObjectTransactionResponse, error) { + return s3a.objectTxnOnFiler(owner, &filer_pb.ObjectTransactionRequest{ + LockKey: filePath, + Condition: cond, + Mutations: []*filer_pb.ObjectMutation{{ + Type: filer_pb.ObjectMutation_PUT, + Directory: path.Dir(filePath), + Entry: entry, + }}, + }) +} + +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. + fullpath := util.NewFullPath(s3a.bucketDir(bucket), object) + dir, name := fullpath.DirAndName() + return s3a.objectTxnOnFiler(owner, &filer_pb.ObjectTransactionRequest{ + LockKey: string(fullpath), + Condition: cond, + Mutations: []*filer_pb.ObjectMutation{{ + Type: filer_pb.ObjectMutation_DELETE, + Directory: dir, + Name: name, + IsDeleteData: true, + }}, + }) +} diff --git a/weed/s3api/s3api_object_routed_write_test.go b/weed/s3api/s3api_object_routed_write_test.go new file mode 100644 index 000000000..1efc6078c --- /dev/null +++ b/weed/s3api/s3api_object_routed_write_test.go @@ -0,0 +1,177 @@ +package s3api + +import ( + "net/http" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" +) + +func reqWith(headers map[string]string) *http.Request { + r, _ := http.NewRequest(http.MethodPut, "/b/o", nil) + for k, v := range headers { + r.Header.Set(k, v) + } + return r +} + +// oneClause returns the single clause of cond, failing if it does not hold +// exactly one. +func oneClause(t *testing.T, cond *filer_pb.WriteCondition) *filer_pb.WriteCondition_Clause { + t.Helper() + if cond == nil { + t.Fatal("expected a condition, got nil") + } + if len(cond.Clauses) != 1 { + t.Fatalf("expected 1 clause, got %d", len(cond.Clauses)) + } + return cond.Clauses[0] +} + +func TestBuildWriteCondition(t *testing.T) { + t.Run("no headers is unconditional", func(t *testing.T) { + cond, ok := buildWriteCondition(reqWith(nil)) + if !ok || cond != nil { + t.Fatalf("want (nil, true), got (%v, %v)", cond, ok) + } + }) + t.Run("If-None-Match * to IF_NOT_EXISTS", func(t *testing.T) { + cond, ok := buildWriteCondition(reqWith(map[string]string{s3_constants.IfNoneMatch: "*"})) + if !ok { + t.Fatal("want ok") + } + if c := oneClause(t, cond); c.Kind != filer_pb.WriteCondition_IF_NOT_EXISTS { + t.Fatalf("kind = %v", c.Kind) + } + }) + t.Run("If-Match * to IF_EXISTS", func(t *testing.T) { + cond, ok := buildWriteCondition(reqWith(map[string]string{s3_constants.IfMatch: "*"})) + if !ok { + t.Fatal("want ok") + } + if c := oneClause(t, cond); c.Kind != filer_pb.WriteCondition_IF_EXISTS { + t.Fatalf("kind = %v", c.Kind) + } + }) + t.Run("If-Match strong etag to IF_ETAG_MATCH", func(t *testing.T) { + cond, ok := buildWriteCondition(reqWith(map[string]string{s3_constants.IfMatch: `"abc123"`})) + if !ok { + t.Fatal("want ok") + } + c := oneClause(t, cond) + if c.Kind != filer_pb.WriteCondition_IF_ETAG_MATCH || len(c.Etags) != 1 || c.Etags[0] != "abc123" { + t.Fatalf("clause = %+v", c) + } + }) + t.Run("If-None-Match strong etag to IF_ETAG_NOT_MATCH", func(t *testing.T) { + cond, ok := buildWriteCondition(reqWith(map[string]string{s3_constants.IfNoneMatch: `"abc123"`})) + if !ok { + t.Fatal("want ok") + } + c := oneClause(t, cond) + if c.Kind != filer_pb.WriteCondition_IF_ETAG_NOT_MATCH || len(c.Etags) != 1 || c.Etags[0] != "abc123" { + t.Fatalf("clause = %+v", c) + } + }) + t.Run("weak etag falls back", func(t *testing.T) { + if _, ok := buildWriteCondition(reqWith(map[string]string{s3_constants.IfMatch: `W/"abc"`})); ok { + t.Fatal("weak etag must not take the fast path") + } + }) + t.Run("etag list falls back", func(t *testing.T) { + if _, ok := buildWriteCondition(reqWith(map[string]string{s3_constants.IfMatch: `"a","b"`})); ok { + t.Fatal("etag list must not take the fast path") + } + }) + t.Run("both match and none-match falls back", func(t *testing.T) { + if _, ok := buildWriteCondition(reqWith(map[string]string{ + s3_constants.IfMatch: "*", + s3_constants.IfNoneMatch: "*", + })); ok { + t.Fatal("ambiguous combination must not take the fast path") + } + }) + t.Run("time-based falls back", func(t *testing.T) { + if _, ok := buildWriteCondition(reqWith(map[string]string{ + "If-Unmodified-Since": "Wed, 21 Oct 2015 07:28:00 GMT", + })); ok { + t.Fatal("time condition must not take the fast path") + } + }) +} + +func TestBuildDeleteCondition(t *testing.T) { + t.Run("no If-Match is unconditional", func(t *testing.T) { + cond, ok := buildDeleteCondition(reqWith(nil)) + if !ok || cond != nil { + t.Fatalf("want (nil, true), got (%v, %v)", cond, ok) + } + }) + t.Run("If-Match * to IF_EXISTS", func(t *testing.T) { + cond, ok := buildDeleteCondition(reqWith(map[string]string{s3_constants.IfMatch: "*"})) + if !ok { + t.Fatal("want ok") + } + if c := oneClause(t, cond); c.Kind != filer_pb.WriteCondition_IF_EXISTS { + t.Fatalf("kind = %v", c.Kind) + } + }) + t.Run("If-Match etag to IF_ETAG_MATCH", func(t *testing.T) { + cond, ok := buildDeleteCondition(reqWith(map[string]string{s3_constants.IfMatch: `"e"`})) + if !ok { + t.Fatal("want ok") + } + if c := oneClause(t, cond); c.Kind != filer_pb.WriteCondition_IF_ETAG_MATCH || c.Etags[0] != "e" { + t.Fatalf("clause = %+v", c) + } + }) + t.Run("weak etag falls back", func(t *testing.T) { + if _, ok := buildDeleteCondition(reqWith(map[string]string{s3_constants.IfMatch: `W/"e"`})); ok { + t.Fatal("weak etag must not take the fast path") + } + }) +} + +func TestSingleStrongETag(t *testing.T) { + cases := []struct { + in string + want string + single bool + }{ + {`"abc"`, "abc", true}, + {` "abc" `, "abc", true}, + {`abc`, "abc", true}, + {`W/"abc"`, "", false}, + {`w/"abc"`, "", false}, + {`"a","b"`, "", false}, + } + for _, c := range cases { + got, single := singleStrongETag(c.in) + if single != c.single || (single && got != c.want) { + t.Errorf("singleStrongETag(%q) = (%q, %v), want (%q, %v)", c.in, got, single, c.want, c.single) + } + } +} + +func TestRouteWriteCondition(t *testing.T) { + // Unconditional routes either way. + if c, ok := routeWriteCondition(reqWith(nil), false); !ok || c != nil { + t.Fatalf("overwrite unconditional: got (%v,%v)", c, ok) + } + if c, ok := routeWriteCondition(reqWith(nil), true); !ok || c != nil { + t.Fatalf("unique unconditional: got (%v,%v)", c, ok) + } + // An overwrite carries a reducible condition. + if c, ok := routeWriteCondition(reqWith(map[string]string{s3_constants.IfMatch: `"e"`}), false); !ok || c == nil { + t.Fatalf("overwrite conditional should route: got (%v,%v)", c, ok) + } + // A conditional unique (versioned) write bails to the lock path. + if _, ok := routeWriteCondition(reqWith(map[string]string{s3_constants.IfMatch: `"e"`}), true); ok { + t.Fatal("conditional unique write must not route") + } + // A non-reducible condition bails regardless. + if _, ok := routeWriteCondition(reqWith(map[string]string{s3_constants.IfMatch: `W/"e"`}), false); ok { + t.Fatal("weak etag must not route") + } +}