github/containerd
4
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 2 Wiki Activity

pushWriter: correctly propagate errors

Browse Source 
In the refactor from 926b9c72f61b5be6bf8d952512f1d0932fbaf898, the error
handling was substantially reworked, and changed the types of errors
returned.

Notably, in the case of a network error, instead of propogating the
error through to return from pushWriter.Write (as previously), it would
be propagated through to pushWriter.Commit - however, this is too late,
since we've already closed the io.Pipe by the time we would have reached
this function. Therefore, we get the generic error message  "io:
read/write on closed pipe" for *every network error*.

This patch corrects this behavior to ensure that the correct error
object is always returned as early as possible, by checking the error
result after writing and detecting a closed pipe.

Additionally, we do some additional hardening - specifically we prevent
falling through when resetting the content or detecting errors, and
update the tests to explicitly check for the ErrReset message.

Signed-off-by: Justin Chadwell <me@jedevc.com>
This commit is contained in:
Justin Chadwell
2023-01-24 11:15:52 +00:00
parent 3f565daf68
commit 9f6058d029
2 changed files with 33 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
remotes/docker/pusher.go
View File

@@ -380,17 +380,24 @@ func (pw *pushWriter) Write(p []byte) (n int, err error) {
// If content has already been written, the bytes
// cannot be written and the caller must reset
if status.Offset > 0 {
status.Offset = 0
status.UpdatedAt = time.Now()
pw.tracker.SetStatus(pw.ref, status)
return 0, content.ErrReset
}
status.Offset = 0
status.UpdatedAt = time.Now()
pw.tracker.SetStatus(pw.ref, status)
return 0, content.ErrReset
default:
}
}
n, err = pw.pipe.Write(p)
if errors.Is(err, io.ErrClosedPipe) {
// if the pipe is closed, we might have the original error on the error
// channel - so we should try and get it
select {
case err2 := <-pw.errC:
err = err2
default:
}
}
status.Offset += int64(n)
status.UpdatedAt = time.Now()
pw.tracker.SetStatus(pw.ref, status)
@@ -431,7 +438,7 @@ func (pw *pushWriter) Digest() digest.Digest {
func (pw *pushWriter) Commit(ctx context.Context, size int64, expected digest.Digest, opts ...content.Opt) error {
// Check whether read has already thrown an error
if _, err := pw.pipe.Write([]byte{}); err != nil && err != io.ErrClosedPipe {
if _, err := pw.pipe.Write([]byte{}); err != nil && !errors.Is(err, io.ErrClosedPipe) {
return fmt.Errorf("pipe error before commit: %w", err)
}
@@ -442,9 +449,7 @@ func (pw *pushWriter) Commit(ctx context.Context, size int64, expected digest.Di
var resp *http.Response
select {
case err := <-pw.errC:
if err != nil {
return err
}
return err
case resp = <-pw.respC:
defer resp.Body.Close()
case p, ok := <-pw.pipeC:
@@ -456,18 +461,17 @@ func (pw *pushWriter) Commit(ctx context.Context, size int64, expected digest.Di
}
pw.pipe.CloseWithError(content.ErrReset)
pw.pipe = p
// If content has already been written, the bytes
// cannot be written again and the caller must reset
status, err := pw.tracker.GetStatus(pw.ref)
if err != nil {
return err
}
// If content has already been written, the bytes
// cannot be written again and the caller must reset
if status.Offset > 0 {
status.Offset = 0
status.UpdatedAt = time.Now()
pw.tracker.SetStatus(pw.ref, status)
return content.ErrReset
}
status.Offset = 0
status.UpdatedAt = time.Now()
pw.tracker.SetStatus(pw.ref, status)
return content.ErrReset
}
// 201 is specified return status, some registries return