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
This commit is contained in:
Klaus Post 2022-01-13 14:28:07 -08:00 committed by GitHub
parent f546636c52
commit a2fd8caa69
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 104 additions and 0 deletions

View File

@ -888,13 +888,19 @@ func (s *xlStorage) deleteVersions(ctx context.Context, volume, path string, fis
var ( var (
dataDir string dataDir string
lastVersion bool lastVersion bool
updated bool
) )
for _, fi := range fis { for _, fi := range fis {
dataDir, lastVersion, err = xlMeta.DeleteVersion(fi) dataDir, lastVersion, err = xlMeta.DeleteVersion(fi)
if err != nil { if err != nil {
if !fi.Deleted && (err == errFileNotFound || err == errFileVersionNotFound) {
// Ignore these since
continue
}
return err return err
} }
updated = true
if dataDir != "" { if dataDir != "" {
versionID := fi.VersionID versionID := fi.VersionID
if versionID == "" { if versionID == "" {
@ -918,6 +924,10 @@ func (s *xlStorage) deleteVersions(ctx context.Context, volume, path string, fis
} }
if !lastVersion { if !lastVersion {
if !updated {
return nil
}
buf, err = xlMeta.AppendTo(metaDataPoolGet()) buf, err = xlMeta.AppendTo(metaDataPoolGet())
defer metaDataPoolPut(buf) defer metaDataPoolPut(buf)
if err != nil { if err != nil {

View File

@ -30,6 +30,8 @@ import (
"strings" "strings"
"syscall" "syscall"
"testing" "testing"
"github.com/google/uuid"
) )
func TestCheckPathLength(t *testing.T) { 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() // TestXLStorage xlStorage.StatInfoFile()
func TestXLStorageStatInfoFile(t *testing.T) { func TestXLStorageStatInfoFile(t *testing.T) {
// create xlStorage test setup // create xlStorage test setup