copy: remove max number of ErrResets
If a writer continually asks to be reset then it should always succeed - it should be the responsibility of the underlying content.Writer to stop producing ErrReset after some amount of time and to instead return the underlying issue - which pushWriter already does today, using the doWithRetries function. doWithRetries already has a separate cap for retries of 6 requests (5 retries after the original failure), and it seems like this would be previously overridden by content.Copy's max number of 5 attempts, hiding the original error. Signed-off-by: Justin Chadwell <me@jedevc.com>
This commit is contained in:
parent
651cfa2a2c
commit
e4f91c2df0
@ -31,9 +31,6 @@ import (
|
|||||||
ocispec "github.com/opencontainers/image-spec/specs-go/v1"
|
ocispec "github.com/opencontainers/image-spec/specs-go/v1"
|
||||||
)
|
)
|
||||||
|
|
||||||
// maxResets is the no.of times the Copy() method can tolerate a reset of the body
|
|
||||||
const maxResets = 5
|
|
||||||
|
|
||||||
var ErrReset = errors.New("writer has been reset")
|
var ErrReset = errors.New("writer has been reset")
|
||||||
|
|
||||||
var bufPool = sync.Pool{
|
var bufPool = sync.Pool{
|
||||||
@ -149,7 +146,7 @@ func OpenWriter(ctx context.Context, cs Ingester, opts ...WriterOpt) (Writer, er
|
|||||||
// Copy is buffered, so no need to wrap reader in buffered io.
|
// Copy is buffered, so no need to wrap reader in buffered io.
|
||||||
func Copy(ctx context.Context, cw Writer, or io.Reader, size int64, expected digest.Digest, opts ...Opt) error {
|
func Copy(ctx context.Context, cw Writer, or io.Reader, size int64, expected digest.Digest, opts ...Opt) error {
|
||||||
r := or
|
r := or
|
||||||
for i := 0; i < maxResets; i++ {
|
for i := 0; ; i++ {
|
||||||
if i >= 1 {
|
if i >= 1 {
|
||||||
log.G(ctx).WithField("digest", expected).Debugf("retrying copy due to reset")
|
log.G(ctx).WithField("digest", expected).Debugf("retrying copy due to reset")
|
||||||
}
|
}
|
||||||
@ -189,9 +186,6 @@ func Copy(ctx context.Context, cw Writer, or io.Reader, size int64, expected dig
|
|||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
log.G(ctx).WithField("digest", expected).Errorf("failed to copy after %d retries", maxResets)
|
|
||||||
return fmt.Errorf("failed to copy after %d retries", maxResets)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// CopyReaderAt copies to a writer from a given reader at for the given
|
// CopyReaderAt copies to a writer from a given reader at for the given
|
||||||
|
@ -20,7 +20,7 @@ import (
|
|||||||
"bytes"
|
"bytes"
|
||||||
"context"
|
"context"
|
||||||
_ "crypto/sha256" // required by go-digest
|
_ "crypto/sha256" // required by go-digest
|
||||||
"fmt"
|
"errors"
|
||||||
"io"
|
"io"
|
||||||
"strings"
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
@ -42,7 +42,7 @@ func TestCopy(t *testing.T) {
|
|||||||
cf1 := func(buf *bytes.Buffer, st Status) commitFunction {
|
cf1 := func(buf *bytes.Buffer, st Status) commitFunction {
|
||||||
i := 0
|
i := 0
|
||||||
return func() error {
|
return func() error {
|
||||||
// function resets the first time
|
// function resets the first time, but then succeeds after
|
||||||
if i == 0 {
|
if i == 0 {
|
||||||
// this is the case where, the pipewriter to which the data was being written has
|
// this is the case where, the pipewriter to which the data was being written has
|
||||||
// changed. which means we need to clear the buffer
|
// changed. which means we need to clear the buffer
|
||||||
@ -55,11 +55,28 @@ func TestCopy(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
cf2err := errors.New("commit failed")
|
||||||
cf2 := func(buf *bytes.Buffer, st Status) commitFunction {
|
cf2 := func(buf *bytes.Buffer, st Status) commitFunction {
|
||||||
i := 0
|
i := 0
|
||||||
return func() error {
|
return func() error {
|
||||||
// function resets for more than the maxReset value
|
// function resets a lot of times, and eventually fails
|
||||||
if i < maxResets+1 {
|
if i < 10 {
|
||||||
|
// this is the case where, the pipewriter to which the data was being written has
|
||||||
|
// changed. which means we need to clear the buffer
|
||||||
|
i++
|
||||||
|
buf.Reset()
|
||||||
|
st.Offset = 0
|
||||||
|
return ErrReset
|
||||||
|
}
|
||||||
|
return cf2err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
cf3 := func(buf *bytes.Buffer, st Status) commitFunction {
|
||||||
|
i := 0
|
||||||
|
return func() error {
|
||||||
|
// function resets a lot of times, and eventually succeeds
|
||||||
|
if i < 10 {
|
||||||
// this is the case where, the pipewriter to which the data was being written has
|
// this is the case where, the pipewriter to which the data was being written has
|
||||||
// changed. which means we need to clear the buffer
|
// changed. which means we need to clear the buffer
|
||||||
i++
|
i++
|
||||||
@ -73,8 +90,10 @@ func TestCopy(t *testing.T) {
|
|||||||
|
|
||||||
s1 := Status{}
|
s1 := Status{}
|
||||||
s2 := Status{}
|
s2 := Status{}
|
||||||
|
s3 := Status{}
|
||||||
b1 := bytes.Buffer{}
|
b1 := bytes.Buffer{}
|
||||||
b2 := bytes.Buffer{}
|
b2 := bytes.Buffer{}
|
||||||
|
b3 := bytes.Buffer{}
|
||||||
|
|
||||||
var testcases = []struct {
|
var testcases = []struct {
|
||||||
name string
|
name string
|
||||||
@ -130,7 +149,7 @@ func TestCopy(t *testing.T) {
|
|||||||
expected: "content to copy",
|
expected: "content to copy",
|
||||||
},
|
},
|
||||||
{
|
{
|
||||||
name: "write fails more than maxReset times due to reset",
|
name: "write fails after lots of resets",
|
||||||
source: newCopySource("content to copy"),
|
source: newCopySource("content to copy"),
|
||||||
writer: fakeWriter{
|
writer: fakeWriter{
|
||||||
Buffer: &b2,
|
Buffer: &b2,
|
||||||
@ -138,7 +157,17 @@ func TestCopy(t *testing.T) {
|
|||||||
commitFunc: cf2(&b2, s2),
|
commitFunc: cf2(&b2, s2),
|
||||||
},
|
},
|
||||||
expected: "",
|
expected: "",
|
||||||
expectedErr: fmt.Errorf("failed to copy after %d retries", maxResets),
|
expectedErr: cf2err,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "write succeeds after lots of resets",
|
||||||
|
source: newCopySource("content to copy"),
|
||||||
|
writer: fakeWriter{
|
||||||
|
Buffer: &b3,
|
||||||
|
status: s3,
|
||||||
|
commitFunc: cf3(&b3, s3),
|
||||||
|
},
|
||||||
|
expected: "content to copy",
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -153,7 +182,7 @@ func TestCopy(t *testing.T) {
|
|||||||
|
|
||||||
// if an error is expected then further comparisons are not required
|
// if an error is expected then further comparisons are not required
|
||||||
if testcase.expectedErr != nil {
|
if testcase.expectedErr != nil {
|
||||||
assert.Equal(t, testcase.expectedErr, err)
|
assert.ErrorIs(t, err, testcase.expectedErr)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
Loading…
Reference in New Issue
Block a user