diff --git a/core/remotes/docker/fetcher_test.go b/core/remotes/docker/fetcher_test.go
index 624a9b2d2b..ed8fccbd10 100644
--- a/core/remotes/docker/fetcher_test.go
+++ b/core/remotes/docker/fetcher_test.go
@@ -30,6 +30,8 @@ import (
 	"net/http/httptest"
 	"net/textproto"
 	"net/url"
+	"os"
+	"path/filepath"
 	"strconv"
 	"strings"
 	"sync"
@@ -38,19 +40,124 @@ import (
 	"time"
 
 	"github.com/klauspost/compress/zstd"
+	"github.com/opencontainers/go-digest"
 	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"golang.org/x/sync/semaphore"
 
+	"github.com/containerd/containerd/v2/core/content"
+	"github.com/containerd/containerd/v2/core/remotes"
 	"github.com/containerd/containerd/v2/core/transfer"
 	"github.com/containerd/containerd/v2/pkg/reference"
+	"github.com/containerd/containerd/v2/plugins/content/local"
 )
 
 type writeFunc func(p []byte) (int, error)
 
 func (f writeFunc) Write(p []byte) (int, error) { return f(p) }
 
+func TestHTTPReadSeekerClosesBodyAtEOF(t *testing.T) {
+	payload := []byte("layer content")
+	path := filepath.Join(t.TempDir(), "body")
+	require.NoError(t, os.WriteFile(path, payload, 0600))
+	var bodies []*os.File
+	rc, err := newHTTPReadSeeker(int64(len(payload)), func(offset int64) (io.ReadCloser, error) {
+		body, err := os.Open(path)
+		if err != nil {
+			return nil, err
+		}
+		if _, err := body.Seek(offset, io.SeekStart); err != nil {
+			body.Close()
+			return nil, err
+		}
+		bodies = append(bodies, body)
+		return body, nil
+	})
+	require.NoError(t, err)
+	defer rc.Close()
+
+	for i := range 2 {
+		data, err := io.ReadAll(rc)
+		require.NoError(t, err)
+		require.Equal(t, payload, data)
+		require.Len(t, bodies, i+1)
+		// The HTTP body must be closed before the caller commits its writer,
+		// without closing the seekable reader needed for a possible reset.
+		_, err = bodies[i].Read(make([]byte, 1))
+		require.ErrorIs(t, err, os.ErrClosed)
+		_, err = rc.(io.Seeker).Seek(0, io.SeekStart)
+		require.NoError(t, err)
+	}
+}
+
+func TestFetchClosesHTTPBodyBeforeCommit(t *testing.T) {
+	ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
+	defer cancel()
+	payload := []byte("layer content")
+	desc := ocispec.Descriptor{
+		MediaType: ocispec.MediaTypeImageLayer,
+		Digest:    digest.FromBytes(payload),
+		Size:      int64(len(payload)),
+	}
+	registry := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Length", strconv.Itoa(len(payload)))
+		w.Write(payload)
+	}))
+	defer registry.Close()
+	var closed atomic.Int32
+	client := registry.Client()
+	transport := client.Transport
+	client.Transport = rtFunc(func(r *http.Request) (*http.Response, error) {
+		resp, err := transport.RoundTrip(r)
+		if err == nil {
+			resp.Body = &fnOnClose{
+				ReadCloser:  resp.Body,
+				BeforeClose: func() { closed.Add(1) },
+			}
+		}
+		return resp, err
+	})
+	resolver := NewResolver(ResolverOptions{Client: client, PlainHTTP: true})
+	fetcher, err := resolver.Fetcher(ctx, strings.TrimPrefix(registry.URL, "http://")+"/image:latest")
+	require.NoError(t, err)
+	store, err := local.NewStore(t.TempDir())
+	require.NoError(t, err)
+	committed := false
+	ingester := beforeCommitIngester{Ingester: store, beforeCommit: func() {
+		// Even if Commit blocks, CRI must no longer count this HTTP request
+		// as active. Check before Fetch's deferred reader.Close can run.
+		require.EqualValues(t, 1, closed.Load())
+		committed = true
+	}}
+	require.NoError(t, remotes.Fetch(ctx, ingester, fetcher, desc))
+	require.True(t, committed)
+	require.EqualValues(t, 1, closed.Load())
+}
+
+type beforeCommitIngester struct {
+	content.Ingester
+	beforeCommit func()
+}
+
+func (i beforeCommitIngester) Writer(ctx context.Context, opts ...content.WriterOpt) (content.Writer, error) {
+	w, err := i.Ingester.Writer(ctx, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return beforeCommitWriter{Writer: w, beforeCommit: i.beforeCommit}, nil
+}
+
+type beforeCommitWriter struct {
+	content.Writer
+	beforeCommit func()
+}
+
+func (w beforeCommitWriter) Commit(ctx context.Context, size int64, expected digest.Digest, opts ...content.Opt) error {
+	w.beforeCommit()
+	return w.Writer.Commit(ctx, size, expected, opts...)
+}
+
 func TestFetcherOpen(t *testing.T) {
 	content := make([]byte, 128)
 	rand.New(rand.NewSource(1)).Read(content)
diff --git a/core/remotes/docker/pusher_test.go b/core/remotes/docker/pusher_test.go
index 2e2a8e8070..e614ffc71d 100644
--- a/core/remotes/docker/pusher_test.go
+++ b/core/remotes/docker/pusher_test.go
@@ -26,13 +26,17 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"net/url"
+	"os"
+	"path/filepath"
 	"reflect"
 	"regexp"
 	"slices"
+	"strconv"
 	"strings"
 	"sync"
 	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/containerd/containerd/v2/core/content"
 	"github.com/containerd/containerd/v2/core/remotes"
@@ -46,6 +50,59 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+// Fetch can also copy local content to a pusher-backed ingester, as BuildKit's
+// registry cache exporter does. The source must remain seekable after EOF so
+// content.Copy can replay it when the push writer returns content.ErrReset.
+func TestFetchPushRetryAfterEOF(t *testing.T) {
+	for _, size := range []int{128, 2 << 20} {
+		t.Run(strconv.Itoa(size), func(t *testing.T) {
+			ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
+			defer cancel()
+			payload := bytes.Repeat([]byte("x"), size)
+			desc := ocispec.Descriptor{
+				MediaType: ocispec.MediaTypeImageLayer,
+				Digest:    digest.FromBytes(payload),
+				Size:      int64(len(payload)),
+			}
+			path := filepath.Join(t.TempDir(), "layer")
+			require.NoError(t, os.WriteFile(path, payload, 0600))
+			var attempts atomic.Int32
+			registry := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				switch r.Method {
+				case http.MethodPost:
+					w.Header().Set("Location", "/v2/cache/blobs/uploads/test")
+					w.WriteHeader(http.StatusAccepted)
+				case http.MethodPut:
+					body, err := io.ReadAll(r.Body)
+					if !assert.NoError(t, err) || !assert.Equal(t, payload, body) {
+						w.WriteHeader(http.StatusBadRequest)
+						return
+					}
+					if attempts.Add(1) == 1 {
+						w.WriteHeader(http.StatusRequestTimeout)
+						return
+					}
+					w.Header().Set("Docker-Content-Digest", desc.Digest.String())
+					w.WriteHeader(http.StatusCreated)
+				default:
+					w.WriteHeader(http.StatusNotFound)
+				}
+			}))
+			defer registry.Close()
+			resolver := NewResolver(ResolverOptions{PlainHTTP: true})
+			pusher, err := resolver.Pusher(ctx, strings.TrimPrefix(registry.URL, "http://")+"/cache:latest")
+			require.NoError(t, err)
+			ingester, ok := pusher.(content.Ingester)
+			require.True(t, ok)
+			fetcher := remotes.FetcherFunc(func(context.Context, ocispec.Descriptor) (io.ReadCloser, error) {
+				return os.Open(path)
+			})
+			require.NoError(t, remotes.Fetch(ctx, ingester, fetcher, desc))
+			require.EqualValues(t, 2, attempts.Load())
+		})
+	}
+}
+
 func TestGetManifestPath(t *testing.T) {
 	for _, tc := range []struct {
 		object   string
diff --git a/core/remotes/handlers.go b/core/remotes/handlers.go
index 4214672ae1..b01e189d72 100644
--- a/core/remotes/handlers.go
+++ b/core/remotes/handlers.go
@@ -173,6 +173,9 @@ func Fetch(ctx context.Context, ingester content.Ingester, fetcher Fetcher, desc
 	if err != nil {
 		return err
 	}
+
+	// Keep the reader usable until Copy finishes: a writer may request a
+	// reset after EOF, including while committing the content.
 	defer rc.Close()
 
 	return content.Copy(ctx, cw, rc, desc.Size, desc.Digest)