fix(storage): also remove the empty object directory when the xl.meta commit fails

The xl.meta rename at the end of RenameData creates the object directory
the same way the data rename does, and its failure branch left that
directory behind for writes without a data directory, such as inline
and zero-byte objects. Remove it and its empty parents there too, with
the same trailing-slash call that PR #286 adds to the data branch for
issue #285.

Making the cleanup safe against concurrent writes needs two more
changes:

- deleteFile now walks up with a trailing slash, so each parent step
  removes only an empty directory. Between two steps of a walk, another
  walk can remove the rest of the chain and a writer can commit an
  object whose xl.meta sits at the path the first walk removes next,
  for example the object "parent" under the key "parent/xl.meta/child".
  The walk used to delete that file. Ordinary deletes could already do
  this; the new cleanup made it reachable with failed writes.

- The two renames of RenameData retry when the directory creation or
  the rename reports "not found" while the source is still in place,
  up to once per path level. A late, failed RenameData can now remove
  the directory another write of the same key has just created, and
  that write used to fail on the drive although its data was intact.
  The retry decides on the error of the attempt itself, because another
  write can recreate the directories in between. renameAll's error
  mapping moves unchanged into renameAllErr so that the retry can see
  the raw error.

The stagedDataGoneDisks test helper now swaps the drives behind the
set's disk list under the lock that the list takes, instead of
replacing the getDisks callback that background tasks read unlocked.

Tests:
- TestXLStorageRenameDataCommitMetaMissingRemovesEmptyDirs: zero-byte
  and inline commit failures leave no empty directories, a sibling and
  a moved data directory stay.
- TestXLStorageWalkUpKeepsCommittedObject: a committed object at a
  parent path survives the walk, through DeleteVersion and both
  cleanup branches.
- TestRenameDataSurvivesConcurrentEmptyDirCleanup: PutObject (large
  and inline) and CompleteMultipartUpload succeed when a late failed
  RenameData of the same key removes their fresh directory on all 16
  drives.
- TestXLStorageRenameDataRetriesWhileCleanupWalksUp: a write survives
  a cleanup that removes each level of its path in turn, also when a
  nested-key write recreates the directories in between.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Signed-off-by: Feng Ruohang <rh@vonng.com>

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)
+			}
+		})
+	}
+}