heal: Single object heal to look for older versions as well (#203) (#20723)

`mc admin heal ALIAS/bucket/object` does not have any flag to heal
object noncurrent versions, this commit will make healing of the object
noncurrent versions implicitly asked.

This also fixes the 'mc admin heal ALIAS/bucket/object' that does not work 
correctly when the bucket is versioned. This has been broken since Apr 2023.

cmd/admin-heal-ops.go

@@ -841,6 +841,7 @@ func (h *healSequence) healMinioSysMeta(objAPI ObjectLayer, metaPrefix string) f
 		// NOTE: Healing on meta is run regardless
 		// of any bucket being selected, this is to ensure that
 		// meta are always upto date and correct.
+		h.settings.Recursive = true
 		return objAPI.HealObjects(h.ctx, minioMetaBucket, metaPrefix, h.settings, func(bucket, object, versionID string, scanMode madmin.HealScanMode) error {
 			if h.isQuitting() {
 				return errHealStopSignalled
@@ -896,16 +897,6 @@ func (h *healSequence) healBucket(objAPI ObjectLayer, bucket string, bucketsOnly
 		return nil
 	}
 
-	if !h.settings.Recursive {
-		if h.object != "" {
-			if err := h.healObject(bucket, h.object, "", h.settings.ScanMode); err != nil {
-				return err
-			}
-		}
-
-		return nil
-	}
-
 	if err := objAPI.HealObjects(h.ctx, bucket, h.object, h.settings, h.healObject); err != nil {
 		return errFnHealFromAPIErr(h.ctx, err)
 	}

cmd/erasure-healing.go

@@ -44,7 +44,8 @@ const (
 	healingMetricCheckAbandonedParts
 )
 
-func (er erasureObjects) listAndHeal(ctx context.Context, bucket, prefix string, scanMode madmin.HealScanMode, healEntry func(string, metaCacheEntry, madmin.HealScanMode) error) error {
+// List a prefix or a single object versions and heal
+func (er erasureObjects) listAndHeal(ctx context.Context, bucket, prefix string, recursive bool, scanMode madmin.HealScanMode, healEntry func(string, metaCacheEntry, madmin.HealScanMode) error) error {
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
@@ -77,11 +78,14 @@ func (er erasureObjects) listAndHeal(ctx context.Context, bucket, prefix string,
 		bucket:         bucket,
 		path:           path,
 		filterPrefix:   filterPrefix,
-		recursive:      true,
+		recursive:      recursive,
 		forwardTo:      "",
 		minDisks:       1,
 		reportNotFound: false,
 		agreed: func(entry metaCacheEntry) {
+			if !recursive && prefix != entry.name {
+				return
+			}
 			if err := healEntry(bucket, entry, scanMode); err != nil {
 				cancel()
 			}
@@ -93,7 +97,9 @@ func (er erasureObjects) listAndHeal(ctx context.Context, bucket, prefix string,
 				// proceed to heal nonetheless.
 				entry, _ = entries.firstFound()
 			}
-
+			if !recursive && prefix != entry.name {
+				return
+			}
 			if err := healEntry(bucket, *entry, scanMode); err != nil {
 				cancel()
 				return

cmd/erasure-healing_test.go

@@ -733,7 +733,7 @@ func TestHealingDanglingObject(t *testing.T) {
 		t.Fatalf("Expected versions 1, got %d", fileInfoPreHeal.NumVersions)
 	}
 
-	if err = objLayer.HealObjects(ctx, bucket, "", madmin.HealOpts{Remove: true},
+	if err = objLayer.HealObjects(ctx, bucket, "", madmin.HealOpts{Recursive: true, Remove: true},
 		func(bucket, object, vid string, scanMode madmin.HealScanMode) error {
 			_, err := objLayer.HealObject(ctx, bucket, object, vid, madmin.HealOpts{ScanMode: scanMode, Remove: true})
 			return err
@@ -780,7 +780,7 @@ func TestHealingDanglingObject(t *testing.T) {
 		t.Fatalf("Expected versions 1, got %d", fileInfoPreHeal.NumVersions)
 	}
 
-	if err = objLayer.HealObjects(ctx, bucket, "", madmin.HealOpts{Remove: true},
+	if err = objLayer.HealObjects(ctx, bucket, "", madmin.HealOpts{Recursive: true, Remove: true},
 		func(bucket, object, vid string, scanMode madmin.HealScanMode) error {
 			_, err := objLayer.HealObject(ctx, bucket, object, vid, madmin.HealOpts{ScanMode: scanMode, Remove: true})
 			return err
@@ -829,7 +829,7 @@ func TestHealingDanglingObject(t *testing.T) {
 		t.Fatalf("Expected versions 3, got %d", fileInfoPreHeal.NumVersions)
 	}
 
-	if err = objLayer.HealObjects(ctx, bucket, "", madmin.HealOpts{Remove: true},
+	if err = objLayer.HealObjects(ctx, bucket, "", madmin.HealOpts{Recursive: true, Remove: true},
 		func(bucket, object, vid string, scanMode madmin.HealScanMode) error {
 			_, err := objLayer.HealObject(ctx, bucket, object, vid, madmin.HealOpts{ScanMode: scanMode, Remove: true})
 			return err

cmd/erasure-server-pool.go

@@ -2478,6 +2478,7 @@ func (z *erasureServerPools) Walk(ctx context.Context, bucket, prefix string, re
 // HealObjectFn closure function heals the object.
 type HealObjectFn func(bucket, object, versionID string, scanMode madmin.HealScanMode) error
 
+// List a prefix or a single object versions and heal
 func (z *erasureServerPools) HealObjects(ctx context.Context, bucket, prefix string, opts madmin.HealOpts, healObjectFn HealObjectFn) error {
 	healEntry := func(bucket string, entry metaCacheEntry, scanMode madmin.HealScanMode) error {
 		if entry.isDir() {
@@ -2541,7 +2542,7 @@ func (z *erasureServerPools) HealObjects(ctx context.Context, bucket, prefix str
 			go func(idx int, set *erasureObjects) {
 				defer wg.Done()
 
-				errs[idx] = set.listAndHeal(ctx, bucket, prefix, opts.ScanMode, healEntry)
+				errs[idx] = set.listAndHeal(ctx, bucket, prefix, opts.Recursive, opts.ScanMode, healEntry)
 			}(idx, set)
 		}
 		wg.Wait()