From 8ed74506dea30ced443ce1d2cc7035e17fbcf1b0 Mon Sep 17 00:00:00 2001 From: Oscar Wieman Date: Wed, 22 Jul 2026 10:12:05 +0200 Subject: [PATCH 1/6] fix(proxycache): retry interrupted upstream blob fetches, verify digest before caching The proxy cache's blob fetch made a single-shot HTTP GET to the upstream registry with no retry: a connection dropped mid-stream (e.g. by a network intermediary killing the TCP session) just failed the pull, with no equivalent of docker/containerd's client-side retry of an interrupted layer. This affected both controller.putBlobToLocal (background cache population after a client pull) and the identical ManifestCache.putBlobToLocal used when caching a manifest list's dependent blobs. Add a resumingBlobReader that, on a retryable mid-stream read error, closes the broken connection and resumes the fetch from the last successfully read byte offset via an HTTP range request (RemoteInterface.BlobReaderAt, backed by the existing PullBlobChunk), with bounded exponential backoff. Applied to the client-serving fetch and both cache-populate fetches. Also add a verifyingReadCloser that recomputes the digest of the bytes streamed into each cache-populate path and surfaces a mismatch as a Read error (instead of io.EOF), so a corrupted-but-well-formed body can't be committed to local storage as if it were a complete, valid blob. Signed-off-by: Oscar Wieman --- src/controller/proxy/blobretry.go | 150 ++++++++++++++++ src/controller/proxy/blobretry_test.go | 163 ++++++++++++++++++ src/controller/proxy/controller.go | 13 ++ src/controller/proxy/controller_test.go | 35 +++- src/controller/proxy/manifestcache.go | 9 + src/controller/proxy/manifestcache_test.go | 32 ++++ src/controller/proxy/remote.go | 15 ++ .../controller/proxy/remote_interface.go | 30 ++++ 8 files changed, 445 insertions(+), 2 deletions(-) create mode 100644 src/controller/proxy/blobretry.go create mode 100644 src/controller/proxy/blobretry_test.go diff --git a/src/controller/proxy/blobretry.go b/src/controller/proxy/blobretry.go new file mode 100644 index 00000000000..a8dcebf27e0 --- /dev/null +++ b/src/controller/proxy/blobretry.go @@ -0,0 +1,150 @@ +// Copyright Project Harbor Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package proxy + +import ( + "context" + "errors" + "fmt" + "io" + "time" + + "github.com/opencontainers/go-digest" + + "github.com/goharbor/harbor/src/lib/log" +) + +const ( + // maxBlobFetchRetries bounds how many times an interrupted upstream blob + // fetch is resumed before the read error is returned to the caller. + maxBlobFetchRetries = 3 +) + +// blobFetchRetryBackoff is the delay before the first resume attempt; it +// doubles on each subsequent attempt (1s, 2s, 4s by default). It's a var +// rather than a const so tests can shrink it. +var blobFetchRetryBackoff = time.Second + +// resumeBlobFunc opens a reader for the remaining bytes of a blob, starting +// at offset. It's used to resume a fetch that broke mid-stream. +type resumeBlobFunc func(offset int64) (io.ReadCloser, error) + +// resumingBlobReader wraps an upstream blob body and transparently resumes +// the fetch, via resume, if the connection breaks mid-stream - instead of +// failing the whole blob on a single dropped connection, the way a plain +// docker/OCI client retries an interrupted layer pull. Resuming is bounded by +// maxBlobFetchRetries and disabled when size is unknown (<= 0), since the +// read offset can't be validated against the blob length in that case. +type resumingBlobReader struct { + reader io.ReadCloser + resume resumeBlobFunc + size int64 + offset int64 + retries int +} + +// newResumingBlobReader wraps reader so a mid-stream read failure is retried +// up to maxBlobFetchRetries times by resuming the fetch from the last +// successfully read byte offset (via an HTTP range request through resume). +func newResumingBlobReader(reader io.ReadCloser, size int64, resume resumeBlobFunc) io.ReadCloser { + return &resumingBlobReader{reader: reader, size: size, resume: resume} +} + +func (r *resumingBlobReader) Read(p []byte) (int, error) { + for { + n, err := r.reader.Read(p) + r.offset += int64(n) + if err == nil || err == io.EOF { // nolint:errorlint + return n, err + } + if !r.canRetry(err) { + return n, err + } + if rErr := r.reconnect(err); rErr != nil { + return n, rErr + } + if n > 0 { + return n, nil + } + // n == 0: loop and read immediately from the freshly resumed reader. + } +} + +// reconnect closes the broken reader and resumes the fetch from the current +// offset, retrying the resume itself (bounded by the same retry budget) if +// establishing the new connection also fails. +func (r *resumingBlobReader) reconnect(cause error) error { + _ = r.reader.Close() + for { + r.retries++ + backoff := blobFetchRetryBackoff * time.Duration(uint64(1)<= maxBlobFetchRetries { + return fmt.Errorf("failed to resume interrupted blob fetch after %v: %w", cause, err) + } + cause = err + } +} + +func (r *resumingBlobReader) canRetry(err error) bool { + return r.size > 0 && r.retries < maxBlobFetchRetries && r.offset < r.size && isRetryableBlobReadErr(err) +} + +func (r *resumingBlobReader) Close() error { + return r.reader.Close() +} + +// isRetryableBlobReadErr reports whether err looks like a transient failure +// of the upstream connection - dropped, reset, or timed out mid-transfer - +// worth retrying, as opposed to the pulling client itself having given up. +func isRetryableBlobReadErr(err error) bool { + return err != nil && !errors.Is(err, context.Canceled) +} + +// verifyingReadCloser wraps a blob reader and, once the wrapped reader signals +// a genuine end of stream, verifies the bytes read so far match the expected +// digest. A mismatch is surfaced as a Read error instead of io.EOF, so a +// corrupted or truncated blob that still parses as a well-formed HTTP +// response is never mistaken for a complete, valid one by the caller (e.g. +// the local registry committing a blob push). +type verifyingReadCloser struct { + io.ReadCloser + verifier digest.Verifier + dig digest.Digest +} + +// newVerifyingReadCloser wraps reader so that reaching the end of the stream +// without the content matching dig surfaces as a Read error. +func newVerifyingReadCloser(reader io.ReadCloser, dig digest.Digest) io.ReadCloser { + return &verifyingReadCloser{ReadCloser: reader, verifier: dig.Verifier(), dig: dig} +} + +func (v *verifyingReadCloser) Read(p []byte) (int, error) { + n, err := v.ReadCloser.Read(p) + if n > 0 { + _, _ = v.verifier.Write(p[:n]) + } + if err == io.EOF && !v.verifier.Verified() { // nolint:errorlint + return n, fmt.Errorf("blob content does not match expected digest %s after fetching from upstream", v.dig) + } + return n, err +} diff --git a/src/controller/proxy/blobretry_test.go b/src/controller/proxy/blobretry_test.go new file mode 100644 index 00000000000..cd400f44d6e --- /dev/null +++ b/src/controller/proxy/blobretry_test.go @@ -0,0 +1,163 @@ +// Copyright Project Harbor Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package proxy + +import ( + "context" + "errors" + "io" + "testing" + "time" + + "github.com/opencontainers/go-digest" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// stubReader serves data and then, once exhausted, returns err (or io.EOF if +// err is nil). It records whether it was closed. +type stubReader struct { + data []byte + err error + closed bool +} + +func (s *stubReader) Read(p []byte) (int, error) { + if len(s.data) > 0 { + n := copy(p, s.data) + s.data = s.data[n:] + return n, nil + } + if s.err != nil { + return 0, s.err + } + return 0, io.EOF +} + +func (s *stubReader) Close() error { + s.closed = true + return nil +} + +func withShortBlobRetryBackoff(t *testing.T) { + orig := blobFetchRetryBackoff + blobFetchRetryBackoff = time.Millisecond + t.Cleanup(func() { blobFetchRetryBackoff = orig }) +} + +func TestResumingBlobReader_ResumesAfterMidStreamError(t *testing.T) { + withShortBlobRetryBackoff(t) + + first := &stubReader{data: []byte("hello "), err: io.ErrUnexpectedEOF} + second := &stubReader{data: []byte("world")} + + var resumeOffsets []int64 + reader := newResumingBlobReader(first, 11, func(offset int64) (io.ReadCloser, error) { + resumeOffsets = append(resumeOffsets, offset) + return second, nil + }) + + got, err := io.ReadAll(reader) + require.NoError(t, err) + assert.Equal(t, "hello world", string(got)) + assert.Equal(t, []int64{6}, resumeOffsets) + assert.True(t, first.closed, "the broken reader should be closed before resuming") +} + +func TestResumingBlobReader_GivesUpAfterMaxRetries(t *testing.T) { + withShortBlobRetryBackoff(t) + + first := &stubReader{err: io.ErrUnexpectedEOF} + resumeAttempts := 0 + reader := newResumingBlobReader(first, 100, func(offset int64) (io.ReadCloser, error) { + resumeAttempts++ + return nil, errors.New("upstream still unreachable") + }) + + _, err := io.ReadAll(reader) + require.Error(t, err) + assert.Equal(t, maxBlobFetchRetries, resumeAttempts) +} + +func TestResumingBlobReader_DoesNotRetryOnContextCanceled(t *testing.T) { + withShortBlobRetryBackoff(t) + + first := &stubReader{err: context.Canceled} + resumeAttempts := 0 + reader := newResumingBlobReader(first, 100, func(offset int64) (io.ReadCloser, error) { + resumeAttempts++ + return &stubReader{}, nil + }) + + _, err := io.ReadAll(reader) + require.Error(t, err) + assert.Equal(t, 0, resumeAttempts, "a canceled request should not be retried") +} + +func TestResumingBlobReader_DoesNotRetryWithUnknownSize(t *testing.T) { + withShortBlobRetryBackoff(t) + + first := &stubReader{err: io.ErrUnexpectedEOF} + resumeAttempts := 0 + reader := newResumingBlobReader(first, 0, func(offset int64) (io.ReadCloser, error) { + resumeAttempts++ + return &stubReader{}, nil + }) + + _, err := io.ReadAll(reader) + require.Error(t, err) + assert.Equal(t, 0, resumeAttempts, "resuming an unknown-length blob can't be validated, so it shouldn't be attempted") +} + +func TestResumingBlobReader_RecoversFromMultipleInterruptions(t *testing.T) { + withShortBlobRetryBackoff(t) + + first := &stubReader{data: []byte("aaa"), err: io.ErrUnexpectedEOF} + second := &stubReader{data: []byte("bbb"), err: io.ErrUnexpectedEOF} + third := &stubReader{data: []byte("ccc")} + var resumeOffsets []int64 + readers := []*stubReader{second, third} + reader := newResumingBlobReader(first, 9, func(offset int64) (io.ReadCloser, error) { + resumeOffsets = append(resumeOffsets, offset) + next := readers[0] + readers = readers[1:] + return next, nil + }) + + got, err := io.ReadAll(reader) + require.NoError(t, err) + assert.Equal(t, "aaabbbccc", string(got)) + assert.Equal(t, []int64{3, 6}, resumeOffsets) +} + +func TestVerifyingReadCloser_PassesThroughValidContent(t *testing.T) { + content := []byte("valid blob content") + dig := digest.FromBytes(content) + + reader := newVerifyingReadCloser(io.NopCloser(&stubReader{data: content}), dig) + got, err := io.ReadAll(reader) + require.NoError(t, err) + assert.Equal(t, content, got) +} + +func TestVerifyingReadCloser_RejectsMismatchedContent(t *testing.T) { + content := []byte("corrupted blob content") + wrongDigest := digest.FromBytes([]byte("something else entirely")) + + reader := newVerifyingReadCloser(io.NopCloser(&stubReader{data: content}), wrongDigest) + _, err := io.ReadAll(reader) + require.Error(t, err) + assert.Contains(t, err.Error(), "does not match expected digest") +} diff --git a/src/controller/proxy/controller.go b/src/controller/proxy/controller.go index efff91a7317..00604b7a466 100644 --- a/src/controller/proxy/controller.go +++ b/src/controller/proxy/controller.go @@ -295,6 +295,11 @@ func (c *controller) ProxyBlob(ctx context.Context, p *proModels.Project, art li log.Errorf("failed to pull blob, error %v", err) return 0, nil, err } + // Resume the fetch from the last read byte offset if the upstream + // connection breaks mid-stream, instead of failing the pull outright. + bReader = newResumingBlobReader(bReader, size, func(offset int64) (io.ReadCloser, error) { + return rHelper.BlobReaderAt(remoteRepo, art.Digest, size, offset) + }) desc := distribution.Descriptor{Size: size, Digest: digest.Digest(art.Digest)} go func() { err := c.putBlobToLocal(remoteRepo, art.Repository, desc, rHelper) @@ -312,6 +317,14 @@ func (c *controller) putBlobToLocal(remoteRepo string, localRepo string, desc di log.Errorf("failed to create blob reader, error %v", err) return err } + // Resume on a dropped upstream connection, same as the client-facing + // fetch above, and verify the assembled content against the expected + // digest so a corrupted/truncated blob that still parses as a valid HTTP + // response is never pushed to local storage as if it were complete. + bReader = newResumingBlobReader(bReader, desc.Size, func(offset int64) (io.ReadCloser, error) { + return r.BlobReaderAt(remoteRepo, string(desc.Digest), desc.Size, offset) + }) + bReader = newVerifyingReadCloser(bReader, desc.Digest) defer bReader.Close() err = c.local.PushBlob(localRepo, desc, bReader) return err diff --git a/src/controller/proxy/controller_test.go b/src/controller/proxy/controller_test.go index 9a3fe7f0236..a64a1ca95da 100644 --- a/src/controller/proxy/controller_test.go +++ b/src/controller/proxy/controller_test.go @@ -60,8 +60,11 @@ func (l *localInterfaceMock) BlobExist(ctx context.Context, art lib.ArtifactInfo return args.Bool(0), args.Error(1) } -func (l *localInterfaceMock) PushBlob(localRepo string, desc distribution.Descriptor, bReader io.ReadCloser) error { - panic("implement me") +func (l *localInterfaceMock) PushBlob(_ string, _ distribution.Descriptor, bReader io.ReadCloser) error { + // Drain the reader the way the real registry client would, so tests can + // exercise the retry/digest-verification wrapping around the reader chain. + _, err := io.ReadAll(bReader) + return err } func (l *localInterfaceMock) PushManifest(repo string, tag string, manifest distribution.Manifest) error { @@ -202,6 +205,34 @@ func (p *proxyControllerTestSuite) TestUseLocalBlob_False() { p.Assert().False(result) } +func (p *proxyControllerTestSuite) TestPutBlobToLocal_ResumesAfterInterruptionAndVerifiesDigest() { + withShortBlobRetryBackoff(p.T()) + content := []byte("hello dependency blob") + dig := digest.FromBytes(content) + desc := distribution.Descriptor{Size: int64(len(content)), Digest: dig} + + first := &stubReader{data: content[:5], err: io.ErrUnexpectedEOF} + second := &stubReader{data: content[5:]} + p.remote.On("BlobReader", "library/hello-world", string(dig)).Return(desc.Size, io.ReadCloser(first), nil) + p.remote.On("BlobReaderAt", "library/hello-world", string(dig), desc.Size, int64(5)).Return(io.ReadCloser(second), nil) + + err := p.ctr.(*controller).putBlobToLocal("library/hello-world", "proxy/library/hello-world", desc, p.remote) + p.Assert().NoError(err) + p.remote.AssertExpectations(p.T()) +} + +func (p *proxyControllerTestSuite) TestPutBlobToLocal_RejectsDigestMismatch() { + content := []byte("actual upstream content") + wrongDigest := digest.FromBytes([]byte("a different blob entirely")) + desc := distribution.Descriptor{Size: int64(len(content)), Digest: wrongDigest} + + p.remote.On("BlobReader", "library/hello-world", string(wrongDigest)).Return(desc.Size, io.ReadCloser(&stubReader{data: content}), nil) + + err := p.ctr.(*controller).putBlobToLocal("library/hello-world", "proxy/library/hello-world", desc, p.remote) + p.Assert().Error(err) + p.remote.AssertExpectations(p.T()) +} + func TestProxyControllerTestSuite(t *testing.T) { suite.Run(t, &proxyControllerTestSuite{}) } diff --git a/src/controller/proxy/manifestcache.go b/src/controller/proxy/manifestcache.go index f3a40ad339c..bb8aebbd47b 100644 --- a/src/controller/proxy/manifestcache.go +++ b/src/controller/proxy/manifestcache.go @@ -18,6 +18,7 @@ import ( "context" "errors" "fmt" + "io" "strings" "time" @@ -244,6 +245,14 @@ func (m *ManifestCache) putBlobToLocal(remoteRepo string, localRepo string, desc log.Errorf("failed to create blob reader, error %v", err) return err } + // Resume on a dropped upstream connection and verify the assembled + // content against the expected digest, same as controller.putBlobToLocal, + // so a corrupted/truncated blob is never pushed to local storage as if + // it were complete. + bReader = newResumingBlobReader(bReader, desc.Size, func(offset int64) (io.ReadCloser, error) { + return r.BlobReaderAt(remoteRepo, string(desc.Digest), desc.Size, offset) + }) + bReader = newVerifyingReadCloser(bReader, desc.Digest) defer bReader.Close() err = m.local.PushBlob(localRepo, desc, bReader) return err diff --git a/src/controller/proxy/manifestcache_test.go b/src/controller/proxy/manifestcache_test.go index cf54e043d0e..72ae67e9373 100644 --- a/src/controller/proxy/manifestcache_test.go +++ b/src/controller/proxy/manifestcache_test.go @@ -17,6 +17,7 @@ package proxy import ( "context" "fmt" + "io" "testing" "github.com/docker/distribution" @@ -28,6 +29,7 @@ import ( "github.com/goharbor/harbor/src/controller/artifact" "github.com/goharbor/harbor/src/lib" + testproxy "github.com/goharbor/harbor/src/testing/controller/proxy" "github.com/goharbor/harbor/src/testing/mock" v1 "github.com/opencontainers/image-spec/specs-go/v1" @@ -251,6 +253,36 @@ func (suite *CacheTestSuite) TestManifestCache_push_fails() { suite.Assert().Contains(errs, tagErr) } +func (suite *CacheTestSuite) TestManifestCache_putBlobToLocal_ResumesAfterInterruption() { + withShortBlobRetryBackoff(suite.T()) + content := []byte("dependency blob content") + dig := digest.FromBytes(content) + desc := distribution.Descriptor{Size: int64(len(content)), Digest: dig} + + first := &stubReader{data: content[:7], err: io.ErrUnexpectedEOF} + second := &stubReader{data: content[7:]} + remote := &testproxy.RemoteInterface{} + remote.On("BlobReader", "library/hello-world", string(dig)).Return(desc.Size, io.ReadCloser(first), nil) + remote.On("BlobReaderAt", "library/hello-world", string(dig), desc.Size, int64(7)).Return(io.ReadCloser(second), nil) + + err := suite.mCache.putBlobToLocal("library/hello-world", "proxy/library/hello-world", desc, remote) + suite.Assert().NoError(err) + remote.AssertExpectations(suite.T()) +} + +func (suite *CacheTestSuite) TestManifestCache_putBlobToLocal_RejectsDigestMismatch() { + content := []byte("dependency blob content") + wrongDigest := digest.FromBytes([]byte("something else")) + desc := distribution.Descriptor{Size: int64(len(content)), Digest: wrongDigest} + + remote := &testproxy.RemoteInterface{} + remote.On("BlobReader", "library/hello-world", string(wrongDigest)).Return(desc.Size, io.ReadCloser(&stubReader{data: content}), nil) + + err := suite.mCache.putBlobToLocal("library/hello-world", "proxy/library/hello-world", desc, remote) + suite.Assert().Error(err) + remote.AssertExpectations(suite.T()) +} + func TestCacheTestSuite(t *testing.T) { suite.Run(t, &CacheTestSuite{}) } diff --git a/src/controller/proxy/remote.go b/src/controller/proxy/remote.go index 781f19a13db..e36ce212642 100644 --- a/src/controller/proxy/remote.go +++ b/src/controller/proxy/remote.go @@ -32,6 +32,10 @@ import ( type RemoteInterface interface { // BlobReader create a reader for remote blob BlobReader(repo, dig string) (int64, io.ReadCloser, error) + // BlobReaderAt creates a reader for the remaining bytes of a remote blob, + // starting at the given offset (via an HTTP range request). It's used to + // resume a blob fetch that was interrupted mid-stream. + BlobReaderAt(repo, dig string, size, start int64) (io.ReadCloser, error) // Manifest get manifest by reference Manifest(repo string, ref string) (distribution.Manifest, string, error) // ManifestExist checks manifest exist, if exist, returns digest @@ -100,6 +104,17 @@ func (r *remoteHelper) BlobReader(repo, dig string) (int64, io.ReadCloser, error return sz, bReader, err } +func (r *remoteHelper) BlobReaderAt(repo, dig string, size, start int64) (io.ReadCloser, error) { + _, bReader, err := r.registry.PullBlobChunk(repo, dig, size, start, size-1) + if err != nil { + return nil, err + } + if r.opts != nil && r.opts.Speed > 0 { + bReader = lib.NewReader(bReader, r.opts.Speed) + } + return bReader, nil +} + func (r *remoteHelper) Manifest(repo string, ref string) (distribution.Manifest, string, error) { return r.registry.PullManifest(repo, ref) } diff --git a/src/testing/controller/proxy/remote_interface.go b/src/testing/controller/proxy/remote_interface.go index fffd363118e..7ed1dd9b5dc 100644 --- a/src/testing/controller/proxy/remote_interface.go +++ b/src/testing/controller/proxy/remote_interface.go @@ -53,6 +53,36 @@ func (_m *RemoteInterface) BlobReader(repo string, dig string) (int64, io.ReadCl return r0, r1, r2 } +// BlobReaderAt provides a mock function with given fields: repo, dig, size, start +func (_m *RemoteInterface) BlobReaderAt(repo string, dig string, size int64, start int64) (io.ReadCloser, error) { + ret := _m.Called(repo, dig, size, start) + + if len(ret) == 0 { + panic("no return value specified for BlobReaderAt") + } + + var r0 io.ReadCloser + var r1 error + if rf, ok := ret.Get(0).(func(string, string, int64, int64) (io.ReadCloser, error)); ok { + return rf(repo, dig, size, start) + } + if rf, ok := ret.Get(0).(func(string, string, int64, int64) io.ReadCloser); ok { + r0 = rf(repo, dig, size, start) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(io.ReadCloser) + } + } + + if rf, ok := ret.Get(1).(func(string, string, int64, int64) error); ok { + r1 = rf(repo, dig, size, start) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + // ListReferrers provides a mock function with given fields: repo, digest, rawQuery func (_m *RemoteInterface) ListReferrers(repo string, digest string, rawQuery string) (*v1.Index, map[string][]string, error) { ret := _m.Called(repo, digest, rawQuery) From 81c56288b42981051527eeb198f2befb44616bef Mon Sep 17 00:00:00 2001 From: Oscar Wieman Date: Wed, 22 Jul 2026 11:11:49 +0200 Subject: [PATCH 2/6] fix(proxycache): validate ranged pulls, retry premature EOF, verify digest at size Addresses three gaps found in review of the initial fix: - PullBlobChunk never checked that the server actually honored the Range request. A registry is allowed to ignore Range and return a full 200 instead of a 206; resumingBlobReader would then splice the full blob (from byte 0) onto bytes already delivered, producing a same-length but corrupted blob - the exact failure mode this fix exists to prevent. Require a 206 whose Content-Range starts at the requested offset, or a plain 200 only when the whole blob was requested from the start. - resumingBlobReader accepted a clean io.EOF as a complete blob even when short of the known size. A network intermediary that terminates a stream gracefully (rather than resetting the connection) produces exactly this, and it would previously bypass retry entirely. Treat a premature clean EOF the same as a dropped connection. - verifyingReadCloser only checked the digest when its own Read returned io.EOF, but the real consumer (an HTTP request body capped by Content-Length) wraps it in io.LimitReader, which returns EOF itself once the declared length is reached and may never call back into the wrapped reader again if the last real read already returned its final bytes with a nil error. Track the expected size and verify as soon as it's reached, not only on a subsequent EOF call. Signed-off-by: Oscar Wieman --- src/controller/proxy/blobretry.go | 50 ++++++++++---- src/controller/proxy/blobretry_test.go | 42 +++++++++++- src/controller/proxy/controller.go | 2 +- src/controller/proxy/manifestcache.go | 2 +- src/pkg/registry/client.go | 36 +++++++++++ src/pkg/registry/client_test.go | 90 ++++++++++++++++++++++++++ 6 files changed, 205 insertions(+), 17 deletions(-) diff --git a/src/controller/proxy/blobretry.go b/src/controller/proxy/blobretry.go index a8dcebf27e0..c7d6f0c5ca5 100644 --- a/src/controller/proxy/blobretry.go +++ b/src/controller/proxy/blobretry.go @@ -66,9 +66,21 @@ func (r *resumingBlobReader) Read(p []byte) (int, error) { for { n, err := r.reader.Read(p) r.offset += int64(n) - if err == nil || err == io.EOF { // nolint:errorlint + if err == nil { return n, err } + if err == io.EOF { // nolint:errorlint + if r.size > 0 && r.offset < r.size { + // The stream ended cleanly but short of the expected size - + // e.g. a network intermediary that terminates a chunked + // response gracefully instead of resetting the connection. + // Treat it the same as a dropped connection so it consumes + // the retry budget instead of being accepted as complete. + err = io.ErrUnexpectedEOF + } else { + return n, err + } + } if !r.canRetry(err) { return n, err } @@ -120,31 +132,43 @@ func isRetryableBlobReadErr(err error) bool { return err != nil && !errors.Is(err, context.Canceled) } -// verifyingReadCloser wraps a blob reader and, once the wrapped reader signals -// a genuine end of stream, verifies the bytes read so far match the expected -// digest. A mismatch is surfaced as a Read error instead of io.EOF, so a -// corrupted or truncated blob that still parses as a well-formed HTTP -// response is never mistaken for a complete, valid one by the caller (e.g. -// the local registry committing a blob push). +// verifyingReadCloser wraps a blob reader and verifies the bytes read so far +// match the expected digest as soon as the declared size has been read - not +// only when the wrapped reader happens to signal io.EOF on some later call. +// A consumer with its own framing (e.g. an HTTP request body capped by +// Content-Length) may never issue that extra call, since it can determine +// on its own that it has all the bytes it declared and stop reading. A +// mismatch is surfaced as a Read error, so a corrupted or truncated blob +// that still parses as a well-formed HTTP response is never mistaken for a +// complete, valid one by the caller (e.g. the local registry committing a +// blob push). type verifyingReadCloser struct { io.ReadCloser verifier digest.Verifier dig digest.Digest + size int64 + read int64 + verified bool } -// newVerifyingReadCloser wraps reader so that reaching the end of the stream -// without the content matching dig surfaces as a Read error. -func newVerifyingReadCloser(reader io.ReadCloser, dig digest.Digest) io.ReadCloser { - return &verifyingReadCloser{ReadCloser: reader, verifier: dig.Verifier(), dig: dig} +// newVerifyingReadCloser wraps reader so that, once size bytes have been +// read (or the stream ends, if size is unknown), content not matching dig +// surfaces as a Read error. +func newVerifyingReadCloser(reader io.ReadCloser, dig digest.Digest, size int64) io.ReadCloser { + return &verifyingReadCloser{ReadCloser: reader, verifier: dig.Verifier(), dig: dig, size: size} } func (v *verifyingReadCloser) Read(p []byte) (int, error) { n, err := v.ReadCloser.Read(p) if n > 0 { _, _ = v.verifier.Write(p[:n]) + v.read += int64(n) } - if err == io.EOF && !v.verifier.Verified() { // nolint:errorlint - return n, fmt.Errorf("blob content does not match expected digest %s after fetching from upstream", v.dig) + if !v.verified && ((v.size > 0 && v.read >= v.size) || err == io.EOF) { // nolint:errorlint + v.verified = true + if !v.verifier.Verified() { + return n, fmt.Errorf("blob content does not match expected digest %s after fetching from upstream", v.dig) + } } return n, err } diff --git a/src/controller/proxy/blobretry_test.go b/src/controller/proxy/blobretry_test.go index cd400f44d6e..31a0fad2204 100644 --- a/src/controller/proxy/blobretry_test.go +++ b/src/controller/proxy/blobretry_test.go @@ -15,6 +15,7 @@ package proxy import ( + "bytes" "context" "errors" "io" @@ -142,11 +143,32 @@ func TestResumingBlobReader_RecoversFromMultipleInterruptions(t *testing.T) { assert.Equal(t, []int64{3, 6}, resumeOffsets) } +func TestResumingBlobReader_RetriesOnPrematureCleanEOF(t *testing.T) { + withShortBlobRetryBackoff(t) + + // first ends with a clean nil error (io.EOF on the next call), not a + // transport error - e.g. a network intermediary that terminates a + // chunked response gracefully instead of resetting the connection. + first := &stubReader{data: []byte("hello ")} + second := &stubReader{data: []byte("world")} + + var resumeOffsets []int64 + reader := newResumingBlobReader(first, 11, func(offset int64) (io.ReadCloser, error) { + resumeOffsets = append(resumeOffsets, offset) + return second, nil + }) + + got, err := io.ReadAll(reader) + require.NoError(t, err) + assert.Equal(t, "hello world", string(got)) + assert.Equal(t, []int64{6}, resumeOffsets, "a clean EOF short of the declared size must still trigger a resume") +} + func TestVerifyingReadCloser_PassesThroughValidContent(t *testing.T) { content := []byte("valid blob content") dig := digest.FromBytes(content) - reader := newVerifyingReadCloser(io.NopCloser(&stubReader{data: content}), dig) + reader := newVerifyingReadCloser(io.NopCloser(&stubReader{data: content}), dig, int64(len(content))) got, err := io.ReadAll(reader) require.NoError(t, err) assert.Equal(t, content, got) @@ -156,8 +178,24 @@ func TestVerifyingReadCloser_RejectsMismatchedContent(t *testing.T) { content := []byte("corrupted blob content") wrongDigest := digest.FromBytes([]byte("something else entirely")) - reader := newVerifyingReadCloser(io.NopCloser(&stubReader{data: content}), wrongDigest) + reader := newVerifyingReadCloser(io.NopCloser(&stubReader{data: content}), wrongDigest, int64(len(content))) _, err := io.ReadAll(reader) require.Error(t, err) assert.Contains(t, err.Error(), "does not match expected digest") } + +func TestVerifyingReadCloser_VerifiesAssoonAsSizeReachedWithoutFurtherEOFCall(t *testing.T) { + // bytes.Reader returns the final chunk with a nil error and only signals + // io.EOF on a subsequent call - the same pattern net/http's + // io.LimitReader-wrapped request body uses. A consumer capped by + // Content-Length (e.g. the real blob upload) may never make that + // subsequent call, so verification must not depend on it. + content := []byte("AAAAAAAAAA") + wrongDigest := digest.FromBytes([]byte("BBBBBBBBBB")) + + reader := newVerifyingReadCloser(io.NopCloser(bytes.NewReader(content)), wrongDigest, int64(len(content))) + buf := make([]byte, len(content)) + n, err := reader.Read(buf) + require.Error(t, err, "the read call that reaches the declared size must itself surface the mismatch") + assert.Equal(t, len(content), n) +} diff --git a/src/controller/proxy/controller.go b/src/controller/proxy/controller.go index 00604b7a466..b0635b23bfc 100644 --- a/src/controller/proxy/controller.go +++ b/src/controller/proxy/controller.go @@ -324,7 +324,7 @@ func (c *controller) putBlobToLocal(remoteRepo string, localRepo string, desc di bReader = newResumingBlobReader(bReader, desc.Size, func(offset int64) (io.ReadCloser, error) { return r.BlobReaderAt(remoteRepo, string(desc.Digest), desc.Size, offset) }) - bReader = newVerifyingReadCloser(bReader, desc.Digest) + bReader = newVerifyingReadCloser(bReader, desc.Digest, desc.Size) defer bReader.Close() err = c.local.PushBlob(localRepo, desc, bReader) return err diff --git a/src/controller/proxy/manifestcache.go b/src/controller/proxy/manifestcache.go index bb8aebbd47b..57ada615778 100644 --- a/src/controller/proxy/manifestcache.go +++ b/src/controller/proxy/manifestcache.go @@ -252,7 +252,7 @@ func (m *ManifestCache) putBlobToLocal(remoteRepo string, localRepo string, desc bReader = newResumingBlobReader(bReader, desc.Size, func(offset int64) (io.ReadCloser, error) { return r.BlobReaderAt(remoteRepo, string(desc.Digest), desc.Size, offset) }) - bReader = newVerifyingReadCloser(bReader, desc.Digest) + bReader = newVerifyingReadCloser(bReader, desc.Digest, desc.Size) defer bReader.Close() err = m.local.PushBlob(localRepo, desc, bReader) return err diff --git a/src/pkg/registry/client.go b/src/pkg/registry/client.go index 0b59b5582cb..0295702be3f 100644 --- a/src/pkg/registry/client.go +++ b/src/pkg/registry/client.go @@ -398,6 +398,31 @@ func (c *client) PullBlobChunk(repository, digest string, _ int64, start, end in return 0, nil, err } + // A server is allowed to ignore the Range header and return the full blob + // with a 200 instead of a 206. Callers use this method to resume a fetch + // from a byte offset, so silently accepting a mismatched range here would + // splice bytes from the wrong offset into whatever they're assembling. + // Require a 206 whose Content-Range actually starts at the requested + // offset, or a plain 200 only when the whole blob was requested from the + // start. + switch resp.StatusCode { + case http.StatusPartialContent: + cr := resp.Header.Get("Content-Range") + got, crErr := parseContentRangeStart(cr) + if crErr != nil || got != start { + defer resp.Body.Close() + return 0, nil, fmt.Errorf("registry returned Content-Range %q for requested range bytes=%d-%d, refusing to treat it as that chunk", cr, start, end) + } + case http.StatusOK: + if start != 0 { + defer resp.Body.Close() + return 0, nil, fmt.Errorf("registry ignored range request bytes=%d-%d and returned a full 200 response", start, end) + } + default: + defer resp.Body.Close() + return 0, nil, fmt.Errorf("unexpected status code %d for ranged blob pull", resp.StatusCode) + } + var size int64 n := resp.Header.Get("Content-Length") // no content-length is acceptable, which can taken from manifests @@ -412,6 +437,17 @@ func (c *client) PullBlobChunk(repository, digest string, _ int64, start, end in return size, resp.Body, nil } +// parseContentRangeStart extracts the start offset from a GET response's +// Content-Range header, e.g. "bytes 100-499/1234" or "bytes 100-499/*". +func parseContentRangeStart(cr string) (int64, error) { + cr = strings.TrimPrefix(cr, "bytes ") + dash := strings.IndexByte(cr, '-') + if dash <= 0 { + return -1, fmt.Errorf("invalid Content-Range format: %q", cr) + } + return strconv.ParseInt(cr[:dash], 10, 64) +} + func (c *client) PushBlob(repository, digest string, size int64, blob io.Reader) error { location, _, err := c.initiateBlobUpload(repository) if err != nil { diff --git a/src/pkg/registry/client_test.go b/src/pkg/registry/client_test.go index 5bde2325bdc..56136b5a823 100644 --- a/src/pkg/registry/client_test.go +++ b/src/pkg/registry/client_test.go @@ -325,6 +325,96 @@ func (c *clientTestSuite) TestPullBlob() { c.EqualValues(data, b) } +func (c *clientTestSuite) TestPullBlobChunk_PartialContentMatchingRange() { + data := []byte("world") + server := test.NewServer( + &test.RequestHandlerMapping{ + Method: "GET", + Pattern: "/v2/library/hello-world/blobs/digest", + Handler: test.Handler(&test.Response{ + StatusCode: http.StatusPartialContent, + Headers: map[string]string{ + "Content-Length": strconv.Itoa(len(data)), + "Content-Range": "bytes 6-10/11", + }, + Body: data, + }), + }) + defer server.Close() + + size, blob, err := NewClient(server.URL, "", "", true).PullBlobChunk("library/hello-world", "digest", 11, 6, 10) + c.Require().NoError(err) + c.Equal(int64(len(data)), size) + b, err := io.ReadAll(blob) + c.Require().NoError(err) + c.EqualValues(data, b) +} + +func (c *clientTestSuite) TestPullBlobChunk_FullContentAtZeroOffset() { + data := []byte("hello world") + server := test.NewServer( + &test.RequestHandlerMapping{ + Method: "GET", + Pattern: "/v2/library/hello-world/blobs/digest", + Handler: test.Handler(&test.Response{ + StatusCode: http.StatusOK, + Headers: map[string]string{ + "Content-Length": strconv.Itoa(len(data)), + }, + Body: data, + }), + }) + defer server.Close() + + size, blob, err := NewClient(server.URL, "", "", true).PullBlobChunk("library/hello-world", "digest", 11, 0, 10) + c.Require().NoError(err) + c.Equal(int64(len(data)), size) + b, err := io.ReadAll(blob) + c.Require().NoError(err) + c.EqualValues(data, b) +} + +func (c *clientTestSuite) TestPullBlobChunk_RejectsIgnoredRangeAtNonZeroOffset() { + data := []byte("hello world") + server := test.NewServer( + &test.RequestHandlerMapping{ + Method: "GET", + Pattern: "/v2/library/hello-world/blobs/digest", + Handler: test.Handler(&test.Response{ + StatusCode: http.StatusOK, + Headers: map[string]string{ + "Content-Length": strconv.Itoa(len(data)), + }, + Body: data, + }), + }) + defer server.Close() + + _, _, err := NewClient(server.URL, "", "", true).PullBlobChunk("library/hello-world", "digest", 11, 6, 10) + c.Require().Error(err) +} + +func (c *clientTestSuite) TestPullBlobChunk_RejectsMismatchedContentRange() { + data := []byte("wrong chunk") + server := test.NewServer( + &test.RequestHandlerMapping{ + Method: "GET", + Pattern: "/v2/library/hello-world/blobs/digest", + Handler: test.Handler(&test.Response{ + StatusCode: http.StatusPartialContent, + Headers: map[string]string{ + "Content-Length": strconv.Itoa(len(data)), + "Content-Range": "bytes 0-10/11", + }, + Body: data, + }), + }) + defer server.Close() + + _, _, err := NewClient(server.URL, "", "", true).PullBlobChunk("library/hello-world", "digest", 11, 6, 10) + c.Require().Error(err) +} + func (c *clientTestSuite) TestPushBlob() { server := test.NewServer( &test.RequestHandlerMapping{ From 2a476c78002d4ab1a0f35c2d813a343d816fae11 Mon Sep 17 00:00:00 2001 From: Oscar Wieman Date: Wed, 22 Jul 2026 11:23:26 +0200 Subject: [PATCH 3/6] fix(proxycache): tighten chunk-range validation, cancel retry loop on ctx Addresses two more gaps found in the second round of review: - PullBlobChunk's 200 fallback only checked start == 0, not that the whole blob was requested. copyBlobByChunk (replication) requests a start=0 first chunk that is shorter than the full blob for any blob bigger than one chunk, so an origin ignoring Range could return a full 200 that gets accepted as if it were just that first chunk. Use the (previously discarded) blobSize parameter to require start == 0 && end == blobSize-1 before accepting a 200, falling back to the weaker check only when blobSize is unknown. - resumingBlobReader's backoff wait and resume attempts weren't cancelable. If the pulling client disconnects mid-retry, the reader would still sleep out the remaining backoff and open further upstream requests - each unbounded by any context, since RemoteInterface doesn't accept one - while holding a connection-limit slot. Thread a context.Context through resumingBlobReader so backoff waits and not-yet-started resume attempts respect cancellation. The client-serving fetch uses the request context; both cache-populate fetches use context.Background(), since that background work is meant to keep going after the client disconnects. Signed-off-by: Oscar Wieman --- src/controller/proxy/blobretry.go | 29 +++++++++++++++++++++---- src/controller/proxy/blobretry_test.go | 30 ++++++++++++++++++++------ src/controller/proxy/controller.go | 12 ++++++++--- src/controller/proxy/manifestcache.go | 5 +++-- src/pkg/registry/client.go | 11 ++++++++-- src/pkg/registry/client_test.go | 23 ++++++++++++++++++++ 6 files changed, 93 insertions(+), 17 deletions(-) diff --git a/src/controller/proxy/blobretry.go b/src/controller/proxy/blobretry.go index c7d6f0c5ca5..d1cdeeca8d9 100644 --- a/src/controller/proxy/blobretry.go +++ b/src/controller/proxy/blobretry.go @@ -47,7 +47,14 @@ type resumeBlobFunc func(offset int64) (io.ReadCloser, error) // docker/OCI client retries an interrupted layer pull. Resuming is bounded by // maxBlobFetchRetries and disabled when size is unknown (<= 0), since the // read offset can't be validated against the blob length in that case. +// +// resume itself isn't cancelable once a request is in flight - the +// underlying RemoteInterface doesn't accept a context - but the backoff wait +// between attempts, and any attempt not yet started, respect ctx, so a +// canceled caller (e.g. a disconnected pulling client) stops this reader +// from waiting out further backoffs and opening further upstream requests. type resumingBlobReader struct { + ctx context.Context reader io.ReadCloser resume resumeBlobFunc size int64 @@ -58,8 +65,12 @@ type resumingBlobReader struct { // newResumingBlobReader wraps reader so a mid-stream read failure is retried // up to maxBlobFetchRetries times by resuming the fetch from the last // successfully read byte offset (via an HTTP range request through resume). -func newResumingBlobReader(reader io.ReadCloser, size int64, resume resumeBlobFunc) io.ReadCloser { - return &resumingBlobReader{reader: reader, size: size, resume: resume} +// ctx bounds the retry loop itself (see resumingBlobReader); pass the +// pulling client's request context to give up promptly if it disconnects, +// or a detached context (e.g. context.Background()) for background work +// that should keep retrying regardless. +func newResumingBlobReader(ctx context.Context, reader io.ReadCloser, size int64, resume resumeBlobFunc) io.ReadCloser { + return &resumingBlobReader{ctx: ctx, reader: reader, size: size, resume: resume} } func (r *resumingBlobReader) Read(p []byte) (int, error) { @@ -96,15 +107,25 @@ func (r *resumingBlobReader) Read(p []byte) (int, error) { // reconnect closes the broken reader and resumes the fetch from the current // offset, retrying the resume itself (bounded by the same retry budget) if -// establishing the new connection also fails. +// establishing the new connection also fails. It gives up early, without +// waiting out the backoff or starting another attempt, if ctx is canceled. func (r *resumingBlobReader) reconnect(cause error) error { _ = r.reader.Close() for { + if ctxErr := r.ctx.Err(); ctxErr != nil { + return fmt.Errorf("blob fetch canceled while resuming after %v: %w", cause, ctxErr) + } r.retries++ backoff := blobFetchRetryBackoff * time.Duration(uint64(1)< Date: Wed, 22 Jul 2026 11:34:34 +0200 Subject: [PATCH 4/6] test(proxycache): fix typo in test name (AsSoon) Signed-off-by: Oscar Wieman --- src/controller/proxy/blobretry_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/controller/proxy/blobretry_test.go b/src/controller/proxy/blobretry_test.go index e2b4391ef46..a4054a2cb3a 100644 --- a/src/controller/proxy/blobretry_test.go +++ b/src/controller/proxy/blobretry_test.go @@ -202,7 +202,7 @@ func TestVerifyingReadCloser_RejectsMismatchedContent(t *testing.T) { assert.Contains(t, err.Error(), "does not match expected digest") } -func TestVerifyingReadCloser_VerifiesAssoonAsSizeReachedWithoutFurtherEOFCall(t *testing.T) { +func TestVerifyingReadCloser_VerifiesAsSoonAsSizeReachedWithoutFurtherEOFCall(t *testing.T) { // bytes.Reader returns the final chunk with a nil error and only signals // io.EOF on a subsequent call - the same pattern net/http's // io.LimitReader-wrapped request body uses. A consumer capped by From 638cb24d9bb3a0c9503a28dace19c1c43f9a2b0c Mon Sep 17 00:00:00 2001 From: Oscar Wieman Date: Wed, 22 Jul 2026 11:44:17 +0200 Subject: [PATCH 5/6] fix(proxycache): treat an error accompanying the final blob bytes as success An io.Reader may legally return its last bytes together with a non-nil, non-EOF error in the same call, instead of returning them cleanly and erroring only on a subsequent call. resumingBlobReader didn't account for this: if those bytes brought offset to the full declared size, canRetry returned false (since offset is no longer < size) and the error was propagated anyway, even though the complete blob had just been assembled. Check for having reached the declared size before looking at the accompanying error, and treat that as success regardless. A completed flag makes the reader self-terminate (return io.EOF) on any further call, instead of re-reading a reader that may now be stale or exhausted - which a naive "just don't retry once size is reached" fix would otherwise turn into a spurious (0, nil) reads. Signed-off-by: Oscar Wieman --- src/controller/proxy/blobretry.go | 48 ++++++++++++++++++-------- src/controller/proxy/blobretry_test.go | 34 ++++++++++++++++++ 2 files changed, 67 insertions(+), 15 deletions(-) diff --git a/src/controller/proxy/blobretry.go b/src/controller/proxy/blobretry.go index d1cdeeca8d9..b1756863a1a 100644 --- a/src/controller/proxy/blobretry.go +++ b/src/controller/proxy/blobretry.go @@ -54,12 +54,13 @@ type resumeBlobFunc func(offset int64) (io.ReadCloser, error) // canceled caller (e.g. a disconnected pulling client) stops this reader // from waiting out further backoffs and opening further upstream requests. type resumingBlobReader struct { - ctx context.Context - reader io.ReadCloser - resume resumeBlobFunc - size int64 - offset int64 - retries int + ctx context.Context + reader io.ReadCloser + resume resumeBlobFunc + size int64 + offset int64 + retries int + completed bool } // newResumingBlobReader wraps reader so a mid-stream read failure is retried @@ -74,23 +75,40 @@ func newResumingBlobReader(ctx context.Context, reader io.ReadCloser, size int64 } func (r *resumingBlobReader) Read(p []byte) (int, error) { + if r.completed { + // A prior call already reached the declared size; self-terminate + // instead of calling into a reader that may now be stale/exhausted. + return 0, io.EOF + } for { n, err := r.reader.Read(p) r.offset += int64(n) + + if r.size > 0 && r.offset >= r.size { + // We now have the full declared size, regardless of what error + // (if any) came with these final bytes - an io.Reader may + // legally pair its last bytes with a non-nil error instead of + // returning them cleanly and erroring only on a subsequent + // call. Report success once; further calls are handled by the + // completed check above rather than re-reading this reader. + r.completed = true + return n, nil + } + if err == nil { - return n, err + return n, nil } if err == io.EOF { // nolint:errorlint - if r.size > 0 && r.offset < r.size { - // The stream ended cleanly but short of the expected size - - // e.g. a network intermediary that terminates a chunked - // response gracefully instead of resetting the connection. - // Treat it the same as a dropped connection so it consumes - // the retry budget instead of being accepted as complete. - err = io.ErrUnexpectedEOF - } else { + if r.size <= 0 { + // Nothing to validate the EOF against; trust it. return n, err } + // The stream ended cleanly but short of the expected size - + // e.g. a network intermediary that terminates a chunked + // response gracefully instead of resetting the connection. + // Treat it the same as a dropped connection so it consumes + // the retry budget instead of being accepted as complete. + err = io.ErrUnexpectedEOF } if !r.canRetry(err) { return n, err diff --git a/src/controller/proxy/blobretry_test.go b/src/controller/proxy/blobretry_test.go index a4054a2cb3a..8f3460bc1e9 100644 --- a/src/controller/proxy/blobretry_test.go +++ b/src/controller/proxy/blobretry_test.go @@ -52,6 +52,22 @@ func (s *stubReader) Close() error { return nil } +// oneShotReader returns all of data together with err in a single Read call, +// unlike stubReader, which always separates the final data from a +// subsequent terminal error/EOF call - both are legal io.Reader behavior. +type oneShotReader struct { + data []byte + err error +} + +func (o *oneShotReader) Read(p []byte) (int, error) { + n := copy(p, o.data) + o.data = o.data[n:] + return n, o.err +} + +func (o *oneShotReader) Close() error { return nil } + func withShortBlobRetryBackoff(t *testing.T) { orig := blobFetchRetryBackoff blobFetchRetryBackoff = time.Millisecond @@ -182,6 +198,24 @@ func TestResumingBlobReader_RetriesOnPrematureCleanEOF(t *testing.T) { assert.Equal(t, []int64{6}, resumeOffsets, "a clean EOF short of the declared size must still trigger a resume") } +func TestResumingBlobReader_TreatsErrorAccompanyingFinalBytesAsSuccess(t *testing.T) { + // An io.Reader may legally return its final bytes together with a + // non-nil, non-EOF error in the same call, instead of returning them + // cleanly and erroring only on a subsequent call. If those bytes bring + // the reader to the full declared size, that's a complete blob and must + // not be treated as a failed/retryable read. + content := []byte("complete blob") + reader := newResumingBlobReader(context.Background(), &oneShotReader{data: content, err: io.ErrUnexpectedEOF}, int64(len(content)), + func(offset int64) (io.ReadCloser, error) { + t.Fatal("resume should not be called when the blob was already fully read") + return nil, nil + }) + + got, err := io.ReadAll(reader) + require.NoError(t, err) + assert.Equal(t, content, got) +} + func TestVerifyingReadCloser_PassesThroughValidContent(t *testing.T) { content := []byte("valid blob content") dig := digest.FromBytes(content) From bbc3c780dabb855190c040bbcbd7211b7a0543e6 Mon Sep 17 00:00:00 2001 From: Oscar Wieman Date: Thu, 23 Jul 2026 13:54:29 +0200 Subject: [PATCH 6/6] fix(proxycache): don't retry a resume attempt that gets rate-limited Raised in PR review: an interrupted blob whose resume attempt hits a 429 would previously be retried like any other transient failure. A few seconds of exponential backoff won't clear a real rate limit and just adds more rejected requests to an already-throttled upstream, so give up as soon as errors.IsRateLimitError reports the resume's error, instead of burning through the rest of the retry budget on attempts likely to fail the same way. Signed-off-by: Oscar Wieman --- src/controller/proxy/blobretry.go | 9 +++++++++ src/controller/proxy/blobretry_test.go | 18 ++++++++++++++++++ 2 files changed, 27 insertions(+) diff --git a/src/controller/proxy/blobretry.go b/src/controller/proxy/blobretry.go index b1756863a1a..c03144617ab 100644 --- a/src/controller/proxy/blobretry.go +++ b/src/controller/proxy/blobretry.go @@ -23,6 +23,7 @@ import ( "github.com/opencontainers/go-digest" + libErrors "github.com/goharbor/harbor/src/lib/errors" "github.com/goharbor/harbor/src/lib/log" ) @@ -149,6 +150,14 @@ func (r *resumingBlobReader) reconnect(cause error) error { r.reader = next return nil } + if libErrors.IsRateLimitError(err) { + // A registry that's rate-limiting won't be helped by retrying + // within a few seconds of backoff - real rate limits typically + // need a much longer cool-down - so give up instead of burning + // through the retry budget on requests likely to fail the same + // way and adding more load to an already-throttled upstream. + return fmt.Errorf("upstream rate-limited the resume attempt after %v: %w", cause, err) + } if r.retries >= maxBlobFetchRetries { return fmt.Errorf("failed to resume interrupted blob fetch after %v: %w", cause, err) } diff --git a/src/controller/proxy/blobretry_test.go b/src/controller/proxy/blobretry_test.go index 8f3460bc1e9..d10cc1365c2 100644 --- a/src/controller/proxy/blobretry_test.go +++ b/src/controller/proxy/blobretry_test.go @@ -25,6 +25,8 @@ import ( "github.com/opencontainers/go-digest" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + + libErrors "github.com/goharbor/harbor/src/lib/errors" ) // stubReader serves data and then, once exhausted, returns err (or io.EOF if @@ -108,6 +110,22 @@ func TestResumingBlobReader_GivesUpAfterMaxRetries(t *testing.T) { assert.Equal(t, maxBlobFetchRetries, resumeAttempts) } +func TestResumingBlobReader_GivesUpImmediatelyOnRateLimitedResume(t *testing.T) { + withShortBlobRetryBackoff(t) + + first := &stubReader{err: io.ErrUnexpectedEOF} + resumeAttempts := 0 + rateLimitErr := libErrors.New("too many requests").WithCode(libErrors.RateLimitCode) + reader := newResumingBlobReader(context.Background(), first, 100, func(offset int64) (io.ReadCloser, error) { + resumeAttempts++ + return nil, rateLimitErr + }) + + _, err := io.ReadAll(reader) + require.Error(t, err) + assert.Equal(t, 1, resumeAttempts, "a rate-limited resume shouldn't be retried - a few seconds of backoff won't help and it just adds more load") +} + func TestResumingBlobReader_DoesNotRetryOnContextCanceled(t *testing.T) { withShortBlobRetryBackoff(t)