diff --git a/xl-erasure-v1.go b/xl-erasure-v1.go index 6b8dc3457..a70aa3365 100644 --- a/xl-erasure-v1.go +++ b/xl-erasure-v1.go @@ -558,25 +558,29 @@ func (xl XL) DeleteFile(volume, path string) error { versions, err := listFileVersions(partsMetadata, errs) // Get highest file version. higherVersion := highestInt(versions) - // Increment to have next higher version. - higherVersion++ - // Take last meta data to use later - var mdata xlMetaV1 - onlineDisksCount := 0 + // find online disks and meta data of higherVersion + var mdata *xlMetaV1 + onlineDiskCount := 0 for index, metadata := range partsMetadata { if errs[index] == nil { - mdata = metadata - onlineDisksCount++ + onlineDiskCount++ + if metadata.Stat.Version == higherVersion && mdata == nil { + mdata = &metadata + } } } - if onlineDisksCount < xl.writeQuorum { + // return error if mdata is empty or onlineDiskCount doesn't meet write quorum + if mdata == nil || onlineDiskCount < xl.writeQuorum { return errWriteQuorum } + // Increment to have next higher version. + higherVersion++ + xlMetaV1FilePath := slashpath.Join(path, xlMetaV1File) - deleteMetaData := (onlineDisksCount == len(xl.storageDisks)) + deleteMetaData := (onlineDiskCount == len(xl.storageDisks)) // Set higher version to indicate file operation mdata.Stat.Version = higherVersion @@ -585,44 +589,52 @@ func (xl XL) DeleteFile(volume, path string) error { nsMutex.Lock(volume, path) defer nsMutex.Unlock(volume, path) - // Loop through and delete each chunks. + errCount := 0 + // Update meta data file and remove part file for index, disk := range xl.storageDisks { + // no need to operate on failed disks if errs[index] != nil { + errCount++ continue } - // TODO: errors can be allowed up to len(xl.storageDisks) - xl.writeQuorum - // TODO: Fix: meta data will be lost if deleteMetaData is true and anyone DeleteFile failure. + // update meta data about delete operation + var metadataWriter io.WriteCloser + metadataWriter, err = disk.CreateFile(volume, xlMetaV1FilePath) + if err != nil { + log.WithFields(logrus.Fields{ + "volume": volume, + "path": path, + }).Errorf("CreateFile failed with %s", err) - if deleteMetaData { - err = disk.DeleteFile(volume, xlMetaV1FilePath) - if err != nil { - log.WithFields(logrus.Fields{ - "volume": volume, - "path": path, - }).Errorf("DeleteFile failed with %s", err) - return err - } - } else { - var metadataWriter io.WriteCloser - metadataWriter, err = disk.CreateFile(volume, xlMetaV1FilePath) - if err != nil { - log.WithFields(logrus.Fields{ - "volume": volume, - "path": path, - }).Errorf("CreateFile failed with %s", err) - return err + errCount++ + + // We can safely allow CreateFile errors up to len(xl.storageDisks) - xl.writeQuorum + // otherwise return failure. + if errCount <= len(xl.storageDisks)-xl.writeQuorum { + continue } - err = mdata.Write(metadataWriter) - if err != nil { - log.WithFields(logrus.Fields{ - "volume": volume, - "path": path, - "diskIndex": index, - }).Errorf("Writing metadata failed with %s", err) - return err + return err + } + + err = mdata.Write(metadataWriter) + if err != nil { + log.WithFields(logrus.Fields{ + "volume": volume, + "path": path, + "diskIndex": index, + }).Errorf("Writing metadata failed with %s", err) + + errCount++ + + // We can safely allow CreateFile errors up to len(xl.storageDisks) - xl.writeQuorum + // otherwise return failure. + if errCount <= len(xl.storageDisks)-xl.writeQuorum { + continue } + + return err } erasureFilePart := slashpath.Join(path, fmt.Sprintf("file.%d", index)) @@ -632,9 +644,38 @@ func (xl XL) DeleteFile(volume, path string) error { "volume": volume, "path": path, }).Errorf("DeleteFile failed with %s", err) + + errCount++ + + // We can safely allow CreateFile errors up to len(xl.storageDisks) - xl.writeQuorum + // otherwise return failure. + if errCount <= len(xl.storageDisks)-xl.writeQuorum { + continue + } + return err } } + + // Remove meta data file only if deleteMetaData is true. + if deleteMetaData { + for index, disk := range xl.storageDisks { + // no need to operate on failed disks + if errs[index] != nil { + continue + } + + err = disk.DeleteFile(volume, xlMetaV1FilePath) + if err != nil { + // No need to return the error as we updated the meta data file previously + log.WithFields(logrus.Fields{ + "volume": volume, + "path": path, + }).Errorf("DeleteFile failed with %s", err) + } + } + } + return nil }