fix: wrapped io.EOF during ListObjects() (#17842)

When listing getObjectFileInfo can return `io.EOF` if file is being written.

When we wrap the error it will *not* retry upstream, since `io.EOF` is a valid return value.

Allow one retry before returning errors and canceling the listing.
This commit is contained in:
Klaus Post 2023-08-11 09:47:16 -07:00 committed by GitHub
parent 6c59b33fb1
commit 96a22bfcbb
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23

View File

@ -436,31 +436,29 @@ func (er *erasureObjects) streamMetadataParts(ctx context.Context, o listPathOpt
break break
} }
} }
retryWait := func() {
retries++
if retries == 1 {
time.Sleep(retryDelay)
} else {
time.Sleep(retryDelay250)
}
}
// Load first part metadata... // Load first part metadata...
// Read metadata associated with the object from all disks. // Read metadata associated with the object from all disks.
fi, metaArr, onlineDisks, err := er.getObjectFileInfo(ctx, minioMetaBucket, o.objectPath(0), ObjectOptions{}, true) fi, metaArr, onlineDisks, err := er.getObjectFileInfo(ctx, minioMetaBucket, o.objectPath(0), ObjectOptions{}, true)
if err != nil { if err != nil {
switch toObjectErr(err, minioMetaBucket, o.objectPath(0)).(type) { switch toObjectErr(err, minioMetaBucket, o.objectPath(0)).(type) {
case ObjectNotFound: case ObjectNotFound, InsufficientReadQuorum:
retries++ retryWait()
if retries == 1 {
time.Sleep(retryDelay)
} else {
time.Sleep(retryDelay250)
}
continue continue
case InsufficientReadQuorum:
retries++
if retries == 1 {
time.Sleep(retryDelay)
} else {
time.Sleep(retryDelay250)
}
continue
default:
return entries, fmt.Errorf("reading first part metadata: %w", err)
} }
// Allow one fast retry for other errors.
if retries > 0 {
return entries, fmt.Errorf("reading first part metadata: %v", err)
}
retryWait()
continue
} }
partN, err := o.findFirstPart(fi) partN, err := o.findFirstPart(fi)
@ -474,8 +472,7 @@ func (er *erasureObjects) streamMetadataParts(ctx context.Context, o listPathOpt
} }
retries = -1 retries = -1
} }
retries++ retryWait()
time.Sleep(retryDelay250)
continue continue
case errors.Is(err, io.EOF): case errors.Is(err, io.EOF):
return entries, io.EOF return entries, io.EOF