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 <mudler@localai.io>
Co-authored-by: Ettore Di Giacinto <mudler@localai.io>

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)

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))
+	})
+})

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))
 	})

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))
+})