fix: make sure to delete dangling objects during heal (#13138)

heal with --remove was not removing dangling versions
on versioned buckets, this PR fixes this properly.

this is a regression introduced in PR #12617

cmd/erasure-healing_test.go

@@ -24,6 +24,7 @@ import (
 	"os"
 	"path"
 	"reflect"
+	"runtime"
 	"testing"
 	"time"
 
@@ -145,6 +146,150 @@ func TestHealing(t *testing.T) {
 	}
 }
 
+func TestHealingDanglingObject(t *testing.T) {
+	if runtime.GOOS == "windows" {
+		t.Skip()
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	resetGlobalHealState()
+	defer resetGlobalHealState()
+
+	nDisks := 16
+	fsDirs, err := getRandomDisks(nDisks)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	//defer removeRoots(fsDirs)
+
+	// Everything is fine, should return nil
+	objLayer, disks, err := initObjectLayer(ctx, mustGetPoolEndpoints(fsDirs...))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	bucket := getRandomBucketName()
+	object := getRandomObjectName()
+	data := bytes.Repeat([]byte("a"), 128*1024)
+
+	err = objLayer.MakeBucketWithLocation(ctx, bucket, BucketOptions{})
+	if err != nil {
+		t.Fatalf("Failed to make a bucket - %v", err)
+	}
+
+	// Enable versioning.
+	globalBucketMetadataSys.Update(bucket, bucketVersioningConfig, []byte(`<VersioningConfiguration><Status>Enabled</Status></VersioningConfiguration>`))
+
+	_, err = objLayer.PutObject(ctx, bucket, object, mustGetPutObjReader(t, bytes.NewReader(data), int64(len(data)), "", ""), ObjectOptions{
+		Versioned: true,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	for _, fsDir := range fsDirs[:4] {
+		if err = os.Chmod(fsDir, 0400); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	// Create delete marker under quorum.
+	objInfo, err := objLayer.DeleteObject(ctx, bucket, object, ObjectOptions{Versioned: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	for _, fsDir := range fsDirs[:4] {
+		if err = os.Chmod(fsDir, 0755); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	fileInfoPreHeal, err := disks[0].ReadVersion(context.Background(), bucket, object, "", false)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if fileInfoPreHeal.NumVersions != 1 {
+		t.Fatalf("Expected versions 1, got %d", fileInfoPreHeal.NumVersions)
+	}
+
+	if err = objLayer.HealObjects(ctx, bucket, "", madmin.HealOpts{Remove: true},
+		func(bucket, object, vid string) error {
+			_, err := objLayer.HealObject(ctx, bucket, object, vid, madmin.HealOpts{Remove: true})
+			return err
+		}); err != nil {
+		t.Fatal(err)
+	}
+
+	fileInfoPostHeal, err := disks[0].ReadVersion(context.Background(), bucket, object, "", false)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if fileInfoPostHeal.NumVersions != 2 {
+		t.Fatalf("Expected versions 2, got %d", fileInfoPreHeal.NumVersions)
+	}
+
+	if objInfo.DeleteMarker {
+		if _, err = objLayer.DeleteObject(ctx, bucket, object, ObjectOptions{
+			Versioned: true,
+			VersionID: objInfo.VersionID,
+		}); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	for _, fsDir := range fsDirs[:4] {
+		if err = os.Chmod(fsDir, 0400); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	rd := mustGetPutObjReader(t, bytes.NewReader(data), int64(len(data)), "", "")
+	_, err = objLayer.PutObject(ctx, bucket, object, rd, ObjectOptions{
+		Versioned: true,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	for _, fsDir := range fsDirs[:4] {
+		if err = os.Chmod(fsDir, 0755); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	fileInfoPreHeal, err = disks[0].ReadVersion(context.Background(), bucket, object, "", false)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if fileInfoPreHeal.NumVersions != 1 {
+		t.Fatalf("Expected versions 1, got %d", fileInfoPreHeal.NumVersions)
+	}
+
+	if err = objLayer.HealObjects(ctx, bucket, "", madmin.HealOpts{Remove: true},
+		func(bucket, object, vid string) error {
+			_, err := objLayer.HealObject(ctx, bucket, object, vid, madmin.HealOpts{Remove: true})
+			return err
+		}); err != nil {
+		t.Fatal(err)
+	}
+
+	fileInfoPostHeal, err = disks[0].ReadVersion(context.Background(), bucket, object, "", false)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if fileInfoPostHeal.NumVersions != 2 {
+		t.Fatalf("Expected versions 2, got %d", fileInfoPreHeal.NumVersions)
+	}
+}
+
 func TestHealObjectCorrupted(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -166,8 +311,8 @@ func TestHealObjectCorrupted(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	bucket := "bucket"
-	object := "object"
+	bucket := getRandomBucketName()
+	object := getRandomObjectName()
 	data := bytes.Repeat([]byte("a"), 5*1024*1024)
 	var opts ObjectOptions