fix: healing to list, purge dangling objects (#621)

in a specific corner case when you only have dangling
objects with single shard left over, we end up a situation
where healing is unable to list this dangling object to
purge due to the fact that listing logic expected only
`len(disks)/2+1` - where as when you make this choice you
end up with a situation that the drive where this object
is present is not part of your expected disks list, causing
it to be never listed and ignored into perpetuity.

change the logic such that HealObjects() would be able
to listAndHeal() per set properly on all its drives, since
there is really no other way to do this cleanly, however
instead of "listing" on all erasure sets simultaneously, we
list on '3' at a time. So in a large enough cluster this is
fairly staggered.

cmd/erasure-healing.go

@@ -55,10 +55,6 @@ func (er erasureObjects) listAndHeal(ctx context.Context, bucket, prefix string,
 		return errors.New("listAndHeal: No non-healing drives found")
 	}
 
-	expectedDisks := len(disks)/2 + 1
-	fallbackDisks := disks[expectedDisks:]
-	disks = disks[:expectedDisks]
-
 	// How to resolve partial results.
 	resolver := metadataResolutionParams{
 		dirQuorum: 1,
@@ -75,7 +71,6 @@ func (er erasureObjects) listAndHeal(ctx context.Context, bucket, prefix string,
 
 	lopts := listPathRawOptions{
 		disks:          disks,
-		fallbackDisks:  fallbackDisks,
 		bucket:         bucket,
 		path:           path,
 		filterPrefix:   filterPrefix,

cmd/erasure-server-pool.go

@@ -45,6 +45,7 @@ import (
 	"github.com/minio/minio/internal/logger"
 	"github.com/minio/pkg/v3/sync/errgroup"
 	"github.com/minio/pkg/v3/wildcard"
+	"github.com/minio/pkg/v3/workers"
 	"github.com/puzpuzpuz/xsync/v3"
 )
 
@@ -2467,7 +2468,7 @@ func (z *erasureServerPools) HealObjects(ctx context.Context, bucket, prefix str
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
-	var poolErrs [][]error
+	poolErrs := make([][]error, len(z.serverPools))
 	for idx, erasureSet := range z.serverPools {
 		if opts.Pool != nil && *opts.Pool != idx {
 			continue
@@ -2476,20 +2477,20 @@ func (z *erasureServerPools) HealObjects(ctx context.Context, bucket, prefix str
 			continue
 		}
 		errs := make([]error, len(erasureSet.sets))
-		var wg sync.WaitGroup
+		wk, _ := workers.New(3)
 		for idx, set := range erasureSet.sets {
 			if opts.Set != nil && *opts.Set != idx {
 				continue
 			}
-			wg.Add(1)
+			wk.Take()
 			go func(idx int, set *erasureObjects) {
-				defer wg.Done()
+				defer wk.Give()
 
 				errs[idx] = set.listAndHeal(ctx, bucket, prefix, opts.Recursive, opts.ScanMode, healEntry)
 			}(idx, set)
 		}
-		wg.Wait()
-		poolErrs = append(poolErrs, errs)
+		wk.Wait()
+		poolErrs[idx] = errs
 	}
 	for _, errs := range poolErrs {
 		for _, err := range errs {

cmd/global-heal.go

@@ -352,10 +352,6 @@ func (er *erasureObjects) healErasureSet(ctx context.Context, buckets []string,
 			disks[i], disks[j] = disks[j], disks[i]
 		})
 
-		expectedDisks := len(disks)/2 + 1
-		fallbackDisks := disks[expectedDisks:]
-		disks = disks[:expectedDisks]
-
 		filterLifecycle := func(bucket, object string, fi FileInfo) bool {
 			if lc == nil {
 				return false
@@ -518,7 +514,6 @@ func (er *erasureObjects) healErasureSet(ctx context.Context, buckets []string,
 
 		err = listPathRaw(ctx, listPathRawOptions{
 			disks:          disks,
-			fallbackDisks:  fallbackDisks,
 			bucket:         bucket,
 			recursive:      true,
 			forwardTo:      forwardTo,
@@ -540,7 +535,7 @@ func (er *erasureObjects) healErasureSet(ctx context.Context, buckets []string,
 			},
 			finished: func(errs []error) {
 				success := countErrs(errs, nil)
-				if success < expectedDisks {
+				if success < len(disks)/2+1 {
 					retErr = fmt.Errorf("one or more errors reported during listing: %v", errors.Join(errs...))
 				}
 			},