serve s3: fix buckets being undeletable after an aborted multipart upload

Starting a multipart upload makes the directory its object will be in,
but aborting the upload didn't remove it. The empty directory isn't
visible to S3 clients, so the bucket looked empty, but deleting it
failed with BucketNotEmpty.

Now aborting an upload removes the directories made for it if they are
still empty, and deleting a bucket holding nothing but empty
directories removes them and succeeds. Deleting a bucket which fails
for some other reason than holding objects now returns that error
rather than BucketNotEmpty.

Removing the empty directories left after deleting an object now
checks they are empty first, rather than logging an error for each
directory which isn't.

cmd/serve/s3/backend.go

@@ -4,6 +4,7 @@ package s3
 import (
 	"context"
 	"encoding/hex"
+	"errors"
 	"io"
 	"maps"
 	"math"
@@ -471,7 +472,7 @@ func (b *s3Backend) PutObject(
 	// }
 
 	if objectDir != "." {
-		if err := mkdirRecursive(objectDir, _vfs); err != nil {
+		if _, err := mkdirRecursive(objectDir, _vfs); err != nil {
 			return result, err
 		}
 	}
@@ -625,8 +626,13 @@ func (b *s3Backend) DeleteBucket(ctx context.Context, name string) error {
 		return gofakes3.BucketNotFound(name)
 	}
 
-	if err := _vfs.Remove(name); err != nil {
-		return gofakes3.ErrBucketNotEmpty
+	// The bucket may hold empty directories, which S3 clients can't
+	// see, so is removed if it has nothing else.
+	if err := removeEmptyTree(name, _vfs); err != nil {
+		if errors.Is(err, errNotEmpty) {
+			return gofakes3.ErrBucketNotEmpty
+		}
+		return err
 	}
 
 	return nil

cmd/serve/s3/multipart.go

@@ -58,6 +58,7 @@ type multipartUpload struct {
 	bucket, key string
 	fp          string // final object path
 	streamFp    string // path the parts are written to (fp when the remote has no server-side move or copy)
+	madeDir     string // top-most directory made for the upload, or "" if none
 	meta        map[string]string
 
 	fh     io.WriteCloser // sink the in-order parts are written to, opened by sink (guarded by mu)
@@ -293,8 +294,10 @@ func (b *s3Backend) CreateMultipartUpload(ctx context.Context, bucketName, objec
 	}
 
 	objectDir := path.Dir(fp)
+	var madeDir string
 	if objectDir != "." {
-		if err := mkdirRecursive(objectDir, _vfs); err != nil {
+		madeDir, err = mkdirRecursive(objectDir, _vfs)
+		if err != nil {
 			b.releaseUpload(t)
 			return "", err
 		}
@@ -315,6 +318,7 @@ func (b *s3Backend) CreateMultipartUpload(ctx context.Context, bucketName, objec
 	up := newMultipartUpload(bucketName, objectName, fp, streamFp, meta, int64(b.s.opt.MultipartStreamingBufferLimit))
 	up.budget = t.budget
 	up.vfs = _vfs
+	up.madeDir = madeDir
 
 	if err := b.addUpload(t, uploadID, up); err != nil {
 		b.releaseUpload(t)
@@ -1020,10 +1024,19 @@ func (b *s3Backend) forgetAllUploads() {
 // remote commits the partial data instead - see abort.)
 func (b *s3Backend) discardUpload(up *multipartUpload) {
 	b.forgetPath(up.vfs, up.streamFp)
-	if up.streamFp == up.fp {
-		return
+	if up.streamFp != up.fp {
+		_ = up.vfs.Remove(up.streamFp)
+	}
+	// Remove the directories made for the upload if they are still
+	// empty, as an empty directory left behind would stop the bucket
+	// being deleted.
+	if up.madeDir != "" {
+		for dir := path.Dir(up.fp); isEmptyDir(dir, up.vfs); dir = path.Dir(dir) {
+			if up.vfs.Remove(dir) != nil || dir == up.madeDir {
+				break
+			}
+		}
 	}
-	_ = up.vfs.Remove(up.streamFp)
 }
 
 // forgetPath invalidates the parent directory's cached VFS listing so that

cmd/serve/s3/multipart_test.go

@@ -1864,3 +1864,49 @@ func TestMultipartMaxUploads(t *testing.T) {
 	_, err = b.CreateMultipartUpload(ctx, bucket, "5", nil)
 	require.NoError(t, err)
 }
+
+// TestMultipartAbortRemovesDirs checks that aborting an upload removes
+// the directories made for it, but not those which existed already, so
+// the bucket can be deleted.
+func TestMultipartAbortRemovesDirs(t *testing.T) {
+	core, f, bucket := newMultipartTestServer(t, false)
+	ctx := context.Background()
+
+	// A directory which exists already with an object in
+	putTestObject(t, f, bucket, "existing/keep.txt")
+
+	for _, object := range []string{"new/dir/aborted.bin", "existing/dir/aborted.bin"} {
+		uploadID, err := core.NewMultipartUpload(ctx, bucket, object, minio.PutObjectOptions{})
+		require.NoError(t, err)
+		data := []byte(random.String(1024))
+		_, err = core.PutObjectPart(ctx, bucket, object, uploadID, 1, bytes.NewReader(data), int64(len(data)), minio.PutObjectPartOptions{})
+		require.NoError(t, err)
+		require.NoError(t, core.AbortMultipartUpload(ctx, bucket, object, uploadID))
+	}
+	requireOnly(t, f, bucket, "existing")
+	requireOnly(t, f, path.Join(bucket, "existing"), "keep.txt")
+
+	// Once the object is gone the bucket can be deleted
+	require.NoError(t, core.RemoveObject(ctx, bucket, "existing/keep.txt", minio.RemoveObjectOptions{}))
+	require.NoError(t, core.RemoveBucket(ctx, bucket))
+}
+
+// TestDeleteBucketEmptyDirs checks a bucket holding only empty
+// directories, which S3 clients can't see, can be deleted, but not one
+// holding an object.
+func TestDeleteBucketEmptyDirs(t *testing.T) {
+	core, f, bucket := newMultipartTestServer(t, false)
+	ctx := context.Background()
+
+	putTestObject(t, f, bucket, "a/b/object.txt")
+	require.NoError(t, f.Mkdir(ctx, path.Join(bucket, "c/d")))
+	err := core.RemoveBucket(ctx, bucket)
+	require.Error(t, err)
+	assert.Equal(t, "BucketNotEmpty", minio.ToErrorResponse(err).Code)
+	requireOnly(t, f, bucket, "a", "c")
+
+	require.NoError(t, core.RemoveObject(ctx, bucket, "a/b/object.txt", minio.RemoveObjectOptions{}))
+	require.NoError(t, core.RemoveBucket(ctx, bucket))
+	_, err = f.List(ctx, bucket)
+	assert.ErrorIs(t, err, fs.ErrorDirNotFound)
+}

cmd/serve/s3/utils.go

@@ -138,8 +138,12 @@ func prefixParser(p *gofakes3.Prefix) (path, remaining string) {
 	return p.Prefix[:idx], p.Prefix[idx+1:]
 }
 
+// mkdirRecursive makes the directory path and any of its parents which
+// don't exist, returning the top-most directory it made, or "" if path
+// existed already.
+//
 // FIXME this could be implemented by VFS.MkdirAll()
-func mkdirRecursive(path string, VFS *vfs.VFS) error {
+func mkdirRecursive(path string, VFS *vfs.VFS) (made string, err error) {
 	path = strings.Trim(path, "/")
 	dirs := strings.Split(path, "/")
 	var dir strings.Builder
@@ -148,26 +152,86 @@ func mkdirRecursive(path string, VFS *vfs.VFS) error {
 		if _, err := VFS.Stat(dir.String()); err != nil {
 			err := VFS.Mkdir(dir.String(), 0777)
 			if err != nil {
-				return err
+				return made, err
+			}
+			if made == "" {
+				made = strings.TrimPrefix(dir.String(), "/")
 			}
 		}
 	}
-	return nil
+	return made, nil
 }
 
+// isEmptyDir reports whether dir is an empty directory.
+func isEmptyDir(dir string, VFS *vfs.VFS) bool {
+	entries, err := VFS.ReadDir(dir)
+	return err == nil && len(entries) == 0
+}
+
+// rmdirRecursive removes the parent directories of p which are empty,
+// stopping at the bucket.
 func rmdirRecursive(p string, VFS *vfs.VFS) {
 	dir := path.Dir(p)
 	if !strings.ContainsAny(dir, "/\\") {
 		// might be bucket(root)
 		return
 	}
-	if _, err := VFS.Stat(dir); err == nil {
-		err := VFS.Remove(dir)
-		if err != nil {
-			return
-		}
-		rmdirRecursive(dir, VFS)
+	// Check the directory is empty first as the VFS logs an error
+	// removing one which isn't
+	if !isEmptyDir(dir, VFS) {
+		return
 	}
+	if err := VFS.Remove(dir); err != nil {
+		return
+	}
+	rmdirRecursive(dir, VFS)
+}
+
+// errNotEmpty is returned by removeEmptyTree for a directory holding
+// files.
+var errNotEmpty = errors.New("directory not empty")
+
+// isEmptyTree reports whether dir holds nothing but empty directories,
+// returning errNotEmpty if it holds any files.
+func isEmptyTree(dir string, VFS *vfs.VFS) error {
+	entries, err := VFS.ReadDir(dir)
+	if err != nil {
+		return err
+	}
+	for _, entry := range entries {
+		if !entry.IsDir() {
+			return errNotEmpty
+		}
+		if err := isEmptyTree(path.Join(dir, entry.Name()), VFS); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// removeEmptyTree removes dir if it holds nothing but empty
+// directories, along with them, returning errNotEmpty if it holds any
+// files.
+func removeEmptyTree(dir string, VFS *vfs.VFS) error {
+	// Check the whole tree first so nothing is removed if it isn't
+	// empty
+	if err := isEmptyTree(dir, VFS); err != nil {
+		return err
+	}
+	var remove func(dir string) error
+	remove = func(dir string) error {
+		entries, err := VFS.ReadDir(dir)
+		if err != nil {
+			return err
+		}
+		for _, entry := range entries {
+			if err := remove(path.Join(dir, entry.Name())); err != nil {
+				return err
+			}
+		}
+		return VFS.Remove(dir)
+	}
+	return remove(dir)
 }
 
 func authlistResolver(list []string) (map[string]string, error) {