From a2fd8caa69a3b545ad33f1b721884efa03070a9d Mon Sep 17 00:00:00 2001
From: Klaus Post <klauspost@gmail.com>
Date: Thu, 13 Jan 2022 14:28:07 -0800
Subject: [PATCH] Ignore version not found in deleteVersions (#14093)

When deleting multiple versions it "gives" up with an errFileVersionNotFound if
a version cannot be found. This effectively skips deleting other versions
sent in the same request.

This can happen on inconsistent objects. We should ignore errFileVersionNotFound
and continue with others.

We already ignore these at the caller level, this PR is continuation of 54a9877
---
 cmd/xl-storage.go      | 10 +++++
 cmd/xl-storage_test.go | 94 ++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 104 insertions(+)

diff --git a/cmd/xl-storage.go b/cmd/xl-storage.go
index 74496b59c..9ea010248 100644
--- a/cmd/xl-storage.go
+++ b/cmd/xl-storage.go
@@ -888,13 +888,19 @@ func (s *xlStorage) deleteVersions(ctx context.Context, volume, path string, fis
 	var (
 		dataDir     string
 		lastVersion bool
+		updated     bool
 	)
 
 	for _, fi := range fis {
 		dataDir, lastVersion, err = xlMeta.DeleteVersion(fi)
 		if err != nil {
+			if !fi.Deleted && (err == errFileNotFound || err == errFileVersionNotFound) {
+				// Ignore these since
+				continue
+			}
 			return err
 		}
+		updated = true
 		if dataDir != "" {
 			versionID := fi.VersionID
 			if versionID == "" {
@@ -918,6 +924,10 @@ func (s *xlStorage) deleteVersions(ctx context.Context, volume, path string, fis
 	}
 
 	if !lastVersion {
+		if !updated {
+			return nil
+		}
+
 		buf, err = xlMeta.AppendTo(metaDataPoolGet())
 		defer metaDataPoolPut(buf)
 		if err != nil {
diff --git a/cmd/xl-storage_test.go b/cmd/xl-storage_test.go
index e76fe7e01..04ef07664 100644
--- a/cmd/xl-storage_test.go
+++ b/cmd/xl-storage_test.go
@@ -30,6 +30,8 @@ import (
 	"strings"
 	"syscall"
 	"testing"
+
+	"github.com/google/uuid"
 )
 
 func TestCheckPathLength(t *testing.T) {
@@ -1665,6 +1667,98 @@ func TestXLStorageRenameFile(t *testing.T) {
 	}
 }
 
+// TestXLStorageDeleteVersion will test if version deletes and bulk deletes work as expected.
+func TestXLStorageDeleteVersion(t *testing.T) {
+	// create xlStorage test setup
+	xl, path, err := newXLStorageTestSetup()
+	if err != nil {
+		t.Fatalf("Unable to create xlStorage test setup, %s", err)
+	}
+	defer os.RemoveAll(path)
+	ctx := context.Background()
+
+	volume := "myvol-vol"
+	object := "my-object"
+	if err := xl.MakeVol(ctx, volume); err != nil {
+		t.Fatalf("Unable to create volume, %s", err)
+	}
+	var versions [50]string
+	for i := range versions {
+		versions[i] = uuid.New().String()
+		fi := FileInfo{
+			Name: object, Volume: volume, VersionID: versions[i], ModTime: UTCNow(), DataDir: uuid.NewString(), Size: 10000,
+			Erasure: ErasureInfo{
+				Algorithm:    erasureAlgorithm,
+				DataBlocks:   4,
+				ParityBlocks: 4,
+				BlockSize:    blockSizeV2,
+				Index:        1,
+				Distribution: []int{0, 1, 2, 3, 4, 5, 6, 7},
+				Checksums:    nil,
+			},
+		}
+		if err := xl.WriteMetadata(ctx, volume, object, fi); err != nil {
+			t.Fatalf("Unable to create object, %s", err)
+		}
+	}
+	var deleted [len(versions)]bool
+	checkVerExist := func(t testing.TB) {
+		t.Helper()
+		for i := range versions {
+			shouldExist := !deleted[i]
+			fi, err := xl.ReadVersion(ctx, volume, object, versions[i], false)
+			if shouldExist {
+				if err != nil {
+					t.Fatalf("Version %s should exist, but got err %v", versions[i], err)
+				}
+				return
+			}
+			if err != errFileVersionNotFound {
+				t.Fatalf("Version %s should not exist, but returned: %#v", versions[i], fi)
+			}
+		}
+	}
+
+	// Delete version 0...
+	checkVerExist(t)
+	err = xl.DeleteVersion(ctx, volume, object, FileInfo{Name: object, Volume: volume, VersionID: versions[0]}, false)
+	if err != nil {
+		t.Fatal(err)
+	}
+	deleted[0] = true
+	checkVerExist(t)
+
+	// Delete 10 in bulk, including a non-existing.
+	fis := []FileInfoVersions{{Name: object, Volume: volume}}
+	for i := range versions[:10] {
+		fis[0].Versions = append(fis[0].Versions, FileInfo{Name: object, Volume: volume, VersionID: versions[i]})
+		deleted[i] = true
+	}
+	errs := xl.DeleteVersions(ctx, volume, fis)
+	if errs[0] != nil {
+		t.Fatalf("expected nil error, got %v", errs[0])
+	}
+	checkVerExist(t)
+
+	// Delete them all... (some again)
+	fis[0].Versions = nil
+	for i := range versions[:] {
+		fis[0].Versions = append(fis[0].Versions, FileInfo{Name: object, Volume: volume, VersionID: versions[i]})
+		deleted[i] = true
+	}
+	errs = xl.DeleteVersions(ctx, volume, fis)
+	if errs[0] != nil {
+		t.Fatalf("expected nil error, got %v", errs[0])
+	}
+	checkVerExist(t)
+
+	// Meta should be deleted now...
+	fi, err := xl.ReadVersion(ctx, volume, object, "", false)
+	if err != errFileNotFound {
+		t.Fatalf("Object %s should not exist, but returned: %#v", object, fi)
+	}
+}
+
 // TestXLStorage xlStorage.StatInfoFile()
 func TestXLStorageStatInfoFile(t *testing.T) {
 	// create xlStorage test setup