fix: invalidate outdated disks appropriately during readAllXL (#15002)

readAllXL would return inlined data for outdated disks
causing "read" to return incorrect content to the client,

this PR fixes this behavior by making sure we skip such
outdated disks appropriately based on the latest ModTime
on the disk.
This commit is contained in:
Harshavardhana 2022-05-30 12:43:54 -07:00 committed by GitHub
parent f1abb92f0c
commit d480022711
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23

View file

@ -509,23 +509,45 @@ func readAllXL(ctx context.Context, disks []StorageAPI, bucket, object string, r
} }
readQuorum := (len(disks) + 1) / 2 readQuorum := (len(disks) + 1) / 2
merged := mergeXLV2Versions(readQuorum, false, 1, metadataShallowVersions...) meta := &xlMetaV2{versions: mergeXLV2Versions(readQuorum, false, 1, metadataShallowVersions...)}
lfi, err := meta.ToFileInfo(bucket, object, "")
if err != nil {
for i := range errs {
if errs[i] == nil {
errs[i] = err
}
}
return metaFileInfos, errs
}
if !lfi.IsValid() {
for i := range errs {
if errs[i] == nil {
errs[i] = errCorruptedFormat
}
}
return metaFileInfos, errs
}
for index := range metadataArray { for index := range metadataArray {
if metadataArray[index] == nil { if metadataArray[index] == nil {
continue continue
} }
metadataArray[index].versions = merged
// make sure to preserve this for diskmtime based healing bugfix. // make sure to preserve this for diskmtime based healing bugfix.
diskMTime := metaFileInfos[index].DiskMTime diskMTime := metaFileInfos[index].DiskMTime
metaFileInfos[index], errs[index] = metadataArray[index].ToFileInfo(bucket, object, "") metaFileInfos[index], errs[index] = metadataArray[index].ToFileInfo(bucket, object, "")
if errs[index] == nil { if errs[index] != nil {
continue
}
if metaFileInfos[index].IsValid() && metaFileInfos[index].ModTime.Equal(lfi.ModTime) {
versionID := metaFileInfos[index].VersionID versionID := metaFileInfos[index].VersionID
if versionID == "" { if versionID == "" {
versionID = nullVersionID versionID = nullVersionID
} }
metaFileInfos[index].Data = metadataArray[index].data.find(versionID) if readData {
metaFileInfos[index].Data = metadataArray[index].data.find(versionID)
}
metaFileInfos[index].DiskMTime = diskMTime metaFileInfos[index].DiskMTime = diskMTime
} }
} }
@ -593,6 +615,8 @@ func (er erasureObjects) getObjectFileInfo(ctx context.Context, bucket, object s
if metaArr[i].IsValid() && metaArr[i].ModTime.Equal(fi.ModTime) { if metaArr[i].IsValid() && metaArr[i].ModTime.Equal(fi.ModTime) {
continue continue
} }
metaArr[i] = FileInfo{}
onlineDisks[i] = nil
missingBlocks++ missingBlocks++
} }