diff --git a/other/java/client/src/main/proto/filer.proto b/other/java/client/src/main/proto/filer.proto index 2a21f1e947..a72377096d 100644 --- a/other/java/client/src/main/proto/filer.proto +++ b/other/java/client/src/main/proto/filer.proto @@ -286,6 +286,7 @@ message WriteCondition { IF_EXTENDED_TIME_ELAPSED = 8; // fail if present and extended[ext_key] (unix seconds) is in the future IF_CHUNKS_EQUAL = 9; // fail unless the stored chunk fid multiset equals fids (absent entry = no chunks) IF_ENTRY_EQUAL = 10; // fail unless the stored entry equals expected_entry + IF_OBJECT_EXISTS = 11; // fail if absent or a plain directory, one holding no object of its own (If-Match: * on a null version) } // Clause is one primitive comparison. IF_ETAG_MATCH holds when the current // entry's ETag equals any value in etags; IF_ETAG_NOT_MATCH holds when it diff --git a/weed/pb/filer.proto b/weed/pb/filer.proto index dd1d6edb05..18486d3c7c 100644 --- a/weed/pb/filer.proto +++ b/weed/pb/filer.proto @@ -286,6 +286,7 @@ message WriteCondition { IF_EXTENDED_TIME_ELAPSED = 8; // fail if present and extended[ext_key] (unix seconds) is in the future IF_CHUNKS_EQUAL = 9; // fail unless the stored chunk fid multiset equals fids (absent entry = no chunks) IF_ENTRY_EQUAL = 10; // fail unless the stored entry equals expected_entry + IF_OBJECT_EXISTS = 11; // fail if absent or a plain directory, one holding no object of its own (If-Match: * on a null version) } // Clause is one primitive comparison. IF_ETAG_MATCH holds when the current // entry's ETag equals any value in etags; IF_ETAG_NOT_MATCH holds when it diff --git a/weed/pb/filer_pb/filer.pb.go b/weed/pb/filer_pb/filer.pb.go index e4efbf7528..6fafc60625 100644 --- a/weed/pb/filer_pb/filer.pb.go +++ b/weed/pb/filer_pb/filer.pb.go @@ -211,6 +211,7 @@ const ( WriteCondition_IF_EXTENDED_TIME_ELAPSED WriteCondition_Kind = 8 // fail if present and extended[ext_key] (unix seconds) is in the future WriteCondition_IF_CHUNKS_EQUAL WriteCondition_Kind = 9 // fail unless the stored chunk fid multiset equals fids (absent entry = no chunks) WriteCondition_IF_ENTRY_EQUAL WriteCondition_Kind = 10 // fail unless the stored entry equals expected_entry + WriteCondition_IF_OBJECT_EXISTS WriteCondition_Kind = 11 // fail if absent or a plain directory, one holding no object of its own (If-Match: * on a null version) ) // Enum value maps for WriteCondition_Kind. @@ -227,6 +228,7 @@ var ( 8: "IF_EXTENDED_TIME_ELAPSED", 9: "IF_CHUNKS_EQUAL", 10: "IF_ENTRY_EQUAL", + 11: "IF_OBJECT_EXISTS", } WriteCondition_Kind_value = map[string]int32{ "NONE": 0, @@ -240,6 +242,7 @@ var ( "IF_EXTENDED_TIME_ELAPSED": 8, "IF_CHUNKS_EQUAL": 9, "IF_ENTRY_EQUAL": 10, + "IF_OBJECT_EXISTS": 11, } ) @@ -7500,7 +7503,7 @@ const file_filer_proto_rawDesc = "" + "signatures\x12=\n" + "\x1bskip_check_parent_directory\x18\x06 \x01(\bR\x18skipCheckParentDirectory\x126\n" + "\tcondition\x18\a \x01(\v2\x18.filer_pb.WriteConditionR\tcondition\x12\x19\n" + - "\bis_moved\x18\b \x01(\bR\aisMoved\"\x88\x05\n" + + "\bis_moved\x18\b \x01(\bR\aisMoved\"\x9e\x05\n" + "\x0eWriteCondition\x129\n" + "\aclauses\x18\x01 \x03(\v2\x1f.filer_pb.WriteCondition.ClauseR\aclauses\x1a\xc9\x02\n" + "\x06Clause\x121\n" + @@ -7516,7 +7519,7 @@ const file_filer_proto_rawDesc = "" + "gate_value\x18\b \x01(\tR\tgateValue\x12\x12\n" + "\x04fids\x18\t \x03(\tR\x04fids\x126\n" + "\x0eexpected_entry\x18\n" + - " \x01(\v2\x0f.filer_pb.EntryR\rexpectedEntry\"\xee\x01\n" + + " \x01(\v2\x0f.filer_pb.EntryR\rexpectedEntry\"\x84\x02\n" + "\x04Kind\x12\b\n" + "\x04NONE\x10\x00\x12\x11\n" + "\rIF_NOT_EXISTS\x10\x01\x12\r\n" + @@ -7529,7 +7532,8 @@ const file_filer_proto_rawDesc = "" + "\x18IF_EXTENDED_TIME_ELAPSED\x10\b\x12\x13\n" + "\x0fIF_CHUNKS_EQUAL\x10\t\x12\x12\n" + "\x0eIF_ENTRY_EQUAL\x10\n" + - "\"\xad\x06\n" + + "\x12\x14\n" + + "\x10IF_OBJECT_EXISTS\x10\v\"\xad\x06\n" + "\x0eObjectMutation\x121\n" + "\x04type\x18\x01 \x01(\x0e2\x1d.filer_pb.ObjectMutation.TypeR\x04type\x12\x1c\n" + "\tdirectory\x18\x02 \x01(\tR\tdirectory\x12\x12\n" + diff --git a/weed/s3api/s3api_internal_lifecycle.go b/weed/s3api/s3api_internal_lifecycle.go index 4d267e0780..720c626f7e 100644 --- a/weed/s3api/s3api_internal_lifecycle.go +++ b/weed/s3api/s3api_internal_lifecycle.go @@ -94,7 +94,7 @@ func (s3a *S3ApiServer) lifecycleDispatch(ctx context.Context, req *s3_lifecycle if owner := s3a.routableWriteOwner(req.Bucket, req.ObjectPath); owner != "" { worm, lockErr := s3a.isObjectLockEnabled(req.Bucket) if lockErr == nil { - code, settled := s3a.routedDeleteNullVersion(owner, req.Bucket, req.ObjectPath, worm, false, metadataOnly) + code, settled := s3a.routedDeleteNullVersion(owner, req.Bucket, req.ObjectPath, worm, false, metadataOnly, nil) if settled { if code == s3err.ErrAccessDenied { return retryLater("ACCESS_DENIED: deleteNullVersion: object lock"), nil @@ -175,7 +175,7 @@ func (s3a *S3ApiServer) lifecycleDispatch(ctx context.Context, req *s3_lifecycle if lockErr == nil { // The "null" version is the object entry, not a .versions file. if req.VersionId == "null" { - code, settled := s3a.routedDeleteNullVersion(owner, req.Bucket, req.ObjectPath, worm, false, metadataOnly) + code, settled := s3a.routedDeleteNullVersion(owner, req.Bucket, req.ObjectPath, worm, false, metadataOnly, nil) if settled { if code == s3err.ErrAccessDenied { return retryLater("ACCESS_DENIED: deleteSpecificVersion: object lock"), nil @@ -186,7 +186,7 @@ func (s3a *S3ApiServer) lifecycleDispatch(ctx context.Context, req *s3_lifecycle versionDeleted = true } } else { - switch code := s3a.routedDeleteSpecificVersion(owner, req.Bucket, req.ObjectPath, req.VersionId, worm && !isDeleteMarkerEntry(entry), false, metadataOnly); code { + switch code := s3a.routedDeleteSpecificVersion(owner, req.Bucket, req.ObjectPath, req.VersionId, worm && !isDeleteMarkerEntry(entry), false, metadataOnly, nil); code { case s3err.ErrNone: versionDeleted = true case s3err.ErrAccessDenied: diff --git a/weed/s3api/s3api_object_handlers_delete.go b/weed/s3api/s3api_object_handlers_delete.go index 535ef4a884..0c6d7d68b3 100644 --- a/weed/s3api/s3api_object_handlers_delete.go +++ b/weed/s3api/s3api_object_handlers_delete.go @@ -173,7 +173,7 @@ func (s3a *S3ApiServer) deleteVersionedObject(r *http.Request, bucket, object, v if owner := s3a.routableWriteOwner(bucket, object); owner != "" { worm, lockErr := s3a.isObjectLockEnabled(bucket) if lockErr == nil { - code, settled := s3a.routedDeleteNullVersion(owner, bucket, object, worm, worm && governanceBypassAllowed, false) + code, settled := s3a.routedDeleteNullVersion(owner, bucket, object, worm, worm && governanceBypassAllowed, false, nil) if settled { if code != s3err.ErrNone { return result, code @@ -288,29 +288,28 @@ func (s3a *S3ApiServer) DeleteObjectHandler(w http.ResponseWriter, r *http.Reque // Fast path: route the delete to the owner filer under its per-path lock; // routedObjectOwner excludes versioned/object-lock buckets. if !deleteHandled && !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 - } + if owner, ownerOk := s3a.routedObjectOwner(bucket, object); ownerOk { + resp, err := s3a.routedDelete(owner, bucket, object, buildDeleteCondition(r)) + 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 } } } // Versioned/suspended delete with no specific version: route off the lock when // the bucket has an owner. createDeleteMarker routes its own pointer flip; a // delete marker never removes a locked version, so object-lock buckets route - // here too. The If-Match precondition was already checked above. - if !deleteHandled && versioningConfigured && versionId == "" { + // here too. A conditional delete stays on the lock path, which re-checks + // If-Match under the lock, so concurrent ones cannot all pass the check above. + if !deleteHandled && versioningConfigured && versionId == "" && r.Header.Get(s3_constants.IfMatch) == "" { if owner := s3a.routableWriteOwner(bucket, object); owner != "" { deleteResult, deleteCode = s3a.deleteVersionedObject(r, bucket, object, versionId, versioningState) deleteHandled = true @@ -322,10 +321,13 @@ func (s3a *S3ApiServer) DeleteObjectHandler(w http.ResponseWriter, r *http.Reque // gate the delete on the version's WORM guards, evaluated on the owner — for // governance bypass the retention guard is scoped to COMPLIANCE so the filer // allows a governance-mode delete while still denying compliance and legal - // hold, without the gateway reading the version. + // hold, without the gateway reading the version. If-Match rides the same + // transaction, so concurrent conditional deletes of one version cannot both + // pass. if !deleteHandled && versionId != "" { worm, lockErr := s3a.isObjectLockEnabled(bucket) bypass := worm && s3a.evaluateGovernanceBypassRequest(r, bucket, object) + ifMatch := buildDeleteCondition(r) if lockErr == nil { if owner := s3a.objectWriteOwner(bucket, object); owner != "" { deleteResult.versionId = versionId @@ -335,11 +337,11 @@ func (s3a *S3ApiServer) DeleteObjectHandler(w http.ResponseWriter, r *http.Reque } } if versionId == "null" { - deleteCode, deleteHandled = s3a.routedDeleteNullVersion(owner, bucket, object, worm, bypass, false) + deleteCode, deleteHandled = s3a.routedDeleteNullVersion(owner, bucket, object, worm, bypass, false, ifMatch) } else { // A version id never changes kind, so a marker read here is // the one deleted; markers carry no WORM guard. - deleteCode, deleteHandled = s3a.routedDeleteSpecificVersion(owner, bucket, object, versionId, worm && !deleteResult.deleteMarker, bypass, false), true + deleteCode, deleteHandled = s3a.routedDeleteSpecificVersion(owner, bucket, object, versionId, worm && !deleteResult.deleteMarker, bypass, false, ifMatch), true } } } diff --git a/weed/s3api/s3api_object_handlers_delete_test.go b/weed/s3api/s3api_object_handlers_delete_test.go index 6c7c956af0..0cee9fbe69 100644 --- a/weed/s3api/s3api_object_handlers_delete_test.go +++ b/weed/s3api/s3api_object_handlers_delete_test.go @@ -3,6 +3,7 @@ package s3api import ( "context" "errors" + "fmt" "net/http" "net/http/httptest" "strings" @@ -175,3 +176,101 @@ func TestVersionedDeleteMarkerCommitsOnOwner(t *testing.T) { require.Nil(t, replica.entries[markerKey], "the replica must still be behind") replica.mu.Unlock() } + +// Two If-Match deletes can both pass the check made before the object write +// lock. The one that takes the lock second must see the first one's delete and +// fail its precondition instead of reporting a second successful delete. +func TestVersionedConditionalDeleteRechecksUnderLock(t *testing.T) { + for _, state := range []string{s3_constants.VersioningEnabled, s3_constants.VersioningSuspended} { + for _, raced := range []bool{false, true} { + t.Run(fmt.Sprintf("%s/raced=%v", state, raced), func(t *testing.T) { + filerImpl, s3a := newVersionedPutFiler(t) + s3a.iam = &IdentityAccessManagement{} + filerImpl.mu.Lock() + filerImpl.entries["/buckets/b"].Extended[s3_constants.ExtVersioningKey] = []byte(state) + filerImpl.mu.Unlock() + filerAddr := s3a.option.Filers[0] + s3a.objectWriteLockClient = cluster.NewLockClient(s3a.option.GrpcDialOption, filerAddr) + s3a.objectWriteLockClient.SetRing([]pb.ServerAddress{filerAddr}, 1) + + locked := false + s3a.newObjectWriteLock = func(bucket, object string) (objectWriteLock, error) { + if raced && !locked { + other := httptest.NewRequest(http.MethodDelete, "/b/k", nil) + if _, code := s3a.deleteVersionedObject(other, bucket, object, "", state); code != s3err.ErrNone { + t.Fatalf("competing delete: %v", code) + } + } + locked = true + return noopObjectWriteLock{}, nil + } + + r := httptest.NewRequest(http.MethodDelete, "/b/k", nil) + r = mux.SetURLVars(r, map[string]string{"bucket": "b", "object": "k"}) + r.Header.Set(s3_constants.IfMatch, `"abc123"`) + w := httptest.NewRecorder() + s3a.DeleteObjectHandler(w, r) + + require.True(t, locked, "the conditional delete must take the object write lock") + want := http.StatusNoContent + if raced { + want = http.StatusPreconditionFailed + } + require.Equal(t, want, w.Code, w.Body.String()) + markers := 0 + filerImpl.mu.Lock() + for key, entry := range filerImpl.entries { + if strings.HasPrefix(key, "/buckets/b/k.versions/") && string(entry.Extended[s3_constants.ExtDeleteMarkerKey]) == "true" { + markers++ + } + } + filerImpl.mu.Unlock() + require.Equal(t, 1, markers, "exactly one delete may leave a marker") + }) + } + } +} + +// A conditional delete of a specific version carries If-Match, ETag lists +// included, into the owner's transaction. +func TestVersionIdConditionalDeleteGuardsOnOwner(t *testing.T) { + for _, tc := range []struct { + ifMatch string + etags []string + }{{`"abc123"`, []string{`"abc123"`}}, {`"other", "abc123"`, []string{`"other"`, `"abc123"`}}} { + t.Run(tc.ifMatch, func(t *testing.T) { + filerImpl, s3a := newVersionedPutFiler(t) + s3a.iam = &IdentityAccessManagement{} + filerAddr := s3a.option.Filers[0] + s3a.objectWriteLockClient = cluster.NewLockClient(s3a.option.GrpcDialOption, filerAddr) + s3a.objectWriteLockClient.SetRing([]pb.ServerAddress{filerAddr}, 1) + s3a.newObjectWriteLock = func(bucket, object string) (objectWriteLock, error) { + return nil, errors.New("a conditional version delete must not take the gateway lock") + } + versionId := latestVersionId(filerImpl) + versionKey := "/buckets/b/k.versions/v_" + versionId + + var txns []*filer_pb.ObjectTransactionRequest + filerImpl.mu.Lock() + filerImpl.txnCheck = func(req *filer_pb.ObjectTransactionRequest) { txns = append(txns, req) } + filerImpl.mu.Unlock() + + r := httptest.NewRequest(http.MethodDelete, "/b/k?versionId="+versionId, nil) + r = mux.SetURLVars(r, map[string]string{"bucket": "b", "object": "k"}) + r.Header.Set(s3_constants.IfMatch, tc.ifMatch) + w := httptest.NewRecorder() + s3a.DeleteObjectHandler(w, r) + + require.Equal(t, http.StatusNoContent, w.Code, w.Body.String()) + filerImpl.mu.Lock() + defer filerImpl.mu.Unlock() + require.Nil(t, filerImpl.entries[versionKey], "the version must be deleted") + require.Len(t, txns, 1) + guard := txns[0].Mutations[0] + require.Equal(t, versionKey, guard.ConditionKey) + require.Len(t, guard.GetCondition().GetClauses(), 1) + require.Equal(t, filer_pb.WriteCondition_IF_ETAG_MATCH, guard.GetCondition().GetClauses()[0].Kind) + require.Equal(t, tc.etags, guard.GetCondition().GetClauses()[0].Etags) + }) + } +} diff --git a/weed/s3api/s3api_object_routed_write.go b/weed/s3api/s3api_object_routed_write.go index 4b89d01239..c90e313dcc 100644 --- a/weed/s3api/s3api_object_routed_write.go +++ b/weed/s3api/s3api_object_routed_write.go @@ -113,20 +113,18 @@ func buildWriteCondition(r *http.Request) (*filer_pb.WriteCondition, bool) { return &filer_pb.WriteCondition{Clauses: clauses}, true } -// 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) { +// buildDeleteCondition reduces a DeleteObject's If-Match header to a condition, +// nil when unconditional; DeleteObject honors only If-Match, matching +// checkDeleteIfMatch, whose comparison the filer applies to every listed ETag. +func buildDeleteCondition(r *http.Request) *filer_pb.WriteCondition { ifMatch := strings.TrimSpace(r.Header.Get(s3_constants.IfMatch)) switch { case ifMatch == "": - return nil, true + return nil case ifMatch == "*": - return clause(filer_pb.WriteCondition_IF_EXISTS), true + return clause(filer_pb.WriteCondition_IF_EXISTS) default: - if etag, single := singleStrongETag(ifMatch); single { - return etagClause(filer_pb.WriteCondition_IF_ETAG_MATCH, etag), true - } - return nil, false + return &filer_pb.WriteCondition{Clauses: []*filer_pb.WriteCondition_Clause{{Kind: filer_pb.WriteCondition_IF_ETAG_MATCH, Etags: etagList(ifMatch)}}} } } @@ -134,23 +132,6 @@ 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 -} - // etagList splits a conditional ETag header into its candidates, each kept // verbatim — the evaluator's canonicalization handles quotes and weak markers. func etagList(v string) []string { diff --git a/weed/s3api/s3api_object_routed_write_test.go b/weed/s3api/s3api_object_routed_write_test.go index 4986fcaac0..e2ea779f8b 100644 --- a/weed/s3api/s3api_object_routed_write_test.go +++ b/weed/s3api/s3api_object_routed_write_test.go @@ -9,6 +9,7 @@ import ( "testing" "time" + "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" @@ -18,6 +19,7 @@ import ( "google.golang.org/grpc/codes" "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/status" + "google.golang.org/protobuf/proto" ) func reqWith(headers map[string]string) *http.Request { @@ -252,57 +254,28 @@ func TestValidateConditionalCopyHeadersAcceptsHTTPDateFormats(t *testing.T) { 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) + if cond := buildDeleteCondition(reqWith(nil)); cond != nil { + t.Fatalf("want nil, got %v", cond) } }) 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 { + if c := oneClause(t, buildDeleteCondition(reqWith(map[string]string{s3_constants.IfMatch: "*"}))); 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" { + if c := oneClause(t, buildDeleteCondition(reqWith(map[string]string{s3_constants.IfMatch: `"e"`}))); c.Kind != filer_pb.WriteCondition_IF_ETAG_MATCH || len(c.Etags) != 1 || 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") + t.Run("an ETag list carries every candidate", func(t *testing.T) { + c := oneClause(t, buildDeleteCondition(reqWith(map[string]string{s3_constants.IfMatch: `"a", W/"b"`}))) + if c.Kind != filer_pb.WriteCondition_IF_ETAG_MATCH || len(c.Etags) != 2 || c.Etags[0] != `"a"` || c.Etags[1] != `W/"b"` { + t.Fatalf("clause = %+v", c) } }) } -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) - } - } -} - // fakeTxnFiler is a minimal SeaweedFiler gRPC server that records ObjectTransaction // calls, standing in for a live filer that a routed write fails over to. type fakeTxnFiler struct { @@ -367,7 +340,7 @@ func TestRoutedDeleteSpecificVersionUsesWormCondition(t *testing.T) { }, } - if code := s3a.routedDeleteSpecificVersion(owner, "b", "obj", versionId, true, false, false); code != s3err.ErrAccessDenied { + if code := s3a.routedDeleteSpecificVersion(owner, "b", "obj", versionId, true, false, false, nil); code != s3err.ErrAccessDenied { t.Fatalf("locked version delete returned %v, want %v", code, s3err.ErrAccessDenied) } @@ -438,7 +411,7 @@ func TestRoutedDeleteSpecificVersionGuardsRepoint(t *testing.T) { t.Run("one transaction behind the version's lock checks", func(t *testing.T) { filer := &fakeTxnFiler{} owner := startFakeFiler(t, filer) - if code := newS3a().routedDeleteSpecificVersion(owner, "b", "obj", versionId, false, false, false); code != s3err.ErrNone { + if code := newS3a().routedDeleteSpecificVersion(owner, "b", "obj", versionId, false, false, false, nil); code != s3err.ErrNone { t.Fatalf("delete returned %v, want %v", code, s3err.ErrNone) } if len(filer.reqs) != 1 { @@ -459,7 +432,7 @@ func TestRoutedDeleteSpecificVersionGuardsRepoint(t *testing.T) { t.Run("metadata-only delete keeps the data", func(t *testing.T) { filer := &fakeTxnFiler{} owner := startFakeFiler(t, filer) - if code := newS3a().routedDeleteSpecificVersion(owner, "b", "obj", versionId, false, false, true); code != s3err.ErrNone { + if code := newS3a().routedDeleteSpecificVersion(owner, "b", "obj", versionId, false, false, true, nil); code != s3err.ErrNone { t.Fatalf("delete returned %v, want %v", code, s3err.ErrNone) } if filer.lastReq.Mutations[1].IsDeleteData { @@ -740,3 +713,234 @@ func TestRouteWriteCondition(t *testing.T) { t.Fatal("unparseable headers must not route") } } + +// A conditional version delete guards the version inside its own transaction, +// ahead of the repoint, and reports that guard failing as a failed +// precondition rather than an object-lock denial. +func TestRoutedVersionDeleteCarriesIfMatch(t *testing.T) { + versionId := "672a75d526ef29c79fe6b5680cad4a0d" + ifMatch := &filer_pb.WriteCondition{Clauses: []*filer_pb.WriteCondition_Clause{{Kind: filer_pb.WriteCondition_IF_ETAG_MATCH, Etags: []string{"abc"}}}} + newS3a := func() *S3ApiServer { + return &S3ApiServer{option: &S3ApiServerOption{ + BucketsPath: "/buckets", + GrpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + }} + } + guarded := func(t *testing.T, m *filer_pb.ObjectMutation, conditionKey string) { + t.Helper() + clauses := m.GetCondition().GetClauses() + if len(clauses) != 1 || clauses[0].Kind != filer_pb.WriteCondition_IF_ETAG_MATCH || len(clauses[0].Etags) != 1 || clauses[0].Etags[0] != "abc" { + t.Fatalf("mutation condition = %+v, want If-Match abc", m.GetCondition()) + } + if m.ConditionKey != conditionKey { + t.Fatalf("mutation condition key = %q, want %q", m.ConditionKey, conditionKey) + } + } + firstMutation := int32(0) + cases := []struct { + name string + resp *filer_pb.ObjectTransactionResponse + want s3err.ErrorCode + }{ + {"applied", nil, s3err.ErrNone}, + {"if-match failed", &filer_pb.ObjectTransactionResponse{ErrorCode: filer_pb.FilerError_PRECONDITION_FAILED, FailedMutation: &firstMutation}, s3err.ErrPreconditionFailed}, + {"object lock denied", &filer_pb.ObjectTransactionResponse{ErrorCode: filer_pb.FilerError_PRECONDITION_FAILED}, s3err.ErrAccessDenied}, + } + for _, tc := range cases { + t.Run("version/"+tc.name, func(t *testing.T) { + filer := &fakeTxnFiler{resp: tc.resp} + owner := startFakeFiler(t, filer) + if code := newS3a().routedDeleteSpecificVersion(owner, "b", "obj", versionId, true, false, false, ifMatch); code != tc.want { + t.Fatalf("delete returned %v, want %v", code, tc.want) + } + if filer.lastReq.Mutations[0].Type != filer_pb.ObjectMutation_RECOMPUTE_LATEST { + t.Fatalf("first mutation = %v, want the repoint", filer.lastReq.Mutations[0].Type) + } + guarded(t, filer.lastReq.Mutations[0], "/buckets/b/obj.versions/v_"+versionId) + }) + t.Run("null/"+tc.name, func(t *testing.T) { + filer := &fakeTxnFiler{resp: tc.resp} + owner := startFakeFiler(t, filer) + code, settled := newS3a().routedDeleteNullVersion(owner, "b", "obj", true, false, false, ifMatch) + if !settled || code != tc.want { + t.Fatalf("delete returned (%v, %v), want (%v, true)", code, settled, tc.want) + } + guarded(t, filer.lastReq.Mutations[0], "") + }) + } + // The owner may have committed before the error, so only an unconditional + // null delete may retry on the lock path. + t.Run("null/transport error", func(t *testing.T) { + filer := &fakeTxnFiler{err: status.Error(codes.Unavailable, "response lost")} + owner := startFakeFiler(t, filer) + if code, settled := newS3a().routedDeleteNullVersion(owner, "b", "obj", false, false, false, ifMatch); !settled || code != s3err.ErrInternalError { + t.Fatalf("conditional delete returned (%v, %v), want (%v, true)", code, settled, s3err.ErrInternalError) + } + if code, settled := newS3a().routedDeleteNullVersion(owner, "b", "obj", false, false, false, nil); settled || code != s3err.ErrNone { + t.Fatalf("unconditional delete returned (%v, %v), want a lock-path fallback", code, settled) + } + }) + // After an owner failure other than the nested-entry refusal, a + // conditional null delete must not re-check on the lock path. + t.Run("null/owner error", func(t *testing.T) { + for _, msg := range []string{ + "mutation 0: delete file /buckets/b/obj: store unavailable", + "mutation 0: delete file /buckets/b/" + filer.MsgFailDelNonEmptyFolder + ": store unavailable", + } { + fake := &fakeTxnFiler{resp: &filer_pb.ObjectTransactionResponse{Error: msg, FailedMutation: &firstMutation}} + owner := startFakeFiler(t, fake) + if code, settled := newS3a().routedDeleteNullVersion(owner, "b", "obj", false, false, false, ifMatch); !settled || code != s3err.ErrInternalError { + t.Fatalf("%q: conditional delete returned (%v, %v), want (%v, true)", msg, code, settled, s3err.ErrInternalError) + } + if code, settled := newS3a().routedDeleteNullVersion(owner, "b", "obj", false, false, false, nil); settled || code != s3err.ErrNone { + t.Fatalf("%q: unconditional delete returned (%v, %v), want a lock-path fallback", msg, code, settled) + } + } + }) + // An entry other keys are nested under is stripped on the owner under the + // same guard plus the entry it was read as, so a routed delete that runs + // once the last nested key is gone cannot pass the check as well. + t.Run("null/nested entry", func(t *testing.T) { + object := &filer_pb.Entry{ + Name: "obj", + IsDirectory: true, + Attributes: &filer_pb.FuseAttributes{Mime: "text/plain", FileSize: 3, Md5: []byte{0xab}}, + Chunks: []*filer_pb.FileChunk{{FileId: "1,ab", Size: 3}}, + Extended: map[string][]byte{s3_constants.ExtETagKey: []byte(`"abc"`), "xattr-user.k": []byte("v")}, + } + for _, tc := range []struct { + name string + entry *filer_pb.Entry + demote *filer_pb.ObjectTransactionResponse + want s3err.ErrorCode + txns int + }{ + {"demoted", object, nil, s3err.ErrNone, 2}, + {"guard failed", object, &filer_pb.ObjectTransactionResponse{ErrorCode: filer_pb.FilerError_PRECONDITION_FAILED, FailedMutation: &firstMutation}, s3err.ErrPreconditionFailed, 2}, + {"object lock denied", object, &filer_pb.ObjectTransactionResponse{ErrorCode: filer_pb.FilerError_PRECONDITION_FAILED}, s3err.ErrAccessDenied, 2}, + {"deleted by a routed delete", nil, nil, s3err.ErrPreconditionFailed, 1}, + {"already demoted", &filer_pb.Entry{Name: "obj", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}}, nil, s3err.ErrPreconditionFailed, 1}, + } { + fake := &fakeNestedFiler{entry: tc.entry, demote: tc.demote} + owner := startFakeFiler(t, fake) + if code, settled := newS3a().routedDeleteNullVersion(owner, "b", "obj", true, false, false, ifMatch); !settled || code != tc.want { + t.Fatalf("%s: delete returned (%v, %v), want (%v, true)", tc.name, code, settled, tc.want) + } + if len(fake.reqs) != tc.txns { + t.Fatalf("%s: %d transactions, want %d", tc.name, len(fake.reqs), tc.txns) + } + if tc.txns < 2 { + continue + } + req := fake.reqs[1] + if len(req.GetCondition().GetClauses()) == 0 || req.LockKey != "/buckets/b/obj" { + t.Fatalf("%s: demote request = %+v, want the object lock guards on its key", tc.name, req) + } + m := req.Mutations[0] + if m.Type != filer_pb.ObjectMutation_PUT || m.Directory != "/buckets/b" || m.Entry.GetName() != "obj" || !m.Entry.IsDirectory { + t.Fatalf("%s: demote mutation = %+v", tc.name, m) + } + if m.Entry.IsDirectoryKeyObject() || len(m.Entry.Attributes.Md5) > 0 || m.Entry.Extended[s3_constants.ExtETagKey] != nil || string(m.Entry.Extended["xattr-user.k"]) != "v" { + t.Fatalf("%s: demoted entry %+v still holds the object or lost the directory xattr", tc.name, m.Entry) + } + clauses := m.GetCondition().GetClauses() + if len(clauses) != 2 || clauses[0].Kind != filer_pb.WriteCondition_IF_ETAG_MATCH || + clauses[1].Kind != filer_pb.WriteCondition_IF_ENTRY_EQUAL || !proto.Equal(clauses[1].ExpectedEntry, tc.entry) { + t.Fatalf("%s: demote condition = %+v, want If-Match and the entry it was read as", tc.name, m.GetCondition()) + } + } + fake := &fakeNestedFiler{entry: object} + owner := startFakeFiler(t, fake) + if code, settled := newS3a().routedDeleteNullVersion(owner, "b", "obj", false, false, false, nil); settled || code != s3err.ErrNone || len(fake.reqs) != 1 { + t.Fatalf("unconditional delete returned (%v, %v) after %d transactions, want a lock-path fallback", code, settled, len(fake.reqs)) + } + }) + // A stripped nested entry stays behind as a plain directory, which + // IF_EXISTS still matches, so If-Match: * asks the owner for an object. + t.Run("null/If-Match *", func(t *testing.T) { + star := &filer_pb.WriteCondition{Clauses: []*filer_pb.WriteCondition_Clause{{Kind: filer_pb.WriteCondition_IF_EXISTS}}} + objectKind := func(t *testing.T, m *filer_pb.ObjectMutation) { + t.Helper() + if c := m.GetCondition().GetClauses(); len(c) == 0 || c[0].Kind != filer_pb.WriteCondition_IF_OBJECT_EXISTS { + t.Fatalf("mutation condition = %+v, want IF_OBJECT_EXISTS first", m.GetCondition()) + } + } + del := &fakeTxnFiler{} + if code, settled := newS3a().routedDeleteNullVersion(startFakeFiler(t, del), "b", "obj", false, false, false, star); !settled || code != s3err.ErrNone { + t.Fatalf("delete returned (%v, %v), want (%v, true)", code, settled, s3err.ErrNone) + } + objectKind(t, del.lastReq.Mutations[0]) + nested := &fakeNestedFiler{entry: &filer_pb.Entry{Name: "obj", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{Mime: "text/plain"}}} + if code, settled := newS3a().routedDeleteNullVersion(startFakeFiler(t, nested), "b", "obj", false, false, false, star); !settled || code != s3err.ErrNone || len(nested.reqs) != 2 { + t.Fatalf("nested delete returned (%v, %v) after %d transactions, want a demote", code, settled, len(nested.reqs)) + } + objectKind(t, nested.reqs[0].Mutations[0]) + if star.Clauses[0].Kind != filer_pb.WriteCondition_IF_EXISTS { + t.Fatalf("caller's condition changed to %v", star.Clauses[0].Kind) + } + // A filer that predates IF_OBJECT_EXISTS fails it closed; the delete is + // then settled on the entry as read, under IF_EXISTS and that entry. + guardFailed := &filer_pb.ObjectTransactionResponse{ErrorCode: filer_pb.FilerError_PRECONDITION_FAILED, FailedMutation: &firstMutation} + file := &filer_pb.Entry{Name: "obj", Attributes: &filer_pb.FuseAttributes{FileSize: 3}} + for _, tc := range []struct { + name string + entry *filer_pb.Entry + want s3err.ErrorCode + kind filer_pb.ObjectMutation_Type + }{ + {"file", file, s3err.ErrNone, filer_pb.ObjectMutation_DELETE}, + {"nested object", nested.entry, s3err.ErrNone, filer_pb.ObjectMutation_PUT}, + {"plain directory", &filer_pb.Entry{Name: "obj", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}}, s3err.ErrPreconditionFailed, 0}, + {"absent", nil, s3err.ErrPreconditionFailed, 0}, + } { + old := &fakeNestedFiler{entry: tc.entry, first: guardFailed} + if code, settled := newS3a().routedDeleteNullVersion(startFakeFiler(t, old), "b", "obj", false, false, false, star); !settled || code != tc.want { + t.Fatalf("older filer, %s: delete returned (%v, %v), want (%v, true)", tc.name, code, settled, tc.want) + } + if tc.want != s3err.ErrNone { + if len(old.reqs) != 1 { + t.Fatalf("older filer, %s: %d transactions, want 1", tc.name, len(old.reqs)) + } + continue + } + m := old.reqs[1].Mutations[0] + if c := m.GetCondition().GetClauses(); m.Type != tc.kind || len(c) != 2 || c[0].Kind != filer_pb.WriteCondition_IF_EXISTS || + c[1].Kind != filer_pb.WriteCondition_IF_ENTRY_EQUAL || !proto.Equal(c[1].ExpectedEntry, tc.entry) { + t.Fatalf("older filer, %s: retry = %+v, want a %v under IF_EXISTS and the entry read", tc.name, m, tc.kind) + } + } + }) +} + +// fakeNestedFiler answers the first transaction with first, else refuses its +// delete because other keys are nested under the entry, serves entry to +// lookups, and answers later transactions with demote. +type fakeNestedFiler struct { + filer_pb.UnimplementedSeaweedFilerServer + entry *filer_pb.Entry + first *filer_pb.ObjectTransactionResponse + demote *filer_pb.ObjectTransactionResponse + reqs []*filer_pb.ObjectTransactionRequest +} + +func (f *fakeNestedFiler) ObjectTransaction(ctx context.Context, req *filer_pb.ObjectTransactionRequest) (*filer_pb.ObjectTransactionResponse, error) { + f.reqs = append(f.reqs, req) + if len(f.reqs) == 1 && f.first != nil { + return f.first, nil + } + if len(f.reqs) == 1 { + failed := int32(0) + return &filer_pb.ObjectTransactionResponse{Error: fmt.Sprintf("mutation 0: %v", fmt.Errorf("%w: %s", filer.ErrNonEmptyFolder, req.LockKey)), FailedMutation: &failed}, nil + } + if f.demote != nil { + return f.demote, nil + } + return &filer_pb.ObjectTransactionResponse{}, nil +} + +func (f *fakeNestedFiler) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) { + if f.entry == nil { + return nil, status.Error(codes.NotFound, filer_pb.ErrNotFound.Error()) + } + return &filer_pb.LookupDirectoryEntryResponse{Entry: f.entry}, nil +} diff --git a/weed/s3api/s3api_object_versioned_finalize.go b/weed/s3api/s3api_object_versioned_finalize.go index 5104b9610f..a687ed4a64 100644 --- a/weed/s3api/s3api_object_versioned_finalize.go +++ b/weed/s3api/s3api_object_versioned_finalize.go @@ -1,17 +1,22 @@ package s3api import ( + "context" "errors" "fmt" + "slices" "strconv" + "strings" "time" + "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/glog" "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" + "google.golang.org/protobuf/proto" ) // objectWriteOwner resolves the filer that owns all of an object's writes, @@ -205,8 +210,9 @@ func wormDeleteCondition(worm, bypass bool) *filer_pb.WriteCondition { // pointer still names the deleted version — a noncurrent delete must leave it // alone, since it can point at an older version than a re-derive would pick, or // at nothing, as when a suspended bucket's null write cleared it while the null -// object stays current. -func (s3a *S3ApiServer) routedDeleteSpecificVersion(owner pb.ServerAddress, bucket, object, versionId string, worm, bypass, metadataOnly bool) s3err.ErrorCode { +// object stays current. ifMatch, when set, guards the version under the same +// lock and fails before anything applies. +func (s3a *S3ApiServer) routedDeleteSpecificVersion(owner pb.ServerAddress, bucket, object, versionId string, worm, bypass, metadataOnly bool, ifMatch *filer_pb.WriteCondition) s3err.ErrorCode { if !isValidVersionID(versionId) { return s3err.ErrInvalidRequest } @@ -214,6 +220,9 @@ func (s3a *S3ApiServer) routedDeleteSpecificVersion(owner pb.ServerAddress, buck recompute := s3a.latestPointerRecompute(bucket, object, isNewFormatVersionId(versionId), versionFileName, false) recompute.GetRecompute().OnlyIfCurrentName = versionFileName versionsPath := s3a.toFilerPath(bucket, object+s3_constants.VersionsFolder) + if ifMatch != nil { + recompute.Condition, recompute.ConditionKey = ifMatch, versionsPath+"/"+versionFileName + } cond := wormDeleteCondition(worm, bypass) req := &filer_pb.ObjectTransactionRequest{ LockKey: s3a.toFilerPath(bucket, object), @@ -231,6 +240,9 @@ func (s3a *S3ApiServer) routedDeleteSpecificVersion(owner pb.ServerAddress, buck glog.Errorf("routedDeleteSpecificVersion: %s/%s %s on %s: %v", bucket, object, versionId, owner, err) return s3err.ErrInternalError case resp.ErrorCode == filer_pb.FilerError_PRECONDITION_FAILED: + if resp.FailedMutation != nil { + return s3err.ErrPreconditionFailed + } // Legal hold or retention in force on the version. return s3err.ErrAccessDenied case resp.Error != "": @@ -247,23 +259,41 @@ func (s3a *S3ApiServer) routedDeleteSpecificVersion(owner pb.ServerAddress, buck // (condition defaults to lock_key). The second return reports whether the delete // was settled here: the raw delete cannot remove an entry other keys are nested // under, which the lock path handles by stripping the object off it instead. -func (s3a *S3ApiServer) routedDeleteNullVersion(owner pb.ServerAddress, bucket, object string, worm, bypass, metadataOnly bool) (s3err.ErrorCode, bool) { +// ifMatch, when set, guards the delete itself and keeps the whole decision on +// the owner: the lock path does not exclude routed deletes, so a nested entry +// is stripped there under the same guard, and any other failure settles as +// one, since the transaction may have committed. +func (s3a *S3ApiServer) routedDeleteNullVersion(owner pb.ServerAddress, bucket, object string, worm, bypass, metadataOnly bool, ifMatch *filer_pb.WriteCondition) (s3err.ErrorCode, bool) { fullpath := util.NewFullPath(s3a.bucketDir(bucket), object) dir, name := fullpath.DirAndName() + guard, star := nullVersionCondition(ifMatch) resp, err := s3a.objectTxnOnFiler(owner, &filer_pb.ObjectTransactionRequest{ LockKey: string(fullpath), RouteKey: s3a.objectRouteKey(bucket, object), Condition: wormDeleteCondition(worm, bypass), Mutations: []*filer_pb.ObjectMutation{ - {Type: filer_pb.ObjectMutation_DELETE, Directory: dir, Name: name, IsDeleteData: !metadataOnly}, + {Type: filer_pb.ObjectMutation_DELETE, Directory: dir, Name: name, IsDeleteData: !metadataOnly, Condition: guard}, }, }) switch { + case err != nil && ifMatch != nil: + glog.Errorf("routedDeleteNullVersion: %s/%s on %s: %v", bucket, object, owner, err) + return s3err.ErrInternalError, true case err != nil: glog.Warningf("routedDeleteNullVersion: %s/%s on %s, falling back to lock: %v", bucket, object, owner, err) return s3err.ErrNone, false - case resp.ErrorCode == filer_pb.FilerError_PRECONDITION_FAILED: + case resp.ErrorCode == filer_pb.FilerError_PRECONDITION_FAILED && resp.FailedMutation == nil: return s3err.ErrAccessDenied, true + case resp.ErrorCode == filer_pb.FilerError_PRECONDITION_FAILED && star: + // A filer that predates IF_OBJECT_EXISTS fails it closed. + return s3a.routedDemoteNullVersion(owner, bucket, object, worm, bypass, metadataOnly, ifMatch), true + case resp.ErrorCode == filer_pb.FilerError_PRECONDITION_FAILED: + return s3err.ErrPreconditionFailed, true + case resp.Error != "" && ifMatch != nil && ownerRefusedNonEmpty(resp): + return s3a.routedDemoteNullVersion(owner, bucket, object, worm, bypass, metadataOnly, ifMatch), true + case resp.Error != "" && ifMatch != nil: + glog.Errorf("routedDeleteNullVersion: %s/%s on %s returned %q", bucket, object, owner, resp.Error) + return s3err.ErrInternalError, true case resp.Error != "": glog.Warningf("routedDeleteNullVersion: %s/%s returned %q, falling back to lock", bucket, object, resp.Error) return s3err.ErrNone, false @@ -272,6 +302,80 @@ func (s3a *S3ApiServer) routedDeleteNullVersion(owner pb.ServerAddress, bucket, } } +// routedDemoteNullVersion removes the null version from the entry as read on +// the owner: a directory, which other keys may be nested under, is stripped as +// the lock path's demote does, and a file is deleted. The write is guarded by +// ifMatch and by the entry it was derived from, so once the last nested key is +// gone a concurrent routed delete cannot pass the same check. +func (s3a *S3ApiServer) routedDemoteNullVersion(owner pb.ServerAddress, bucket, object string, worm, bypass, metadataOnly bool, ifMatch *filer_pb.WriteCondition) s3err.ErrorCode { + fullpath := util.NewFullPath(s3a.bucketDir(bucket), object) + dir, name := fullpath.DirAndName() + entry, err := s3a.lookupEntryOnFiler(context.Background(), owner, dir, name) + switch { + case errors.Is(err, filer_pb.ErrNotFound): + return s3err.ErrPreconditionFailed + case err != nil: + glog.Errorf("routedDemoteNullVersion: %s/%s on %s: %v", bucket, object, owner, err) + return s3err.ErrInternalError + case entry.IsDirectory && !entry.IsDirectoryKeyObject(): + return s3err.ErrPreconditionFailed + } + guard := &filer_pb.WriteCondition{Clauses: append(slices.Clone(ifMatch.GetClauses()), + &filer_pb.WriteCondition_Clause{Kind: filer_pb.WriteCondition_IF_ENTRY_EQUAL, ExpectedEntry: proto.Clone(entry).(*filer_pb.Entry)})} + mutation := &filer_pb.ObjectMutation{Type: filer_pb.ObjectMutation_DELETE, Directory: dir, Name: name, IsDeleteData: !metadataOnly, Condition: guard} + if entry.IsDirectory { + clearDirectoryMarkerMetadata(entry) + mutation = &filer_pb.ObjectMutation{Type: filer_pb.ObjectMutation_PUT, Directory: dir, Entry: entry, Condition: guard} + } + resp, err := s3a.objectTxnOnFiler(owner, &filer_pb.ObjectTransactionRequest{ + LockKey: string(fullpath), + RouteKey: s3a.objectRouteKey(bucket, object), + Condition: wormDeleteCondition(worm, bypass), + Mutations: []*filer_pb.ObjectMutation{mutation}, + }) + switch { + case err != nil: + glog.Errorf("routedDemoteNullVersion: %s/%s on %s: %v", bucket, object, owner, err) + return s3err.ErrInternalError + case resp.ErrorCode == filer_pb.FilerError_PRECONDITION_FAILED: + if resp.FailedMutation != nil { + return s3err.ErrPreconditionFailed + } + return s3err.ErrAccessDenied + case resp.Error != "": + glog.Errorf("routedDemoteNullVersion: %s/%s on %s returned %q", bucket, object, owner, resp.Error) + return s3err.ErrInternalError + default: + return s3err.ErrNone + } +} + +// nullVersionCondition narrows If-Match: * to an entry that carries an +// object, reporting whether it did: a nested entry stripped of its null +// version stays behind as a plain directory, which IF_EXISTS still matches. +func nullVersionCondition(ifMatch *filer_pb.WriteCondition) (*filer_pb.WriteCondition, bool) { + if ifMatch == nil { + return nil, false + } + cond, narrowed := proto.Clone(ifMatch).(*filer_pb.WriteCondition), false + for _, c := range cond.Clauses { + if c.Kind == filer_pb.WriteCondition_IF_EXISTS { + c.Kind, narrowed = filer_pb.WriteCondition_IF_OBJECT_EXISTS, true + } + } + return cond, narrowed +} + +// ownerRefusedNonEmpty reports a transaction whose delete mutation the owner +// refused because other keys are nested under the entry, before applying it. +func ownerRefusedNonEmpty(resp *filer_pb.ObjectTransactionResponse) bool { + if resp.FailedMutation == nil { + return false + } + msg := strings.TrimPrefix(resp.Error, fmt.Sprintf("mutation %d: ", *resp.FailedMutation)) + return errors.Is(filer.DeleteEntryError(msg), filer.ErrNonEmptyFolder) +} + // versionedFinalize flips the .versions latest pointer for a versioned PutObject: // on the routed path RECOMPUTE_LATEST rides in the version file's PUT transaction, // committing atomically under the object's per-path lock; off the ring diff --git a/weed/server/filer_grpc_server_condition.go b/weed/server/filer_grpc_server_condition.go index 0eaca1b316..31442b18fb 100644 --- a/weed/server/filer_grpc_server_condition.go +++ b/weed/server/filer_grpc_server_condition.go @@ -43,6 +43,8 @@ func clauseSatisfied(c *filer_pb.WriteCondition_Clause, current *filer.Entry) bo return !exists case filer_pb.WriteCondition_IF_EXISTS: return exists + case filer_pb.WriteCondition_IF_OBJECT_EXISTS: + return exists && (!current.IsDirectory() || current.ToProtoEntry().IsDirectoryKeyObject()) case filer_pb.WriteCondition_IF_ETAG_MATCH: return exists && etagInSet(storedEntryETag(current), c.Etags, c.AllowWeak) case filer_pb.WriteCondition_IF_ETAG_NOT_MATCH: diff --git a/weed/server/filer_grpc_server_condition_test.go b/weed/server/filer_grpc_server_condition_test.go index 3673c76b55..548f153e93 100644 --- a/weed/server/filer_grpc_server_condition_test.go +++ b/weed/server/filer_grpc_server_condition_test.go @@ -2,6 +2,7 @@ package weed_server import ( "context" + "os" "strconv" "testing" "time" @@ -31,6 +32,8 @@ func one(c *filer_pb.WriteCondition_Clause) *filer_pb.WriteCondition { func TestWriteConditionSatisfied(t *testing.T) { base := time.Unix(1700000000, 0) present := entryWithETag("abc", base) + plainDir := &filer.Entry{FullPath: "/test/obj", Attr: filer.Attr{Mode: os.ModeDir | 0755}} + objectDir := &filer.Entry{FullPath: "/test/obj", Attr: filer.Attr{Mode: os.ModeDir | 0755, Mime: "text/plain"}} cases := []struct { name string @@ -43,6 +46,11 @@ func TestWriteConditionSatisfied(t *testing.T) { {"ifnotexists-present", one(&filer_pb.WriteCondition_Clause{Kind: filer_pb.WriteCondition_IF_NOT_EXISTS}), present, false}, {"ifexists-absent", one(&filer_pb.WriteCondition_Clause{Kind: filer_pb.WriteCondition_IF_EXISTS}), nil, false}, {"ifexists-present", one(&filer_pb.WriteCondition_Clause{Kind: filer_pb.WriteCondition_IF_EXISTS}), present, true}, + {"ifexists-plain-dir", one(&filer_pb.WriteCondition_Clause{Kind: filer_pb.WriteCondition_IF_EXISTS}), plainDir, true}, + {"ifobjectexists-absent", one(&filer_pb.WriteCondition_Clause{Kind: filer_pb.WriteCondition_IF_OBJECT_EXISTS}), nil, false}, + {"ifobjectexists-present", one(&filer_pb.WriteCondition_Clause{Kind: filer_pb.WriteCondition_IF_OBJECT_EXISTS}), present, true}, + {"ifobjectexists-plain-dir", one(&filer_pb.WriteCondition_Clause{Kind: filer_pb.WriteCondition_IF_OBJECT_EXISTS}), plainDir, false}, + {"ifobjectexists-object-dir", one(&filer_pb.WriteCondition_Clause{Kind: filer_pb.WriteCondition_IF_OBJECT_EXISTS}), objectDir, true}, {"etagmatch-hit", one(&filer_pb.WriteCondition_Clause{Kind: filer_pb.WriteCondition_IF_ETAG_MATCH, Etags: []string{`"abc"`}}), present, true}, {"etagmatch-miss", one(&filer_pb.WriteCondition_Clause{Kind: filer_pb.WriteCondition_IF_ETAG_MATCH, Etags: []string{`"zzz"`}}), present, false}, {"etagmatch-absent", one(&filer_pb.WriteCondition_Clause{Kind: filer_pb.WriteCondition_IF_ETAG_MATCH, Etags: []string{`"abc"`}}), nil, false},