diff --git a/cmd/serve/s3/backend.go b/cmd/serve/s3/backend.go index 731195900..cdd44d98e 100644 --- a/cmd/serve/s3/backend.go +++ b/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 diff --git a/cmd/serve/s3/multipart.go b/cmd/serve/s3/multipart.go index 35e94433c..32eb0ff50 100644 --- a/cmd/serve/s3/multipart.go +++ b/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 diff --git a/cmd/serve/s3/multipart_test.go b/cmd/serve/s3/multipart_test.go index e671bc953..aee900bc2 100644 --- a/cmd/serve/s3/multipart_test.go +++ b/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) +} diff --git a/cmd/serve/s3/utils.go b/cmd/serve/s3/utils.go index 0d778dc0d..e5f21a14c 100644 --- a/cmd/serve/s3/utils.go +++ b/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) {