From d7ff43781d79254b07861e29781ee97d553efd99 Mon Sep 17 00:00:00 2001 From: "mudler's LocalAI [bot]" <139863280+localai-bot@users.noreply.github.com> Date: Mon, 24 Aug 2026 09:33:44 +0200 Subject: [PATCH] fix(oci): resume interrupted layer downloads (#11688) quay.io redirects blob downloads to pre-signed S3/Akamai URLs that expire after about 10 minutes. On a slow connection a multi-GiB backend layer cannot finish inside that window, so the connection drops mid-stream on every attempt. The retry added for #10577 restarted each attempt from byte zero, which replayed the same failure until the budget ran out and the install failed with "unexpected EOF". A retry now keeps the bytes already on disk and re-requests the blob with "Range: bytes=N-". Each request goes back to the registry, so it gets a fresh redirect URL and auth token. The retry budget only counts attempts that made no forward progress, so a slow link that keeps advancing keeps downloading. A resumed file is spliced from separate responses and bypasses the digest check in layer.Compressed(), so the assembled file is re-verified against the layer digest before it is trusted; on a mismatch the download starts over through the verified reader. Fixes #10577 Assisted-by: Claude Code:claude-fable-5 Signed-off-by: Ettore Di Giacinto Co-authored-by: Ettore Di Giacinto --- pkg/oci/image.go | 207 +++++++++++++++++++++++--- pkg/oci/image_resume_internal_test.go | 151 +++++++++++++++++++ pkg/oci/layer_internal_test.go | 18 ++- pkg/oci/layer_resume_internal_test.go | 191 ++++++++++++++++++++++++ 4 files changed, 538 insertions(+), 29 deletions(-) create mode 100644 pkg/oci/image_resume_internal_test.go create mode 100644 pkg/oci/layer_resume_internal_test.go diff --git a/pkg/oci/image.go b/pkg/oci/image.go index 8e1546647..c44f7e5eb 100644 --- a/pkg/oci/image.go +++ b/pkg/oci/image.go @@ -80,27 +80,133 @@ var layerRetryBackoff = func(attempt int) time.Duration { return d } +// blobRangeOpener re-opens a layer blob at a byte offset. It returns the +// stream and the offset it actually starts at: the requested offset when the +// server honoured the Range request, or 0 when it ignored it and is sending +// the blob from the first byte again. +type blobRangeOpener func(ctx context.Context, offset int64) (io.ReadCloser, int64, error) + +// newBlobRangeOpener returns a blobRangeOpener that re-fetches the layer's +// blob from its registry with an HTTP Range request. Registries like quay.io +// redirect blob downloads to pre-signed S3/CDN URLs that expire after ~10 +// minutes; on a slow connection a multi-GiB layer cannot finish inside that +// window, so restarting from byte zero can never succeed while resuming from +// the current offset can (docker pull survives the same expiry this way). +// Each call goes back to the registry, so it obtains a fresh redirect URL and +// a fresh auth token. Returns nil when imageRef does not name a registry blob +// (e.g. local tarballs), which disables resuming. See issue #10577. +func newBlobRangeOpener(imageRef string, layer v1.Layer, auth *registrytypes.AuthConfig, base http.RoundTripper) blobRangeOpener { + ref, err := name.ParseReference(imageRef) + if err != nil { + return nil + } + digest, err := layer.Digest() + if err != nil || digest.Hex == "" { + return nil + } + repo := ref.Context() + if base == nil { + base = http.DefaultTransport + } + var authenticator authn.Authenticator + if auth != nil { + authenticator = staticAuth{auth} + } else if authenticator, err = authn.DefaultKeychain.Resolve(repo.Registry); err != nil { + authenticator = authn.Anonymous + } + blobURL := fmt.Sprintf("%s://%s/v2/%s/blobs/%s", repo.Registry.Scheme(), repo.RegistryStr(), repo.RepositoryStr(), digest.String()) + + return func(ctx context.Context, offset int64) (io.ReadCloser, int64, error) { + tr, err := transport.NewWithContext(ctx, repo.Registry, authenticator, base, []string{repo.Scope(transport.PullScope)}) + if err != nil { + return nil, 0, err + } + req, err := http.NewRequestWithContext(ctx, http.MethodGet, blobURL, nil) + if err != nil { + return nil, 0, err + } + if offset > 0 { + req.Header.Set("Range", fmt.Sprintf("bytes=%d-", offset)) + } + req.Header.Set("User-Agent", UserAgent()) + resp, err := (&http.Client{Transport: tr}).Do(req) + if err != nil { + return nil, 0, err + } + switch resp.StatusCode { + case http.StatusPartialContent: + return resp.Body, offset, nil + case http.StatusOK: + return resp.Body, 0, nil + default: + _ = resp.Body.Close() + return nil, 0, fmt.Errorf("unexpected status %d resuming blob %s", resp.StatusCode, digest.String()) + } + } +} + +// verifyLayerFile proves the assembled layer file matches the digest the +// registry advertised. A resumed download splices bytes from independent HTTP +// responses and bypasses the verified reader layer.Compressed() provides, so +// the whole file must be re-checked before it is trusted. +func verifyLayerFile(layer v1.Layer, f *os.File) error { + digest, err := layer.Digest() + if err != nil || digest.Hex == "" || digest.Algorithm != "sha256" { + return nil + } + if _, err := f.Seek(0, io.SeekStart); err != nil { + return err + } + got, _, err := v1.SHA256(f) + if err != nil { + return err + } + if got.Hex != digest.Hex { + return fmt.Errorf("resumed layer digest mismatch: got %s, want %s", got, digest) + } + return nil +} + // downloadLayerToFile streams a single compressed layer into dst, retrying on // transient network errors (unexpected EOF, connection reset, ...). Large // backend images (e.g. vLLM) are several GiB and a single dropped connection // mid-stream previously failed the whole install with "unexpected EOF" and no -// recovery. The registry transport already retries manifest fetches via -// defaultRetryPredicate (see GetImage/GetImageDigest); this extends the same -// behaviour to the layer data stream. See issue #10577. -func downloadLayerToFile(ctx context.Context, layer v1.Layer, dst *os.File, progress *progressWriter) error { +// recovery. When resume is non-nil, a retry keeps the bytes already on disk +// and continues from that offset instead of starting over: registries that +// serve blobs through expiring pre-signed URLs (quay.io + S3/Akamai) cut off +// every full-length transfer on slow connections, so restarting can never +// finish while resuming makes progress each round. The retry budget only +// counts attempts that made no forward progress, so a download that keeps +// advancing keeps going. See issue #10577. +func downloadLayerToFile(ctx context.Context, layer v1.Layer, dst *os.File, progress *progressWriter, resume blobRangeOpener) error { var lastErr error + // written tracks the valid bytes currently in dst across attempts, and + // bestWritten the furthest offset any attempt has reached: only beating + // it counts as forward progress for the retry budget, so a server that + // ignores Range requests and keeps dropping mid-stream still runs out + // of attempts instead of looping forever. + var written, bestWritten int64 + // resumed records whether any byte in dst came from a resumed raw blob + // fetch, which requires re-verifying the assembled file at the end. + resumed := false + + truncate := func() error { + if _, err := dst.Seek(0, io.SeekStart); err != nil { + return err + } + if err := dst.Truncate(0); err != nil { + return err + } + written = 0 + resumed = false + if progress != nil { + progress.written = 0 + } + return nil + } + for attempt := 0; attempt <= layerDownloadRetries; attempt++ { if attempt > 0 { - // Discard any partial data from the previous failed attempt. - if _, err := dst.Seek(0, io.SeekStart); err != nil { - return err - } - if err := dst.Truncate(0); err != nil { - return err - } - if progress != nil { - progress.written = 0 - } select { case <-ctx.Done(): return ctx.Err() @@ -108,19 +214,69 @@ func downloadLayerToFile(ctx context.Context, layer v1.Layer, dst *os.File, prog } } - var w io.Writer = dst - if progress != nil { - w = io.MultiWriter(dst, progress) + var reader io.ReadCloser + if attempt > 0 && resume != nil && written > 0 { + r, offset, rerr := resume(ctx, written) + switch { + case rerr != nil: + // Keep the partial bytes: opening the resume stream can + // fail transiently (token refresh, connection refused) + // and the next attempt can still continue from here. + lastErr = rerr + case offset != written: + // The server ignored the Range request and is sending + // the blob from the first byte: drop the partial data. + if err := truncate(); err != nil { + _ = r.Close() + return err + } + reader = r + resumed = true + default: + reader = r + resumed = true + } + } else { + // First attempt, or no way to resume: restart from scratch + // through the digest-verifying layer reader. + if err := truncate(); err != nil { + return err + } + reader, lastErr = layer.Compressed() } - var reader io.ReadCloser - reader, lastErr = layer.Compressed() - if lastErr == nil { - _, lastErr = xio.Copy(ctx, w, reader) + if reader != nil { + var w io.Writer = dst + if progress != nil { + w = io.MultiWriter(dst, progress) + } + var n int64 + n, lastErr = xio.Copy(ctx, w, reader) + written += n _ = reader.Close() + if written > bestWritten { + // Forward progress: don't charge this round against the + // retry budget, or slow links would still exhaust it. + bestWritten = written + attempt = 0 + } } + if lastErr == nil { - return nil + if !resumed { + return nil + } + verr := verifyLayerFile(layer, dst) + if verr == nil { + return nil + } + // The spliced file is corrupt: discard it and retry cleanly. + logs.Warn.Printf("discarding resumed layer download: %v", verr) + lastErr = verr + if err := truncate(); err != nil { + return err + } + continue } // Stop early on context cancellation or non-retryable errors. @@ -382,8 +538,11 @@ func DownloadOCIImageTar(ctx context.Context, img v1.Image, imageRef string, tar } } - // Download the compressed layer, retrying on transient network errors. - err = downloadLayerToFile(ctx, layer, file, progress) + // Download the compressed layer, retrying on transient network + // errors and resuming from the last byte received where possible. + // Anonymous/default-keychain credentials match what GetImage uses + // for every in-tree caller (they all pass a nil auth). + err = downloadLayerToFile(ctx, layer, file, progress, newBlobRangeOpener(imageRef, layer, nil, nil)) file.Close() if err != nil { return fmt.Errorf("failed to download layer %d: %v", i, err) diff --git a/pkg/oci/image_resume_internal_test.go b/pkg/oci/image_resume_internal_test.go new file mode 100644 index 000000000..dec545d96 --- /dev/null +++ b/pkg/oci/image_resume_internal_test.go @@ -0,0 +1,151 @@ +package oci + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strconv" + "strings" + "sync" + "time" + + "github.com/google/go-containerregistry/pkg/name" + "github.com/google/go-containerregistry/pkg/registry" + "github.com/google/go-containerregistry/pkg/v1/random" + "github.com/google/go-containerregistry/pkg/v1/remote" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +// droppingBlobRegistry emulates how quay.io serves layer blobs from S3/Akamai +// with a short-lived pre-signed URL: a full-blob GET on a slow connection is +// always cut off mid-transfer, so a client that restarts from byte zero can +// never complete the download. Only a client that resumes with a Range request +// (like docker pull does) receives the remaining bytes and can finish. +type droppingBlobRegistry struct { + inner http.Handler + + mu sync.Mutex + rangeRequests []int64 + fullRequests int +} + +// dropThreshold separates real layer blobs from small metadata blobs (image +// config), which are served untouched. +const dropThreshold = 1024 + +func (h *droppingBlobRegistry) ServeHTTP(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet || !strings.Contains(r.URL.Path, "/blobs/sha256:") { + h.inner.ServeHTTP(w, r) + return + } + + // Fetch the full blob from the inner registry (which does not speak + // Range) and apply the Range semantics here. + inner := r.Clone(r.Context()) + inner.Header.Del("Range") + rec := httptest.NewRecorder() + h.inner.ServeHTTP(rec, inner) + body := rec.Body.Bytes() + if rec.Code != http.StatusOK || len(body) <= dropThreshold { + for k, vv := range rec.Header() { + for _, v := range vv { + w.Header().Add(k, v) + } + } + w.WriteHeader(rec.Code) + _, _ = w.Write(body) + return + } + + if rh := r.Header.Get("Range"); rh != "" { + offset, err := strconv.ParseInt(strings.TrimSuffix(strings.TrimPrefix(rh, "bytes="), "-"), 10, 64) + if err != nil || offset < 0 || offset >= int64(len(body)) { + w.WriteHeader(http.StatusRequestedRangeNotSatisfiable) + return + } + h.mu.Lock() + h.rangeRequests = append(h.rangeRequests, offset) + h.mu.Unlock() + w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", offset, len(body)-1, len(body))) + w.Header().Set("Content-Length", strconv.Itoa(len(body)-int(offset))) + w.WriteHeader(http.StatusPartialContent) + _, _ = w.Write(body[offset:]) + return + } + + h.mu.Lock() + h.fullRequests++ + h.mu.Unlock() + + // Announce the full size but deliver only half, then sever the + // connection, like a pre-signed URL expiring mid-download. + w.Header().Set("Content-Length", strconv.Itoa(len(body))) + w.WriteHeader(http.StatusOK) + _, _ = w.Write(body[:len(body)/2]) + if f, ok := w.(http.Flusher); ok { + f.Flush() + } + panic(http.ErrAbortHandler) +} + +var _ = Describe("DownloadOCIImageTar resume", func() { + var ( + server *httptest.Server + reg *droppingBlobRegistry + tmpDir string + restoreWait func() + ) + + BeforeEach(func() { + reg = &droppingBlobRegistry{inner: registry.New()} + server = httptest.NewServer(reg) + + var err error + tmpDir, err = os.MkdirTemp("", "oci-resume-e2e-*") + Expect(err).NotTo(HaveOccurred()) + + prev := layerRetryBackoff + layerRetryBackoff = func(int) time.Duration { return 0 } + restoreWait = func() { layerRetryBackoff = prev } + }) + + AfterEach(func() { + restoreWait() + server.Close() + _ = os.RemoveAll(tmpDir) + }) + + It("completes the download by resuming interrupted layer transfers with Range requests", func() { + img, err := random.Image(4096, 1) + Expect(err).NotTo(HaveOccurred()) + + imageRef := strings.TrimPrefix(server.URL, "http://") + "/testrepo/backend:latest" + ref, err := name.ParseReference(imageRef) + Expect(err).NotTo(HaveOccurred()) + Expect(remote.Write(ref, img)).To(Succeed()) + + pulled, err := GetImage(imageRef, "", nil, nil) + Expect(err).NotTo(HaveOccurred()) + + tarPath := filepath.Join(tmpDir, "image.tar") + err = DownloadOCIImageTar(context.Background(), pulled, imageRef, tarPath, nil) + Expect(err).NotTo(HaveOccurred()) + + // The full-blob attempt was cut off, so success is only possible + // through at least one Range request picking up where it stopped. + reg.mu.Lock() + defer reg.mu.Unlock() + Expect(reg.rangeRequests).NotTo(BeEmpty()) + for _, off := range reg.rangeRequests { + Expect(off).To(BeNumerically(">", 0)) + } + + fi, err := os.Stat(tarPath) + Expect(err).NotTo(HaveOccurred()) + Expect(fi.Size()).To(BeNumerically(">", 0)) + }) +}) diff --git a/pkg/oci/layer_internal_test.go b/pkg/oci/layer_internal_test.go index faa8d5a45..23ba48160 100644 --- a/pkg/oci/layer_internal_test.go +++ b/pkg/oci/layer_internal_test.go @@ -33,14 +33,18 @@ func (r *failingReader) Read(p []byte) (int, error) { // fakeLayer is a minimal v1.Layer whose Compressed() fails failUntil times with // err (after emitting a partial prefix) before finally returning data in full. +// The failing attempts emit prefix when set, or placeholder garbage otherwise. +// digest, when set, is what Digest() reports. type fakeLayer struct { data []byte + prefix []byte + digest v1.Hash failUntil int err error calls int } -func (f *fakeLayer) Digest() (v1.Hash, error) { return v1.Hash{}, nil } +func (f *fakeLayer) Digest() (v1.Hash, error) { return f.digest, nil } func (f *fakeLayer) DiffID() (v1.Hash, error) { return v1.Hash{}, nil } func (f *fakeLayer) Size() (int64, error) { return int64(len(f.data)), nil } func (f *fakeLayer) MediaType() (types.MediaType, error) { return types.DockerLayer, nil } @@ -51,7 +55,11 @@ func (f *fakeLayer) Uncompressed() (io.ReadCloser, error) { func (f *fakeLayer) Compressed() (io.ReadCloser, error) { f.calls++ if f.calls <= f.failUntil { - return io.NopCloser(&failingReader{prefix: []byte("partial-garbage"), err: f.err}), nil + prefix := f.prefix + if prefix == nil { + prefix = []byte("partial-garbage") + } + return io.NopCloser(&failingReader{prefix: prefix, err: f.err}), nil } return io.NopCloser(bytes.NewReader(f.data)), nil } @@ -86,7 +94,7 @@ var _ = Describe("downloadLayerToFile", func() { err: io.ErrUnexpectedEOF, } - err := downloadLayerToFile(context.Background(), layer, dst, nil) + err := downloadLayerToFile(context.Background(), layer, dst, nil, nil) Expect(err).NotTo(HaveOccurred()) Expect(layer.calls).To(Equal(3)) @@ -104,7 +112,7 @@ var _ = Describe("downloadLayerToFile", func() { err: errors.New("permission denied"), } - err := downloadLayerToFile(context.Background(), layer, dst, nil) + err := downloadLayerToFile(context.Background(), layer, dst, nil, nil) Expect(err).To(HaveOccurred()) Expect(layer.calls).To(Equal(1)) }) @@ -116,7 +124,7 @@ var _ = Describe("downloadLayerToFile", func() { err: io.ErrUnexpectedEOF, } - err := downloadLayerToFile(context.Background(), layer, dst, nil) + err := downloadLayerToFile(context.Background(), layer, dst, nil, nil) Expect(err).To(MatchError(io.ErrUnexpectedEOF)) Expect(layer.calls).To(Equal(layerDownloadRetries + 1)) }) diff --git a/pkg/oci/layer_resume_internal_test.go b/pkg/oci/layer_resume_internal_test.go new file mode 100644 index 000000000..3f9e032d9 --- /dev/null +++ b/pkg/oci/layer_resume_internal_test.go @@ -0,0 +1,191 @@ +package oci + +import ( + "bytes" + "context" + "io" + "os" + "time" + + v1 "github.com/google/go-containerregistry/pkg/v1" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +// recordingOpener is a test blobRangeOpener that records the offsets it was +// asked to resume from and delegates the stream to open. +type recordingOpener struct { + offsets []int64 + open func(offset int64) (io.ReadCloser, int64, error) +} + +func (o *recordingOpener) opener() blobRangeOpener { + return func(_ context.Context, offset int64) (io.ReadCloser, int64, error) { + o.offsets = append(o.offsets, offset) + return o.open(offset) + } +} + +func sha256Of(data []byte) v1.Hash { + h, _, err := v1.SHA256(bytes.NewReader(data)) + Expect(err).NotTo(HaveOccurred()) + return h +} + +var _ = Describe("downloadLayerToFile resume", func() { + var ( + dst *os.File + data []byte + restoreWait func() + ) + + readDst := func() string { + got, err := os.ReadFile(dst.Name()) + Expect(err).NotTo(HaveOccurred()) + return string(got) + } + + BeforeEach(func() { + var err error + dst, err = os.CreateTemp("", "layer-resume-*.tar.gz") + Expect(err).NotTo(HaveOccurred()) + + data = []byte("0123456789abcdefghijklmnopqrstuvwxyzABCD") + + prev := layerRetryBackoff + layerRetryBackoff = func(int) time.Duration { return 0 } + restoreWait = func() { layerRetryBackoff = prev } + }) + + AfterEach(func() { + restoreWait() + _ = dst.Close() + _ = os.Remove(dst.Name()) + }) + + It("continues from the interruption offset instead of restarting", func() { + layer := &fakeLayer{ + data: data, + prefix: data[:15], + digest: sha256Of(data), + failUntil: 1, + err: io.ErrUnexpectedEOF, + } + rec := &recordingOpener{open: func(offset int64) (io.ReadCloser, int64, error) { + return io.NopCloser(bytes.NewReader(data[offset:])), offset, nil + }} + + err := downloadLayerToFile(context.Background(), layer, dst, nil, rec.opener()) + Expect(err).NotTo(HaveOccurred()) + Expect(readDst()).To(Equal(string(data))) + // The interrupted first attempt left 15 bytes; the resume must ask + // for exactly the rest, without a second full-stream attempt. + Expect(rec.offsets).To(Equal([]int64{15})) + Expect(layer.calls).To(Equal(1)) + }) + + It("restarts cleanly when the server ignores the Range request", func() { + layer := &fakeLayer{ + data: data, + prefix: data[:15], + digest: sha256Of(data), + failUntil: 1, + err: io.ErrUnexpectedEOF, + } + rec := &recordingOpener{open: func(int64) (io.ReadCloser, int64, error) { + // A 200 response: the whole blob from the first byte. + return io.NopCloser(bytes.NewReader(data)), 0, nil + }} + + err := downloadLayerToFile(context.Background(), layer, dst, nil, rec.opener()) + Expect(err).NotTo(HaveOccurred()) + // The partial bytes must have been discarded, not prepended. + Expect(readDst()).To(Equal(string(data))) + Expect(rec.offsets).To(HaveLen(1)) + Expect(layer.calls).To(Equal(1)) + }) + + It("discards a resumed download whose digest does not match", func() { + layer := &fakeLayer{ + data: data, + prefix: data[:15], + digest: sha256Of(data), + failUntil: 1, + err: io.ErrUnexpectedEOF, + } + rec := &recordingOpener{open: func(offset int64) (io.ReadCloser, int64, error) { + corrupt := bytes.Repeat([]byte("x"), len(data)-int(offset)) + return io.NopCloser(bytes.NewReader(corrupt)), offset, nil + }} + + err := downloadLayerToFile(context.Background(), layer, dst, nil, rec.opener()) + Expect(err).NotTo(HaveOccurred()) + // The spliced file failed verification, so the download must have + // started over through the verified layer reader and succeeded. + Expect(readDst()).To(Equal(string(data))) + Expect(rec.offsets).To(Equal([]int64{15})) + Expect(layer.calls).To(Equal(2)) + }) + + It("keeps retrying beyond the budget while each resume makes progress", func() { + const step = 5 + layer := &fakeLayer{ + data: data, + prefix: data[:step], + digest: sha256Of(data), + failUntil: 1, + err: io.ErrUnexpectedEOF, + } + rec := &recordingOpener{open: func(offset int64) (io.ReadCloser, int64, error) { + if offset+step >= int64(len(data)) { + return io.NopCloser(bytes.NewReader(data[offset:])), offset, nil + } + return io.NopCloser(&failingReader{prefix: data[offset : offset+step], err: io.ErrUnexpectedEOF}), offset, nil + }} + + err := downloadLayerToFile(context.Background(), layer, dst, nil, rec.opener()) + Expect(err).NotTo(HaveOccurred()) + Expect(readDst()).To(Equal(string(data))) + // 40 bytes delivered 5 at a time: 7 resumes, far more rounds than + // the retry budget allows for stalled attempts. + Expect(len(rec.offsets)).To(BeNumerically(">", layerDownloadRetries)) + }) + + It("gives up when resumes stop making progress", func(ctx SpecContext) { + layer := &fakeLayer{ + data: data, + prefix: data[:15], + digest: sha256Of(data), + failUntil: 1000, + err: io.ErrUnexpectedEOF, + } + rec := &recordingOpener{open: func(offset int64) (io.ReadCloser, int64, error) { + // Resume accepted but the connection dies before any byte. + return io.NopCloser(&failingReader{err: io.ErrUnexpectedEOF}), offset, nil + }} + + err := downloadLayerToFile(ctx, layer, dst, nil, rec.opener()) + Expect(err).To(MatchError(io.ErrUnexpectedEOF)) + Expect(len(rec.offsets)).To(Equal(layerDownloadRetries)) + }, NodeTimeout(10*time.Second)) + + It("terminates when the server ignores Range and keeps dropping mid-stream", func(ctx SpecContext) { + // Each round delivers some bytes from the start and dies: the file + // never gets further than before, so this must exhaust the budget + // rather than count the repeated partial bytes as progress. + layer := &fakeLayer{ + data: data, + prefix: data[:15], + digest: sha256Of(data), + failUntil: 1000, + err: io.ErrUnexpectedEOF, + } + rec := &recordingOpener{open: func(int64) (io.ReadCloser, int64, error) { + return io.NopCloser(&failingReader{prefix: data[:15], err: io.ErrUnexpectedEOF}), 0, nil + }} + + err := downloadLayerToFile(ctx, layer, dst, nil, rec.opener()) + Expect(err).To(MatchError(io.ErrUnexpectedEOF)) + Expect(len(rec.offsets)).To(Equal(layerDownloadRetries)) + }, NodeTimeout(10*time.Second)) +})