s3: keep empty-folder cleanup out of the multipart .uploads staging tree (#10273)

* s3: keep empty-folder cleanup out of the multipart .uploads staging tree

The async EmptyFolderCleaner deleted <bucket>/.uploads once it looked
empty. A concurrent CreateMultipartUpload inserting its marker between
the cleaner's emptiness check and the bulk child delete had its row
wiped, so the upload silently vanished: ListMultipartUploads then omits
it and the following part/complete/copy calls fail. Skip the .uploads
subtree in both the queue and the delete path (including the eager
parent cascade); the multipart upload lifecycle owns it.

* s3: slice the bucket-relative path instead of reallocating it

weed/filer/empty_folder_cleanup/empty_folder_cleaner.go

@@ -128,6 +128,11 @@ func (efc *EmptyFolderCleaner) OnDeleteEvent(directory string, entryName string,
 		return
 	}
 
+	// Never queue the S3 multipart staging area; the upload lifecycle owns it.
+	if isMultipartUploadsPath(efc.bucketPath, directory) {
+		return
+	}
+
 	// Check if we own this folder
 	if !efc.ownsFolder(directory) {
 		glog.V(4).Infof("EmptyFolderCleaner: not owner of %s, skipping", directory)
@@ -239,6 +244,15 @@ func (efc *EmptyFolderCleaner) processCleanupQueue() {
 
 // executeCleanup performs the actual cleanup of an empty folder
 func (efc *EmptyFolderCleaner) executeCleanup(folder string, triggeredBy string) {
+	// The bucket-shared .uploads staging tree holds in-progress multipart uploads.
+	// Deleting <bucket>/.uploads (reached here directly or via the parent cascade
+	// below) races a concurrent CreateMultipartUpload: the new upload's marker row
+	// is inserted between the emptiness check and the folder delete, then wiped,
+	// so the upload silently vanishes. Leave this tree to the upload lifecycle.
+	if isMultipartUploadsPath(efc.bucketPath, folder) {
+		return
+	}
+
 	efc.mu.Lock()
 
 	// Quick check: if we have cached count and it's > 0, skip
@@ -422,6 +436,21 @@ func isUnderPath(child, parent string) bool {
 	return child[len(parent)] == '/'
 }
 
+// isMultipartUploadsPath reports whether directory is the S3 multipart staging
+// root <bucket>/.uploads or anything beneath it.
+func isMultipartUploadsPath(bucketPath, directory string) bool {
+	if bucketPath == "" {
+		return false
+	}
+	dir, ok := util.ExtractBucketPath(bucketPath, directory, true)
+	if !ok {
+		return false
+	}
+	// requireChild guarantees directory starts with dir + "/", so slice past it.
+	first, _, _ := strings.Cut(directory[len(dir)+1:], "/")
+	return first == s3_constants.MultipartUploadsFolder
+}
+
 // isUnderBucketPath checks if directory is inside a bucket (under /buckets/<bucket>/...)
 // This ensures we only clean up folders inside buckets, not the buckets themselves
 func isUnderBucketPath(directory, bucketPath string) bool {

weed/filer/empty_folder_cleanup/empty_folder_cleaner_test.go

@@ -104,6 +104,102 @@ func Test_isUnderBucketPath(t *testing.T) {
 	}
 }
 
+func Test_isMultipartUploadsPath(t *testing.T) {
+	uploads := s3_constants.MultipartUploadsFolder
+	tests := []struct {
+		name       string
+		directory  string
+		bucketPath string
+		expected   bool
+	}{
+		{"staging root", "/buckets/mybucket/" + uploads, "/buckets", true},
+		{"upload marker dir", "/buckets/mybucket/" + uploads + "/abc123", "/buckets", true},
+		{"nested under staging", "/buckets/mybucket/" + uploads + "/abc123/hashstates", "/buckets", true},
+		{"normal folder", "/buckets/mybucket/folder", "/buckets", false},
+		{"registry own uploads dir", "/buckets/mybucket/docker/registry/v2/_uploads/x", "/buckets", false},
+		{"bucket itself", "/buckets/mybucket", "/buckets", false},
+		{"outside buckets", "/other/" + uploads, "/buckets", false},
+		{"empty bucket path", "/buckets/mybucket/" + uploads, "", false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := isMultipartUploadsPath(tt.bucketPath, tt.directory); got != tt.expected {
+				t.Errorf("isMultipartUploadsPath(%q, %q) = %v, want %v", tt.bucketPath, tt.directory, got, tt.expected)
+			}
+		})
+	}
+}
+
+func TestEmptyFolderCleaner_OnDeleteEvent_skipsMultipartUploads(t *testing.T) {
+	lockRing := lock_manager.NewLockRing(5 * time.Second)
+	lockRing.SetSnapshot([]pb.ServerAddress{"filer1:8888"}, 0)
+
+	cleaner := &EmptyFolderCleaner{
+		lockRing:     lockRing,
+		host:         "filer1:8888",
+		bucketPath:   "/buckets",
+		enabled:      true,
+		folderCounts: make(map[string]*folderState),
+		cleanupQueue: NewCleanupQueue(1000, 10*time.Minute),
+		stopCh:       make(chan struct{}),
+	}
+
+	now := time.Now()
+	uploads := "/buckets/mybucket/" + s3_constants.MultipartUploadsFolder
+	cleaner.OnDeleteEvent(uploads, "abc123", true, now)
+	cleaner.OnDeleteEvent(uploads+"/abc123", "0001.part", false, now)
+	if cleaner.GetPendingCleanupCount() != 0 {
+		t.Fatalf("multipart staging paths must not be queued, got %d pending", cleaner.GetPendingCleanupCount())
+	}
+
+	// A normal folder is still queued, proving the guard is scoped to .uploads.
+	cleaner.OnDeleteEvent("/buckets/mybucket/folder", "file.txt", false, now)
+	if cleaner.GetPendingCleanupCount() != 1 {
+		t.Fatalf("normal folder should be queued, got %d pending", cleaner.GetPendingCleanupCount())
+	}
+
+	cleaner.Stop()
+}
+
+func TestEmptyFolderCleaner_executeCleanup_skipsMultipartUploads(t *testing.T) {
+	lockRing := lock_manager.NewLockRing(5 * time.Second)
+	lockRing.SetSnapshot([]pb.ServerAddress{"filer1:8888"}, 0)
+
+	var deleted []string
+	mock := &mockFilerOps{
+		countFn:  func(util.FullPath) (int, error) { return 0, nil },
+		deleteFn: func(p util.FullPath) error { deleted = append(deleted, string(p)); return nil },
+	}
+
+	cleaner := &EmptyFolderCleaner{
+		filer:                 mock,
+		lockRing:              lockRing,
+		host:                  "filer1:8888",
+		bucketPath:            "/buckets",
+		enabled:               true,
+		maxCountCheck:         DefaultMaxCountCheck,
+		cacheExpiry:           DefaultCacheExpiry,
+		folderCounts:          make(map[string]*folderState),
+		bucketCleanupPolicies: make(map[string]*bucketCleanupPolicyState),
+		cleanupQueue:          NewCleanupQueue(1000, 10*time.Minute),
+		stopCh:                make(chan struct{}),
+	}
+
+	uploads := "/buckets/mybucket/" + s3_constants.MultipartUploadsFolder
+	cleaner.executeCleanup(uploads, "abc123")
+	cleaner.executeCleanup(uploads+"/abc123", "0001.part")
+	if len(deleted) != 0 {
+		t.Fatalf("multipart staging paths must not be deleted, got %v", deleted)
+	}
+
+	// An empty normal folder is deleted, proving the guard did not disable cleanup.
+	cleaner.executeCleanup("/buckets/mybucket/folder", "file.txt")
+	if len(deleted) != 1 || deleted[0] != "/buckets/mybucket/folder" {
+		t.Fatalf("normal empty folder should be deleted, got %v", deleted)
+	}
+}
+
 func Test_autoRemoveEmptyFoldersEnabled(t *testing.T) {
 	tests := []struct {
 		name      string