diff --git a/cmd/erasure-object_test.go b/cmd/erasure-object_test.go
index 14a30d3bb..bb0f896e2 100644
--- a/cmd/erasure-object_test.go
+++ b/cmd/erasure-object_test.go
@@ -28,10 +28,15 @@ import (
 	"os"
 	"path/filepath"
 	"runtime"
+	"slices"
 	"strconv"
+	"sync"
+	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/dustin/go-humanize"
+	"github.com/minio/madmin-go/v3"
 	"github.com/minio/minio/internal/config/storageclass"
 )
 
@@ -1302,3 +1307,247 @@ func TestJoinErrs(t *testing.T) {
 		}
 	}
 }
+
+type stagedDataGoneDisk struct {
+	StorageAPI
+}
+
+func (d stagedDataGoneDisk) RenameData(ctx context.Context, srcVolume, srcPath string, fi FileInfo, dstVolume, dstPath string, opts RenameOptions) (RenameDataResp, error) {
+	if err := d.Delete(ctx, srcVolume, srcPath, DeleteOptions{Recursive: true, Immediate: true}); err != nil {
+		return RenameDataResp{}, err
+	}
+	return d.StorageAPI.RenameData(ctx, srcVolume, srcPath, fi, dstVolume, dstPath, opts)
+}
+
+// stagedDataGoneDisks makes every drive of the only set lose the staged data
+// of a write right before RenameData, as when the caller has already given up
+// and removed it. It returns a function that restores the drives.
+func stagedDataGoneDisks(z *erasureServerPools) func() {
+	// Swap the drives behind the set's getDisks under the mutex that
+	// getDisks takes, so that background readers do not race the swap.
+	sets := z.serverPools[0]
+	sets.erasureDisksMu.Lock()
+	disks := sets.erasureDisks[0]
+	original := slices.Clone(disks)
+	for i, disk := range disks {
+		disks[i] = stagedDataGoneDisk{StorageAPI: disk}
+	}
+	sets.erasureDisksMu.Unlock()
+	return func() {
+		sets.erasureDisksMu.Lock()
+		copy(sets.erasureDisks[0], original)
+		sets.erasureDisksMu.Unlock()
+	}
+}
+
+// A write whose staged data is gone before the drives rename it fails. The
+// object directory the drives create for it must not stay behind: it is
+// invisible while its parent holds other objects, but once they are deleted
+// the parent can no longer be removed and is listed as an empty prefix.
+func TestRenameDataStagedDataGoneLeavesNoPrefix(t *testing.T) {
+	ctx := t.Context()
+	obj, dirs, err := prepareErasure16(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer obj.Shutdown(context.Background())
+	defer removeRoots(dirs)
+	z := obj.(*erasureServerPools)
+	const bucket = "bucket"
+	const sibling, putObject, mpuObject = "day=1/b=9/sibling", "day=1/b=9/put", "day=1/b=9/mpu"
+	if err := z.MakeBucket(ctx, bucket, MakeBucketOptions{}); err != nil {
+		t.Fatal(err)
+	}
+	body := bytes.Repeat([]byte{'a'}, 8<<20) // above the inline limit: data goes to a data dir
+	if _, err := z.PutObject(ctx, bucket, sibling, mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), ObjectOptions{}); err != nil {
+		t.Fatal(err)
+	}
+	res, err := z.NewMultipartUpload(ctx, bucket, mpuObject, ObjectOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	pi, err := z.PutObjectPart(ctx, bucket, mpuObject, res.UploadID, 1, mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), ObjectOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	restore := stagedDataGoneDisks(z)
+	if _, err := z.PutObject(ctx, bucket, putObject, mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), ObjectOptions{}); err == nil {
+		t.Error("PutObject succeeded without its staged data")
+	}
+	if _, err := z.CompleteMultipartUpload(ctx, bucket, mpuObject, res.UploadID, []CompletePart{{PartNumber: 1, ETag: pi.ETag}}, ObjectOptions{}); err == nil {
+		t.Error("CompleteMultipartUpload succeeded without its staged data")
+	}
+	restore()
+
+	if _, err := z.DeleteObject(ctx, bucket, sibling, ObjectOptions{}); err != nil {
+		t.Fatal(err)
+	}
+	for _, prefix := range []string{"", "day=1/", "day=1/b=9/"} {
+		res, err := z.ListObjectsV2(ctx, bucket, prefix, "", "/", 1000, false, "")
+		if err != nil {
+			t.Fatal(err)
+		}
+		if len(res.Objects) != 0 || len(res.Prefixes) != 0 {
+			t.Errorf("ListObjectsV2 %q: %d objects, prefixes %v, want none", prefix, len(res.Objects), res.Prefixes)
+		}
+	}
+	for _, dir := range dirs {
+		if _, err := os.Stat(filepath.Join(dir, bucket, "day=1")); !os.IsNotExist(err) {
+			t.Errorf("%s: day=1 is still on the drive (%v)", dir, err)
+		}
+	}
+}
+
+// A failed overwrite whose staged data is gone must keep the existing object.
+func TestRenameDataStagedDataGoneKeepsExistingObject(t *testing.T) {
+	ctx := t.Context()
+	obj, dirs, err := prepareErasure16(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer obj.Shutdown(context.Background())
+	defer removeRoots(dirs)
+	z := obj.(*erasureServerPools)
+	const bucket, object = "bucket", "day=1/b=9/object"
+	if err := z.MakeBucket(ctx, bucket, MakeBucketOptions{}); err != nil {
+		t.Fatal(err)
+	}
+	body := bytes.Repeat([]byte{'a'}, 8<<20) // above the inline limit: data goes to a data dir
+	if _, err := z.PutObject(ctx, bucket, object, mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), ObjectOptions{}); err != nil {
+		t.Fatal(err)
+	}
+
+	restore := stagedDataGoneDisks(z)
+	other := bytes.Repeat([]byte{'b'}, 8<<20)
+	if _, err := z.PutObject(ctx, bucket, object, mustGetPutObjReader(t, bytes.NewReader(other), int64(len(other)), "", ""), ObjectOptions{}); err == nil {
+		t.Error("overwrite succeeded without its staged data")
+	}
+	restore()
+
+	var buf bytes.Buffer
+	if err := GetObject(ctx, z, bucket, object, 0, int64(len(body)), &buf, "", ObjectOptions{}); err != nil {
+		t.Fatal(err)
+	}
+	if !bytes.Equal(buf.Bytes(), body) {
+		t.Error("existing object content changed")
+	}
+}
+
+// A failed RenameData that runs late removes its object directory while it is
+// empty. When it does so on a drive between the directory creation and the
+// rename of another write of the same key, that write must create the
+// directory again and succeed instead of failing on the drive.
+func TestRenameDataSurvivesConcurrentEmptyDirCleanup(t *testing.T) {
+	if runtime.GOOS == globalWindowsOSName {
+		t.Skip("directory creation is not traced on Windows")
+	}
+	for _, tc := range []struct {
+		name      string
+		size      int
+		multipart bool
+	}{
+		{"put", 8 << 20, false},
+		{"inline_put", 1 << 10, false},
+		{"multipart", 8 << 20, true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx := t.Context()
+			obj, dirs, err := prepareErasure16(ctx)
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer obj.Shutdown(context.Background())
+			defer removeRoots(dirs)
+			z := obj.(*erasureServerPools)
+			const bucket, object = "bucket", "day=1/b=9/object"
+			if err := z.MakeBucket(ctx, bucket, MakeBucketOptions{}); err != nil {
+				t.Fatal(err)
+			}
+			body := bytes.Repeat([]byte{'a'}, tc.size)
+			var uploadID, etag string
+			if tc.multipart {
+				res, err := z.NewMultipartUpload(ctx, bucket, object, ObjectOptions{})
+				if err != nil {
+					t.Fatal(err)
+				}
+				pi, err := z.PutObjectPart(ctx, bucket, object, res.UploadID, 1, mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), ObjectOptions{})
+				if err != nil {
+					t.Fatal(err)
+				}
+				uploadID, etag = res.UploadID, pi.ETag
+			}
+
+			drives := make(map[string]*xlStorage)
+			for _, disk := range z.serverPools[0].sets[0].getDisks() {
+				s := disk.(*xlStorageDiskIDCheck).storage
+				drives[filepath.Join(s.drivePath, bucket, filepath.FromSlash(object))] = s
+			}
+			var mu sync.Mutex
+			hit := make(map[string]bool)
+			var hits atomic.Int32
+			var lateErrs []error
+			traceOSCalls(t, func(tr madmin.TraceInfo) {
+				if tr.FuncName != "os.Mkdir" || tr.Error != "" {
+					return
+				}
+				s := drives[tr.Path]
+				if s == nil {
+					return
+				}
+				mu.Lock()
+				first := !hit[tr.Path]
+				hit[tr.Path] = true
+				mu.Unlock()
+				if !first {
+					return
+				}
+				// The write has just created the object directory on this
+				// drive. A late RenameData of the same key, whose staged data
+				// is gone, fails now and removes the empty directory.
+				fi := newFileInfo(object, 8, 8)
+				fi.DataDir, fi.VersionID = mustGetUUID(), mustGetUUID()
+				fi.Erasure.Index = 1
+				fi.ModTime = time.Now().UTC()
+				fi.Size = 1
+				fi.AddObjectPart(1, "", 1, 1, fi.ModTime, nil, nil)
+				_, err := s.RenameData(context.Background(), minioMetaTmpBucket, mustGetUUID(), fi, bucket, object, RenameOptions{})
+				if _, serr := os.Stat(tr.Path); !os.IsNotExist(serr) {
+					err = fmt.Errorf("late RenameData left %s (%v, rename error %v)", tr.Path, serr, err)
+				} else if errors.Is(err, errFileNotFound) {
+					err = nil
+				}
+				mu.Lock()
+				if err != nil {
+					lateErrs = append(lateErrs, err)
+				}
+				mu.Unlock()
+				hits.Add(1)
+			})
+
+			if tc.multipart {
+				_, err = z.CompleteMultipartUpload(ctx, bucket, object, uploadID, []CompletePart{{PartNumber: 1, ETag: etag}}, ObjectOptions{})
+			} else {
+				_, err = z.PutObject(ctx, bucket, object, mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), ObjectOptions{})
+			}
+			if n := hits.Load(); n != int32(len(dirs)) {
+				t.Fatalf("late cleanup ran on %d drives, want %d", n, len(dirs))
+			}
+			mu.Lock()
+			for _, lateErr := range lateErrs {
+				t.Error(lateErr)
+			}
+			mu.Unlock()
+			if err != nil {
+				t.Fatalf("write failed after a concurrent cleanup: %v", err)
+			}
+			var buf bytes.Buffer
+			if err := GetObject(ctx, z, bucket, object, 0, int64(len(body)), &buf, "", ObjectOptions{}); err != nil {
+				t.Fatal(err)
+			}
+			if !bytes.Equal(buf.Bytes(), body) {
+				t.Fatal("object content differs")
+			}
+		})
+	}
+}
diff --git a/cmd/os-reliable.go b/cmd/os-reliable.go
index 3561cbd33..b333af63c 100644
--- a/cmd/os-reliable.go
+++ b/cmd/os-reliable.go
@@ -142,32 +142,37 @@ func renameAll(srcFilePath, dstFilePath, baseDir string) (err error) {
 		return err
 	}
 
-	if err = reliableRename(srcFilePath, dstFilePath, baseDir); err != nil {
-		switch {
-		case isSysErrNotDir(err) && !osIsNotExist(err):
-			// Windows can have both isSysErrNotDir(err) and osIsNotExist(err) returning
-			// true if the source file path contains an non-existent directory. In that case,
-			// we want to return errFileNotFound instead, which will honored in subsequent
-			// switch cases
-			return errFileAccessDenied
-		case isSysErrPathNotFound(err):
-			// This is a special case should be handled only for
-			// windows, because windows API does not return "not a
-			// directory" error message. Handle this specifically here.
-			return errFileAccessDenied
-		case isSysErrCrossDevice(err):
-			return fmt.Errorf("%w (%s)->(%s)", errCrossDeviceLink, srcFilePath, dstFilePath)
-		case osIsNotExist(err):
-			return errFileNotFound
-		case osIsExist(err):
-			// This is returned only when destination is a directory and we
-			// are attempting a rename from file to directory.
-			return errIsNotRegular
-		default:
-			return err
-		}
+	return renameAllErr(reliableRename(srcFilePath, dstFilePath, baseDir), srcFilePath, dstFilePath)
+}
+
+// renameAllErr converts an error of reliableRename into the error that
+// renameAll returns.
+func renameAllErr(err error, srcFilePath, dstFilePath string) error {
+	switch {
+	case err == nil:
+		return nil
+	case isSysErrNotDir(err) && !osIsNotExist(err):
+		// Windows can have both isSysErrNotDir(err) and osIsNotExist(err) returning
+		// true if the source file path contains an non-existent directory. In that case,
+		// we want to return errFileNotFound instead, which will honored in subsequent
+		// switch cases
+		return errFileAccessDenied
+	case isSysErrPathNotFound(err):
+		// This is a special case should be handled only for
+		// windows, because windows API does not return "not a
+		// directory" error message. Handle this specifically here.
+		return errFileAccessDenied
+	case isSysErrCrossDevice(err):
+		return fmt.Errorf("%w (%s)->(%s)", errCrossDeviceLink, srcFilePath, dstFilePath)
+	case osIsNotExist(err):
+		return errFileNotFound
+	case osIsExist(err):
+		// This is returned only when destination is a directory and we
+		// are attempting a rename from file to directory.
+		return errIsNotRegular
+	default:
+		return err
 	}
-	return nil
 }
 
 // Reliably retries os.RenameAll if for some reason os.RenameAll returns
diff --git a/cmd/xl-storage-errors_test.go b/cmd/xl-storage-errors_test.go
index b9d12fc4d..547ab6812 100644
--- a/cmd/xl-storage-errors_test.go
+++ b/cmd/xl-storage-errors_test.go
@@ -18,6 +18,7 @@
 package cmd
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"net/http"
@@ -25,9 +26,12 @@ import (
 	"path/filepath"
 	"runtime"
 	"sync"
+	"sync/atomic"
 	"syscall"
 	"testing"
 	"time"
+
+	"github.com/minio/madmin-go/v3"
 )
 
 func TestSysErrors(t *testing.T) {
@@ -179,3 +183,346 @@ func TestXLStorageRenameDataCommitMetaMissing(t *testing.T) {
 		})
 	}
 }
+
+// A failed commit must not leave behind the object directory that renameAll()
+// created for it when nothing was moved into it: once the other objects under
+// its parent are gone, it keeps the parent and every ancestor listed as an
+// empty prefix. A data directory that was already moved keeps the directory.
+func TestXLStorageRenameDataCommitMetaMissingRemovesEmptyDirs(t *testing.T) {
+	const object = "day=1/b=9/object"
+	for _, tc := range []struct {
+		name    string
+		data    string // "" (zero bytes), "inline" or "dir"
+		sibling bool   // another object under day=1/b=9/
+	}{
+		{"empty_object", "", false},
+		{"inline_object_with_sibling", "inline", true},
+		{"data_directory", "dir", false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			disk, root, err := newXLStorageTestSetup(t)
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer disk.Close()
+			s := disk.storage
+			if err := s.MakeVol(t.Context(), "bucket"); err != nil {
+				t.Fatal(err)
+			}
+			bucketDir := filepath.Join(root, "bucket")
+			objDir := filepath.Join(bucketDir, filepath.FromSlash(object))
+			siblingMeta := filepath.Join(bucketDir, "day=1", "b=9", "sibling", xlStorageFormatFile)
+			if tc.sibling {
+				if err := os.MkdirAll(filepath.Dir(siblingMeta), 0o755); err != nil {
+					t.Fatal(err)
+				}
+				if err := os.WriteFile(siblingMeta, []byte("sibling"), 0o644); err != nil {
+					t.Fatal(err)
+				}
+			}
+			fi := newFileInfo(object, 2, 2)
+			fi.DataDir, fi.VersionID = mustGetUUID(), mustGetUUID()
+			fi.Erasure.Index = 1
+			fi.ModTime = time.Now().UTC()
+			fi.Fresh = true
+			ctx := &missingCommitMetaContext{
+				Context: t.Context(),
+				meta:    filepath.Join(root, minioMetaTmpBucket, "staged", xlStorageFormatFile),
+			}
+			switch tc.data {
+			case "inline":
+				fi.Size = 1
+				fi.Data = []byte("a")
+			case "dir":
+				fi.Size = 1
+				fi.AddObjectPart(1, "", 1, 1, fi.ModTime, nil, nil)
+				part := filepath.Join(root, minioMetaTmpBucket, "staged", fi.DataDir, "part.1")
+				if err := os.MkdirAll(filepath.Dir(part), 0o755); err != nil {
+					t.Fatal(err)
+				}
+				if err := os.WriteFile(part, []byte("a"), 0o644); err != nil {
+					t.Fatal(err)
+				}
+				ctx.dataDest = filepath.Join(objDir, fi.DataDir)
+			}
+			_, err = s.RenameData(ctx, minioMetaTmpBucket, "staged", fi, "bucket", object, RenameOptions{})
+			if !ctx.removed {
+				t.Fatalf("metadata removal did not execute: removal error = %v, RenameData error = %v", ctx.removeErr, err)
+			}
+			if !errors.Is(err, errCommitMetaMissing) {
+				t.Fatalf("expected commit metadata missing, got %v", err)
+			}
+
+			if tc.data == "dir" {
+				// The moved data directory is not deleted, so the object
+				// directory stays and holds only it.
+				entries, err := os.ReadDir(objDir)
+				if err != nil {
+					t.Fatal(err)
+				}
+				if len(entries) != 1 || entries[0].Name() != fi.DataDir {
+					t.Fatalf("object directory holds %v, want only data directory %s", entries, fi.DataDir)
+				}
+				if _, err := os.Stat(filepath.Join(objDir, fi.DataDir, "part.1")); err != nil {
+					t.Fatalf("moved data must be kept: %v", err)
+				}
+				return
+			}
+			if _, err := os.Stat(objDir); !os.IsNotExist(err) {
+				t.Fatalf("empty object directory left after a failed commit (%v)", err)
+			}
+			if tc.sibling {
+				if _, err := os.Stat(siblingMeta); err != nil {
+					t.Fatalf("sibling object must be kept: %v", err)
+				}
+			} else if _, err := os.Stat(filepath.Join(bucketDir, "day=1")); !os.IsNotExist(err) {
+				t.Fatalf("empty parent day=1 left after a failed commit (%v)", err)
+			}
+			if _, err := os.Stat(bucketDir); err != nil {
+				t.Fatalf("bucket directory must be kept: %v", err)
+			}
+		})
+	}
+}
+
+// traceOSCalls calls hook synchronously after every instrumented OS call
+// until the test ends, so that a test can run other storage calls at an
+// exact point inside a storage call without changing production code.
+func traceOSCalls(t *testing.T, hook func(madmin.TraceInfo)) {
+	t.Helper()
+	before := globalTrace.Subscribers()
+	done := make(chan struct{})
+	if err := globalTrace.Subscribe(madmin.TraceOS, make(chan madmin.TraceInfo, 1), done, func(tr madmin.TraceInfo) bool {
+		hook(tr)
+		return false
+	}); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		close(done)
+		// Wait for the unsubscribe, which takes the pubsub write lock, so
+		// that it cannot block a nested trace of a later test.
+		for globalTrace.Subscribers() > before {
+			time.Sleep(time.Millisecond)
+		}
+	})
+}
+
+// The walk up from a removed path removes only empty directories. Another
+// walk, for example the cleanup of a second failed write of the same key,
+// can remove the rest of the chain while this one is between two steps, and
+// a writer can then commit an object whose xl.meta is the path this walk
+// removes next. That object must stay.
+func TestXLStorageWalkUpKeepsCommittedObject(t *testing.T) {
+	const nested, parent = "parent/xl.meta/child", "parent"
+	for _, tc := range []string{"delete_version", "data_rename", "commit"} {
+		t.Run(tc, func(t *testing.T) {
+			disk, root, err := newXLStorageTestSetup(t)
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer disk.Close()
+			s := disk.storage
+			ctx := t.Context()
+			if err := s.MakeVol(ctx, "bucket"); err != nil {
+				t.Fatal(err)
+			}
+			newFI := func(object string) FileInfo {
+				fi := newFileInfo(object, 2, 2)
+				fi.DataDir, fi.VersionID = mustGetUUID(), mustGetUUID()
+				fi.Erasure.Index = 1
+				fi.ModTime = time.Now().UTC()
+				fi.Fresh = true
+				return fi
+			}
+			inline := func(object string) FileInfo {
+				fi := newFI(object)
+				fi.Size, fi.Data = 1, []byte("x")
+				fi.AddObjectPart(1, "", 1, 1, fi.ModTime, nil, nil)
+				fi.SetInlineData()
+				return fi
+			}
+			childDir := filepath.Join(root, "bucket", filepath.FromSlash(nested))
+			metaDir := filepath.Dir(childDir)
+			parentDir := filepath.Dir(metaDir)
+
+			var nestedFI FileInfo
+			if tc == "delete_version" {
+				nestedFI = inline(nested)
+				if _, err := s.RenameData(ctx, minioMetaTmpBucket, "nested", nestedFI, "bucket", nested, RenameOptions{}); err != nil {
+					t.Fatal(err)
+				}
+			}
+			parentFI := inline(parent)
+			var fired atomic.Bool
+			var hookErr error
+			traceOSCalls(t, func(tr madmin.TraceInfo) {
+				if tr.FuncName != "os.Remove" || tr.Error != "" || filepath.Clean(tr.Path) != childDir || !fired.CompareAndSwap(false, true) {
+					return
+				}
+				// This walk has removed child/. Another one removes the rest
+				// of the chain, then a writer commits the object parent,
+				// whose metadata file is parent/xl.meta.
+				for _, dir := range []string{metaDir, parentDir} {
+					if err := os.Remove(dir); err != nil {
+						hookErr = err
+						return
+					}
+				}
+				if _, err := s.RenameData(context.Background(), minioMetaTmpBucket, "writer", parentFI, "bucket", parent, RenameOptions{}); err != nil {
+					hookErr = err
+				}
+			})
+
+			switch tc {
+			case "delete_version":
+				err = s.DeleteVersion(ctx, "bucket", nested, nestedFI, false, DeleteOptions{})
+			case "data_rename":
+				// The staged data directory is gone.
+				fi := newFI(nested)
+				fi.Size = 1
+				fi.AddObjectPart(1, "", 1, 1, fi.ModTime, nil, nil)
+				_, err = s.RenameData(ctx, minioMetaTmpBucket, "late", fi, "bucket", nested, RenameOptions{})
+			case "commit":
+				mctx := &missingCommitMetaContext{
+					Context: ctx,
+					meta:    filepath.Join(root, minioMetaTmpBucket, "late", xlStorageFormatFile),
+				}
+				_, err = s.RenameData(mctx, minioMetaTmpBucket, "late", newFI(nested), "bucket", nested, RenameOptions{})
+			}
+			if tc == "delete_version" && err != nil {
+				t.Fatal(err)
+			}
+			if tc != "delete_version" && err == nil {
+				t.Fatal("RenameData succeeded without its staged data")
+			}
+			if !fired.Load() {
+				t.Fatalf("%s did not remove the empty directory %s", tc, nested)
+			}
+			if hookErr != nil {
+				t.Fatal(hookErr)
+			}
+			got, err := s.ReadVersion(ctx, "", "bucket", parent, parentFI.VersionID, ReadOptions{ReadData: true})
+			if err != nil {
+				t.Fatalf("committed object %s lost: %v", parent, err)
+			}
+			if !bytes.Equal(got.Data, parentFI.Data) {
+				t.Fatalf("committed object %s has data %q, want %q", parent, got.Data, parentFI.Data)
+			}
+		})
+	}
+}
+
+// A cleanup that walks up removes each empty directory of a write's
+// destination path once. When it removes them one by one while the write
+// creates them, so that every directory the write has just made or found is
+// gone before it uses it, the write must keep retrying until the walk has
+// passed and then succeed. That holds also when another write recreates the
+// directories after a failed attempt, which must not hide the failure.
+func TestXLStorageRenameDataRetriesWhileCleanupWalksUp(t *testing.T) {
+	if runtime.GOOS == globalWindowsOSName {
+		t.Skip("directory creation is not traced on Windows")
+	}
+	const object = "a/b/object"
+	for _, tc := range []struct {
+		name           string
+		inline, nested bool
+	}{
+		{"data_dir", false, false},
+		{"inline", true, false},
+		{"data_dir_nested_writer", false, true},
+	} {
+		inline := tc.inline
+		t.Run(tc.name, func(t *testing.T) {
+			disk, root, err := newXLStorageTestSetup(t)
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer disk.Close()
+			s := disk.storage
+			ctx := t.Context()
+			if err := s.MakeVol(ctx, "bucket"); err != nil {
+				t.Fatal(err)
+			}
+			fi := newFileInfo(object, 2, 2)
+			fi.DataDir, fi.VersionID = mustGetUUID(), mustGetUUID()
+			fi.Erasure.Index = 1
+			fi.ModTime = time.Now().UTC()
+			fi.Fresh = true
+			fi.Size = 1
+			fi.AddObjectPart(1, "", 1, 1, fi.ModTime, nil, nil)
+			if inline {
+				fi.Data = []byte("x")
+				fi.SetInlineData()
+			} else {
+				part := filepath.Join(root, minioMetaTmpBucket, "staged", fi.DataDir, "part.1")
+				if err := os.MkdirAll(filepath.Dir(part), 0o755); err != nil {
+					t.Fatal(err)
+				}
+				if err := os.WriteFile(part, []byte("x"), 0o644); err != nil {
+					t.Fatal(err)
+				}
+			}
+			objDir := filepath.Join(root, "bucket", filepath.FromSlash(object))
+			bDir := filepath.Dir(objDir)
+			aDir := filepath.Dir(bDir)
+			// The walk removes object/, then b/, then a/. Each step happens
+			// right after the write's next directory creation reached the
+			// directory, whether it created it or found it in place.
+			remove := func(dir string) func() error { return func() error { return os.Remove(dir) } }
+			steps := []struct {
+				after string
+				do    func() error
+			}{
+				{objDir, remove(objDir)},
+				{bDir, remove(bDir)},
+				{aDir, remove(aDir)},
+			}
+			if tc.nested {
+				// The next creation of b/ fails. A write of the key
+				// a/b/object/child then creates the directories again.
+				steps = append(steps, struct {
+					after string
+					do    func() error
+				}{bDir, func() error { return os.MkdirAll(filepath.Join(objDir, "child"), 0o755) }})
+			}
+			var mu sync.Mutex
+			step := 0
+			var hookErr error
+			traceOSCalls(t, func(tr madmin.TraceInfo) {
+				if tr.FuncName != "os.Mkdir" {
+					return
+				}
+				mu.Lock()
+				defer mu.Unlock()
+				if step == len(steps) || filepath.Clean(tr.Path) != steps[step].after {
+					return
+				}
+				if err := steps[step].do(); err != nil && hookErr == nil {
+					hookErr = err
+				}
+				step++
+			})
+			_, err = s.RenameData(ctx, minioMetaTmpBucket, "staged", fi, "bucket", object, RenameOptions{})
+			mu.Lock()
+			done := step
+			mu.Unlock()
+			if hookErr != nil {
+				t.Fatal(hookErr)
+			}
+			if done != len(steps) {
+				t.Fatalf("the walk made %d of %d steps", done, len(steps))
+			}
+			if err != nil {
+				t.Fatalf("write failed while a cleanup walked up its path: %v", err)
+			}
+			got, err := s.ReadVersion(ctx, "", "bucket", object, fi.VersionID, ReadOptions{ReadData: true})
+			if err != nil {
+				t.Fatal(err)
+			}
+			if got.Size != fi.Size {
+				t.Fatalf("committed size %d, want %d", got.Size, fi.Size)
+			}
+		})
+	}
+}
diff --git a/cmd/xl-storage.go b/cmd/xl-storage.go
index 0758396b6..81d9a2cf3 100644
--- a/cmd/xl-storage.go
+++ b/cmd/xl-storage.go
@@ -2511,8 +2511,11 @@ func (s *xlStorage) deleteFile(basePath, deletePath string, recursive, immediate
 	}
 
 	// Delete parent directory obviously not recursively. Errors for
-	// parent directories shouldn't trickle down.
-	s.deleteFile(basePath, pathutil.Dir(pathutil.Clean(deletePath)), false, false)
+	// parent directories shouldn't trickle down. The trailing slash
+	// removes only an empty directory: a concurrent writer can have
+	// replaced the parent with a file, for example the xl.meta of an
+	// object whose name is that parent path, and it must stay.
+	s.deleteFile(basePath, pathutil.Dir(pathutil.Clean(deletePath))+SlashSeparator, false, false)
 
 	return nil
 }
@@ -2580,6 +2583,40 @@ func skipAccessChecks(volume string) (ok bool) {
 	return strings.HasPrefix(volume, minioMetaBucket)
 }
 
+// renameDataAll is renameAll for the two renames of RenameData. A failed
+// RenameData, typically one that runs after its caller gave up, removes its
+// object directory and the parents that are empty, so it can remove the
+// directories that renameAll is creating for this rename. The directory
+// creation or the rename then reports "not found" although the source is
+// still in place: retry, which creates the directories again. A cleanup
+// removes each directory of the path at most once and each failed attempt
+// needs at least one such removal, so one retry per path level is enough
+// against any single cleanup. The decision uses the error of the attempt
+// itself, since another write can recreate the directories in between.
+func renameDataAll(srcPath, dstPath, skipParent string) error {
+	if srcPath == "" || dstPath == "" {
+		return errInvalidArgument
+	}
+	if err := checkPathLength(srcPath); err != nil {
+		return err
+	}
+	if err := checkPathLength(dstPath); err != nil {
+		return err
+	}
+	retries := strings.Count(strings.TrimPrefix(pathutil.Clean(dstPath), pathutil.Clean(skipParent)), SlashSeparator)
+	for {
+		err := reliableRename(srcPath, dstPath, skipParent)
+		if err == nil || retries == 0 || !osIsNotExist(err) {
+			return renameAllErr(err, srcPath, dstPath)
+		}
+		if _, serr := Lstat(pathutil.Clean(srcPath)); serr != nil {
+			// The source is gone: a late call, which the caller cleans up.
+			return renameAllErr(err, srcPath, dstPath)
+		}
+		retries--
+	}
+}
+
 // RenameData - rename source path to destination path atomically, metadata and data directory.
 func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath string, fi FileInfo, dstVolume, dstPath string, opts RenameOptions) (res RenameDataResp, err error) {
 	defer func() {
@@ -2886,13 +2923,18 @@ func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath string, f
 		if contextCanceled(ctx) {
 			return res, ctx.Err()
 		}
-		if err = renameAll(srcDataPath, dstDataPath, skipParent); err != nil {
+		if err = renameDataAll(srcDataPath, dstDataPath, skipParent); err != nil {
 			if legacyPreserved {
 				// Any failed rename calls un-roll previous transaction.
 				s.deleteFile(dstVolumeDir, legacyDataPath, true, false)
 			}
 			// if its a partial rename() do not attempt to delete recursively.
 			s.deleteFile(dstVolumeDir, dstDataPath, false, false)
+			// renameAll() creates the object directory before the rename; if the
+			// staged data was already gone, drop it and its parents while empty,
+			// otherwise they outlive the other objects there and are listed as an
+			// empty prefix. The trailing slash removes only an empty directory.
+			s.deleteFile(dstVolumeDir, pathutil.Dir(dstFilePath)+SlashSeparator, false, false)
 			return res, osErrToFileErr(err)
 		}
 		diskHealthCheckOK(ctx, err)
@@ -2920,7 +2962,7 @@ func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath string, f
 	}
 
 	// Commit meta-file
-	if err = renameAll(srcFilePath, dstFilePath, skipParent); err != nil {
+	if err = renameDataAll(srcFilePath, dstFilePath, skipParent); err != nil {
 		if legacyPreserved {
 			// Any failed rename calls un-roll previous transaction.
 			s.deleteFile(dstVolumeDir, legacyDataPath, true, false)
@@ -2928,6 +2970,11 @@ func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath string, f
 		// if its a partial rename() do not attempt to delete recursively.
 		// this can be healed since all parts are available.
 		s.deleteFile(dstVolumeDir, dstDataPath, false, false)
+		// As with the data rename above, renameAll() may have created the
+		// object directory; drop it and its parents while empty. A moved
+		// data directory keeps it, since the trailing slash removes only an
+		// empty directory.
+		s.deleteFile(dstVolumeDir, pathutil.Dir(dstFilePath)+SlashSeparator, false, false)
 		err = osErrToFileErr(err)
 		if errors.Is(err, errFileNotFound) {
 			// The staged xl.meta disappeared between staging and commit.