Do not pro-actively return false in isObjectDir() (#7246)

We should change the logic for both isObject()
and isObjectDir() leaf detection to be done
with quorum, due to how our directory navigation
works - this allows for properly deleting all
the dangling directories or objects if any.

cmd/xl-v1-common.go

@@ -20,8 +20,6 @@ import (
 	"context"
 	"path"
 	"sync"
-
-	"github.com/minio/minio/cmd/logger"
 )
 
 // getLoadBalancedDisks - fetches load balanced (sufficiently randomized) disk slice.
@@ -52,31 +50,44 @@ func (xl xlObjects) parentDirIsObject(ctx context.Context, bucket, parent string
 	return isParentDirObject(parent)
 }
 
-var xlTreeWalkIgnoredErrs = append(baseIgnoredErrs, errDiskAccessDenied, errVolumeNotFound, errFileNotFound)
-
 // isObject - returns `true` if the prefix is an object i.e if
 // `xl.json` exists at the leaf, false otherwise.
 func (xl xlObjects) isObject(bucket, prefix string) (ok bool) {
-	for _, disk := range xl.getLoadBalancedDisks() {
+	var errs = make([]error, len(xl.getDisks()))
+	var wg sync.WaitGroup
+	for index, disk := range xl.getDisks() {
 		if disk == nil {
 			continue
 		}
-		// Check if 'prefix' is an object on this 'disk', else continue the check the next disk
-		_, err := disk.StatFile(bucket, path.Join(prefix, xlMetaJSONFile))
-		if err == nil {
-			return true
-		}
-		// Ignore for file not found,  disk not found or faulty disk.
-		if IsErrIgnored(err, xlTreeWalkIgnoredErrs...) {
-			continue
-		}
-	} // Exhausted all disks - return false.
-	return false
+		wg.Add(1)
+		go func(index int, disk StorageAPI) {
+			defer wg.Done()
+			// Check if 'prefix' is an object on this 'disk', else continue the check the next disk
+			fi, err := disk.StatFile(bucket, path.Join(prefix, xlMetaJSONFile))
+			if err != nil {
+				errs[index] = err
+				return
+			}
+			if fi.Size == 0 {
+				errs[index] = errCorruptedFormat
+				return
+			}
+		}(index, disk)
+	}
+
+	wg.Wait()
+
+	// NOTE: Observe we are not trying to read `xl.json` and figure out the actual
+	// quorum intentionally, but rely on the default case scenario. Actual quorum
+	// verification will happen by top layer by using getObjectInfo() and will be
+	// ignored if necessary.
+	readQuorum := len(xl.getDisks()) / 2
+
+	return reduceReadQuorumErrs(context.Background(), errs, objectOpIgnoredErrs, readQuorum) == nil
 }
 
 // isObjectDir returns if the specified path represents an empty directory.
 func (xl xlObjects) isObjectDir(bucket, prefix string) (ok bool) {
-	var objectDirs = make([]bool, len(xl.getDisks()))
 	var errs = make([]error, len(xl.getDisks()))
 	var wg sync.WaitGroup
 	for index, disk := range xl.getDisks() {
@@ -92,7 +103,10 @@ func (xl xlObjects) isObjectDir(bucket, prefix string) (ok bool) {
 				errs[index] = err
 				return
 			}
-			objectDirs[index] = len(entries) == 0
+			if len(entries) > 0 {
+				errs[index] = errVolumeNotEmpty
+				return
+			}
 		}(index, disk)
 	}
 
@@ -100,22 +114,5 @@ func (xl xlObjects) isObjectDir(bucket, prefix string) (ok bool) {
 
 	readQuorum := len(xl.getDisks()) / 2
 
-	err := reduceReadQuorumErrs(context.Background(), errs, xlTreeWalkIgnoredErrs, readQuorum)
-	if err != nil {
-		reqInfo := &logger.ReqInfo{BucketName: bucket}
-		reqInfo.AppendTags("prefix", prefix)
-		ctx := logger.SetReqInfo(context.Background(), reqInfo)
-		logger.LogIf(ctx, err)
-		return false
-	}
-
-	for _, ok := range objectDirs {
-		if !ok {
-			// We perhaps found some entries in a prefix.
-			return false
-		}
-	}
-
-	// Return true if and only if the all directories are empty.
-	return true
+	return reduceReadQuorumErrs(context.Background(), errs, objectOpIgnoredErrs, readQuorum) == nil
 }