shell: s3.clean.uploads frees an open upload under the object write lock (#12558)

* shell: s3.clean.uploads deletes uploads over filer gRPC

Same DeleteEntry the HTTP handler runs, without minting a filer JWT,
and the gRPC path is the one a locked, routed delete needs next.

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

* shell: s3.clean.uploads frees an open upload under the object write lock

The cleaner deleted a stale upload recursively with no lock, so a
CompleteMultipartUpload could pass its upload-exists check after the
part entries and their chunks were gone, and publish an object over
freed chunks. Do what AbortMultipartUpload does: hold the object's
distributed write lock, and free an open upload in an ObjectTransaction
routed to the object's owner filer, conditioned on the object not
carrying the upload; if it does, delete the metadata only.

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

* shell: s3.clean.uploads keeps the object write lock for the whole cleanup

A large upload's recursive delete can outlast one 15 second lease, and
a completion on the distributed-lock path could then take the lock and
commit over chunks being freed. Renew the lock in the background and
skip the delete once it is lost.

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

* shell: s3.clean.uploads locks the path a completion commits under

For a trailing-slash key a non-versioned completion writes and locks the
entry inside the directory the key names, not the object path, so the
cleaner's transaction did not serialize with it. Lock and check the
same path the completion uses for the bucket's versioning state; the
route key stays the object's.

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

* s3api: abort locks the path a completion commits under

A routed abort locked and checked the object path, but for a
trailing-slash key a non-versioned completion commits under the entry
inside the directory the key names. Use the same path the completion
locks for the bucket's versioning state.

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

* shell: end s3.clean.uploads bucket failure line with a newline

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

* cluster: NewBlockingLongLivedLockWithin bounds the wait on a held lock

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

* shell: s3.clean.uploads skips an upload whose object lock stays held

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

* cluster: share the contended lock filer setup across lock tests

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

* shell: s3.clean.uploads reads the bucket's versioning state per upload

The commit path a routed completion locks depends on versioning for a
trailing-slash key. Reading it once per bucket before the run left a
window as long as the run; read it just before the delete, as the
gateway's abort does.

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

---------

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

weed/cluster/lock_client.go

@@ -182,7 +182,7 @@ func (lc *LockClient) NewShortLivedLock(key string, owner string) (lock *LiveLoc
 		self:           owner,
 		lc:             lc,
 	}
-	if err := lock.retryUntilLocked(5 * time.Second); err != nil {
+	if err := lock.retryUntilLocked(5*time.Second, 0); err != nil {
 		glog.Warningf("create lock %s: %v", key, err)
 		return nil
 	}
@@ -194,6 +194,17 @@ func (lc *LockClient) NewShortLivedLock(key string, owner string) (lock *LiveLoc
 // synchronous acquisition of NewShortLivedLock with the auto-renewal of
 // StartLongLivedLock. Release with Stop().
 func (lc *LockClient) NewBlockingLongLivedLock(key, owner string, lockTTL time.Duration) *LiveLock {
+	lock, err := lc.NewBlockingLongLivedLockWithin(key, owner, lockTTL, 0)
+	if err != nil {
+		glog.Warningf("create lock %s: %v", key, err)
+		return nil
+	}
+	return lock
+}
+
+// NewBlockingLongLivedLockWithin is NewBlockingLongLivedLock that gives up once
+// another owner has held the lock for maxWait. Zero waits without bound.
+func (lc *LockClient) NewBlockingLongLivedLockWithin(key, owner string, lockTTL, maxWait time.Duration) (*LiveLock, error) {
 	if lockTTL == 0 {
 		lockTTL = lock_manager.LiveLockTTL
 	}
@@ -207,10 +218,8 @@ func (lc *LockClient) NewBlockingLongLivedLock(key, owner string, lockTTL time.D
 		lc:             lc,
 		lockTTL:        lockTTL,
 	}
-	// Block until acquired
-	if err := lock.retryUntilLocked(lockTTL); err != nil {
-		glog.Warningf("create lock %s: %v", key, err)
-		return nil
+	if err := lock.retryUntilLocked(lockTTL, maxWait); err != nil {
+		return nil, err
 	}
 	// Start renewal goroutine using a ticker for interruptible sleep
 	lock.renewalDone = make(chan struct{})
@@ -231,7 +240,7 @@ func (lc *LockClient) NewBlockingLongLivedLock(key, owner string, lockTTL time.D
 			}
 		}
 	}()
-	return lock
+	return lock, nil
 }
 
 // StartLongLivedLock starts a goroutine to lock the key and returns immediately.
@@ -310,7 +319,8 @@ func (lc *LockClient) StartLongLivedLock(key string, owner string, onLockOwnerCh
 // several seconds): when a holder on another mount releases the lock, the
 // waiter must pick it up promptly, otherwise cross-mount write handoff stalls
 // long enough to time out clients.
-func (lock *LiveLock) retryUntilLocked(lockDuration time.Duration) error {
+func (lock *LiveLock) retryUntilLocked(lockDuration, maxWait time.Duration) error {
+	started := time.Now()
 	// The retry window bounds the attempts against one route: the clock
 	// starts before the first attempt and each attempt gets what is left
 	// of it. A spent window on a non-seed target falls back to the seed
@@ -356,7 +366,10 @@ func (lock *LiveLock) retryUntilLocked(lockDuration time.Duration) error {
 		glog.V(1).Infof("create lock %s: %v", lock.key, err)
 		if strings.Contains(err.Error(), "lock already owned") {
 			// Ordinary contention: a reachable server holds the lock, so
-			// waiting is the point and has no bound.
+			// waiting is the point and is bounded only by maxWait.
+			if maxWait > 0 && time.Since(started) >= maxWait {
+				return err
+			}
 			unavailableSince = time.Now()
 			continue
 		}

weed/cluster/lock_client_test.go

@@ -225,9 +225,8 @@ func (s *contendedLockFiler) DistributedLock(ctx context.Context, req *filer_pb.
 	return &filer_pb.LockResponse{Error: "lock already owned by someone"}, nil
 }
 
-// A lock held by another owner is ordinary contention: acquisition waits it
-// out rather than failing on the unavailability bound.
-func TestNewShortLivedLockWaitsOutContention(t *testing.T) {
+func newContendedLockClient(t *testing.T) *LockClient {
+	t.Helper()
 	listener, err := net.Listen("tcp", "127.0.0.1:0")
 	if err != nil {
 		t.Fatalf("listen: %v", err)
@@ -235,14 +234,19 @@ func TestNewShortLivedLockWaitsOutContention(t *testing.T) {
 	grpcServer := grpc.NewServer()
 	filer_pb.RegisterSeaweedFilerServer(grpcServer, &contendedLockFiler{})
 	go grpcServer.Serve(listener)
-	defer grpcServer.Stop()
+	t.Cleanup(grpcServer.Stop)
 
-	dialOption := grpc.WithTransportCredentials(insecure.NewCredentials())
 	host, port, err := net.SplitHostPort(listener.Addr().String())
 	if err != nil {
 		t.Fatalf("split host port: %v", err)
 	}
-	lc := NewLockClient(dialOption, pb.ServerAddress(fmt.Sprintf("%s:0.%s", host, port)))
+	return NewLockClient(grpc.WithTransportCredentials(insecure.NewCredentials()), pb.ServerAddress(fmt.Sprintf("%s:0.%s", host, port)))
+}
+
+// A lock held by another owner is ordinary contention: acquisition waits it
+// out rather than failing on the unavailability bound.
+func TestNewShortLivedLockWaitsOutContention(t *testing.T) {
+	lc := newContendedLockClient(t)
 	lc.noLockServerRetryPeriod = 200 * time.Millisecond
 
 	done := make(chan *LiveLock, 1)
@@ -256,6 +260,22 @@ func TestNewShortLivedLockWaitsOutContention(t *testing.T) {
 	}
 }
 
+func TestNewBlockingLongLivedLockWithinGivesUpOnContention(t *testing.T) {
+	lc := newContendedLockClient(t)
+
+	maxWait := 1500 * time.Millisecond
+	start := time.Now()
+	lock, err := lc.NewBlockingLongLivedLockWithin("test-key", "test-owner", 15*time.Second, maxWait)
+	elapsed := time.Since(start)
+
+	if lock != nil || err == nil {
+		t.Fatalf("got lock %v err %v, want no lock and the contention error", lock, err)
+	}
+	if elapsed < maxWait || elapsed > maxWait+5*time.Second {
+		t.Fatalf("gave up after %v, want just past %v", elapsed, maxWait)
+	}
+}
+
 // A ring member that refuses connections is unavailability, not contention:
 // acquisition fails on the same bound as "no lock server found".
 func TestNewShortLivedLockFailsFastOnUnreachableFiler(t *testing.T) {
@@ -519,27 +539,12 @@ func TestDoLockKeepsRenewTokenOnError(t *testing.T) {
 // A definitive rejection answers with an empty token, which clears the stale
 // one — only transport failures keep it.
 func TestDoLockClearsRenewTokenOnRejection(t *testing.T) {
-	listener, err := net.Listen("tcp", "127.0.0.1:0")
-	if err != nil {
-		t.Fatalf("listen: %v", err)
-	}
-	grpcServer := grpc.NewServer()
-	filer_pb.RegisterSeaweedFilerServer(grpcServer, &contendedLockFiler{})
-	go grpcServer.Serve(listener)
-	defer grpcServer.Stop()
-
-	dialOption := grpc.WithTransportCredentials(insecure.NewCredentials())
-	host, port, err := net.SplitHostPort(listener.Addr().String())
-	if err != nil {
-		t.Fatalf("split host port: %v", err)
-	}
-	filerAddr := pb.ServerAddress(fmt.Sprintf("%s:0.%s", host, port))
-	lc := NewLockClient(dialOption, filerAddr)
+	lc := newContendedLockClient(t)
 	lock := &LiveLock{
 		key:            "test-key",
 		renewToken:     "tok",
-		hostFiler:      filerAddr,
-		grpcDialOption: dialOption,
+		hostFiler:      lc.seedFiler,
+		grpcDialOption: lc.grpcDialOption,
 		lc:             lc,
 	}
 	errorMessage, _, err := lock.doLock(time.Second, time.Second)

weed/s3api/filer_multipart.go

@@ -1246,11 +1246,15 @@ func (s3a *S3ApiServer) routedUploadDirDelete(bucket, uploadId, object string) s
 	if owner == "" {
 		return rmUploadDir(true)
 	}
-	objectPath := s3a.toFilerPath(bucket, object)
+	commitPath, err := s3a.uploadCommitPath(bucket, object)
+	if err != nil {
+		glog.Errorf("bucket %s abort upload %s: %v", bucket, uploadId, err)
+		return s3err.ErrInternalError
+	}
 	resp, err := s3a.objectTxnOnFiler(owner, &filer_pb.ObjectTransactionRequest{
-		LockKey:      objectPath,
+		LockKey:      commitPath,
 		RouteKey:     s3a.objectRouteKey(bucket, object),
-		ConditionKey: objectPath,
+		ConditionKey: commitPath,
 		Condition: &filer_pb.WriteCondition{Clauses: []*filer_pb.WriteCondition_Clause{{
 			Kind:     filer_pb.WriteCondition_IF_EXTENDED_NOT_EQUAL,
 			ExtKey:   s3_constants.SeaweedFSUploadId,
@@ -1278,6 +1282,21 @@ func (s3a *S3ApiServer) routedUploadDirDelete(bucket, uploadId, object string) s
 	return s3err.ErrNone
 }
 
+// uploadCommitPath is the path a routed completion of object locks: the
+// object path in a versioned bucket, else the entry it writes, which for a
+// trailing-slash key lies inside the directory the key names.
+func (s3a *S3ApiServer) uploadCommitPath(bucket, object string) (string, error) {
+	versioningState, err := s3a.getVersioningState(bucket)
+	if err != nil {
+		return "", err
+	}
+	if versioningState == s3_constants.VersioningEnabled {
+		return s3a.toFilerPath(bucket, object), nil
+	}
+	entryName, dirName := s3a.getEntryNameAndDir(&s3.CompleteMultipartUploadInput{Bucket: &bucket, Key: &object})
+	return dirName + "/" + entryName, nil
+}
+
 // uploadCompleted reports whether the upload assembled into an object: the
 // object entry, or any version file under <key>.versions, still carries the
 // upload id completion stamps on it.

weed/s3api/filer_multipart_abort_test.go

@@ -3,9 +3,12 @@ package s3api
 import (
 	"context"
 	"testing"
+	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/s3"
+	"github.com/seaweedfs/seaweedfs/weed/cluster"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
@@ -281,3 +284,52 @@ func TestLifecycleAbortUndecidableCheckRetriesLater(t *testing.T) {
 		t.Fatalf("delete issued despite undecidable check: %+v", f.deleteReq)
 	}
 }
+
+type fakeRoutedAbortFiler struct {
+	fakeAbortFiler
+	txnReq *filer_pb.ObjectTransactionRequest
+}
+
+func (f *fakeRoutedAbortFiler) ObjectTransaction(ctx context.Context, req *filer_pb.ObjectTransactionRequest) (*filer_pb.ObjectTransactionResponse, error) {
+	f.txnReq = req
+	return &filer_pb.ObjectTransactionResponse{}, nil
+}
+
+// A routed abort must take the path lock the completion commits under. For a
+// trailing-slash key a non-versioned completion writes the entry inside the
+// directory the key names, while a versioned one locks the object path.
+func TestRoutedAbortLocksTheCompletionCommitPath(t *testing.T) {
+	for _, tc := range []struct {
+		versioning string
+		lockKey    string
+	}{
+		{"", "/dir/dir"},
+		{s3_constants.VersioningSuspended, "/dir/dir"},
+		{s3_constants.VersioningEnabled, "/dir/"},
+	} {
+		f := &fakeRoutedAbortFiler{}
+		ownerAddr := startFakeFiler(t, f)
+		s3a := newFailoverTestServer(t, ownerAddr)
+		bucketDir := s3a.bucketDir("b")
+		f.uploadsDir = bucketDir + "/" + s3_constants.MultipartUploadsFolder
+		f.uploadEntry = uploadRecordEntry("up1")
+		f.uploadEntry.Extended[s3_constants.ExtMultipartObjectKey] = []byte("dir/")
+		f.objectDir = bucketDir
+		f.versionsDir = bucketDir + "/dir/dir" + s3_constants.VersionsFolder
+		s3a.bucketConfigCache = NewBucketConfigCache(time.Minute)
+		s3a.bucketConfigCache.Set("b", &BucketConfig{Name: "b", Versioning: tc.versioning})
+		s3a.objectWriteLockClient = cluster.NewLockClient(s3a.option.GrpcDialOption, ownerAddr)
+		s3a.objectWriteLockClient.SetRing([]pb.ServerAddress{ownerAddr}, 1)
+
+		if _, code := s3a.abortMultipartUpload(&s3.AbortMultipartUploadInput{Bucket: aws.String("b"), Key: aws.String("dir/"), UploadId: aws.String("up1")}); code != s3err.ErrNone {
+			t.Fatalf("versioning %q: code = %v", tc.versioning, code)
+		}
+		req := f.txnReq
+		if req == nil {
+			t.Fatalf("versioning %q: no transaction", tc.versioning)
+		}
+		if req.LockKey != bucketDir+tc.lockKey || req.ConditionKey != req.LockKey || req.RouteKey != s3a.objectRouteKey("b", "dir/") {
+			t.Fatalf("versioning %q: lock %q condition %q route %q", tc.versioning, req.LockKey, req.ConditionKey, req.RouteKey)
+		}
+	}
+}

weed/shell/command_s3_clean_uploads.go

@@ -11,12 +11,14 @@ import (
 	"strings"
 	"time"
 
+	"github.com/google/uuid"
+
+	"github.com/seaweedfs/seaweedfs/weed/cluster"
+	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
-	"github.com/seaweedfs/seaweedfs/weed/security"
 	"github.com/seaweedfs/seaweedfs/weed/util"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
-	util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
 )
 
 func init() {
@@ -49,34 +51,34 @@ func (c *commandS3CleanUploads) Do(args []string, commandEnv *CommandEnv, writer
 		return nil
 	}
 
-	signingKey := util.GetViper().GetString("jwt.filer_signing.key")
-
 	var filerBucketsPath string
 	filerBucketsPath, err = readFilerBucketsPath(commandEnv)
 	if err != nil {
 		return fmt.Errorf("read buckets: %w", err)
 	}
 
-	var buckets []string
+	var buckets []*filer_pb.Entry
 	err = filer_pb.List(context.Background(), commandEnv, filerBucketsPath, "", func(entry *filer_pb.Entry, isLast bool) error {
-		buckets = append(buckets, entry.Name)
+		buckets = append(buckets, entry)
 		return nil
 	}, "", false, math.MaxUint32)
 	if err != nil {
 		return fmt.Errorf("list buckets under %v: %w", filerBucketsPath, err)
 	}
 
+	lock := newObjectWriteLock(commandEnv)
 	for _, bucket := range buckets {
-		if err := c.cleanupUploads(commandEnv, writer, filerBucketsPath, bucket, *uploadedTimeAgo, signingKey); err != nil {
-			fmt.Fprintf(writer, "failed cleanup uploads for bucket %s: %v", bucket, err)
+		if err := c.cleanupUploads(commandEnv, writer, filerBucketsPath, bucket.Name, *uploadedTimeAgo, lock); err != nil {
+			fmt.Fprintf(writer, "failed cleanup uploads for bucket %s: %v\n", bucket.Name, err)
 		}
 	}
 
 	return err
 }
 
-func (c *commandS3CleanUploads) cleanupUploads(commandEnv *CommandEnv, writer io.Writer, filerBucketsPath string, bucket string, timeAgo time.Duration, signingKey string) error {
-	uploadsDir := filerBucketsPath + "/" + bucket + "/" + s3_constants.MultipartUploadsFolder
+func (c *commandS3CleanUploads) cleanupUploads(commandEnv *CommandEnv, writer io.Writer, filerBucketsPath string, bucket string, timeAgo time.Duration, lock objectWriteLock) error {
+	bucketDir := filerBucketsPath + "/" + bucket
+	uploadsDir := bucketDir + "/" + s3_constants.MultipartUploadsFolder
 	var staleUploads []*filer_pb.Entry
 	now := time.Now()
 	err := filer_pb.List(context.Background(), commandEnv, uploadsDir, "", func(entry *filer_pb.Entry, isLast bool) error {
@@ -90,37 +92,166 @@ func (c *commandS3CleanUploads) cleanupUploads(commandEnv *CommandEnv, writer io
 		return fmt.Errorf("list uploads under %v: %w", uploadsDir, err)
 	}
 
-	var encodedJwt security.EncodedJwt
-	if signingKey != "" {
-		encodedJwt = security.GenJwtForFilerServer(security.SigningKey(signingKey), 15*60)
-	}
-
 	for _, staleUpload := range staleUploads {
-		// A completed upload's part entries share chunks with the finished
-		// object, so purging their data corrupts it. Completion normally
-		// removes this directory itself; a survivor means that cleanup failed
-		// and only the metadata should go. An undecidable lookup is left for
-		// the next run rather than risk live chunks.
-		completed, checkErr := c.uploadCompleted(commandEnv, filerBucketsPath+"/"+bucket, staleUpload)
-		if checkErr != nil {
-			fmt.Fprintf(writer, "skip %s: %v\n", staleUpload.Name, checkErr)
+		held, release, lockErr := lock(uploadObjectPath(bucketDir, staleUpload))
+		if lockErr != nil {
+			fmt.Fprintf(writer, "skip %s: %v\n", staleUpload.Name, lockErr)
 			continue
 		}
-		deleteUrl := fmt.Sprintf("http://%s%s/%s?recursive=true&ignoreRecursiveError=true", commandEnv.FilerAddress().ToHttpAddress(), uploadsDir, staleUpload.Name)
-		if completed {
-			deleteUrl += "&skipChunkDeletion=true"
-		}
-		fmt.Fprintf(writer, "purge %s\n", deleteUrl)
-
-		err = util_http.Delete(deleteUrl, string(encodedJwt))
-		if err != nil && err.Error() != "" {
-			return fmt.Errorf("purge %s/%s: %v", uploadsDir, staleUpload.Name, err)
+		err := c.removeUploadDir(commandEnv, writer, bucketDir, uploadsDir, staleUpload, held)
+		release()
+		if err != nil {
+			return err
 		}
 	}
 
 	return nil
 }
 
+// objectWriteLock takes the distributed lock the S3 gateway commits a
+// completion under when it is not routed to the object's owner filer, and
+// renews it until release. It gives up after a minute of contention.
+type objectWriteLock func(objectPath string) (held func() bool, release func(), err error)
+
+func newObjectWriteLock(commandEnv *CommandEnv) objectWriteLock {
+	lockClient := cluster.NewLockClient(commandEnv.option.GrpcDialOption, commandEnv.FilerAddress())
+	owner := fmt.Sprintf("s3.clean.uploads-%d", util.RandomInt32())
+	return func(objectPath string) (func() bool, func(), error) {
+		if objectPath == "" {
+			return func() bool { return true }, func() {}, nil
+		}
+		key := objectRouteKey(objectPath)
+		lock, err := lockClient.NewBlockingLongLivedLockWithin(key, owner, 15*time.Second, time.Minute)
+		if err != nil {
+			return nil, nil, fmt.Errorf("lock %s: %w", key, err)
+		}
+		return lock.IsLocked, func() {
+			if err := lock.Stop(); err != nil {
+				glog.Warningf("release lock %s: %v", key, err)
+			}
+		}, nil
+	}
+}
+
+// removeUploadDir mirrors the S3 gateway's abort, under the object write lock.
+// A completed upload's part entries share chunks with the finished object, so
+// only the metadata goes; an open upload's chunks are freed by
+// routedUploadDirDelete. An undecidable lookup, or a lost lock, is left for
+// the next run.
+func (c *commandS3CleanUploads) removeUploadDir(filerClient filer_pb.FilerClient, writer io.Writer, bucketDir, uploadsDir string, upload *filer_pb.Entry, lockHeld func() bool) error {
+	completed, err := c.uploadCompleted(filerClient, bucketDir, upload)
+	if err != nil {
+		fmt.Fprintf(writer, "skip %s: %v\n", upload.Name, err)
+		return nil
+	}
+	var commitPath string
+	if !completed {
+		if commitPath, err = uploadCommitPath(filerClient, bucketDir, upload); err != nil {
+			fmt.Fprintf(writer, "skip %s: %v\n", upload.Name, err)
+			return nil
+		}
+	}
+	if !lockHeld() {
+		fmt.Fprintf(writer, "skip %s: lost the object write lock\n", upload.Name)
+		return nil
+	}
+	if !completed {
+		freed, err := routedUploadDirDelete(filerClient, uploadObjectPath(bucketDir, upload), commitPath, uploadsDir, upload.Name)
+		if err != nil {
+			return fmt.Errorf("purge %s/%s: %v", uploadsDir, upload.Name, err)
+		}
+		if freed {
+			fmt.Fprintf(writer, "purge %s/%s\n", uploadsDir, upload.Name)
+			return nil
+		}
+	}
+	fmt.Fprintf(writer, "purge %s/%s metadata only\n", uploadsDir, upload.Name)
+	if err := filer_pb.Remove(context.Background(), filerClient, uploadsDir, upload.Name, false, true, true, false, nil); err != nil {
+		return fmt.Errorf("purge %s/%s: %v", uploadsDir, upload.Name, err)
+	}
+	return nil
+}
+
+// routedUploadDirDelete frees an open upload's chunks in an ObjectTransaction
+// on the object's owner filer, under the path lock a routed completion
+// commits under. It deletes nothing and reports false when the object
+// already carries the upload.
+func routedUploadDirDelete(filerClient filer_pb.FilerClient, objectPath, commitPath, uploadsDir, uploadId string) (bool, error) {
+	if objectPath == "" {
+		return true, filer_pb.Remove(context.Background(), filerClient, uploadsDir, uploadId, true, true, true, false, nil)
+	}
+	var resp *filer_pb.ObjectTransactionResponse
+	err := filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+		var txnErr error
+		resp, txnErr = client.ObjectTransaction(context.Background(), &filer_pb.ObjectTransactionRequest{
+			LockKey:      commitPath,
+			RouteKey:     objectRouteKey(objectPath),
+			ConditionKey: commitPath,
+			Condition: &filer_pb.WriteCondition{Clauses: []*filer_pb.WriteCondition_Clause{{
+				Kind:     filer_pb.WriteCondition_IF_EXTENDED_NOT_EQUAL,
+				ExtKey:   s3_constants.SeaweedFSUploadId,
+				ExtValue: uploadId,
+			}}},
+			Mutations: []*filer_pb.ObjectMutation{{
+				Type:         filer_pb.ObjectMutation_DELETE,
+				Directory:    uploadsDir,
+				Name:         uploadId,
+				IsDeleteData: true,
+				IsRecursive:  true,
+			}},
+			RequestId: uuid.NewString(),
+		})
+		return txnErr
+	})
+	if err != nil {
+		return false, err
+	}
+	if resp.ErrorCode == filer_pb.FilerError_PRECONDITION_FAILED {
+		return false, nil
+	}
+	if resp.Error != "" {
+		return false, errors.New(resp.Error)
+	}
+	return true, nil
+}
+
+// uploadObjectPath is the filer path of the object the upload completes into, as
+// the gateway's toFilerPath derives it, or "" when the upload names none.
+func uploadObjectPath(bucketDir string, upload *filer_pb.Entry) string {
+	objectKey := string(upload.Extended[s3_constants.ExtMultipartObjectKey])
+	if objectKey == "" {
+		return ""
+	}
+	return bucketDir + "/" + s3_constants.NormalizeObjectKey(objectKey)
+}
+
+// uploadCommitPath is the path a routed completion of the upload locks: the
+// object path in a versioned bucket, else the entry it writes, which for a
+// trailing-slash key lies inside the directory the key names. Like the
+// gateway's, it reads the bucket's versioning state just before the delete.
+func uploadCommitPath(filerClient filer_pb.FilerClient, bucketDir string, upload *filer_pb.Entry) (string, error) {
+	objectKey := string(upload.Extended[s3_constants.ExtMultipartObjectKey])
+	if objectKey == "" {
+		return "", nil
+	}
+	bucket, _, _, err := filer_pb.GetEntry(context.Background(), filerClient, util.FullPath(bucketDir))
+	if err != nil {
+		return "", fmt.Errorf("read bucket %s: %w", bucketDir, err)
+	}
+	if versioning, lockEnabled := getBucketVersioningState(bucket); versioning == s3_constants.VersioningEnabled || lockEnabled {
+		return uploadObjectPath(bucketDir, upload), nil
+	}
+	dir := path.Dir(objectKey)
+	if dir == "." {
+		dir = ""
+	}
+	return path.Join(bucketDir, dir, path.Base(objectKey)), nil
+}
+
+func objectRouteKey(objectPath string) string {
+	return s3_constants.ObjectWriteRouteKeyPrefix + strings.TrimSuffix(objectPath, "/")
+}
+
 // uploadCompleted reports whether the upload assembled into an object: the
 // object entry, or any version file under <key>.versions, still carries the
 // upload id completion stamps on it.

weed/shell/command_s3_clean_uploads_test.go

@@ -26,6 +26,7 @@ func (s stubFilerClient) GetDataCenter() string { return "" }
 
 type stubFilerSvc struct {
 	filer_pb.SeaweedFilerClient
+	entries    map[string]*filer_pb.Entry
 	lookupResp *filer_pb.LookupDirectoryEntryResponse
 	lookupErr  error
 	listResp   []*filer_pb.Entry
@@ -33,9 +34,25 @@ type stubFilerSvc struct {
 	listErr    error
 	lookupReq  *filer_pb.LookupDirectoryEntryRequest
 	listReq    *filer_pb.ListEntriesRequest
+	txnResp    *filer_pb.ObjectTransactionResponse
+	txnReqs    []*filer_pb.ObjectTransactionRequest
+	deleteReqs []*filer_pb.DeleteEntryRequest
+}
+
+func (s *stubFilerSvc) ObjectTransaction(ctx context.Context, req *filer_pb.ObjectTransactionRequest, opts ...grpc.CallOption) (*filer_pb.ObjectTransactionResponse, error) {
+	s.txnReqs = append(s.txnReqs, req)
+	return s.txnResp, nil
+}
+
+func (s *stubFilerSvc) DeleteEntry(ctx context.Context, req *filer_pb.DeleteEntryRequest, opts ...grpc.CallOption) (*filer_pb.DeleteEntryResponse, error) {
+	s.deleteReqs = append(s.deleteReqs, req)
+	return &filer_pb.DeleteEntryResponse{}, nil
 }
 
 func (s *stubFilerSvc) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest, opts ...grpc.CallOption) (*filer_pb.LookupDirectoryEntryResponse, error) {
+	if entry, ok := s.entries[req.Directory+"/"+req.Name]; ok {
+		return &filer_pb.LookupDirectoryEntryResponse{Entry: entry}, nil
+	}
 	s.lookupReq = req
 	return s.lookupResp, s.lookupErr
 }
@@ -201,3 +218,144 @@ func TestUploadCompleted(t *testing.T) {
 		}
 	})
 }
+
+func bucketEntry(versioning string) *filer_pb.Entry {
+	return &filer_pb.Entry{Name: "b", IsDirectory: true, Extended: map[string][]byte{s3_constants.ExtVersioningKey: []byte(versioning)}}
+}
+
+// openUploadSvc answers an open upload's lookups: no object, no versions, and
+// a bucket with the given versioning state.
+func openUploadSvc(versioning string, txnResp *filer_pb.ObjectTransactionResponse) *stubFilerSvc {
+	return &stubFilerSvc{
+		entries:   map[string]*filer_pb.Entry{"/buckets/b": bucketEntry(versioning)},
+		lookupErr: filer_pb.ErrNotFound,
+		listErr:   filer_pb.ErrNotFound,
+		txnResp:   txnResp,
+	}
+}
+
+func TestRemoveUploadDir(t *testing.T) {
+	c := &commandS3CleanUploads{}
+	const uploadsDir = "/buckets/b/" + s3_constants.MultipartUploadsFolder
+	lockHeld := func() bool { return true }
+
+	t.Run("open upload is freed in a transaction on the object's lock", func(t *testing.T) {
+		svc := openUploadSvc("", &filer_pb.ObjectTransactionResponse{})
+		if err := c.removeUploadDir(stubFilerClient{client: svc}, io.Discard, "/buckets/b", uploadsDir, uploadEntry("up1", "/dir//obj"), lockHeld); err != nil {
+			t.Fatal(err)
+		}
+		if len(svc.txnReqs) != 1 || len(svc.deleteReqs) != 0 {
+			t.Fatalf("%d transactions, %d deletes; want 1, 0", len(svc.txnReqs), len(svc.deleteReqs))
+		}
+		req := svc.txnReqs[0]
+		if req.LockKey != "/buckets/b/dir/obj" || req.ConditionKey != req.LockKey || req.RouteKey != s3_constants.ObjectWriteRouteKeyPrefix+req.LockKey {
+			t.Fatalf("lock %q condition %q route %q", req.LockKey, req.ConditionKey, req.RouteKey)
+		}
+		clause := req.Condition.GetClauses()[0]
+		if clause.Kind != filer_pb.WriteCondition_IF_EXTENDED_NOT_EQUAL || clause.ExtKey != s3_constants.SeaweedFSUploadId || string(clause.ExtValue) != "up1" {
+			t.Fatalf("condition %v", clause)
+		}
+		m := req.Mutations[0]
+		if m.Type != filer_pb.ObjectMutation_DELETE || m.Directory != uploadsDir || m.Name != "up1" || !m.IsDeleteData || !m.IsRecursive {
+			t.Fatalf("mutation %v", m)
+		}
+	})
+
+	t.Run("trailing-slash key locks the entry its completion writes as versioning changes", func(t *testing.T) {
+		svc := openUploadSvc("", &filer_pb.ObjectTransactionResponse{})
+		for _, tc := range []struct {
+			versioning string
+			objectLock bool
+			lockKey    string
+		}{
+			{"", false, "/buckets/b/dir/dir"},
+			{s3_constants.VersioningEnabled, false, "/buckets/b/dir/"},
+			{s3_constants.VersioningSuspended, false, "/buckets/b/dir/dir"},
+			{"", true, "/buckets/b/dir/"},
+		} {
+			bucket := bucketEntry(tc.versioning)
+			if tc.objectLock {
+				bucket.Extended[s3_constants.ExtObjectLockEnabledKey] = []byte(s3_constants.ObjectLockEnabled)
+			}
+			svc.entries["/buckets/b"] = bucket
+			svc.txnReqs = nil
+			if err := c.removeUploadDir(stubFilerClient{client: svc}, io.Discard, "/buckets/b", uploadsDir, uploadEntry("up1", "dir/"), lockHeld); err != nil {
+				t.Fatal(err)
+			}
+			req := svc.txnReqs[0]
+			if req.LockKey != tc.lockKey || req.ConditionKey != tc.lockKey || req.RouteKey != s3_constants.ObjectWriteRouteKeyPrefix+"/buckets/b/dir" {
+				t.Fatalf("versioning=%q objectLock=%v: lock %q condition %q route %q", tc.versioning, tc.objectLock, req.LockKey, req.ConditionKey, req.RouteKey)
+			}
+		}
+	})
+
+	t.Run("unreadable bucket deletes nothing", func(t *testing.T) {
+		svc := &stubFilerSvc{lookupErr: filer_pb.ErrNotFound, listErr: filer_pb.ErrNotFound, txnResp: &filer_pb.ObjectTransactionResponse{}}
+		if err := c.removeUploadDir(stubFilerClient{client: svc}, io.Discard, "/buckets/b", uploadsDir, uploadEntry("up1", "obj"), lockHeld); err != nil {
+			t.Fatal(err)
+		}
+		if len(svc.txnReqs) != 0 || len(svc.deleteReqs) != 0 {
+			t.Fatalf("%d transactions, %d deletes", len(svc.txnReqs), len(svc.deleteReqs))
+		}
+	})
+
+	t.Run("upload completed before the transaction keeps its chunks", func(t *testing.T) {
+		svc := openUploadSvc("", &filer_pb.ObjectTransactionResponse{ErrorCode: filer_pb.FilerError_PRECONDITION_FAILED})
+		if err := c.removeUploadDir(stubFilerClient{client: svc}, io.Discard, "/buckets/b", uploadsDir, uploadEntry("up1", "obj"), lockHeld); err != nil {
+			t.Fatal(err)
+		}
+		if len(svc.deleteReqs) != 1 || svc.deleteReqs[0].IsDeleteData || svc.deleteReqs[0].Name != "up1" {
+			t.Fatalf("deletes %v, want one metadata-only delete of up1", svc.deleteReqs)
+		}
+	})
+
+	t.Run("completed upload goes metadata-only without a transaction", func(t *testing.T) {
+		svc := &stubFilerSvc{lookupResp: &filer_pb.LookupDirectoryEntryResponse{Entry: completedEntry("up1")}}
+		if err := c.removeUploadDir(stubFilerClient{client: svc}, io.Discard, "/buckets/b", uploadsDir, uploadEntry("up1", "obj"), lockHeld); err != nil {
+			t.Fatal(err)
+		}
+		if len(svc.txnReqs) != 0 || len(svc.deleteReqs) != 1 || svc.deleteReqs[0].IsDeleteData {
+			t.Fatalf("%d transactions, deletes %v", len(svc.txnReqs), svc.deleteReqs)
+		}
+	})
+
+	t.Run("undecidable lookup deletes nothing", func(t *testing.T) {
+		svc := &stubFilerSvc{lookupErr: errors.New("store unavailable")}
+		if err := c.removeUploadDir(stubFilerClient{client: svc}, io.Discard, "/buckets/b", uploadsDir, uploadEntry("up1", "obj"), lockHeld); err != nil {
+			t.Fatal(err)
+		}
+		if len(svc.txnReqs) != 0 || len(svc.deleteReqs) != 0 {
+			t.Fatalf("%d transactions, %d deletes", len(svc.txnReqs), len(svc.deleteReqs))
+		}
+	})
+
+	t.Run("lost lock deletes nothing", func(t *testing.T) {
+		svc := openUploadSvc("", &filer_pb.ObjectTransactionResponse{})
+		if err := c.removeUploadDir(stubFilerClient{client: svc}, io.Discard, "/buckets/b", uploadsDir, uploadEntry("up1", "obj"), func() bool { return false }); err != nil {
+			t.Fatal(err)
+		}
+		if len(svc.txnReqs) != 0 || len(svc.deleteReqs) != 0 {
+			t.Fatalf("%d transactions, %d deletes", len(svc.txnReqs), len(svc.deleteReqs))
+		}
+	})
+
+	t.Run("transaction error fails the purge", func(t *testing.T) {
+		svc := openUploadSvc("", &filer_pb.ObjectTransactionResponse{Error: "boom"})
+		if err := c.removeUploadDir(stubFilerClient{client: svc}, io.Discard, "/buckets/b", uploadsDir, uploadEntry("up1", "obj"), lockHeld); err == nil {
+			t.Fatal("want an error")
+		}
+		if len(svc.deleteReqs) != 0 {
+			t.Fatalf("deletes %v", svc.deleteReqs)
+		}
+	})
+
+	t.Run("upload without an object key frees its chunks directly", func(t *testing.T) {
+		svc := &stubFilerSvc{}
+		if err := c.removeUploadDir(stubFilerClient{client: svc}, io.Discard, "/buckets/b", uploadsDir, &filer_pb.Entry{Name: "up1"}, lockHeld); err != nil {
+			t.Fatal(err)
+		}
+		if len(svc.txnReqs) != 0 || len(svc.deleteReqs) != 1 || !svc.deleteReqs[0].IsDeleteData {
+			t.Fatalf("%d transactions, deletes %v", len(svc.txnReqs), svc.deleteReqs)
+		}
+	})
+}