proxy: add opt-in Range pass-through

Relay cache-miss blob Range requests to the upstream when
proxy.rangepassthrough is enabled, without populating the blob cache.
Preserve upstream 200, 206 and 416 responses while keeping ordinary GET
cache fills and cache-hit handling unchanged.

Reuse the authenticated client, abort interrupted streams, and exclude
416 responses from blob transfer metrics and pull notifications.
Document the option and cover the HTTP and cache behavior in tests.

Signed-off-by: jeongho-kim <jeongho.kim@toss.im>

configuration/configuration.go

@@ -698,6 +698,10 @@ type Proxy struct {
 	// RemoteURL is the URL of the remote registry
 	RemoteURL string `yaml:"remoteurl"`
 
+	// RangePassthrough forwards GET Range requests to the remote registry on a
+	// cache miss without populating the blob cache. Defaults to false.
+	RangePassthrough bool `yaml:"rangepassthrough,omitempty"`
+
 	// Username of the hub user
 	Username string `yaml:"username"`
 

docs/content/about/configuration.md

@@ -1274,6 +1274,7 @@ proxy:
   username: [username]
   password: [password]
   ttl: 168h
+  rangepassthrough: false
 ```
 
 The `proxy` structure allows a registry to be configured as a pull-through cache
@@ -1286,6 +1287,22 @@ is unsupported.
 |-----------|----------|-------------------------------------------------------|
 | `remoteurl`| yes     | The URL for the repository on Docker Hub.             |
 | `ttl`      | no      | Expire proxy cache configured in "storage" after this time. Cache 168h(7 days) by default, set to 0 to disable cache expiration, The suffix is one of `ns`, `us`, `ms`, `s`, `m`, or `h`. If you specify a value but omit the suffix, the value is interpreted as a number of nanoseconds. |
+| `rangepassthrough` | no | Forward cache-miss blob GET requests with a non-empty Range header to the upstream without caching the response. Defaults to `false`. |
+
+Set `proxy.rangepassthrough: true` or
+`REGISTRY_PROXY_RANGEPASSTHROUGH=true` to enable origin Range pass-through.
+Omitting the setting or setting it to `false` preserves the existing cache-miss
+behavior. This option does not disable Range support for cached blobs: cache
+hits retain local serving or storage redirects. Ordinary GET requests still
+fetch and cache the complete blob, and HEAD requests keep their existing path.
+
+With pass-through enabled, the upstream interprets `Range` and `If-Range`.
+Its `206` response (including multipart ranges) is relayed without assembling
+pieces or verifying the whole blob digest. If the upstream ignores Range or
+`If-Range` selects the full representation, its `200` response is also relayed
+without caching. An unsatisfiable range is relayed as `416`, including its
+`Content-Range` and body. Traffic consisting only of Range requests therefore
+does not populate the registry's complete blob cache.
 
 To enable pulling private repositories (e.g. `batman/robin`), specify one of the
 following authentication methods for the pull-through cache to authenticate with

internal/client/blob_range_test.go

@@ -0,0 +1,154 @@
+package client
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/distribution/distribution/v3"
+	"github.com/distribution/reference"
+	"github.com/opencontainers/go-digest"
+)
+
+func TestServeBlobRange(t *testing.T) {
+	data := []byte("0123456789abcdef")
+	dgst := digest.FromBytes(data)
+	for _, tc := range []struct {
+		name, value, ifRange string
+		status               int
+	}{
+		{"bounded", "bytes=2-5", "", 206},
+		{"open", "bytes=12-", "", 206},
+		{"suffix", "bytes=-4", "", 206},
+		{"multipart", "bytes=0-1,8-9", "", 206},
+		{"if-range fallback", "bytes=2-5", `"old"`, 200},
+		{"ignored", "bytes=2-5", "", 200},
+		{"unsatisfiable", "bytes=99-", "", 416},
+		{"missing", "bytes=2-5", "", 404},
+		{"failure", "bytes=2-5", "", 500},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			responses := make(chan *httptest.ResponseRecorder, 1)
+			// Use a distinct hostname on the redirect destination, as object stores do.
+			target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				if r.Method != http.MethodGet || r.Header.Get("Range") != tc.value || r.Header.Get("If-Range") != tc.ifRange || r.Header.Get("Accept-Encoding") != "identity" {
+					t.Errorf("origin request: %s %v", r.Method, r.Header)
+				}
+				if r.Header.Get("Authorization") != "" || r.Header.Get("Cookie") != "" {
+					t.Error("forwarded downstream credentials")
+				}
+				expected := httptest.NewRecorder()
+				expected.Header().Set("Etag", `"current"`)
+				if tc.status >= 400 && tc.status != 416 {
+					expected.WriteHeader(tc.status)
+				} else {
+					if tc.name == "ignored" {
+						r.Header.Del("Range")
+					}
+					http.ServeContent(expected, r, "", time.Time{}, bytes.NewReader(data))
+				}
+				responses <- expected
+				for k, values := range expected.Header() {
+					w.Header()[k] = values
+				}
+				w.Header().Set("Connection", "X-Origin-Only, Etag")
+				w.Header().Set("X-Origin-Only", "private")
+				w.WriteHeader(expected.Code)
+				// Flush exercises the relay with a streamed response as well.
+				w.(http.Flusher).Flush()
+				_, _ = w.Write(expected.Body.Bytes())
+			}))
+			defer target.Close()
+			origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				http.Redirect(w, r, strings.Replace(target.URL, "127.0.0.1", "localhost", 1), http.StatusTemporaryRedirect)
+			}))
+			defer origin.Close()
+			name, _ := reference.WithName("test/blob")
+			repo, err := NewRepository(name, origin.URL, nil)
+			if err != nil {
+				t.Fatal(err)
+			}
+			req := httptest.NewRequest(http.MethodGet, "/", nil)
+			req.Header.Set("Range", tc.value)
+			req.Header.Set("If-Range", tc.ifRange)
+			req.Header.Set("Authorization", "Bearer downstream-only")
+			req.Header.Set("Cookie", "session=downstream-only")
+			w := httptest.NewRecorder()
+			err = repo.Blobs(context.Background()).ServeBlob(req.Context(), w, req, dgst)
+			if tc.status == 404 {
+				if err != distribution.ErrBlobUnknown {
+					t.Fatalf("missing blob: %v", err)
+				}
+				return
+			}
+			if tc.status == 500 {
+				if _, ok := err.(*UnexpectedHTTPStatusError); !ok {
+					t.Fatalf("origin failure: %v", err)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatal(err)
+			}
+			expected := <-responses
+			if w.Code != tc.status || !bytes.Equal(w.Body.Bytes(), expected.Body.Bytes()) {
+				t.Fatalf("relay: status=%d body=%q, want %d %q", w.Code, w.Body.Bytes(), tc.status, expected.Body.Bytes())
+			}
+			for _, key := range []string{"Content-Range", "Content-Length", "Content-Type"} {
+				if w.Header().Get(key) != expected.Header().Get(key) {
+					t.Errorf("%s: got %q want %q", key, w.Header().Get(key), expected.Header().Get(key))
+				}
+			}
+			for _, key := range []string{"Connection", "X-Origin-Only", "Etag"} {
+				if w.Header().Get(key) != "" {
+					t.Errorf("forwarded hop-by-hop header %s", key)
+				}
+			}
+		})
+	}
+}
+
+func TestServeBlobRangeCanceled(t *testing.T) {
+	started, canceled := make(chan struct{}), make(chan struct{})
+	origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		close(started)
+		<-r.Context().Done()
+		close(canceled)
+	}))
+	defer origin.Close()
+	name, _ := reference.WithName("test/blob")
+	repo, err := NewRepository(name, origin.URL, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	req := httptest.NewRequest(http.MethodGet, "/", nil).WithContext(ctx)
+	req.Header.Set("Range", "bytes=0-1")
+	done := make(chan error, 1)
+	go func() { done <- repo.Blobs(ctx).ServeBlob(ctx, httptest.NewRecorder(), req, digest.FromString("blob")) }()
+	select {
+	case <-started:
+	case <-time.After(5 * time.Second):
+		t.Fatal("origin request did not start")
+	}
+	cancel()
+	select {
+	case err := <-done:
+		if !errors.Is(err, context.Canceled) {
+			t.Fatalf("canceled request: %v", err)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("client did not stop")
+	}
+	select {
+	case <-canceled:
+	case <-time.After(5 * time.Second):
+		t.Fatal("origin was not canceled")
+	}
+}

internal/client/repository.go

@@ -803,6 +803,10 @@ func (bs *blobs) Open(ctx context.Context, dgst digest.Digest) (io.ReadSeekClose
 }
 
 func (bs *blobs) ServeBlob(ctx context.Context, w http.ResponseWriter, r *http.Request, dgst digest.Digest) error {
+	if r.Method == http.MethodGet && r.Header.Get("Range") != "" {
+		return bs.serveBlobRange(ctx, w, r, dgst)
+	}
+
 	desc, err := bs.statter.Stat(ctx, dgst)
 	if err != nil {
 		return err
@@ -827,6 +831,71 @@ func (bs *blobs) ServeBlob(ctx context.Context, w http.ResponseWriter, r *http.R
 	return err
 }
 
+// serveBlobRange relays the origin representation without interpreting ranges
+// or verifying the whole blob digest. The caller must not cache this response.
+func (bs *blobs) serveBlobRange(ctx context.Context, w http.ResponseWriter, r *http.Request, dgst digest.Digest) error {
+	ref, err := reference.WithDigest(bs.name, dgst)
+	if err != nil {
+		return err
+	}
+	blobURL, err := bs.ub.BuildBlobURL(ref)
+	if err != nil {
+		return err
+	}
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, blobURL, nil)
+	if err != nil {
+		return err
+	}
+	req.Header.Set("Range", r.Header.Get("Range"))
+	if value := r.Header.Get("If-Range"); value != "" {
+		req.Header.Set("If-Range", value)
+	}
+	// Byte offsets refer to the encoded representation. Prevent the transport
+	// from requesting gzip and transparently decoding the origin response.
+	req.Header.Set("Accept-Encoding", "identity")
+	resp, err := bs.client.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+	switch resp.StatusCode {
+	case http.StatusOK, http.StatusPartialContent, http.StatusRequestedRangeNotSatisfiable:
+	case http.StatusNotFound:
+		return distribution.ErrBlobUnknown
+	default:
+		if err := HandleHTTPResponseError(resp); err != nil {
+			return err
+		}
+		return &UnexpectedHTTPStatusError{Status: resp.Status}
+	}
+
+	// Forward representation metadata, excluding even allowlisted fields named
+	// by Connection. Never relay origin cookies or hop-by-hop headers.
+	headers := resp.Header.Clone()
+	for _, value := range headers.Values("Connection") {
+		for _, name := range strings.Split(value, ",") {
+			headers.Del(strings.TrimSpace(name))
+		}
+	}
+	for _, name := range []string{
+		"Content-Type", "Content-Length", "Content-Range", "Content-Encoding",
+		"Accept-Ranges", "Etag", "Last-Modified", "Cache-Control", "Expires", "Vary",
+		"Docker-Content-Digest",
+	} {
+		w.Header().Del(name)
+		for _, value := range headers.Values(name) {
+			w.Header().Add(name, value)
+		}
+	}
+	w.WriteHeader(resp.StatusCode)
+	if _, err := io.Copy(w, resp.Body); err != nil {
+		// Headers and possibly binary data have already been sent. Returning an
+		// error would let the registry append JSON to an incomplete response.
+		panic(http.ErrAbortHandler)
+	}
+	return nil
+}
+
 func (bs *blobs) Put(ctx context.Context, mediaType string, p []byte) (v1.Descriptor, error) {
 	writer, err := bs.Create(ctx)
 	if err != nil {

notifications/listener.go

@@ -165,8 +165,10 @@ func (bsl *blobServiceListener) Open(ctx context.Context, dgst digest.Digest) (i
 }
 
 func (bsl *blobServiceListener) ServeBlob(ctx context.Context, w http.ResponseWriter, r *http.Request, dgst digest.Digest) error {
+	ctx, w = dcontext.WithResponseWriter(ctx, w)
 	err := bsl.BlobStore.ServeBlob(ctx, w, r, dgst)
-	if err == nil {
+	// A relayed 416 has already been served successfully, but is not a blob pull.
+	if err == nil && ctx.Value("http.response.status") != http.StatusRequestedRangeNotSatisfiable {
 		// Use a detached context for Stat() since the HTTP request context may be canceled
 		// after ServeBlob completes, but we still want to send the notification.
 		if desc, err := bsl.Stat(context.WithoutCancel(ctx), dgst); err != nil {

registry/proxy/proxyblobstore.go

@@ -19,10 +19,11 @@ import (
 
 type proxyBlobStore struct {
 	localStore        distribution.BlobStore
-	remoteStore       distribution.BlobService
+	remoteStore       distribution.BlobStore
 	scheduler         *scheduler.TTLExpirationScheduler
 	ttl               *time.Duration
 	cacheWriteTimeout time.Duration
+	rangePassthrough  bool
 	repositoryName    reference.Named
 	authChallenger    authChallenger
 }
@@ -80,6 +81,24 @@ func (pbs *proxyBlobStore) serveLocal(ctx context.Context, w http.ResponseWriter
 	return true, pbs.localStore.ServeBlob(ctx, w, r, dgst)
 }
 
+func (pbs *proxyBlobStore) serveBlobRange(ctx context.Context, w http.ResponseWriter, r *http.Request, dgst digest.Digest) (bool, error) {
+	if !pbs.rangePassthrough || r.Method != http.MethodGet || r.Header.Get("Range") == "" {
+		return false, nil
+	}
+
+	// Range responses, including a full 200 fallback, never enter the cache
+	// writer or the digest-wide inflight path.
+	ctx, w = dcontext.WithResponseWriter(ctx, w)
+	err := pbs.remoteStore.ServeBlob(ctx, w, r, dgst)
+	status := ctx.Value("http.response.status")
+	if err == nil && (status == http.StatusOK || status == http.StatusPartialContent) {
+		written := ctx.Value("http.response.written").(int64)
+		proxyMetrics.BlobPull(uint64(written))
+		proxyMetrics.BlobPush(uint64(written), false)
+	}
+	return true, err
+}
+
 func (pbs *proxyBlobStore) ServeBlob(ctx context.Context, w http.ResponseWriter, r *http.Request, dgst digest.Digest) error {
 	served, err := pbs.serveLocal(ctx, w, r, dgst)
 	if err != nil {
@@ -95,6 +114,10 @@ func (pbs *proxyBlobStore) ServeBlob(ctx context.Context, w http.ResponseWriter,
 		return err
 	}
 
+	if served, err := pbs.serveBlobRange(ctx, w, r, dgst); served || err != nil {
+		return err
+	}
+
 	mu.Lock()
 	_, ok := inflight[dgst]
 	if ok {

registry/proxy/proxyblobstore_range_test.go

@@ -0,0 +1,177 @@
+package proxy
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/distribution/distribution/v3"
+	"github.com/distribution/distribution/v3/configuration"
+	"github.com/distribution/distribution/v3/notifications"
+	"github.com/distribution/distribution/v3/registry/storage"
+	"github.com/distribution/distribution/v3/registry/storage/driver/inmemory"
+	"github.com/distribution/reference"
+	"github.com/opencontainers/go-digest"
+	v1 "github.com/opencontainers/image-spec/specs-go/v1"
+)
+
+// Exercise the actual origin client and configuration wiring, including the
+// notification decorator. HEAD requests for metadata are not blob downloads.
+func TestProxyRangePassthrough(t *testing.T) {
+	data := []byte("0123456789abcdef")
+	dgst := digest.FromBytes(data)
+	for _, tc := range []struct {
+		name, setting, env string
+		enabled            bool
+	}{
+		{name: "omitted"},
+		{name: "disabled", setting: "rangepassthrough: false"},
+		{name: "enabled", setting: "rangepassthrough: true", enabled: true},
+		{name: "environment enabled", env: "true", enabled: true},
+		{name: "environment disabled", setting: "rangepassthrough: true", env: "false"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Setenv("REGISTRY_PROXY_RANGEPASSTHROUGH", tc.env)
+			if tc.env == "" {
+				if err := os.Unsetenv("REGISTRY_PROXY_RANGEPASSTHROUGH"); err != nil {
+					t.Fatal(err)
+				}
+			}
+			var ranges []string
+			origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				if r.URL.Path == "/v2/" {
+					w.Header().Set("WWW-Authenticate", `Basic realm="test"`)
+					w.WriteHeader(http.StatusUnauthorized)
+					return
+				}
+				user, password, ok := r.BasicAuth()
+				if !ok || user != "origin-user" || password != "origin-password" {
+					http.Error(w, "missing origin credentials", http.StatusUnauthorized)
+					return
+				}
+				w.Header().Set("Docker-Content-Digest", dgst.String())
+				w.Header().Set("Content-Type", "application/octet-stream")
+				if r.Method == http.MethodHead {
+					w.Header().Set("Content-Length", strconv.Itoa(len(data)))
+					return
+				}
+				ranges = append(ranges, r.Header.Get("Range"))
+				http.ServeContent(w, r, "", time.Time{}, bytes.NewReader(data))
+			}))
+			defer origin.Close()
+			config, err := configuration.Parse(strings.NewReader(fmt.Sprintf(`version: 0.1
+storage:
+  inmemory: {}
+proxy:
+  remoteurl: %s
+  username: origin-user
+  password: origin-password
+  ttl: 0
+  %s
+`, origin.URL, tc.setting)))
+			if err != nil {
+				t.Fatal(err)
+			}
+			ctx := context.Background()
+			driver := inmemory.New()
+			local, err := storage.NewRegistry(ctx, driver)
+			if err != nil {
+				t.Fatal(err)
+			}
+			namespace, err := NewRegistryPullThroughCache(ctx, local, driver, config.Proxy)
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer namespace.(Closer).Close()
+			name, _ := reference.WithName("test/blob")
+			repo, err := namespace.Repository(ctx, name)
+			if err != nil {
+				t.Fatal(err)
+			}
+			pbs := repo.Blobs(ctx).(*proxyBlobStore)
+			stats := statsBlobStore{stats: make(map[string]int), blobs: pbs.localStore}
+			pbs.localStore = stats
+			listener := &rangePullListener{}
+			repo, _ = notifications.Listen(repo, nil, listener)
+			blobs := repo.Blobs(ctx)
+			serve := func(value, ifRange string) *httptest.ResponseRecorder {
+				t.Helper()
+				req := httptest.NewRequest(http.MethodGet, "/", nil)
+				req.Header.Set("Range", value)
+				req.Header.Set("If-Range", ifRange)
+				req.Header.Set("Authorization", "Bearer downstream-only")
+				w := httptest.NewRecorder()
+				if err := blobs.ServeBlob(ctx, w, req, dgst); err != nil {
+					t.Fatal(err)
+				}
+				return w
+			}
+			before := proxyMetrics.blobMetrics
+			w := serve("bytes=2-5", "")
+			wantStatus, wantBody, wantRange := http.StatusOK, string(data), ""
+			if tc.enabled {
+				wantStatus, wantBody, wantRange = http.StatusPartialContent, "2345", "bytes=2-5"
+			}
+			if w.Code != wantStatus || w.Body.String() != wantBody || len(ranges) != 1 || ranges[0] != wantRange {
+				t.Fatalf("response %d %q, origin ranges %v", w.Code, w.Body.String(), ranges)
+			}
+			if listener.pulls != 1 || listener.size != int64(len(data)) {
+				t.Fatalf("pull descriptor: %+v", listener)
+			}
+			if tc.enabled {
+				if w.Header().Get("Content-Range") != "bytes 2-5/16" || w.Header().Get("Content-Length") != "4" {
+					t.Fatal(w.Header())
+				}
+				if _, err := stats.Stat(ctx, dgst); err != distribution.ErrBlobUnknown || stats.stats["create"] != 0 {
+					t.Fatalf("range populated cache: %v %v", err, stats.stats)
+				}
+				after := proxyMetrics.blobMetrics
+				if after.BytesPulled-before.BytesPulled != 4 || after.BytesPushed-before.BytesPushed != 4 {
+					t.Fatalf("range metrics: before=%+v after=%+v", before, after)
+				}
+				w = serve("bytes=99-", "")
+				if w.Code != http.StatusRequestedRangeNotSatisfiable || listener.pulls != 1 || proxyMetrics.blobMetrics != after {
+					t.Fatalf("416 counted as a pull: status=%d pulls=%d metrics=%+v", w.Code, listener.pulls, proxyMetrics.blobMetrics)
+				}
+				w = serve("bytes=2-5", `"stale"`)
+				if w.Code != http.StatusOK || !bytes.Equal(w.Body.Bytes(), data) || stats.stats["create"] != 0 {
+					t.Fatal("origin 200 fallback was changed or cached")
+				}
+				w = serve("", "")
+				if w.Code != http.StatusOK || !bytes.Equal(w.Body.Bytes(), data) {
+					t.Fatal("ordinary GET changed")
+				}
+			}
+			if stats.stats["create"] != 1 {
+				t.Fatalf("cache writers: %v", stats.stats)
+			}
+			if _, err := stats.Stat(ctx, dgst); err != nil {
+				t.Fatal(err)
+			}
+			originGets := len(ranges)
+			w = serve("bytes=6-9", "")
+			if w.Code != http.StatusPartialContent || w.Body.String() != "6789" || len(ranges) != originGets {
+				t.Fatalf("warm range: %d %q, origin ranges %v", w.Code, w.Body.String(), ranges)
+			}
+		})
+	}
+}
+
+type rangePullListener struct {
+	notifications.Listener
+	pulls int
+	size  int64
+}
+
+func (l *rangePullListener) BlobPulled(_ reference.Named, desc v1.Descriptor) error {
+	l.pulls++
+	l.size = desc.Size
+	return nil
+}

registry/proxy/proxyregistry.go

@@ -31,6 +31,7 @@ type proxyingRegistry struct {
 	scheduler         *scheduler.TTLExpirationScheduler
 	ttl               *time.Duration
 	cacheWriteTimeout time.Duration
+	rangePassthrough  bool
 	remoteURL         url.URL
 	authChallenger    authChallenger
 	basicAuth         auth.CredentialStore
@@ -140,6 +141,7 @@ func NewRegistryPullThroughCache(ctx context.Context, registry distribution.Name
 		scheduler:         s,
 		ttl:               ttl,
 		cacheWriteTimeout: cacheWriteTimeout,
+		rangePassthrough:  config.RangePassthrough,
 		remoteURL:         *remoteURL,
 		authChallenger: &remoteAuthChallenger{
 			remoteURL: *remoteURL,
@@ -204,6 +206,7 @@ func (pr *proxyingRegistry) Repository(ctx context.Context, name reference.Named
 			scheduler:         pr.scheduler,
 			ttl:               pr.ttl,
 			cacheWriteTimeout: pr.cacheWriteTimeout,
+			rangePassthrough:  pr.rangePassthrough,
 			repositoryName:    name,
 			authChallenger:    pr.authChallenger,
 		},

registry/proxy_range_test.go

@@ -0,0 +1,89 @@
+package registry
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/distribution/distribution/v3/configuration"
+	"github.com/distribution/distribution/v3/registry/handlers"
+	"github.com/opencontainers/go-digest"
+)
+
+// Include the dispatcher and outer recovery middleware: a truncated origin
+// stream must not become a completed response or acquire a JSON error suffix.
+func TestProxyRangeStreamAbort(t *testing.T) {
+	data := bytes.Repeat([]byte("0123456789abcdef"), 2048)
+	dgst := digest.FromBytes(data)
+	origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/v2/" {
+			return
+		}
+		w.Header().Set("Content-Type", "application/octet-stream")
+		w.Header().Set("Docker-Content-Digest", dgst.String())
+		if r.Method == http.MethodHead {
+			w.Header().Set("Content-Length", strconv.Itoa(len(data)))
+			return
+		}
+		if r.Header.Get("Range") != "bytes=4096-20479" {
+			t.Errorf("origin Range: %q", r.Header.Get("Range"))
+		}
+		w.Header().Set("Content-Range", "bytes 4096-20479/32768")
+		w.Header().Set("Content-Length", "16384")
+		w.WriteHeader(http.StatusPartialContent)
+		_, _ = w.Write(data[4096:12288])
+		w.(http.Flusher).Flush()
+	}))
+	defer origin.Close()
+	config := &configuration.Configuration{
+		Storage: configuration.Storage{
+			"inmemory":    configuration.Parameters{},
+			"maintenance": configuration.Parameters{"uploadpurging": map[any]any{"enabled": false}},
+		},
+		Proxy: configuration.Proxy{RemoteURL: origin.URL, RangePassthrough: true, TTL: new(time.Duration)},
+	}
+	app := handlers.NewApp(context.Background(), config)
+	defer app.Shutdown()
+	aborted := make(chan any, 1)
+	handler := panicHandler(app)
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		defer func() {
+			if p := recover(); p != nil {
+				aborted <- p
+				panic(p)
+			}
+		}()
+		handler.ServeHTTP(w, r)
+	}))
+	defer server.Close()
+	req, err := http.NewRequest(http.MethodGet, server.URL+"/v2/test/blob/blobs/"+dgst.String(), nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	req.Header.Set("Range", "bytes=4096-20479")
+	client := server.Client()
+	client.Timeout = 5 * time.Second
+	resp, err := client.Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+	body, err := io.ReadAll(resp.Body)
+	if resp.StatusCode != http.StatusPartialContent || !errors.Is(err, io.ErrUnexpectedEOF) || len(body) == 0 || len(body) >= 16384 || !bytes.Equal(body, data[4096:4096+len(body)]) {
+		t.Fatalf("stream was not aborted cleanly: status=%d bytes=%d err=%v", resp.StatusCode, len(body), err)
+	}
+	select {
+	case p := <-aborted:
+		if p != http.ErrAbortHandler {
+			t.Fatalf("recovery changed abort sentinel: %v", p)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("handler did not abort")
+	}
+}

registry/registry.go

@@ -444,6 +444,9 @@ func panicHandler(handler http.Handler) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		defer func() {
 			if err := recover(); err != nil {
+				if err == http.ErrAbortHandler {
+					panic(err)
+				}
 				logrus.Panic(fmt.Sprintf("%v", err))
 			}
 		}()