Fix truncated list response on deleted replicated objects (#16504)

This commit is contained in:
Klaus Post 2023-01-30 18:13:53 +01:00 committed by GitHub
parent b923a62425
commit f713436dd0
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 73 additions and 2 deletions

View file

@ -189,6 +189,43 @@ func (e *metaCacheEntry) isLatestDeletemarker() bool {
return xlMeta.versions[0].header.Type == DeleteType
}
// isAllFreeVersions returns if all objects are free versions.
// If metadata is NOT versioned false will always be returned.
// If v2 and UNABLE to load metadata true will be returned.
func (e *metaCacheEntry) isAllFreeVersions() bool {
if e.cached != nil {
if len(e.cached.versions) == 0 {
return true
}
for _, v := range e.cached.versions {
if !v.header.FreeVersion() {
return false
}
}
return true
}
if !isXL2V1Format(e.metadata) {
return false
}
if meta, _, err := isIndexedMetaV2(e.metadata); meta != nil {
return meta.AllHidden(false)
} else if err != nil {
return true
}
// Fall back...
xlMeta, err := e.xlmeta()
if err != nil || len(xlMeta.versions) == 0 {
return true
}
// Check versions..
for _, v := range e.cached.versions {
if !v.header.FreeVersion() {
return false
}
}
return true
}
// fileInfo returns the decoded metadata.
// If entry is a directory it is returned as that.
// If versioned the latest version will be returned.

View file

@ -367,7 +367,10 @@ func (r *metacacheReader) filter(o listPathOptions) (entries metaCacheEntriesSor
return true
}
if !o.InclDeleted && entry.isObject() && entry.isLatestDeletemarker() && !entry.isObjectDir() {
return entries.len() < o.Limit
return true
}
if entry.isAllFreeVersions() {
return true
}
entries.o = append(entries.o, entry)
return entries.len() < o.Limit

View file

@ -93,7 +93,7 @@ func (s *xlStorage) WalkDir(ctx context.Context, opts WalkDirOptions, wr io.Writ
if opts.Limit <= 0 {
return
}
if m, _, _ := isIndexedMetaV2(metadata); m != nil && !m.IsLatestDeleteMarker() {
if m, _, _ := isIndexedMetaV2(metadata); m != nil && !m.AllHidden(true) {
objsReturned++
}
}

View file

@ -2129,3 +2129,34 @@ func (x xlMetaBuf) IsLatestDeleteMarker() bool {
})
return isDeleteMarker
}
// AllHidden returns true are no versions that would show up in a listing (ie all free markers)
// Optionally also return early if top is a delete marker.
func (x xlMetaBuf) AllHidden(topDeleteMarker bool) bool {
vers, headerV, _, buf, err := decodeXLHeaders(x)
if err != nil {
return false
}
if vers == 0 {
return true
}
hidden := true
var xl xlMetaV2VersionHeader
_ = decodeVersions(buf, vers, func(idx int, hdr, _ []byte) error {
if _, err := xl.unmarshalV(headerV, hdr); err != nil {
return errDoneForNow
}
if topDeleteMarker && idx == 0 && xl.Type == DeleteType {
hidden = true
return errDoneForNow
}
if !xl.FreeVersion() {
hidden = false
return errDoneForNow
}
// Check next version
return nil
})
return hidden
}