s3: make conditional DeleteObject atomic on versioned buckets (#12624)

* s3: re-check a versioned DeleteObject If-Match under the object lock

A versioned or suspended DeleteObject without a versionId routed to the
owner filer after checking If-Match outside any lock, so concurrent
conditional deletes could all pass the check and each add a delete
marker. Keep conditional deletes on the lock path, which re-checks the
precondition under the object write lock, as conditional versioned PUTs
already do.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* s3: guard a conditional version delete on the owner filer

DeleteObject with a versionId checked If-Match before routing the
delete to the owner filer, so concurrent conditional deletes of one
version could all report success. Carry the If-Match clause into the
owner's transaction as a guard on the version (the null entry for
versionId=null), evaluated under the object's path lock before anything
applies, and map that guard failing to 412 instead of the object-lock
403. An If-Match the filer cannot evaluate stays on the lock path.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* s3: carry an If-Match ETag list into the routed delete

A DeleteObject whose If-Match listed several ETags fell back to the
gateway's object write lock, while a single-ETag delete of the same key
ran under the owner filer's path lock. The two locks do not exclude each
other, so a list delete and a single-ETag delete could both report
success. Send every candidate to the filer, whose strong comparison
matches checkDeleteIfMatch, so all conditional deletes with an owner are
decided under the owner's lock.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* s3: fail a conditional null delete on an owner transport error

The owner may have committed the delete before the error, and the gateway
lock path it fell back to does not exclude routed deletes, so a second
If-Match delete could pass there too. Answer 500 as the specific-version
path does; unconditional deletes still fall back.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* s3: fail a conditional null delete on any owner error but the nested refusal

Only the owner's non-empty-folder refusal is known to apply nothing. After
any other owner error the delete may have been partly applied, and a lock-path
re-check does not exclude a concurrent routed delete of the same version, so a
conditional null-version delete answers 500 instead of falling back.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* s3: strip a nested conditional null delete on the owner

A conditional versionId=null delete that the owner refuses because other
keys are nested under the entry used to fall back to the gateway lock and
re-check If-Match there. That lock does not exclude routed deletes: once
the last nested key is removed, a concurrent routed delete can pass the
same check on the owner, and the fallback then treats the missing entry as
deleted, so both requests answer 204.

Demote the entry on the owner instead, in a transaction guarded by the
If-Match clauses and by the entry it was read as, with the object-lock
guards on the request. A missing or already demoted entry answers 412.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* filer: add an IF_OBJECT_EXISTS write condition

IF_EXISTS also holds for a plain directory, which is what is left of a key
once the object on it is stripped while other keys stay nested under it.
IF_OBJECT_EXISTS holds only for an entry that carries an object.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* s3: require an object for a conditional null delete with If-Match: *

Stripping the null version off an entry other keys are nested under leaves a
plain directory. Once the last nested key is gone, a routed delete with
If-Match: * still found an entry there and removed the empty directory, so a
second conditional delete could also answer 204. The owner now checks that
the entry carries an object.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* s3: settle a wildcard null delete on the entry when the filer predates IF_OBJECT_EXISTS

A filer older than IF_OBJECT_EXISTS fails it closed, so with the gateway
upgraded first every If-Match: * delete of a null version answered 412. When
the narrowed guard fails, the gateway now reads the entry on the owner: absent
or a plain directory is a 412, otherwise the file is deleted, or the directory
stripped, under IF_EXISTS and IF_ENTRY_EQUAL on the entry it read. The demote
keeps the caller's If-Match clauses, since the entry guard already requires an
object.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

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