fix retry on unauthorized chunk
This commit is contained in:
parent
f0f4943577
commit
bf146fb072
1 changed files with 25 additions and 18 deletions
|
@ -66,31 +66,39 @@ func uploadBlobChunked(ctx context.Context, requestURL *url.URL, layer *Layer, r
|
||||||
|
|
||||||
sectionReader := io.NewSectionReader(f, int64(offset), chunk)
|
sectionReader := io.NewSectionReader(f, int64(offset), chunk)
|
||||||
for try := 0; try < MaxRetries; try++ {
|
for try := 0; try < MaxRetries; try++ {
|
||||||
|
ch := make(chan error, 1)
|
||||||
|
|
||||||
r, w := io.Pipe()
|
r, w := io.Pipe()
|
||||||
defer r.Close()
|
defer r.Close()
|
||||||
go func() {
|
go func() {
|
||||||
defer w.Close()
|
defer w.Close()
|
||||||
|
|
||||||
for chunked := int64(0); chunked < chunk; {
|
for chunked := int64(0); chunked < chunk; {
|
||||||
n, err := io.CopyN(w, sectionReader, 1024*1024)
|
select {
|
||||||
if err != nil && !errors.Is(err, io.EOF) {
|
case err := <-ch:
|
||||||
|
log.Printf("chunk interrupted: %v", err)
|
||||||
|
return
|
||||||
|
default:
|
||||||
|
n, err := io.CopyN(w, sectionReader, 1024*1024)
|
||||||
|
if err != nil && !errors.Is(err, io.EOF) {
|
||||||
|
fn(api.ProgressResponse{
|
||||||
|
Status: fmt.Sprintf("error reading chunk: %v", err),
|
||||||
|
Digest: layer.Digest,
|
||||||
|
Total: layer.Size,
|
||||||
|
Completed: int(offset),
|
||||||
|
})
|
||||||
|
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
chunked += n
|
||||||
fn(api.ProgressResponse{
|
fn(api.ProgressResponse{
|
||||||
Status: fmt.Sprintf("error reading chunk: %v", err),
|
Status: fmt.Sprintf("uploading %s", layer.Digest),
|
||||||
Digest: layer.Digest,
|
Digest: layer.Digest,
|
||||||
Total: layer.Size,
|
Total: layer.Size,
|
||||||
Completed: int(offset),
|
Completed: int(offset) + int(chunked),
|
||||||
})
|
})
|
||||||
|
|
||||||
return
|
|
||||||
}
|
}
|
||||||
|
|
||||||
chunked += n
|
|
||||||
fn(api.ProgressResponse{
|
|
||||||
Status: fmt.Sprintf("uploading %s", layer.Digest),
|
|
||||||
Digest: layer.Digest,
|
|
||||||
Total: layer.Size,
|
|
||||||
Completed: int(offset) + int(chunked),
|
|
||||||
})
|
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
|
||||||
|
@ -113,6 +121,8 @@ func uploadBlobChunked(ctx context.Context, requestURL *url.URL, layer *Layer, r
|
||||||
|
|
||||||
switch {
|
switch {
|
||||||
case resp.StatusCode == http.StatusUnauthorized:
|
case resp.StatusCode == http.StatusUnauthorized:
|
||||||
|
ch <- errors.New("unauthorized")
|
||||||
|
|
||||||
auth := resp.Header.Get("www-authenticate")
|
auth := resp.Header.Get("www-authenticate")
|
||||||
authRedir := ParseAuthRedirectString(auth)
|
authRedir := ParseAuthRedirectString(auth)
|
||||||
token, err := getAuthToken(ctx, authRedir, regOpts)
|
token, err := getAuthToken(ctx, authRedir, regOpts)
|
||||||
|
@ -121,10 +131,7 @@ func uploadBlobChunked(ctx context.Context, requestURL *url.URL, layer *Layer, r
|
||||||
}
|
}
|
||||||
|
|
||||||
regOpts.Token = token
|
regOpts.Token = token
|
||||||
if _, err := sectionReader.Seek(0, io.SeekStart); err != nil {
|
sectionReader = io.NewSectionReader(f, int64(offset), chunk)
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
continue
|
continue
|
||||||
case resp.StatusCode >= http.StatusBadRequest:
|
case resp.StatusCode >= http.StatusBadRequest:
|
||||||
body, _ := io.ReadAll(resp.Body)
|
body, _ := io.ReadAll(resp.Body)
|
||||||
|
|
Loading…
Reference in a new issue