chore(deps): bump github.com/google/go-containerregistry (#2508)

Bumps [github.com/google/go-containerregistry](https://github.com/google/go-containerregistry) from 0.14.0 to 0.15.1. - [Release notes](https://github.com/google/go-containerregistry/releases) - [Changelog](https://github.com/google/go-containerregistry/blob/main/.goreleaser.yml) - [Commits](https://github.com/google/go-containerregistry/compare/v0.14.0...v0.15.1) --- updated-dependencies: - dependency-name: github.com/google/go-containerregistry dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2026-10-21 22:56:30 +02:00 · 2023-05-16 00:42:29 -07:00
parent 8469f7f105
commit 5792a72413
97 changed files with 3469 additions and 1881 deletions
@@ -18,6 +18,7 @@ import (
 	"os"
 	"path/filepath"
 	"sync"
+	"time"

 	"github.com/docker/cli/cli/config"
 	"github.com/docker/cli/cli/config/configfile"
@@ -52,7 +53,7 @@ type defaultKeychain struct {

 var (
 	// DefaultKeychain implements Keychain by interpreting the docker config file.
-	DefaultKeychain Keychain = &defaultKeychain{}
+	DefaultKeychain = RefreshingKeychain(&defaultKeychain{}, 5*time.Minute)
 )

 const (
@@ -178,3 +179,71 @@ func (w wrapper) Resolve(r Resource) (Authenticator, error) {
 	}
 	return FromConfig(AuthConfig{Username: u, Password: p}), nil
 }
+
+func RefreshingKeychain(inner Keychain, duration time.Duration) Keychain {
+	return &refreshingKeychain{
+		keychain: inner,
+		duration: duration,
+	}
+}
+
+type refreshingKeychain struct {
+	keychain Keychain
+	duration time.Duration
+	clock    func() time.Time
+}
+
+func (r *refreshingKeychain) Resolve(target Resource) (Authenticator, error) {
+	last := time.Now()
+	auth, err := r.keychain.Resolve(target)
+	if err != nil || auth == Anonymous {
+		return auth, err
+	}
+	return &refreshing{
+		target:   target,
+		keychain: r.keychain,
+		last:     last,
+		cached:   auth,
+		duration: r.duration,
+		clock:    r.clock,
+	}, nil
+}
+
+type refreshing struct {
+	sync.Mutex
+	target   Resource
+	keychain Keychain
+
+	duration time.Duration
+
+	last   time.Time
+	cached Authenticator
+
+	// for testing
+	clock func() time.Time
+}
+
+func (r *refreshing) Authorization() (*AuthConfig, error) {
+	r.Lock()
+	defer r.Unlock()
+	if r.cached == nil || r.expired() {
+		r.last = r.now()
+		auth, err := r.keychain.Resolve(r.target)
+		if err != nil {
+			return nil, err
+		}
+		r.cached = auth
+	}
+	return r.cached.Authorization()
+}
+
+func (r *refreshing) now() time.Time {
+	if r.clock == nil {
+		return time.Now()
+	}
+	return r.clock()
+}
+
+func (r *refreshing) expired() bool {
+	return r.now().Sub(r.last) > r.duration
+}
@@ -17,6 +17,7 @@ package name
 import (
 	"net"
 	"net/url"
+	"path"
 	"regexp"
 	"strings"
 )
@@ -50,6 +51,11 @@ func (r Registry) String() string {
 	return r.Name()
 }

+// Repo returns a Repository in the Registry with the given name.
+func (r Registry) Repo(repo ...string) Repository {
+	return Repository{Registry: r, repository: path.Join(repo...)}
+}
+
 // Scope returns the scope required to access the registry.
 func (r Registry) Scope(string) string {
 	// The only resource under 'registry' is 'catalog'. http://goo.gl/N9cN9Z
@@ -19,6 +19,10 @@ import (
 	"context"
 	"io"
 	"sync"
+	"time"
+
+	api "github.com/docker/docker/api/types"
+	"github.com/docker/docker/api/types/container"

 	"github.com/google/go-containerregistry/pkg/name"
 	v1 "github.com/google/go-containerregistry/pkg/v1"
@@ -30,7 +34,9 @@ type image struct {
 	ref          name.Reference
 	opener       *imageOpener
 	tarballImage v1.Image
+	computed     bool
 	id           *v1.Hash
+	configFile   *v1.ConfigFile

 	once sync.Once
 	err  error
@@ -121,6 +127,28 @@ func (i *image) initialize() error {
 	return i.err
 }

+func (i *image) compute() error {
+	// Don't re-compute if already computed.
+	if i.computed {
+		return nil
+	}
+
+	inspect, _, err := i.opener.client.ImageInspectWithRaw(i.opener.ctx, i.ref.String())
+	if err != nil {
+		return err
+	}
+
+	configFile, err := i.computeConfigFile(inspect)
+	if err != nil {
+		return err
+	}
+
+	i.configFile = configFile
+	i.computed = true
+
+	return nil
+}
+
 func (i *image) Layers() ([]v1.Layer, error) {
 	if err := i.initialize(); err != nil {
 		return nil, err
@@ -154,16 +182,19 @@ func (i *image) ConfigName() (v1.Hash, error) {
 }

 func (i *image) ConfigFile() (*v1.ConfigFile, error) {
-	if err := i.initialize(); err != nil {
+	if err := i.compute(); err != nil {
 		return nil, err
 	}
-	return i.tarballImage.ConfigFile()
+	return i.configFile.DeepCopy(), nil
 }

 func (i *image) RawConfigFile() ([]byte, error) {
 	if err := i.initialize(); err != nil {
 		return nil, err
 	}
+
+	// RawConfigFile cannot be generated from "docker inspect" because Docker Engine API returns serialized data,
+	// and formatting information of the raw config such as indent and prefix will be lost.
 	return i.tarballImage.RawConfigFile()
 }

@@ -201,3 +232,119 @@ func (i *image) LayerByDiffID(h v1.Hash) (v1.Layer, error) {
 	}
 	return i.tarballImage.LayerByDiffID(h)
 }
+
+func (i *image) configHistory(author string) ([]v1.History, error) {
+	historyItems, err := i.opener.client.ImageHistory(i.opener.ctx, i.ref.String())
+	if err != nil {
+		return nil, err
+	}
+
+	history := make([]v1.History, len(historyItems))
+	for j, h := range historyItems {
+		history[j] = v1.History{
+			Author: author,
+			Created: v1.Time{
+				Time: time.Unix(h.Created, 0).UTC(),
+			},
+			CreatedBy:  h.CreatedBy,
+			Comment:    h.Comment,
+			EmptyLayer: h.Size == 0,
+		}
+	}
+	return history, nil
+}
+
+func (i *image) diffIDs(rootFS api.RootFS) ([]v1.Hash, error) {
+	diffIDs := make([]v1.Hash, len(rootFS.Layers))
+	for j, l := range rootFS.Layers {
+		h, err := v1.NewHash(l)
+		if err != nil {
+			return nil, err
+		}
+		diffIDs[j] = h
+	}
+	return diffIDs, nil
+}
+
+func (i *image) computeConfigFile(inspect api.ImageInspect) (*v1.ConfigFile, error) {
+	diffIDs, err := i.diffIDs(inspect.RootFS)
+	if err != nil {
+		return nil, err
+	}
+
+	history, err := i.configHistory(inspect.Author)
+	if err != nil {
+		return nil, err
+	}
+
+	created, err := time.Parse(time.RFC3339Nano, inspect.Created)
+	if err != nil {
+		return nil, err
+	}
+
+	return &v1.ConfigFile{
+		Architecture:  inspect.Architecture,
+		Author:        inspect.Author,
+		Container:     inspect.Container,
+		Created:       v1.Time{Time: created},
+		DockerVersion: inspect.DockerVersion,
+		History:       history,
+		OS:            inspect.Os,
+		RootFS: v1.RootFS{
+			Type:    inspect.RootFS.Type,
+			DiffIDs: diffIDs,
+		},
+		Config:    i.computeImageConfig(inspect.Config),
+		OSVersion: inspect.OsVersion,
+	}, nil
+}
+
+func (i *image) computeImageConfig(config *container.Config) v1.Config {
+	if config == nil {
+		return v1.Config{}
+	}
+
+	c := v1.Config{
+		AttachStderr:    config.AttachStderr,
+		AttachStdin:     config.AttachStdin,
+		AttachStdout:    config.AttachStdout,
+		Cmd:             config.Cmd,
+		Domainname:      config.Domainname,
+		Entrypoint:      config.Entrypoint,
+		Env:             config.Env,
+		Hostname:        config.Hostname,
+		Image:           config.Image,
+		Labels:          config.Labels,
+		OnBuild:         config.OnBuild,
+		OpenStdin:       config.OpenStdin,
+		StdinOnce:       config.StdinOnce,
+		Tty:             config.Tty,
+		User:            config.User,
+		Volumes:         config.Volumes,
+		WorkingDir:      config.WorkingDir,
+		ArgsEscaped:     config.ArgsEscaped,
+		NetworkDisabled: config.NetworkDisabled,
+		MacAddress:      config.MacAddress,
+		StopSignal:      config.StopSignal,
+		Shell:           config.Shell,
+	}
+
+	if config.Healthcheck != nil {
+		c.Healthcheck = &v1.HealthConfig{
+			Test:        config.Healthcheck.Test,
+			Interval:    config.Healthcheck.Interval,
+			Timeout:     config.Healthcheck.Timeout,
+			StartPeriod: config.Healthcheck.StartPeriod,
+			Retries:     config.Healthcheck.Retries,
+		}
+	}
+
+	if len(config.ExposedPorts) > 0 {
+		c.ExposedPorts = map[string]struct{}{}
+		for port := range c.ExposedPorts {
+			c.ExposedPorts[port] = struct{}{}
+		}
+	}
+
+	return c
+}
@@ -19,6 +19,7 @@ import (
 	"io"

 	"github.com/docker/docker/api/types"
+	api "github.com/docker/docker/api/types/image"
 	"github.com/docker/docker/client"
 )

@@ -100,4 +101,5 @@ type Client interface {
 	ImageLoad(context.Context, io.Reader, bool) (types.ImageLoadResponse, error)
 	ImageTag(context.Context, string, string) error
 	ImageInspectWithRaw(context.Context, string) (types.ImageInspect, []byte, error)
+	ImageHistory(context.Context, string) ([]api.HistoryResponseItem, error)
 }
@@ -60,5 +60,6 @@ func base() *v1.IndexManifest {
 	return &v1.IndexManifest{
 		SchemaVersion: 2,
 		MediaType:     types.OCIImageIndex,
+		Manifests:     []v1.Descriptor{},
 	}
 }
@@ -53,11 +53,7 @@ type googleKeychain struct {
 // gcloud configuration in the scope of this one process.
 func (gk *googleKeychain) Resolve(target authn.Resource) (authn.Authenticator, error) {
 	// Only authenticate GCR and AR so it works with authn.NewMultiKeychain to fallback.
-	host := target.RegistryStr()
-	if host != "gcr.io" &&
-		!strings.HasSuffix(host, ".gcr.io") &&
-		!strings.HasSuffix(host, ".pkg.dev") &&
-		!strings.HasSuffix(host, ".google.com") {
+	if !isGoogle(target.RegistryStr()) {
 		return authn.Anonymous, nil
 	}

@@ -90,3 +86,10 @@ func resolve() authn.Authenticator {
 	}
 	return authn.Anonymous
 }
+
+func isGoogle(host string) bool {
+	return host == "gcr.io" ||
+		strings.HasSuffix(host, ".gcr.io") ||
+		strings.HasSuffix(host, ".pkg.dev") ||
+		strings.HasSuffix(host, ".google.com")
+}
@@ -89,12 +89,16 @@ func newLister(repo name.Repository, options ...Option) (*lister, error) {

 func (l *lister) list(repo name.Repository) (*Tags, error) {
 	uri := &url.URL{
-		Scheme: repo.Registry.Scheme(),
-		Host:   repo.Registry.RegistryStr(),
-		Path:   fmt.Sprintf("/v2/%s/tags/list", repo.RepositoryStr()),
-		// ECR returns an error if n > 1000:
-		// https://github.com/google/go-containerregistry/issues/681
-		RawQuery: "n=1000",
+		Scheme:   repo.Registry.Scheme(),
+		Host:     repo.Registry.RegistryStr(),
+		Path:     fmt.Sprintf("/v2/%s/tags/list", repo.RepositoryStr()),
+		RawQuery: "n=10000",
+	}
+
+	// ECR returns an error if n > 1000:
+	// https://github.com/google/go-containerregistry/issues/681
+	if !isGoogle(repo.RegistryStr()) {
+		uri.RawQuery = "n=1000"
 	}

 	tags := Tags{}
@@ -18,6 +18,7 @@ import (
 	"bytes"
 	"encoding/json"
 	"errors"
+	"sync"

 	v1 "github.com/google/go-containerregistry/pkg/v1"
 	"github.com/google/go-containerregistry/pkg/v1/partial"
@@ -38,6 +39,8 @@ type image struct {
 	diffIDMap       map[v1.Hash]v1.Layer
 	digestMap       map[v1.Hash]v1.Layer
 	subject         *v1.Descriptor
+
+	sync.Mutex
 }

 var _ v1.Image = (*image)(nil)
@@ -50,6 +53,9 @@ func (i *image) MediaType() (types.MediaType, error) {
 }

 func (i *image) compute() error {
+	i.Lock()
+	defer i.Unlock()
+
 	// Don't re-compute if already computed.
 	if i.computed {
 		return nil
@@ -16,12 +16,15 @@ package mutate

 import (
 	"encoding/json"
+	"errors"
 	"fmt"
+	"sync"

 	"github.com/google/go-containerregistry/pkg/logs"
 	v1 "github.com/google/go-containerregistry/pkg/v1"
 	"github.com/google/go-containerregistry/pkg/v1/match"
 	"github.com/google/go-containerregistry/pkg/v1/partial"
+	"github.com/google/go-containerregistry/pkg/v1/stream"
 	"github.com/google/go-containerregistry/pkg/v1/types"
 )

@@ -71,6 +74,8 @@ type index struct {
 	indexMap    map[v1.Hash]v1.ImageIndex
 	layerMap    map[v1.Hash]v1.Layer
 	subject     *v1.Descriptor
+
+	sync.Mutex
 }

 var _ v1.ImageIndex = (*index)(nil)
@@ -85,6 +90,9 @@ func (i *index) MediaType() (types.MediaType, error) {
 func (i *index) Size() (int64, error) { return partial.Size(i) }

 func (i *index) compute() error {
+	i.Lock()
+	defer i.Unlock()
+
 	// Don't re-compute if already computed.
 	if i.computed {
 		return nil
@@ -202,3 +210,23 @@ func (i *index) RawManifest() ([]byte, error) {
 	}
 	return json.Marshal(i.manifest)
 }
+
+func (i *index) Manifests() ([]partial.Describable, error) {
+	if err := i.compute(); errors.Is(err, stream.ErrNotComputed) {
+		// Index contains a streamable layer which has not yet been
+		// consumed. Just return the manifests we have in case the caller
+		// is going to consume the streamable layers.
+		manifests, err := partial.Manifests(i.base)
+		if err != nil {
+			return nil, err
+		}
+		for _, add := range i.adds {
+			manifests = append(manifests, add.Add)
+		}
+		return manifests, nil
+	} else if err != nil {
+		return nil, err
+	}
+
+	return partial.ComputeManifests(i)
+}
@@ -19,6 +19,7 @@ import (

 	v1 "github.com/google/go-containerregistry/pkg/v1"
 	"github.com/google/go-containerregistry/pkg/v1/match"
+	"github.com/google/go-containerregistry/pkg/v1/types"
 )

 // FindManifests given a v1.ImageIndex, find the manifests that fit the matcher.
@@ -83,3 +84,82 @@ func FindIndexes(index v1.ImageIndex, matcher match.Matcher) ([]v1.ImageIndex, e
 	}
 	return matches, nil
 }
+
+type withManifests interface {
+	Manifests() ([]Describable, error)
+}
+
+type withLayer interface {
+	Layer(v1.Hash) (v1.Layer, error)
+}
+
+type describable struct {
+	desc v1.Descriptor
+}
+
+func (d describable) Digest() (v1.Hash, error) {
+	return d.desc.Digest, nil
+}
+
+func (d describable) Size() (int64, error) {
+	return d.desc.Size, nil
+}
+
+func (d describable) MediaType() (types.MediaType, error) {
+	return d.desc.MediaType, nil
+}
+
+func (d describable) Descriptor() (*v1.Descriptor, error) {
+	return &d.desc, nil
+}
+
+// Manifests is analogous to v1.Image.Layers in that it allows values in the
+// returned list to be lazily evaluated, which enables an index to contain
+// an image that contains a streaming layer.
+//
+// This should have been part of the v1.ImageIndex interface, but wasn't.
+// It is instead usable through this extension interface.
+func Manifests(idx v1.ImageIndex) ([]Describable, error) {
+	if wm, ok := idx.(withManifests); ok {
+		return wm.Manifests()
+	}
+
+	return ComputeManifests(idx)
+}
+
+// ComputeManifests provides a fallback implementation for Manifests.
+func ComputeManifests(idx v1.ImageIndex) ([]Describable, error) {
+	m, err := idx.IndexManifest()
+	if err != nil {
+		return nil, err
+	}
+	manifests := []Describable{}
+	for _, desc := range m.Manifests {
+		switch {
+		case desc.MediaType.IsImage():
+			img, err := idx.Image(desc.Digest)
+			if err != nil {
+				return nil, err
+			}
+			manifests = append(manifests, img)
+		case desc.MediaType.IsIndex():
+			idx, err := idx.ImageIndex(desc.Digest)
+			if err != nil {
+				return nil, err
+			}
+			manifests = append(manifests, idx)
+		default:
+			if wl, ok := idx.(withLayer); ok {
+				layer, err := wl.Layer(desc.Digest)
+				if err != nil {
+					return nil, err
+				}
+				manifests = append(manifests, layer)
+			} else {
+				manifests = append(manifests, describable{desc})
+			}
+		}
+	}
+
+	return manifests, nil
+}
@@ -18,12 +18,10 @@ import (
 	"archive/tar"
 	"bytes"
 	"crypto"
-	"crypto/rand"
 	"encoding/hex"
 	"fmt"
 	"io"
-	mrand "math/rand"
-	"time"
+	"math/rand"

 	v1 "github.com/google/go-containerregistry/pkg/v1"
 	"github.com/google/go-containerregistry/pkg/v1/empty"
@@ -57,10 +55,10 @@ func (ul *uncompressedLayer) MediaType() (types.MediaType, error) {
 var _ partial.UncompressedLayer = (*uncompressedLayer)(nil)

 // Image returns a pseudo-randomly generated Image.
-func Image(byteSize, layers int64) (v1.Image, error) {
+func Image(byteSize, layers int64, options ...Option) (v1.Image, error) {
 	adds := make([]mutate.Addendum, 0, 5)
 	for i := int64(0); i < layers; i++ {
-		layer, err := Layer(byteSize, types.DockerLayer)
+		layer, err := Layer(byteSize, types.DockerLayer, options...)
 		if err != nil {
 			return nil, err
 		}
@@ -70,7 +68,6 @@ func Image(byteSize, layers int64) (v1.Image, error) {
 				Author:    "random.Image",
 				Comment:   fmt.Sprintf("this is a random history %d of %d", i, layers),
 				CreatedBy: "random",
-				Created:   v1.Time{Time: time.Now()},
 			},
 		})
 	}
@@ -79,8 +76,11 @@ func Image(byteSize, layers int64) (v1.Image, error) {
 }

 // Layer returns a layer with pseudo-randomly generated content.
-func Layer(byteSize int64, mt types.MediaType) (v1.Layer, error) {
-	fileName := fmt.Sprintf("random_file_%d.txt", mrand.Int()) //nolint: gosec
+func Layer(byteSize int64, mt types.MediaType, options ...Option) (v1.Layer, error) {
+	o := getOptions(options)
+	rng := rand.New(o.source) //nolint:gosec
+
+	fileName := fmt.Sprintf("random_file_%d.txt", rng.Int())

 	// Hash the contents as we write it out to the buffer.
 	var b bytes.Buffer
@@ -96,7 +96,7 @@ func Layer(byteSize int64, mt types.MediaType) (v1.Layer, error) {
 	}); err != nil {
 		return nil, err
 	}
-	if _, err := io.CopyN(tw, rand.Reader, byteSize); err != nil {
+	if _, err := io.CopyN(tw, rng, byteSize); err != nil {
 		return nil, err
 	}
 	if err := tw.Close(); err != nil {
@@ -31,7 +31,7 @@ type randomIndex struct {

 // Index returns a pseudo-randomly generated ImageIndex with count images, each
 // having the given number of layers of size byteSize.
-func Index(byteSize, layers, count int64) (v1.ImageIndex, error) {
+func Index(byteSize, layers, count int64, options ...Option) (v1.ImageIndex, error) {
 	manifest := v1.IndexManifest{
 		SchemaVersion: 2,
 		MediaType:     types.OCIImageIndex,
@@ -40,7 +40,7 @@ func Index(byteSize, layers, count int64) (v1.ImageIndex, error) {

 	images := make(map[v1.Hash]v1.Image)
 	for i := int64(0); i < count; i++ {
-		img, err := Image(byteSize, layers)
+		img, err := Image(byteSize, layers, options...)
 		if err != nil {
 			return nil, err
 		}
@@ -0,0 +1,60 @@
+// Copyright 2018 Google LLC All Rights Reserved.
+//
+// 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 random
+
+import "math/rand"
+
+// Option is an optional parameter to the random functions
+type Option func(opts *options)
+
+type options struct {
+	source rand.Source
+
+	// TODO opens the door to add this in the future
+	// algorithm digest.Algorithm
+}
+
+func getOptions(opts []Option) *options {
+	// get a random seed
+
+	// TODO in go 1.20 this is fine (it will be random)
+	seed := rand.Int63() //nolint:gosec
+	/*
+		// in prior go versions this needs to come from crypto/rand
+		var b [8]byte
+		_, err := crypto_rand.Read(b[:])
+		if err != nil {
+			panic("cryptographically secure random number generator is not working")
+		}
+		seed := int64(binary.LittleEndian.Int64(b[:]))
+	*/
+
+	// defaults
+	o := &options{
+		source: rand.NewSource(seed),
+	}
+
+	for _, opt := range opts {
+		opt(o)
+	}
+	return o
+}
+
+// WithSource sets the random number generator source
+func WithSource(source rand.Source) Option {
+	return func(opts *options) {
+		opts.source = source
+	}
+}
@@ -25,38 +25,35 @@ import (
 	"github.com/google/go-containerregistry/pkg/v1/remote/transport"
 )

-type catalog struct {
+type Catalogs struct {
 	Repos []string `json:"repositories"`
+	Next  string   `json:"next,omitempty"`
 }

 // CatalogPage calls /_catalog, returning the list of repositories on the registry.
 func CatalogPage(target name.Registry, last string, n int, options ...Option) ([]string, error) {
-	o, err := makeOptions(target, options...)
+	o, err := makeOptions(options...)
 	if err != nil {
 		return nil, err
 	}

-	scopes := []string{target.Scope(transport.PullScope)}
-	tr, err := transport.NewWithContext(o.context, target, o.auth, o.transport, scopes)
+	f, err := newPuller(o).fetcher(o.context, target)
 	if err != nil {
 		return nil, err
 	}

-	query := fmt.Sprintf("last=%s&n=%d", url.QueryEscape(last), n)
-
 	uri := url.URL{
 		Scheme:   target.Scheme(),
 		Host:     target.RegistryStr(),
 		Path:     "/v2/_catalog",
-		RawQuery: query,
+		RawQuery: fmt.Sprintf("last=%s&n=%d", url.QueryEscape(last), n),
 	}

-	client := http.Client{Transport: tr}
 	req, err := http.NewRequest(http.MethodGet, uri.String(), nil)
 	if err != nil {
 		return nil, err
 	}
-	resp, err := client.Do(req.WithContext(o.context))
+	resp, err := f.client.Do(req.WithContext(o.context))
 	if err != nil {
 		return nil, err
 	}
@@ -66,7 +63,7 @@ func CatalogPage(target name.Registry, last string, n int, options ...Option) ([
 		return nil, err
 	}

-	var parsed catalog
+	var parsed Catalogs
 	if err := json.NewDecoder(resp.Body).Decode(&parsed); err != nil {
 		return nil, err
 	}
@@ -76,79 +73,87 @@ func CatalogPage(target name.Registry, last string, n int, options ...Option) ([

 // Catalog calls /_catalog, returning the list of repositories on the registry.
 func Catalog(ctx context.Context, target name.Registry, options ...Option) ([]string, error) {
-	o, err := makeOptions(target, options...)
+	o, err := makeOptions(options...)
 	if err != nil {
 		return nil, err
 	}

-	scopes := []string{target.Scope(transport.PullScope)}
-	tr, err := transport.NewWithContext(o.context, target, o.auth, o.transport, scopes)
-	if err != nil {
-		return nil, err
-	}
-
-	uri := &url.URL{
-		Scheme: target.Scheme(),
-		Host:   target.RegistryStr(),
-		Path:   "/v2/_catalog",
-	}
-
-	if o.pageSize > 0 {
-		uri.RawQuery = fmt.Sprintf("n=%d", o.pageSize)
-	}
-
-	client := http.Client{Transport: tr}
-
 	// WithContext overrides the ctx passed directly.
 	if o.context != context.Background() {
 		ctx = o.context
 	}

-	var (
-		parsed   catalog
-		repoList []string
-	)
-
-	// get responses until there is no next page
-	for {
-		select {
-		case <-ctx.Done():
-			return nil, ctx.Err()
-		default:
+	return newPuller(o).catalog(ctx, target, o.pageSize)
+}
+
+func (f *fetcher) catalogPage(ctx context.Context, reg name.Registry, next string, pageSize int) (*Catalogs, error) {
+	if next == "" {
+		uri := &url.URL{
+			Scheme: reg.Scheme(),
+			Host:   reg.RegistryStr(),
+			Path:   "/v2/_catalog",
+		}
+		if pageSize > 0 {
+			uri.RawQuery = fmt.Sprintf("n=%d", pageSize)
 		}
+		next = uri.String()
+	}
+
+	req, err := http.NewRequestWithContext(ctx, "GET", next, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := f.client.Do(req)
+	if err != nil {
+		return nil, err
+	}
+
+	if err := transport.CheckError(resp, http.StatusOK); err != nil {
+		return nil, err
+	}
+
+	parsed := Catalogs{}
+	if err := json.NewDecoder(resp.Body).Decode(&parsed); err != nil {
+		return nil, err
+	}
+
+	if err := resp.Body.Close(); err != nil {
+		return nil, err
+	}
+
+	uri, err := getNextPageURL(resp)
+	if err != nil {
+		return nil, err
+	}

-		req, err := http.NewRequest("GET", uri.String(), nil)
-		if err != nil {
-			return nil, err
-		}
-		req = req.WithContext(ctx)
+	if uri != nil {
+		parsed.Next = uri.String()
+	}

-		resp, err := client.Do(req)
-		if err != nil {
-			return nil, err
-		}
+	return &parsed, nil
+}

-		if err := transport.CheckError(resp, http.StatusOK); err != nil {
-			return nil, err
-		}
+type Catalogger struct {
+	f        *fetcher
+	reg      name.Registry
+	pageSize int

-		if err := json.NewDecoder(resp.Body).Decode(&parsed); err != nil {
-			return nil, err
-		}
-		if err := resp.Body.Close(); err != nil {
-			return nil, err
-		}
+	page *Catalogs
+	err  error

-		repoList = append(repoList, parsed.Repos...)
+	needMore bool
+}

-		uri, err = getNextPageURL(resp)
-		if err != nil {
-			return nil, err
-		}
-		// no next page
-		if uri == nil {
-			break
-		}
+func (l *Catalogger) Next(ctx context.Context) (*Catalogs, error) {
+	if l.needMore {
+		l.page, l.err = l.f.catalogPage(ctx, l.reg, l.page.Next, l.pageSize)
+	} else {
+		l.needMore = true
 	}
-	return repoList, nil
+	return l.page, l.err
+}
+
+func (l *Catalogger) HasNext() bool {
+	return l.page != nil && (!l.needMore || l.page.Next != "")
 }
@@ -15,47 +15,14 @@
 package remote

 import (
-	"fmt"
-	"net/http"
-	"net/url"
-
 	"github.com/google/go-containerregistry/pkg/name"
-	"github.com/google/go-containerregistry/pkg/v1/remote/transport"
 )

 // Delete removes the specified image reference from the remote registry.
 func Delete(ref name.Reference, options ...Option) error {
-	o, err := makeOptions(ref.Context(), options...)
+	o, err := makeOptions(options...)
 	if err != nil {
 		return err
 	}
-	scopes := []string{ref.Scope(transport.DeleteScope)}
-	tr, err := transport.NewWithContext(o.context, ref.Context().Registry, o.auth, o.transport, scopes)
-	if err != nil {
-		return err
-	}
-	c := &http.Client{Transport: tr}
-
-	u := url.URL{
-		Scheme: ref.Context().Registry.Scheme(),
-		Host:   ref.Context().RegistryStr(),
-		Path:   fmt.Sprintf("/v2/%s/manifests/%s", ref.Context().RepositoryStr(), ref.Identifier()),
-	}
-
-	req, err := http.NewRequest(http.MethodDelete, u.String(), nil)
-	if err != nil {
-		return err
-	}
-
-	resp, err := c.Do(req.WithContext(o.context))
-	if err != nil {
-		return err
-	}
-	defer resp.Body.Close()
-
-	return transport.CheckError(resp, http.StatusOK, http.StatusAccepted)
-
-	// TODO(jason): If the manifest had a `subject`, and if the registry
-	// doesn't support Referrers, update the index pointed to by the
-	// subject's fallback tag to remove the descriptor for this manifest.
+	return newPusher(o).Delete(o.context, ref)
 }
@@ -15,26 +15,21 @@
 package remote

 import (
-	"bytes"
 	"context"
-	"encoding/json"
-	"errors"
 	"fmt"
-	"io"
-	"net/http"
-	"net/url"
-	"strings"

-	"github.com/google/go-containerregistry/internal/redact"
-	"github.com/google/go-containerregistry/internal/verify"
 	"github.com/google/go-containerregistry/pkg/logs"
 	"github.com/google/go-containerregistry/pkg/name"
 	v1 "github.com/google/go-containerregistry/pkg/v1"
 	"github.com/google/go-containerregistry/pkg/v1/partial"
-	"github.com/google/go-containerregistry/pkg/v1/remote/transport"
 	"github.com/google/go-containerregistry/pkg/v1/types"
 )

+var allManifestMediaTypes = append(append([]types.MediaType{
+	types.DockerManifestSchema1,
+	types.DockerManifestSchema1Signed,
+}, acceptableImageMediaTypes...), acceptableIndexMediaTypes...)
+
 // ErrSchema1 indicates that we received a schema1 manifest from the registry.
 // This library doesn't have plans to support this legacy image format:
 // https://github.com/google/go-containerregistry/issues/377
@@ -57,14 +52,21 @@ func (e *ErrSchema1) Error() string {
 // Descriptor provides access to metadata about remote artifact and accessors
 // for efficiently converting it into a v1.Image or v1.ImageIndex.
 type Descriptor struct {
-	fetcher
+	fetcher fetcher
 	v1.Descriptor
+
+	ref      name.Reference
 	Manifest []byte
+	ctx      context.Context

 	// So we can share this implementation with Image.
 	platform v1.Platform
 }

+func (d *Descriptor) toDesc() v1.Descriptor {
+	return d.Descriptor
+}
+
 // RawManifest exists to satisfy the Taggable interface.
 func (d *Descriptor) RawManifest() ([]byte, error) {
 	return d.Manifest, nil
@@ -76,14 +78,7 @@ func (d *Descriptor) RawManifest() ([]byte, error) {
 //
 // See Head if you don't need the response body.
 func Get(ref name.Reference, options ...Option) (*Descriptor, error) {
-	acceptable := []types.MediaType{
-		// Just to look at them.
-		types.DockerManifestSchema1,
-		types.DockerManifestSchema1Signed,
-	}
-	acceptable = append(acceptable, acceptableImageMediaTypes...)
-	acceptable = append(acceptable, acceptableIndexMediaTypes...)
-	return get(ref, acceptable, options...)
+	return get(ref, allManifestMediaTypes, options...)
 }

 // Head returns a v1.Descriptor for the given reference by issuing a HEAD
@@ -92,48 +87,22 @@ func Get(ref name.Reference, options ...Option) (*Descriptor, error) {
 // Note that the server response will not have a body, so any errors encountered
 // should be retried with Get to get more details.
 func Head(ref name.Reference, options ...Option) (*v1.Descriptor, error) {
-	acceptable := []types.MediaType{
-		// Just to look at them.
-		types.DockerManifestSchema1,
-		types.DockerManifestSchema1Signed,
-	}
-	acceptable = append(acceptable, acceptableImageMediaTypes...)
-	acceptable = append(acceptable, acceptableIndexMediaTypes...)
-
-	o, err := makeOptions(ref.Context(), options...)
+	o, err := makeOptions(options...)
 	if err != nil {
 		return nil, err
 	}

-	f, err := makeFetcher(ref, o)
-	if err != nil {
-		return nil, err
-	}
-
-	return f.headManifest(ref, acceptable)
+	return newPuller(o).Head(o.context, ref)
 }

 // Handle options and fetch the manifest with the acceptable MediaTypes in the
 // Accept header.
 func get(ref name.Reference, acceptable []types.MediaType, options ...Option) (*Descriptor, error) {
-	o, err := makeOptions(ref.Context(), options...)
+	o, err := makeOptions(options...)
 	if err != nil {
 		return nil, err
 	}
-	f, err := makeFetcher(ref, o)
-	if err != nil {
-		return nil, err
-	}
-	b, desc, err := f.fetchManifest(ref, acceptable)
-	if err != nil {
-		return nil, err
-	}
-	return &Descriptor{
-		fetcher:    *f,
-		Manifest:   b,
-		Descriptor: *desc,
-		platform:   o.platform,
-	}, nil
+	return newPuller(o).get(o.context, ref, acceptable, o.platform)
 }

 // Image converts the Descriptor into a v1.Image.
@@ -169,7 +138,28 @@ func (d *Descriptor) Image() (v1.Image, error) {
 	}
 	return &mountableImage{
 		Image:     imgCore,
-		Reference: d.Ref,
+		Reference: d.ref,
+	}, nil
+}
+
+// Schema1 converts the Descriptor into a v1.Image for v2 schema 1 media types.
+//
+// The v1.Image returned by this method does not implement the entire interface because it would be inefficient.
+// This exists mostly to make it easier to copy schema 1 images around or look at their filesystems.
+// This is separate from Image() to avoid a backward incompatible change for callers expecting ErrSchema1.
+func (d *Descriptor) Schema1() (v1.Image, error) {
+	i := &schema1{
+		ref:        d.ref,
+		fetcher:    d.fetcher,
+		ctx:        d.ctx,
+		manifest:   d.Manifest,
+		mediaType:  d.MediaType,
+		descriptor: &d.Descriptor,
+	}
+
+	return &mountableImage{
+		Image:     i,
+		Reference: d.ref,
 	}, nil
 }

@@ -195,6 +185,8 @@ func (d *Descriptor) ImageIndex() (v1.ImageIndex, error) {

 func (d *Descriptor) remoteImage() *remoteImage {
 	return &remoteImage{
+		ref:        d.ref,
+		ctx:        d.ctx,
 		fetcher:    d.fetcher,
 		manifest:   d.Manifest,
 		mediaType:  d.MediaType,
@@ -204,308 +196,11 @@ func (d *Descriptor) remoteImage() *remoteImage {

 func (d *Descriptor) remoteIndex() *remoteIndex {
 	return &remoteIndex{
+		ref:        d.ref,
+		ctx:        d.ctx,
 		fetcher:    d.fetcher,
 		manifest:   d.Manifest,
 		mediaType:  d.MediaType,
 		descriptor: &d.Descriptor,
 	}
 }
-
-// fetcher implements methods for reading from a registry.
-type fetcher struct {
-	Ref     name.Reference
-	Client  *http.Client
-	context context.Context
-}
-
-func makeFetcher(ref name.Reference, o *options) (*fetcher, error) {
-	tr, err := transport.NewWithContext(o.context, ref.Context().Registry, o.auth, o.transport, []string{ref.Scope(transport.PullScope)})
-	if err != nil {
-		return nil, err
-	}
-	return &fetcher{
-		Ref:     ref,
-		Client:  &http.Client{Transport: tr},
-		context: o.context,
-	}, nil
-}
-
-// url returns a url.Url for the specified path in the context of this remote image reference.
-func (f *fetcher) url(resource, identifier string) url.URL {
-	return url.URL{
-		Scheme: f.Ref.Context().Registry.Scheme(),
-		Host:   f.Ref.Context().RegistryStr(),
-		Path:   fmt.Sprintf("/v2/%s/%s/%s", f.Ref.Context().RepositoryStr(), resource, identifier),
-	}
-}
-
-// https://github.com/opencontainers/distribution-spec/blob/main/spec.md#referrers-tag-schema
-func fallbackTag(d name.Digest) name.Tag {
-	return d.Context().Tag(strings.Replace(d.DigestStr(), ":", "-", 1))
-}
-
-func (f *fetcher) fetchReferrers(ctx context.Context, filter map[string]string, d name.Digest) (*v1.IndexManifest, error) {
-	// Check the Referrers API endpoint first.
-	u := f.url("referrers", d.DigestStr())
-	req, err := http.NewRequestWithContext(ctx, http.MethodGet, u.String(), nil)
-	if err != nil {
-		return nil, err
-	}
-	req.Header.Set("Accept", string(types.OCIImageIndex))
-
-	resp, err := f.Client.Do(req)
-	if err != nil {
-		return nil, err
-	}
-	defer resp.Body.Close()
-
-	if err := transport.CheckError(resp, http.StatusOK, http.StatusNotFound, http.StatusBadRequest); err != nil {
-		return nil, err
-	}
-	if resp.StatusCode == http.StatusOK {
-		var im v1.IndexManifest
-		if err := json.NewDecoder(resp.Body).Decode(&im); err != nil {
-			return nil, err
-		}
-		return filterReferrersResponse(filter, &im), nil
-	}
-
-	// The registry doesn't support the Referrers API endpoint, so we'll use the fallback tag scheme.
-	b, _, err := f.fetchManifest(fallbackTag(d), []types.MediaType{types.OCIImageIndex})
-	if err != nil {
-		return nil, err
-	}
-	var terr *transport.Error
-	if ok := errors.As(err, &terr); ok && terr.StatusCode == http.StatusNotFound {
-		// Not found just means there are no attachments yet. Start with an empty manifest.
-		return &v1.IndexManifest{MediaType: types.OCIImageIndex}, nil
-	}
-
-	var im v1.IndexManifest
-	if err := json.Unmarshal(b, &im); err != nil {
-		return nil, err
-	}
-
-	return filterReferrersResponse(filter, &im), nil
-}
-
-func (f *fetcher) fetchManifest(ref name.Reference, acceptable []types.MediaType) ([]byte, *v1.Descriptor, error) {
-	u := f.url("manifests", ref.Identifier())
-	req, err := http.NewRequest(http.MethodGet, u.String(), nil)
-	if err != nil {
-		return nil, nil, err
-	}
-	accept := []string{}
-	for _, mt := range acceptable {
-		accept = append(accept, string(mt))
-	}
-	req.Header.Set("Accept", strings.Join(accept, ","))
-
-	resp, err := f.Client.Do(req.WithContext(f.context))
-	if err != nil {
-		return nil, nil, err
-	}
-	defer resp.Body.Close()
-
-	if err := transport.CheckError(resp, http.StatusOK); err != nil {
-		return nil, nil, err
-	}
-
-	manifest, err := io.ReadAll(resp.Body)
-	if err != nil {
-		return nil, nil, err
-	}
-
-	digest, size, err := v1.SHA256(bytes.NewReader(manifest))
-	if err != nil {
-		return nil, nil, err
-	}
-
-	mediaType := types.MediaType(resp.Header.Get("Content-Type"))
-	contentDigest, err := v1.NewHash(resp.Header.Get("Docker-Content-Digest"))
-	if err == nil && mediaType == types.DockerManifestSchema1Signed {
-		// If we can parse the digest from the header, and it's a signed schema 1
-		// manifest, let's use that for the digest to appease older registries.
-		digest = contentDigest
-	}
-
-	// Validate the digest matches what we asked for, if pulling by digest.
-	if dgst, ok := ref.(name.Digest); ok {
-		if digest.String() != dgst.DigestStr() {
-			return nil, nil, fmt.Errorf("manifest digest: %q does not match requested digest: %q for %q", digest, dgst.DigestStr(), f.Ref)
-		}
-	}
-
-	var artifactType string
-	mf, _ := v1.ParseManifest(bytes.NewReader(manifest))
-	// Failing to parse as a manifest should just be ignored.
-	// The manifest might not be valid, and that's okay.
-	if mf != nil && !mf.Config.MediaType.IsConfig() {
-		artifactType = string(mf.Config.MediaType)
-	}
-
-	// Do nothing for tags; I give up.
-	//
-	// We'd like to validate that the "Docker-Content-Digest" header matches what is returned by the registry,
-	// but so many registries implement this incorrectly that it's not worth checking.
-	//
-	// For reference:
-	// https://github.com/GoogleContainerTools/kaniko/issues/298
-
-	// Return all this info since we have to calculate it anyway.
-	desc := v1.Descriptor{
-		Digest:       digest,
-		Size:         size,
-		MediaType:    mediaType,
-		ArtifactType: artifactType,
-	}
-
-	return manifest, &desc, nil
-}
-
-func (f *fetcher) headManifest(ref name.Reference, acceptable []types.MediaType) (*v1.Descriptor, error) {
-	u := f.url("manifests", ref.Identifier())
-	req, err := http.NewRequest(http.MethodHead, u.String(), nil)
-	if err != nil {
-		return nil, err
-	}
-	accept := []string{}
-	for _, mt := range acceptable {
-		accept = append(accept, string(mt))
-	}
-	req.Header.Set("Accept", strings.Join(accept, ","))
-
-	resp, err := f.Client.Do(req.WithContext(f.context))
-	if err != nil {
-		return nil, err
-	}
-	defer resp.Body.Close()
-
-	if err := transport.CheckError(resp, http.StatusOK); err != nil {
-		return nil, err
-	}
-
-	mth := resp.Header.Get("Content-Type")
-	if mth == "" {
-		return nil, fmt.Errorf("HEAD %s: response did not include Content-Type header", u.String())
-	}
-	mediaType := types.MediaType(mth)
-
-	size := resp.ContentLength
-	if size == -1 {
-		return nil, fmt.Errorf("GET %s: response did not include Content-Length header", u.String())
-	}
-
-	dh := resp.Header.Get("Docker-Content-Digest")
-	if dh == "" {
-		return nil, fmt.Errorf("HEAD %s: response did not include Docker-Content-Digest header", u.String())
-	}
-	digest, err := v1.NewHash(dh)
-	if err != nil {
-		return nil, err
-	}
-
-	// Validate the digest matches what we asked for, if pulling by digest.
-	if dgst, ok := ref.(name.Digest); ok {
-		if digest.String() != dgst.DigestStr() {
-			return nil, fmt.Errorf("manifest digest: %q does not match requested digest: %q for %q", digest, dgst.DigestStr(), f.Ref)
-		}
-	}
-
-	// Return all this info since we have to calculate it anyway.
-	return &v1.Descriptor{
-		Digest:    digest,
-		Size:      size,
-		MediaType: mediaType,
-	}, nil
-}
-
-func (f *fetcher) fetchBlob(ctx context.Context, size int64, h v1.Hash) (io.ReadCloser, error) {
-	u := f.url("blobs", h.String())
-	req, err := http.NewRequest(http.MethodGet, u.String(), nil)
-	if err != nil {
-		return nil, err
-	}
-
-	resp, err := f.Client.Do(req.WithContext(ctx))
-	if err != nil {
-		return nil, redact.Error(err)
-	}
-
-	if err := transport.CheckError(resp, http.StatusOK); err != nil {
-		resp.Body.Close()
-		return nil, err
-	}
-
-	// Do whatever we can.
-	// If we have an expected size and Content-Length doesn't match, return an error.
-	// If we don't have an expected size and we do have a Content-Length, use Content-Length.
-	if hsize := resp.ContentLength; hsize != -1 {
-		if size == verify.SizeUnknown {
-			size = hsize
-		} else if hsize != size {
-			return nil, fmt.Errorf("GET %s: Content-Length header %d does not match expected size %d", u.String(), hsize, size)
-		}
-	}
-
-	return verify.ReadCloser(resp.Body, size, h)
-}
-
-func (f *fetcher) headBlob(h v1.Hash) (*http.Response, error) {
-	u := f.url("blobs", h.String())
-	req, err := http.NewRequest(http.MethodHead, u.String(), nil)
-	if err != nil {
-		return nil, err
-	}
-
-	resp, err := f.Client.Do(req.WithContext(f.context))
-	if err != nil {
-		return nil, redact.Error(err)
-	}
-
-	if err := transport.CheckError(resp, http.StatusOK); err != nil {
-		resp.Body.Close()
-		return nil, err
-	}
-
-	return resp, nil
-}
-
-func (f *fetcher) blobExists(h v1.Hash) (bool, error) {
-	u := f.url("blobs", h.String())
-	req, err := http.NewRequest(http.MethodHead, u.String(), nil)
-	if err != nil {
-		return false, err
-	}
-
-	resp, err := f.Client.Do(req.WithContext(f.context))
-	if err != nil {
-		return false, redact.Error(err)
-	}
-	defer resp.Body.Close()
-
-	if err := transport.CheckError(resp, http.StatusOK, http.StatusNotFound); err != nil {
-		return false, err
-	}
-
-	return resp.StatusCode == http.StatusOK, nil
-}
-
-// If filter applied, filter out by artifactType.
-// See https://github.com/opencontainers/distribution-spec/blob/main/spec.md#listing-referrers
-func filterReferrersResponse(filter map[string]string, origIndex *v1.IndexManifest) *v1.IndexManifest {
-	newIndex := origIndex
-	if filter == nil {
-		return newIndex
-	}
-	if v, ok := filter["artifactType"]; ok {
-		tmp := []v1.Descriptor{}
-		for _, desc := range newIndex.Manifests {
-			if desc.ArtifactType == v {
-				tmp = append(tmp, desc)
-			}
-		}
-		newIndex.Manifests = tmp
-	}
-	return newIndex
-}
@@ -0,0 +1,318 @@
+// Copyright 2023 Google LLC All Rights Reserved.
+//
+// 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 remote
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"strings"
+
+	"github.com/google/go-containerregistry/internal/redact"
+	"github.com/google/go-containerregistry/internal/verify"
+	"github.com/google/go-containerregistry/pkg/authn"
+	"github.com/google/go-containerregistry/pkg/name"
+	v1 "github.com/google/go-containerregistry/pkg/v1"
+	"github.com/google/go-containerregistry/pkg/v1/remote/transport"
+	"github.com/google/go-containerregistry/pkg/v1/types"
+)
+
+func fetcherFromWriter(w *writer) *fetcher {
+	return &fetcher{
+		target: w.repo,
+		client: w.client,
+	}
+}
+
+// fetcher implements methods for reading from a registry.
+type fetcher struct {
+	target resource
+	client *http.Client
+}
+
+func makeFetcher(ctx context.Context, target resource, o *options) (*fetcher, error) {
+	auth := o.auth
+	if o.keychain != nil {
+		kauth, err := o.keychain.Resolve(target)
+		if err != nil {
+			return nil, err
+		}
+		auth = kauth
+	}
+
+	reg, ok := target.(name.Registry)
+	if !ok {
+		repo, ok := target.(name.Repository)
+		if !ok {
+			return nil, fmt.Errorf("unexpected resource: %T", target)
+		}
+		reg = repo.Registry
+	}
+
+	tr, err := transport.NewWithContext(ctx, reg, auth, o.transport, []string{target.Scope(transport.PullScope)})
+	if err != nil {
+		return nil, err
+	}
+	return &fetcher{
+		target: target,
+		client: &http.Client{Transport: tr},
+	}, nil
+}
+
+func (f *fetcher) Do(req *http.Request) (*http.Response, error) {
+	return f.client.Do(req)
+}
+
+type resource interface {
+	Scheme() string
+	RegistryStr() string
+	Scope(string) string
+
+	authn.Resource
+}
+
+// url returns a url.Url for the specified path in the context of this remote image reference.
+func (f *fetcher) url(resource, identifier string) url.URL {
+	u := url.URL{
+		Scheme: f.target.Scheme(),
+		Host:   f.target.RegistryStr(),
+		// Default path if this is not a repository.
+		Path: "/v2/_catalog",
+	}
+	if repo, ok := f.target.(name.Repository); ok {
+		u.Path = fmt.Sprintf("/v2/%s/%s/%s", repo.RepositoryStr(), resource, identifier)
+	}
+	return u
+}
+
+func (f *fetcher) get(ctx context.Context, ref name.Reference, acceptable []types.MediaType, platform v1.Platform) (*Descriptor, error) {
+	b, desc, err := f.fetchManifest(ctx, ref, acceptable)
+	if err != nil {
+		return nil, err
+	}
+	return &Descriptor{
+		ref:        ref,
+		ctx:        ctx,
+		fetcher:    *f,
+		Manifest:   b,
+		Descriptor: *desc,
+		platform:   platform,
+	}, nil
+}
+
+func (f *fetcher) fetchManifest(ctx context.Context, ref name.Reference, acceptable []types.MediaType) ([]byte, *v1.Descriptor, error) {
+	u := f.url("manifests", ref.Identifier())
+	req, err := http.NewRequest(http.MethodGet, u.String(), nil)
+	if err != nil {
+		return nil, nil, err
+	}
+	accept := []string{}
+	for _, mt := range acceptable {
+		accept = append(accept, string(mt))
+	}
+	req.Header.Set("Accept", strings.Join(accept, ","))
+
+	resp, err := f.client.Do(req.WithContext(ctx))
+	if err != nil {
+		return nil, nil, err
+	}
+	defer resp.Body.Close()
+
+	if err := transport.CheckError(resp, http.StatusOK); err != nil {
+		return nil, nil, err
+	}
+
+	manifest, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	digest, size, err := v1.SHA256(bytes.NewReader(manifest))
+	if err != nil {
+		return nil, nil, err
+	}
+
+	mediaType := types.MediaType(resp.Header.Get("Content-Type"))
+	contentDigest, err := v1.NewHash(resp.Header.Get("Docker-Content-Digest"))
+	if err == nil && mediaType == types.DockerManifestSchema1Signed {
+		// If we can parse the digest from the header, and it's a signed schema 1
+		// manifest, let's use that for the digest to appease older registries.
+		digest = contentDigest
+	}
+
+	// Validate the digest matches what we asked for, if pulling by digest.
+	if dgst, ok := ref.(name.Digest); ok {
+		if digest.String() != dgst.DigestStr() {
+			return nil, nil, fmt.Errorf("manifest digest: %q does not match requested digest: %q for %q", digest, dgst.DigestStr(), ref)
+		}
+	}
+
+	var artifactType string
+	mf, _ := v1.ParseManifest(bytes.NewReader(manifest))
+	// Failing to parse as a manifest should just be ignored.
+	// The manifest might not be valid, and that's okay.
+	if mf != nil && !mf.Config.MediaType.IsConfig() {
+		artifactType = string(mf.Config.MediaType)
+	}
+
+	// Do nothing for tags; I give up.
+	//
+	// We'd like to validate that the "Docker-Content-Digest" header matches what is returned by the registry,
+	// but so many registries implement this incorrectly that it's not worth checking.
+	//
+	// For reference:
+	// https://github.com/GoogleContainerTools/kaniko/issues/298
+
+	// Return all this info since we have to calculate it anyway.
+	desc := v1.Descriptor{
+		Digest:       digest,
+		Size:         size,
+		MediaType:    mediaType,
+		ArtifactType: artifactType,
+	}
+
+	return manifest, &desc, nil
+}
+
+func (f *fetcher) headManifest(ctx context.Context, ref name.Reference, acceptable []types.MediaType) (*v1.Descriptor, error) {
+	u := f.url("manifests", ref.Identifier())
+	req, err := http.NewRequest(http.MethodHead, u.String(), nil)
+	if err != nil {
+		return nil, err
+	}
+	accept := []string{}
+	for _, mt := range acceptable {
+		accept = append(accept, string(mt))
+	}
+	req.Header.Set("Accept", strings.Join(accept, ","))
+
+	resp, err := f.client.Do(req.WithContext(ctx))
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	if err := transport.CheckError(resp, http.StatusOK); err != nil {
+		return nil, err
+	}
+
+	mth := resp.Header.Get("Content-Type")
+	if mth == "" {
+		return nil, fmt.Errorf("HEAD %s: response did not include Content-Type header", u.String())
+	}
+	mediaType := types.MediaType(mth)
+
+	size := resp.ContentLength
+	if size == -1 {
+		return nil, fmt.Errorf("GET %s: response did not include Content-Length header", u.String())
+	}
+
+	dh := resp.Header.Get("Docker-Content-Digest")
+	if dh == "" {
+		return nil, fmt.Errorf("HEAD %s: response did not include Docker-Content-Digest header", u.String())
+	}
+	digest, err := v1.NewHash(dh)
+	if err != nil {
+		return nil, err
+	}
+
+	// Validate the digest matches what we asked for, if pulling by digest.
+	if dgst, ok := ref.(name.Digest); ok {
+		if digest.String() != dgst.DigestStr() {
+			return nil, fmt.Errorf("manifest digest: %q does not match requested digest: %q for %q", digest, dgst.DigestStr(), ref)
+		}
+	}
+
+	// Return all this info since we have to calculate it anyway.
+	return &v1.Descriptor{
+		Digest:    digest,
+		Size:      size,
+		MediaType: mediaType,
+	}, nil
+}
+
+func (f *fetcher) fetchBlob(ctx context.Context, size int64, h v1.Hash) (io.ReadCloser, error) {
+	u := f.url("blobs", h.String())
+	req, err := http.NewRequest(http.MethodGet, u.String(), nil)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := f.client.Do(req.WithContext(ctx))
+	if err != nil {
+		return nil, redact.Error(err)
+	}
+
+	if err := transport.CheckError(resp, http.StatusOK); err != nil {
+		resp.Body.Close()
+		return nil, err
+	}
+
+	// Do whatever we can.
+	// If we have an expected size and Content-Length doesn't match, return an error.
+	// If we don't have an expected size and we do have a Content-Length, use Content-Length.
+	if hsize := resp.ContentLength; hsize != -1 {
+		if size == verify.SizeUnknown {
+			size = hsize
+		} else if hsize != size {
+			return nil, fmt.Errorf("GET %s: Content-Length header %d does not match expected size %d", u.String(), hsize, size)
+		}
+	}
+
+	return verify.ReadCloser(resp.Body, size, h)
+}
+
+func (f *fetcher) headBlob(ctx context.Context, h v1.Hash) (*http.Response, error) {
+	u := f.url("blobs", h.String())
+	req, err := http.NewRequest(http.MethodHead, u.String(), nil)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := f.client.Do(req.WithContext(ctx))
+	if err != nil {
+		return nil, redact.Error(err)
+	}
+
+	if err := transport.CheckError(resp, http.StatusOK); err != nil {
+		resp.Body.Close()
+		return nil, err
+	}
+
+	return resp, nil
+}
+
+func (f *fetcher) blobExists(ctx context.Context, h v1.Hash) (bool, error) {
+	u := f.url("blobs", h.String())
+	req, err := http.NewRequest(http.MethodHead, u.String(), nil)
+	if err != nil {
+		return false, err
+	}
+
+	resp, err := f.client.Do(req.WithContext(ctx))
+	if err != nil {
+		return false, redact.Error(err)
+	}
+	defer resp.Body.Close()
+
+	if err := transport.CheckError(resp, http.StatusOK, http.StatusNotFound); err != nil {
+		return false, err
+	}
+
+	return resp.StatusCode == http.StatusOK, nil
+}
@@ -16,6 +16,7 @@ package remote

 import (
 	"bytes"
+	"context"
 	"io"
 	"net/http"
 	"net/url"
@@ -37,7 +38,9 @@ var acceptableImageMediaTypes = []types.MediaType{

 // remoteImage accesses an image from a remote registry
 type remoteImage struct {
-	fetcher
+	fetcher      fetcher
+	ref          name.Reference
+	ctx          context.Context
 	manifestLock sync.Mutex // Protects manifest
 	manifest     []byte
 	configLock   sync.Mutex // Protects config
@@ -84,7 +87,7 @@ func (r *remoteImage) RawManifest() ([]byte, error) {
 	// NOTE(jonjohnsonjr): We should never get here because the public entrypoints
 	// do type-checking via remote.Descriptor. I've left this here for tests that
 	// directly instantiate a remoteImage.
-	manifest, desc, err := r.fetchManifest(r.Ref, acceptableImageMediaTypes)
+	manifest, desc, err := r.fetcher.fetchManifest(r.ctx, r.ref, acceptableImageMediaTypes)
 	if err != nil {
 		return nil, err
 	}
@@ -117,7 +120,7 @@ func (r *remoteImage) RawConfigFile() ([]byte, error) {
 		return r.config, nil
 	}

-	body, err := r.fetchBlob(r.context, m.Config.Size, m.Config.Digest)
+	body, err := r.fetcher.fetchBlob(r.ctx, m.Config.Size, m.Config.Digest)
 	if err != nil {
 		return nil, err
 	}
@@ -139,9 +142,26 @@ func (r *remoteImage) Descriptor() (*v1.Descriptor, error) {
 	return r.descriptor, err
 }

+func (r *remoteImage) ConfigLayer() (v1.Layer, error) {
+	if _, err := r.RawManifest(); err != nil {
+		return nil, err
+	}
+	m, err := partial.Manifest(r)
+	if err != nil {
+		return nil, err
+	}
+
+	return partial.CompressedToLayer(&remoteImageLayer{
+		ri:     r,
+		ctx:    r.ctx,
+		digest: m.Config.Digest,
+	})
+}
+
 // remoteImageLayer implements partial.CompressedLayer
 type remoteImageLayer struct {
 	ri     *remoteImage
+	ctx    context.Context
 	digest v1.Hash
 }

@@ -152,7 +172,7 @@ func (rl *remoteImageLayer) Digest() (v1.Hash, error) {

 // Compressed implements partial.CompressedLayer
 func (rl *remoteImageLayer) Compressed() (io.ReadCloser, error) {
-	urls := []url.URL{rl.ri.url("blobs", rl.digest.String())}
+	urls := []url.URL{rl.ri.fetcher.url("blobs", rl.digest.String())}

 	// Add alternative layer sources from URLs (usually none).
 	d, err := partial.BlobDescriptor(rl, rl.digest)
@@ -165,7 +185,7 @@ func (rl *remoteImageLayer) Compressed() (io.ReadCloser, error) {
 	}

 	// We don't want to log binary layers -- this can break terminals.
-	ctx := redact.NewContext(rl.ri.context, "omitting binary blobs from logs")
+	ctx := redact.NewContext(rl.ctx, "omitting binary blobs from logs")

 	for _, s := range d.URLs {
 		u, err := url.Parse(s)
@@ -186,7 +206,7 @@ func (rl *remoteImageLayer) Compressed() (io.ReadCloser, error) {
 			return nil, err
 		}

-		resp, err := rl.ri.Client.Do(req.WithContext(ctx))
+		resp, err := rl.ri.fetcher.Do(req.WithContext(ctx))
 		if err != nil {
 			lastErr = err
 			continue
@@ -244,13 +264,14 @@ func (rl *remoteImageLayer) Descriptor() (*v1.Descriptor, error) {

 // See partial.Exists.
 func (rl *remoteImageLayer) Exists() (bool, error) {
-	return rl.ri.blobExists(rl.digest)
+	return rl.ri.fetcher.blobExists(rl.ri.ctx, rl.digest)
 }

 // LayerByDigest implements partial.CompressedLayer
 func (r *remoteImage) LayerByDigest(h v1.Hash) (partial.CompressedLayer, error) {
 	return &remoteImageLayer{
 		ri:     r,
+		ctx:    r.ctx,
 		digest: h,
 	}, nil
 }
@@ -16,6 +16,7 @@ package remote

 import (
 	"bytes"
+	"context"
 	"fmt"
 	"sync"

@@ -33,7 +34,9 @@ var acceptableIndexMediaTypes = []types.MediaType{

 // remoteIndex accesses an index from a remote registry
 type remoteIndex struct {
-	fetcher
+	fetcher      fetcher
+	ref          name.Reference
+	ctx          context.Context
 	manifestLock sync.Mutex // Protects manifest
 	manifest     []byte
 	mediaType    types.MediaType
@@ -75,7 +78,7 @@ func (r *remoteIndex) RawManifest() ([]byte, error) {
 	// NOTE(jonjohnsonjr): We should never get here because the public entrypoints
 	// do type-checking via remote.Descriptor. I've left this here for tests that
 	// directly instantiate a remoteIndex.
-	manifest, desc, err := r.fetchManifest(r.Ref, acceptableIndexMediaTypes)
+	manifest, desc, err := r.fetcher.fetchManifest(r.ctx, r.ref, acceptableIndexMediaTypes)
 	if err != nil {
 		return nil, err
 	}
@@ -133,6 +136,7 @@ func (r *remoteIndex) Layer(h v1.Hash) (v1.Layer, error) {
 		if h == childDesc.Digest {
 			l, err := partial.CompressedToLayer(&remoteLayer{
 				fetcher: r.fetcher,
+				ctx:     r.ctx,
 				digest:  h,
 			})
 			if err != nil {
@@ -140,47 +144,13 @@ func (r *remoteIndex) Layer(h v1.Hash) (v1.Layer, error) {
 			}
 			return &MountableLayer{
 				Layer:     l,
-				Reference: r.Ref.Context().Digest(h.String()),
+				Reference: r.ref.Context().Digest(h.String()),
 			}, nil
 		}
 	}
 	return nil, fmt.Errorf("layer not found: %s", h)
 }

-// Experiment with a better API for v1.ImageIndex. We might want to move this
-// to partial?
-func (r *remoteIndex) Manifests() ([]partial.Describable, error) {
-	m, err := r.IndexManifest()
-	if err != nil {
-		return nil, err
-	}
-	manifests := []partial.Describable{}
-	for _, desc := range m.Manifests {
-		switch {
-		case desc.MediaType.IsImage():
-			img, err := r.Image(desc.Digest)
-			if err != nil {
-				return nil, err
-			}
-			manifests = append(manifests, img)
-		case desc.MediaType.IsIndex():
-			idx, err := r.ImageIndex(desc.Digest)
-			if err != nil {
-				return nil, err
-			}
-			manifests = append(manifests, idx)
-		default:
-			layer, err := r.Layer(desc.Digest)
-			if err != nil {
-				return nil, err
-			}
-			manifests = append(manifests, layer)
-		}
-	}
-
-	return manifests, nil
-}
-
 func (r *remoteIndex) imageByPlatform(platform v1.Platform) (v1.Image, error) {
 	desc, err := r.childByPlatform(platform)
 	if err != nil {
@@ -216,7 +186,7 @@ func (r *remoteIndex) childByPlatform(platform v1.Platform) (*Descriptor, error)
 			return r.childDescriptor(childDesc, platform)
 		}
 	}
-	return nil, fmt.Errorf("no child with platform %+v in index %s", platform, r.Ref)
+	return nil, fmt.Errorf("no child with platform %+v in index %s", platform, r.ref)
 }

 func (r *remoteIndex) childByHash(h v1.Hash) (*Descriptor, error) {
@@ -229,12 +199,12 @@ func (r *remoteIndex) childByHash(h v1.Hash) (*Descriptor, error) {
 			return r.childDescriptor(childDesc, defaultPlatform)
 		}
 	}
-	return nil, fmt.Errorf("no child with digest %s in index %s", h, r.Ref)
+	return nil, fmt.Errorf("no child with digest %s in index %s", h, r.ref)
 }

 // Convert one of this index's child's v1.Descriptor into a remote.Descriptor, with the given platform option.
 func (r *remoteIndex) childDescriptor(child v1.Descriptor, platform v1.Platform) (*Descriptor, error) {
-	ref := r.Ref.Context().Digest(child.Digest.String())
+	ref := r.ref.Context().Digest(child.Digest.String())
 	var (
 		manifest []byte
 		err      error
@@ -245,7 +215,7 @@ func (r *remoteIndex) childDescriptor(child v1.Descriptor, platform v1.Platform)
 		}
 		manifest = child.Data
 	} else {
-		manifest, _, err = r.fetchManifest(ref, []types.MediaType{child.MediaType})
+		manifest, _, err = r.fetcher.fetchManifest(r.ctx, ref, []types.MediaType{child.MediaType})
 		if err != nil {
 			return nil, err
 		}
@@ -261,11 +231,9 @@ func (r *remoteIndex) childDescriptor(child v1.Descriptor, platform v1.Platform)
 	}

 	return &Descriptor{
-		fetcher: fetcher{
-			Ref:     ref,
-			Client:  r.Client,
-			context: r.context,
-		},
+		ref:        ref,
+		ctx:        r.ctx,
+		fetcher:    r.fetcher,
 		Manifest:   manifest,
 		Descriptor: child,
 		platform:   platform,
@@ -15,32 +15,33 @@
 package remote

 import (
+	"context"
 	"io"

 	"github.com/google/go-containerregistry/internal/redact"
 	"github.com/google/go-containerregistry/internal/verify"
 	"github.com/google/go-containerregistry/pkg/name"
 	v1 "github.com/google/go-containerregistry/pkg/v1"
-	"github.com/google/go-containerregistry/pkg/v1/partial"
 	"github.com/google/go-containerregistry/pkg/v1/types"
 )

 // remoteImagelayer implements partial.CompressedLayer
 type remoteLayer struct {
-	fetcher
-	digest v1.Hash
+	ctx     context.Context
+	fetcher fetcher
+	digest  v1.Hash
 }

 // Compressed implements partial.CompressedLayer
 func (rl *remoteLayer) Compressed() (io.ReadCloser, error) {
 	// We don't want to log binary layers -- this can break terminals.
-	ctx := redact.NewContext(rl.context, "omitting binary blobs from logs")
-	return rl.fetchBlob(ctx, verify.SizeUnknown, rl.digest)
+	ctx := redact.NewContext(rl.ctx, "omitting binary blobs from logs")
+	return rl.fetcher.fetchBlob(ctx, verify.SizeUnknown, rl.digest)
 }

 // Compressed implements partial.CompressedLayer
 func (rl *remoteLayer) Size() (int64, error) {
-	resp, err := rl.headBlob(rl.digest)
+	resp, err := rl.fetcher.headBlob(rl.ctx, rl.digest)
 	if err != nil {
 		return -1, err
 	}
@@ -60,7 +61,7 @@ func (rl *remoteLayer) MediaType() (types.MediaType, error) {

 // See partial.Exists.
 func (rl *remoteLayer) Exists() (bool, error) {
-	return rl.blobExists(rl.digest)
+	return rl.fetcher.blobExists(rl.ctx, rl.digest)
 }

 // Layer reads the given blob reference from a registry as a Layer. A blob
@@ -68,27 +69,9 @@ func (rl *remoteLayer) Exists() (bool, error) {
 // digest of the blob to be read and the repository portion is the repo where
 // that blob lives.
 func Layer(ref name.Digest, options ...Option) (v1.Layer, error) {
-	o, err := makeOptions(ref.Context(), options...)
+	o, err := makeOptions(options...)
 	if err != nil {
 		return nil, err
 	}
-	f, err := makeFetcher(ref, o)
-	if err != nil {
-		return nil, err
-	}
-	h, err := v1.NewHash(ref.Identifier())
-	if err != nil {
-		return nil, err
-	}
-	l, err := partial.CompressedToLayer(&remoteLayer{
-		fetcher: *f,
-		digest:  h,
-	})
-	if err != nil {
-		return nil, err
-	}
-	return &MountableLayer{
-		Layer:     l,
-		Reference: ref,
-	}, nil
+	return newPuller(o).Layer(o.context, ref)
 }
@@ -26,11 +26,6 @@ import (
 	"github.com/google/go-containerregistry/pkg/v1/remote/transport"
 )

-type tags struct {
-	Name string   `json:"name"`
-	Tags []string `json:"tags"`
-}
-
 // ListWithContext calls List with the given context.
 //
 // Deprecated: Use List and WithContext. This will be removed in a future release.
@@ -41,73 +36,65 @@ func ListWithContext(ctx context.Context, repo name.Repository, options ...Optio
 // List calls /tags/list for the given repository, returning the list of tags
 // in the "tags" property.
 func List(repo name.Repository, options ...Option) ([]string, error) {
-	o, err := makeOptions(repo, options...)
+	o, err := makeOptions(options...)
 	if err != nil {
 		return nil, err
 	}
-	scopes := []string{repo.Scope(transport.PullScope)}
-	tr, err := transport.NewWithContext(o.context, repo.Registry, o.auth, o.transport, scopes)
+	return newPuller(o).List(o.context, repo)
+}
+
+type Tags struct {
+	Name string   `json:"name"`
+	Tags []string `json:"tags"`
+	Next string   `json:"next,omitempty"`
+}
+
+func (f *fetcher) listPage(ctx context.Context, repo name.Repository, next string, pageSize int) (*Tags, error) {
+	if next == "" {
+		uri := &url.URL{
+			Scheme: repo.Scheme(),
+			Host:   repo.RegistryStr(),
+			Path:   fmt.Sprintf("/v2/%s/tags/list", repo.RepositoryStr()),
+		}
+		if pageSize > 0 {
+			uri.RawQuery = fmt.Sprintf("n=%d", pageSize)
+		}
+		next = uri.String()
+	}
+
+	req, err := http.NewRequestWithContext(ctx, "GET", next, nil)
 	if err != nil {
 		return nil, err
 	}

-	uri := &url.URL{
-		Scheme: repo.Registry.Scheme(),
-		Host:   repo.Registry.RegistryStr(),
-		Path:   fmt.Sprintf("/v2/%s/tags/list", repo.RepositoryStr()),
+	resp, err := f.client.Do(req)
+	if err != nil {
+		return nil, err
 	}

-	if o.pageSize > 0 {
-		uri.RawQuery = fmt.Sprintf("n=%d", o.pageSize)
+	if err := transport.CheckError(resp, http.StatusOK); err != nil {
+		return nil, err
 	}

-	client := http.Client{Transport: tr}
-	tagList := []string{}
-	parsed := tags{}
-
-	// get responses until there is no next page
-	for {
-		select {
-		case <-o.context.Done():
-			return nil, o.context.Err()
-		default:
-		}
-
-		req, err := http.NewRequestWithContext(o.context, "GET", uri.String(), nil)
-		if err != nil {
-			return nil, err
-		}
-
-		resp, err := client.Do(req)
-		if err != nil {
-			return nil, err
-		}
-
-		if err := transport.CheckError(resp, http.StatusOK); err != nil {
-			return nil, err
-		}
-
-		if err := json.NewDecoder(resp.Body).Decode(&parsed); err != nil {
-			return nil, err
-		}
-
-		if err := resp.Body.Close(); err != nil {
-			return nil, err
-		}
-
-		tagList = append(tagList, parsed.Tags...)
-
-		uri, err = getNextPageURL(resp)
-		if err != nil {
-			return nil, err
-		}
-		// no next page
-		if uri == nil {
-			break
-		}
+	parsed := Tags{}
+	if err := json.NewDecoder(resp.Body).Decode(&parsed); err != nil {
+		return nil, err
 	}

-	return tagList, nil
+	if err := resp.Body.Close(); err != nil {
+		return nil, err
+	}
+
+	uri, err := getNextPageURL(resp)
+	if err != nil {
+		return nil, err
+	}
+
+	if uri != nil {
+		parsed.Next = uri.String()
+	}
+
+	return &parsed, nil
 }

 // getNextPageURL checks if there is a Link header in a http.Response which
@@ -139,3 +126,27 @@ func getNextPageURL(resp *http.Response) (*url.URL, error) {
 	linkURL = resp.Request.URL.ResolveReference(linkURL)
 	return linkURL, nil
 }
+
+type Lister struct {
+	f        *fetcher
+	repo     name.Repository
+	pageSize int
+
+	page *Tags
+	err  error
+
+	needMore bool
+}
+
+func (l *Lister) Next(ctx context.Context) (*Tags, error) {
+	if l.needMore {
+		l.page, l.err = l.f.listPage(ctx, l.repo, l.page.Next, l.pageSize)
+	} else {
+		l.needMore = true
+	}
+	return l.page, l.err
+}
+
+func (l *Lister) HasNext() bool {
+	return l.page != nil && (!l.needMore || l.page.Next != "")
+}
@@ -15,288 +15,32 @@
 package remote

 import (
-	"context"
-	"fmt"
-	"net/http"
-
 	"github.com/google/go-containerregistry/pkg/name"
-	v1 "github.com/google/go-containerregistry/pkg/v1"
-	"github.com/google/go-containerregistry/pkg/v1/partial"
-	"github.com/google/go-containerregistry/pkg/v1/remote/transport"
-	"github.com/google/go-containerregistry/pkg/v1/types"
 	"golang.org/x/sync/errgroup"
 )

 // MultiWrite writes the given Images or ImageIndexes to the given refs, as
-// efficiently as possible, by deduping shared layer blobs and uploading layers
-// in parallel, then uploading all manifests in parallel.
-//
-// Current limitations:
-// - All refs must share the same repository.
-// - Images cannot consist of stream.Layers.
-func MultiWrite(m map[name.Reference]Taggable, options ...Option) (rerr error) {
-	// Determine the repository being pushed to; if asked to push to
-	// multiple repositories, give up.
-	var repo, zero name.Repository
-	for ref := range m {
-		if repo == zero {
-			repo = ref.Context()
-		} else if ref.Context() != repo {
-			return fmt.Errorf("MultiWrite can only push to the same repository (saw %q and %q)", repo, ref.Context())
-		}
-	}
-
-	o, err := makeOptions(repo, options...)
+// efficiently as possible, by deduping shared layer blobs while uploading them
+// in parallel.
+func MultiWrite(todo map[name.Reference]Taggable, options ...Option) (rerr error) {
+	o, err := makeOptions(options...)
 	if err != nil {
 		return err
 	}
+	if o.progress != nil {
+		defer func() { o.progress.Close(rerr) }()
+	}
+	p := newPusher(o)

-	// Collect unique blobs (layers and config blobs).
-	blobs := map[v1.Hash]v1.Layer{}
-	newManifests := []map[name.Reference]Taggable{}
-	// Separate originally requested images and indexes, so we can push images first.
-	images, indexes := map[name.Reference]Taggable{}, map[name.Reference]Taggable{}
-	for ref, i := range m {
-		if img, ok := i.(v1.Image); ok {
-			images[ref] = i
-			if err := addImageBlobs(img, blobs, o.allowNondistributableArtifacts); err != nil {
-				return err
-			}
-			continue
-		}
-		if idx, ok := i.(v1.ImageIndex); ok {
-			indexes[ref] = i
-			newManifests, err = addIndexBlobs(idx, blobs, repo, newManifests, 0, o.allowNondistributableArtifacts)
-			if err != nil {
-				return err
-			}
-			continue
-		}
-		return fmt.Errorf("pushable resource was not Image or ImageIndex: %T", i)
-	}
+	g, ctx := errgroup.WithContext(o.context)
+	g.SetLimit(o.jobs)

-	// Determine if any of the layers are Mountable, because if so we need
-	// to request Pull scope too.
-	ls := []v1.Layer{}
-	for _, l := range blobs {
-		ls = append(ls, l)
-	}
-	scopes := scopesForUploadingImage(repo, ls)
-	tr, err := transport.NewWithContext(o.context, repo.Registry, o.auth, o.transport, scopes)
-	if err != nil {
-		return err
-	}
-	w := writer{
-		repo:      repo,
-		client:    &http.Client{Transport: tr},
-		backoff:   o.retryBackoff,
-		predicate: o.retryPredicate,
-	}
-
-	// Collect the total size of blobs and manifests we're about to write.
-	if o.updates != nil {
-		w.progress = &progress{updates: o.updates}
-		w.progress.lastUpdate = &v1.Update{}
-		defer close(o.updates)
-		defer func() { _ = w.progress.err(rerr) }()
-		for _, b := range blobs {
-			size, err := b.Size()
-			if err != nil {
-				return err
-			}
-			w.progress.total(size)
-		}
-		countManifest := func(t Taggable) error {
-			b, err := t.RawManifest()
-			if err != nil {
-				return err
-			}
-			w.progress.total(int64(len(b)))
-			return nil
-		}
-		for _, i := range images {
-			if err := countManifest(i); err != nil {
-				return err
-			}
-		}
-		for _, nm := range newManifests {
-			for _, i := range nm {
-				if err := countManifest(i); err != nil {
-					return err
-				}
-			}
-		}
-		for _, i := range indexes {
-			if err := countManifest(i); err != nil {
-				return err
-			}
-		}
-	}
-
-	// Upload individual blobs and collect any errors.
-	blobChan := make(chan v1.Layer, 2*o.jobs)
-	ctx := o.context
-	g, gctx := errgroup.WithContext(o.context)
-	for i := 0; i < o.jobs; i++ {
-		// Start N workers consuming blobs to upload.
+	for ref, t := range todo {
+		ref, t := ref, t
 		g.Go(func() error {
-			for b := range blobChan {
-				if err := w.uploadOne(gctx, b); err != nil {
-					return err
-				}
-			}
-			return nil
+			return p.Push(ctx, ref, t)
 		})
 	}
-	g.Go(func() error {
-		defer close(blobChan)
-		for _, b := range blobs {
-			select {
-			case blobChan <- b:
-			case <-gctx.Done():
-				return gctx.Err()
-			}
-		}
-		return nil
-	})
-	if err := g.Wait(); err != nil {
-		return err
-	}

-	commitMany := func(ctx context.Context, m map[name.Reference]Taggable) error {
-		g, ctx := errgroup.WithContext(ctx)
-		// With all of the constituent elements uploaded, upload the manifests
-		// to commit the images and indexes, and collect any errors.
-		type task struct {
-			i   Taggable
-			ref name.Reference
-		}
-		taskChan := make(chan task, 2*o.jobs)
-		for i := 0; i < o.jobs; i++ {
-			// Start N workers consuming tasks to upload manifests.
-			g.Go(func() error {
-				for t := range taskChan {
-					if err := w.commitManifest(ctx, t.i, t.ref); err != nil {
-						return err
-					}
-				}
-				return nil
-			})
-		}
-		go func() {
-			for ref, i := range m {
-				taskChan <- task{i, ref}
-			}
-			close(taskChan)
-		}()
-		return g.Wait()
-	}
-	// Push originally requested image manifests. These have no
-	// dependencies.
-	if err := commitMany(ctx, images); err != nil {
-		return err
-	}
-	// Push new manifests from lowest levels up.
-	for i := len(newManifests) - 1; i >= 0; i-- {
-		if err := commitMany(ctx, newManifests[i]); err != nil {
-			return err
-		}
-	}
-	// Push originally requested index manifests, which might depend on
-	// newly discovered manifests.
-
-	return commitMany(ctx, indexes)
-}
-
-// addIndexBlobs adds blobs to the set of blobs we intend to upload, and
-// returns the latest copy of the ordered collection of manifests to upload.
-func addIndexBlobs(idx v1.ImageIndex, blobs map[v1.Hash]v1.Layer, repo name.Repository, newManifests []map[name.Reference]Taggable, lvl int, allowNondistributableArtifacts bool) ([]map[name.Reference]Taggable, error) {
-	if lvl > len(newManifests)-1 {
-		newManifests = append(newManifests, map[name.Reference]Taggable{})
-	}
-
-	im, err := idx.IndexManifest()
-	if err != nil {
-		return nil, err
-	}
-	for _, desc := range im.Manifests {
-		switch desc.MediaType {
-		case types.OCIImageIndex, types.DockerManifestList:
-			idx, err := idx.ImageIndex(desc.Digest)
-			if err != nil {
-				return nil, err
-			}
-			newManifests, err = addIndexBlobs(idx, blobs, repo, newManifests, lvl+1, allowNondistributableArtifacts)
-			if err != nil {
-				return nil, err
-			}
-
-			// Also track the sub-index manifest to upload later by digest.
-			newManifests[lvl][repo.Digest(desc.Digest.String())] = idx
-		case types.OCIManifestSchema1, types.DockerManifestSchema2:
-			img, err := idx.Image(desc.Digest)
-			if err != nil {
-				return nil, err
-			}
-			if err := addImageBlobs(img, blobs, allowNondistributableArtifacts); err != nil {
-				return nil, err
-			}
-
-			// Also track the sub-image manifest to upload later by digest.
-			newManifests[lvl][repo.Digest(desc.Digest.String())] = img
-		default:
-			// Workaround for #819.
-			if wl, ok := idx.(withLayer); ok {
-				layer, err := wl.Layer(desc.Digest)
-				if err != nil {
-					return nil, err
-				}
-				if err := addLayerBlob(layer, blobs, allowNondistributableArtifacts); err != nil {
-					return nil, err
-				}
-			} else {
-				return nil, fmt.Errorf("unknown media type: %v", desc.MediaType)
-			}
-		}
-	}
-	return newManifests, nil
-}
-
-func addLayerBlob(l v1.Layer, blobs map[v1.Hash]v1.Layer, allowNondistributableArtifacts bool) error {
-	// Ignore foreign layers.
-	mt, err := l.MediaType()
-	if err != nil {
-		return err
-	}
-
-	if mt.IsDistributable() || allowNondistributableArtifacts {
-		d, err := l.Digest()
-		if err != nil {
-			return err
-		}
-
-		blobs[d] = l
-	}
-
-	return nil
-}
-
-func addImageBlobs(img v1.Image, blobs map[v1.Hash]v1.Layer, allowNondistributableArtifacts bool) error {
-	ls, err := img.Layers()
-	if err != nil {
-		return err
-	}
-	// Collect all layers.
-	for _, l := range ls {
-		if err := addLayerBlob(l, blobs, allowNondistributableArtifacts); err != nil {
-			return err
-		}
-	}
-
-	// Collect config blob.
-	cl, err := partial.ConfigLayer(img)
-	if err != nil {
-		return err
-	}
-	return addLayerBlob(cl, blobs, allowNondistributableArtifacts)
+	return g.Wait()
 }
@@ -37,16 +37,23 @@ type options struct {
 	auth                           authn.Authenticator
 	keychain                       authn.Keychain
 	transport                      http.RoundTripper
-	platform                       v1.Platform
 	context                        context.Context
 	jobs                           int
 	userAgent                      string
 	allowNondistributableArtifacts bool
-	updates                        chan<- v1.Update
-	pageSize                       int
+	progress                       *progress
 	retryBackoff                   Backoff
 	retryPredicate                 retry.Predicate
-	filter                         map[string]string
+	retryStatusCodes               []int
+
+	// Only these options can overwrite Reuse()d options.
+	platform v1.Platform
+	pageSize int
+	filter   map[string]string
+
+	// Set by Reuse, we currently store one or the other.
+	puller *Puller
+	pusher *Pusher
 }

 var defaultPlatform = v1.Platform{
@@ -60,7 +67,7 @@ type Backoff = retry.Backoff
 var defaultRetryPredicate retry.Predicate = func(err error) bool {
 	// Various failure modes here, as we're often reading from and writing to
 	// the network.
-	if retry.IsTemporary(err) || errors.Is(err, io.ErrUnexpectedEOF) || errors.Is(err, io.EOF) || errors.Is(err, syscall.EPIPE) || errors.Is(err, syscall.ECONNRESET) {
+	if retry.IsTemporary(err) || errors.Is(err, io.ErrUnexpectedEOF) || errors.Is(err, io.EOF) || errors.Is(err, syscall.EPIPE) || errors.Is(err, syscall.ECONNRESET) || errors.Is(err, net.ErrClosed) {
 		logs.Warn.Printf("retrying %v", err)
 		return true
 	}
@@ -83,12 +90,13 @@ var fastBackoff = Backoff{
 	Steps:    3,
 }

-var retryableStatusCodes = []int{
+var defaultRetryStatusCodes = []int{
 	http.StatusRequestTimeout,
 	http.StatusInternalServerError,
 	http.StatusBadGateway,
 	http.StatusServiceUnavailable,
 	http.StatusGatewayTimeout,
+	499,
 }

 const (
@@ -112,17 +120,20 @@ var DefaultTransport http.RoundTripper = &http.Transport{
 	IdleConnTimeout:       90 * time.Second,
 	TLSHandshakeTimeout:   10 * time.Second,
 	ExpectContinueTimeout: 1 * time.Second,
+	// We usually are dealing with 2 hosts (at most), split MaxIdleConns between them.
+	MaxIdleConnsPerHost: 50,
 }

-func makeOptions(target authn.Resource, opts ...Option) (*options, error) {
+func makeOptions(opts ...Option) (*options, error) {
 	o := &options{
-		transport:      DefaultTransport,
-		platform:       defaultPlatform,
-		context:        context.Background(),
-		jobs:           defaultJobs,
-		pageSize:       defaultPageSize,
-		retryPredicate: defaultRetryPredicate,
-		retryBackoff:   defaultRetryBackoff,
+		transport:        DefaultTransport,
+		platform:         defaultPlatform,
+		context:          context.Background(),
+		jobs:             defaultJobs,
+		pageSize:         defaultPageSize,
+		retryPredicate:   defaultRetryPredicate,
+		retryBackoff:     defaultRetryBackoff,
+		retryStatusCodes: defaultRetryStatusCodes,
 	}

 	for _, option := range opts {
@@ -136,12 +147,6 @@ func makeOptions(target authn.Resource, opts ...Option) (*options, error) {
 		// It is a better experience to explicitly tell a caller their auth is misconfigured
 		// than potentially fail silently when the correct auth is overridden by option misuse.
 		return nil, errors.New("provide an option for either authn.Authenticator or authn.Keychain, not both")
-	case o.keychain != nil:
-		auth, err := o.keychain.Resolve(target)
-		if err != nil {
-			return nil, err
-		}
-		o.auth = auth
 	case o.auth == nil:
 		o.auth = authn.Anonymous
 	}
@@ -157,7 +162,7 @@ func makeOptions(target authn.Resource, opts ...Option) (*options, error) {
 		}

 		// Wrap the transport in something that can retry network flakes.
-		o.transport = transport.NewRetry(o.transport, transport.WithRetryPredicate(defaultRetryPredicate), transport.WithRetryStatusCodes(retryableStatusCodes...))
+		o.transport = transport.NewRetry(o.transport, transport.WithRetryPredicate(defaultRetryPredicate), transport.WithRetryStatusCodes(o.retryStatusCodes...))

 		// Wrap this last to prevent transport.New from double-wrapping.
 		if o.userAgent != "" {
@@ -273,7 +278,8 @@ func WithNondistributable(o *options) error {
 // should provide a buffered channel to avoid potential deadlocks.
 func WithProgress(updates chan<- v1.Update) Option {
 	return func(o *options) error {
-		o.updates = updates
+		o.progress = &progress{updates: updates}
+		o.progress.lastUpdate = &v1.Update{}
 		return nil
 	}
 }
@@ -305,6 +311,14 @@ func WithRetryPredicate(predicate retry.Predicate) Option {
 	}
 }

+// WithRetryStatusCodes sets which http response codes will be retried.
+func WithRetryStatusCodes(codes ...int) Option {
+	return func(o *options) error {
+		o.retryStatusCodes = codes
+		return nil
+	}
+}
+
 // WithFilter sets the filter querystring for HTTP operations.
 func WithFilter(key string, value string) Option {
 	return func(o *options) error {
@@ -315,3 +329,20 @@ func WithFilter(key string, value string) Option {
 		return nil
 	}
 }
+
+// Reuse takes a Puller or Pusher and reuses it for remote interactions
+// rather than starting from a clean slate. For example, it will reuse token exchanges
+// when possible and avoid sending redundant HEAD requests.
+//
+// Reuse will take precedence over other options passed to most remote functions because
+// most options deal with setting up auth and transports, which Reuse intetionally skips.
+func Reuse[I *Puller | *Pusher](i I) Option {
+	return func(o *options) error {
+		if puller, ok := any(i).(*Puller); ok {
+			o.puller = puller
+		} else if pusher, ok := any(i).(*Pusher); ok {
+			o.pusher = pusher
+		}
+		return nil
+	}
+}
@@ -29,6 +29,8 @@ type progress struct {
 }

 func (p *progress) total(delta int64) {
+	p.Lock()
+	defer p.Unlock()
 	atomic.AddInt64(&p.lastUpdate.Total, delta)
 }

@@ -48,6 +50,11 @@ func (p *progress) err(err error) error {
 	return err
 }

+func (p *progress) Close(err error) {
+	_ = p.err(err)
+	close(p.updates)
+}
+
 type progressReader struct {
 	rc io.ReadCloser

@@ -0,0 +1,238 @@
+// Copyright 2023 Google LLC All Rights Reserved.
+//
+// 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 remote
+
+import (
+	"context"
+	"sync"
+
+	"github.com/google/go-containerregistry/pkg/logs"
+	"github.com/google/go-containerregistry/pkg/name"
+	v1 "github.com/google/go-containerregistry/pkg/v1"
+	"github.com/google/go-containerregistry/pkg/v1/partial"
+	"github.com/google/go-containerregistry/pkg/v1/types"
+)
+
+type Puller struct {
+	o *options
+
+	// map[resource]*reader
+	readers sync.Map
+}
+
+func NewPuller(options ...Option) (*Puller, error) {
+	o, err := makeOptions(options...)
+	if err != nil {
+		return nil, err
+	}
+
+	return newPuller(o), nil
+}
+
+func newPuller(o *options) *Puller {
+	if o.puller != nil {
+		return o.puller
+	}
+	return &Puller{
+		o: o,
+	}
+}
+
+type reader struct {
+	// in
+	target resource
+	o      *options
+
+	// f()
+	once sync.Once
+
+	// out
+	f   *fetcher
+	err error
+}
+
+// this will run once per reader instance
+func (r *reader) init(ctx context.Context) error {
+	r.once.Do(func() {
+		r.f, r.err = makeFetcher(ctx, r.target, r.o)
+	})
+	return r.err
+}
+
+func (p *Puller) fetcher(ctx context.Context, target resource) (*fetcher, error) {
+	// If we are Reuse()ing a Pusher, we want to use that for token handshakes and scopes,
+	// but we want to do read requests via a fetcher{}.
+	//
+	// TODO(jonjohnsonjr): Unify fetcher, writer, and repoWriter.
+	if p.o.pusher != nil {
+		if repo, ok := target.(name.Repository); ok {
+			w, err := p.o.pusher.writer(ctx, repo, p.o)
+			if err == nil {
+				return fetcherFromWriter(w.w), nil
+			}
+			logs.Debug.Printf("reusing Pusher failed: %v", err)
+		}
+	}
+
+	// Normal path for NewPuller.
+	v, _ := p.readers.LoadOrStore(target, &reader{
+		target: target,
+		o:      p.o,
+	})
+	rr := v.(*reader)
+	return rr.f, rr.init(ctx)
+}
+
+// Head is like remote.Head, but avoids re-authenticating when possible.
+func (p *Puller) Head(ctx context.Context, ref name.Reference) (*v1.Descriptor, error) {
+	f, err := p.fetcher(ctx, ref.Context())
+	if err != nil {
+		return nil, err
+	}
+
+	return f.headManifest(ctx, ref, allManifestMediaTypes)
+}
+
+// Get is like remote.Get, but avoids re-authenticating when possible.
+func (p *Puller) Get(ctx context.Context, ref name.Reference) (*Descriptor, error) {
+	return p.get(ctx, ref, allManifestMediaTypes, p.o.platform)
+}
+
+func (p *Puller) get(ctx context.Context, ref name.Reference, acceptable []types.MediaType, platform v1.Platform) (*Descriptor, error) {
+	f, err := p.fetcher(ctx, ref.Context())
+	if err != nil {
+		return nil, err
+	}
+	return f.get(ctx, ref, acceptable, platform)
+}
+
+// Layer is like remote.Layer, but avoids re-authenticating when possible.
+func (p *Puller) Layer(ctx context.Context, ref name.Digest) (v1.Layer, error) {
+	f, err := p.fetcher(ctx, ref.Context())
+	if err != nil {
+		return nil, err
+	}
+
+	h, err := v1.NewHash(ref.Identifier())
+	if err != nil {
+		return nil, err
+	}
+	l, err := partial.CompressedToLayer(&remoteLayer{
+		fetcher: *f,
+		ctx:     ctx,
+		digest:  h,
+	})
+	if err != nil {
+		return nil, err
+	}
+	return &MountableLayer{
+		Layer:     l,
+		Reference: ref,
+	}, nil
+}
+
+// List lists tags in a repo and handles pagination, returning the full list of tags.
+func (p *Puller) List(ctx context.Context, repo name.Repository) ([]string, error) {
+	lister, err := p.Lister(ctx, repo)
+	if err != nil {
+		return nil, err
+	}
+
+	tagList := []string{}
+	for lister.HasNext() {
+		tags, err := lister.Next(ctx)
+		if err != nil {
+			return nil, err
+		}
+		tagList = append(tagList, tags.Tags...)
+	}
+
+	return tagList, nil
+}
+
+// Lister lists tags in a repo and returns a Lister for paginating through the results.
+func (p *Puller) Lister(ctx context.Context, repo name.Repository) (*Lister, error) {
+	return p.lister(ctx, repo, p.o.pageSize)
+}
+
+func (p *Puller) lister(ctx context.Context, repo name.Repository, pageSize int) (*Lister, error) {
+	f, err := p.fetcher(ctx, repo)
+	if err != nil {
+		return nil, err
+	}
+	page, err := f.listPage(ctx, repo, "", pageSize)
+	if err != nil {
+		return nil, err
+	}
+	return &Lister{
+		f:        f,
+		repo:     repo,
+		pageSize: pageSize,
+		page:     page,
+		err:      err,
+	}, nil
+}
+
+// Catalog lists repos in a registry and handles pagination, returning the full list of repos.
+func (p *Puller) Catalog(ctx context.Context, reg name.Registry) ([]string, error) {
+	return p.catalog(ctx, reg, p.o.pageSize)
+}
+
+func (p *Puller) catalog(ctx context.Context, reg name.Registry, pageSize int) ([]string, error) {
+	catalogger, err := p.catalogger(ctx, reg, pageSize)
+	if err != nil {
+		return nil, err
+	}
+	repoList := []string{}
+	for catalogger.HasNext() {
+		repos, err := catalogger.Next(ctx)
+		if err != nil {
+			return nil, err
+		}
+		repoList = append(repoList, repos.Repos...)
+	}
+	return repoList, nil
+}
+
+// Catalogger lists repos in a registry and returns a Catalogger for paginating through the results.
+func (p *Puller) Catalogger(ctx context.Context, reg name.Registry) (*Catalogger, error) {
+	return p.catalogger(ctx, reg, p.o.pageSize)
+}
+
+func (p *Puller) catalogger(ctx context.Context, reg name.Registry, pageSize int) (*Catalogger, error) {
+	f, err := p.fetcher(ctx, reg)
+	if err != nil {
+		return nil, err
+	}
+	page, err := f.catalogPage(ctx, reg, "", pageSize)
+	if err != nil {
+		return nil, err
+	}
+	return &Catalogger{
+		f:        f,
+		reg:      reg,
+		pageSize: pageSize,
+		page:     page,
+		err:      err,
+	}, nil
+}
+
+func (p *Puller) referrers(ctx context.Context, d name.Digest, filter map[string]string) (v1.ImageIndex, error) {
+	f, err := p.fetcher(ctx, d.Context())
+	if err != nil {
+		return nil, err
+	}
+	return f.fetchReferrers(ctx, filter, d)
+}
@@ -0,0 +1,548 @@
+// Copyright 2023 Google LLC All Rights Reserved.
+//
+// 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 remote
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"net/http"
+	"net/url"
+	"sync"
+
+	"github.com/google/go-containerregistry/pkg/logs"
+	"github.com/google/go-containerregistry/pkg/name"
+	v1 "github.com/google/go-containerregistry/pkg/v1"
+	"github.com/google/go-containerregistry/pkg/v1/partial"
+	"github.com/google/go-containerregistry/pkg/v1/remote/transport"
+	"github.com/google/go-containerregistry/pkg/v1/stream"
+	"github.com/google/go-containerregistry/pkg/v1/types"
+	"golang.org/x/sync/errgroup"
+)
+
+type manifest interface {
+	Taggable
+	partial.Describable
+}
+
+// key is either v1.Hash or v1.Layer (for stream.Layer)
+type workers struct {
+	// map[v1.Hash|v1.Layer]*sync.Once
+	onces sync.Map
+
+	// map[v1.Hash|v1.Layer]error
+	errors sync.Map
+}
+
+func nop() error {
+	return nil
+}
+
+func (w *workers) err(digest v1.Hash) error {
+	v, ok := w.errors.Load(digest)
+	if !ok || v == nil {
+		return nil
+	}
+	return v.(error)
+}
+
+func (w *workers) Do(digest v1.Hash, f func() error) error {
+	// We don't care if it was loaded or not because the sync.Once will do it for us.
+	once, _ := w.onces.LoadOrStore(digest, &sync.Once{})
+
+	once.(*sync.Once).Do(func() {
+		w.errors.Store(digest, f())
+	})
+
+	err := w.err(digest)
+	if err != nil {
+		// Allow this to be retried by another caller.
+		w.onces.Delete(digest)
+	}
+	return err
+}
+
+func (w *workers) Stream(layer v1.Layer, f func() error) error {
+	// We don't care if it was loaded or not because the sync.Once will do it for us.
+	once, _ := w.onces.LoadOrStore(layer, &sync.Once{})
+
+	once.(*sync.Once).Do(func() {
+		w.errors.Store(layer, f())
+	})
+
+	v, ok := w.errors.Load(layer)
+	if !ok || v == nil {
+		return nil
+	}
+
+	return v.(error)
+}
+
+type Pusher struct {
+	o *options
+
+	// map[name.Repository]*repoWriter
+	writers sync.Map
+}
+
+func NewPusher(options ...Option) (*Pusher, error) {
+	o, err := makeOptions(options...)
+	if err != nil {
+		return nil, err
+	}
+
+	return newPusher(o), nil
+}
+
+func newPusher(o *options) *Pusher {
+	if o.pusher != nil {
+		return o.pusher
+	}
+	return &Pusher{
+		o: o,
+	}
+}
+
+func (p *Pusher) writer(ctx context.Context, repo name.Repository, o *options) (*repoWriter, error) {
+	v, _ := p.writers.LoadOrStore(repo, &repoWriter{
+		repo: repo,
+		o:    o,
+	})
+	rw := v.(*repoWriter)
+	return rw, rw.init(ctx)
+}
+
+func (p *Pusher) Push(ctx context.Context, ref name.Reference, t Taggable) error {
+	w, err := p.writer(ctx, ref.Context(), p.o)
+	if err != nil {
+		return err
+	}
+	return w.writeManifest(ctx, ref, t)
+}
+
+func (p *Pusher) Upload(ctx context.Context, repo name.Repository, l v1.Layer) error {
+	w, err := p.writer(ctx, repo, p.o)
+	if err != nil {
+		return err
+	}
+	return w.writeLayer(ctx, l)
+}
+
+func (p *Pusher) Delete(ctx context.Context, ref name.Reference) error {
+	w, err := p.writer(ctx, ref.Context(), p.o)
+	if err != nil {
+		return err
+	}
+
+	u := url.URL{
+		Scheme: ref.Context().Registry.Scheme(),
+		Host:   ref.Context().RegistryStr(),
+		Path:   fmt.Sprintf("/v2/%s/manifests/%s", ref.Context().RepositoryStr(), ref.Identifier()),
+	}
+
+	req, err := http.NewRequest(http.MethodDelete, u.String(), nil)
+	if err != nil {
+		return err
+	}
+
+	resp, err := w.w.client.Do(req.WithContext(ctx))
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+
+	return transport.CheckError(resp, http.StatusOK, http.StatusAccepted)
+
+	// TODO(jason): If the manifest had a `subject`, and if the registry
+	// doesn't support Referrers, update the index pointed to by the
+	// subject's fallback tag to remove the descriptor for this manifest.
+}
+
+type repoWriter struct {
+	repo name.Repository
+	o    *options
+	once sync.Once
+
+	w   *writer
+	err error
+
+	work *workers
+}
+
+// this will run once per repoWriter instance
+func (rw *repoWriter) init(ctx context.Context) error {
+	rw.once.Do(func() {
+		rw.work = &workers{}
+		rw.w, rw.err = makeWriter(ctx, rw.repo, nil, rw.o)
+	})
+	return rw.err
+}
+
+func (rw *repoWriter) writeDeps(ctx context.Context, m manifest) error {
+	if img, ok := m.(v1.Image); ok {
+		return rw.writeLayers(ctx, img)
+	}
+
+	if idx, ok := m.(v1.ImageIndex); ok {
+		return rw.writeChildren(ctx, idx)
+	}
+
+	// This has no deps, not an error (e.g. something you want to just PUT).
+	return nil
+}
+
+type describable struct {
+	desc v1.Descriptor
+}
+
+func (d describable) Digest() (v1.Hash, error) {
+	return d.desc.Digest, nil
+}
+
+func (d describable) Size() (int64, error) {
+	return d.desc.Size, nil
+}
+
+func (d describable) MediaType() (types.MediaType, error) {
+	return d.desc.MediaType, nil
+}
+
+type tagManifest struct {
+	Taggable
+	partial.Describable
+}
+
+func taggableToManifest(t Taggable) (manifest, error) {
+	if m, ok := t.(manifest); ok {
+		return m, nil
+	}
+
+	if d, ok := t.(*Descriptor); ok {
+		if d.MediaType.IsIndex() {
+			return d.ImageIndex()
+		}
+
+		if d.MediaType.IsImage() {
+			return d.Image()
+		}
+
+		if d.MediaType.IsSchema1() {
+			return d.Schema1()
+		}
+
+		return tagManifest{t, describable{d.toDesc()}}, nil
+	}
+
+	desc := v1.Descriptor{
+		// A reasonable default if Taggable doesn't implement MediaType.
+		MediaType: types.DockerManifestSchema2,
+	}
+
+	b, err := t.RawManifest()
+	if err != nil {
+		return nil, err
+	}
+
+	if wmt, ok := t.(withMediaType); ok {
+		desc.MediaType, err = wmt.MediaType()
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	desc.Digest, desc.Size, err = v1.SHA256(bytes.NewReader(b))
+	if err != nil {
+		return nil, err
+	}
+
+	return tagManifest{t, describable{desc}}, nil
+}
+
+func (rw *repoWriter) writeManifest(ctx context.Context, ref name.Reference, t Taggable) error {
+	m, err := taggableToManifest(t)
+	if err != nil {
+		return err
+	}
+
+	needDeps := true
+
+	digest, err := m.Digest()
+	if errors.Is(err, stream.ErrNotComputed) {
+		if err := rw.writeDeps(ctx, m); err != nil {
+			return err
+		}
+
+		needDeps = false
+
+		digest, err = m.Digest()
+		if err != nil {
+			return err
+		}
+	} else if err != nil {
+		return err
+	}
+
+	// This may be a lazy child where we have no ref until digest is computed.
+	if ref == nil {
+		ref = rw.repo.Digest(digest.String())
+	}
+
+	// For tags, we want to do this check outside of our Work.Do closure because
+	// we don't want to dedupe based on the manifest digest.
+	_, byTag := ref.(name.Tag)
+	if byTag {
+		if exists, err := rw.manifestExists(ctx, ref, t); err != nil {
+			return err
+		} else if exists {
+			return nil
+		}
+	}
+
+	// The following work.Do will get deduped by digest, so it won't happen unless
+	// this tag happens to be the first commitManifest to run for that digest.
+	needPut := byTag
+
+	if err := rw.work.Do(digest, func() error {
+		if !byTag {
+			if exists, err := rw.manifestExists(ctx, ref, t); err != nil {
+				return err
+			} else if exists {
+				return nil
+			}
+		}
+
+		if needDeps {
+			if err := rw.writeDeps(ctx, m); err != nil {
+				return err
+			}
+		}
+
+		needPut = false
+		return rw.commitManifest(ctx, ref, m)
+	}); err != nil {
+		return err
+	}
+
+	if !needPut {
+		return nil
+	}
+
+	// Only runs for tags that got deduped by digest.
+	return rw.commitManifest(ctx, ref, m)
+}
+
+func (rw *repoWriter) writeChildren(ctx context.Context, idx v1.ImageIndex) error {
+	children, err := partial.Manifests(idx)
+	if err != nil {
+		return err
+	}
+
+	g, ctx := errgroup.WithContext(ctx)
+	g.SetLimit(rw.o.jobs)
+
+	for _, child := range children {
+		child := child
+		if err := rw.writeChild(ctx, child, g); err != nil {
+			return err
+		}
+	}
+
+	return g.Wait()
+}
+
+func (rw *repoWriter) writeChild(ctx context.Context, child partial.Describable, g *errgroup.Group) error {
+	switch child := child.(type) {
+	case v1.ImageIndex:
+		// For recursive index, we want to do a depth-first launching of goroutines
+		// to avoid deadlocking.
+		//
+		// Note that this is rare, so the impact of this should be really small.
+		return rw.writeManifest(ctx, nil, child)
+	case v1.Image:
+		g.Go(func() error {
+			return rw.writeManifest(ctx, nil, child)
+		})
+	case v1.Layer:
+		g.Go(func() error {
+			return rw.writeLayer(ctx, child)
+		})
+	default:
+		// This can't happen.
+		return fmt.Errorf("encountered unknown child: %T", child)
+	}
+	return nil
+}
+
+func (rw *repoWriter) manifestExists(ctx context.Context, ref name.Reference, t Taggable) (bool, error) {
+	f := &fetcher{
+		target: ref.Context(),
+		client: rw.w.client,
+	}
+
+	m, err := taggableToManifest(t)
+	if err != nil {
+		return false, err
+	}
+
+	digest, err := m.Digest()
+	if err != nil {
+		// Possibly due to streaming layers.
+		return false, nil
+	}
+	got, err := f.headManifest(ctx, ref, allManifestMediaTypes)
+	if err != nil {
+		var terr *transport.Error
+		if errors.As(err, &terr) {
+			if terr.StatusCode == http.StatusNotFound {
+				return false, nil
+			}
+		}
+
+		return false, err
+	}
+
+	if digest != got.Digest {
+		// Mark that we saw this digest in the registry so we don't have to check it again.
+		rw.work.Do(got.Digest, nop)
+
+		return false, nil
+	}
+
+	if tag, ok := ref.(name.Tag); ok {
+		logs.Progress.Printf("existing manifest: %s@%s", tag.Identifier(), got.Digest)
+	} else {
+		logs.Progress.Print("existing manifest: ", got.Digest)
+	}
+
+	return true, nil
+}
+
+func (rw *repoWriter) commitManifest(ctx context.Context, ref name.Reference, m manifest) error {
+	if rw.o.progress != nil {
+		size, err := m.Size()
+		if err != nil {
+			return err
+		}
+		rw.o.progress.total(size)
+	}
+
+	return rw.w.commitManifest(ctx, m, ref)
+}
+
+func (rw *repoWriter) writeLayers(pctx context.Context, img v1.Image) error {
+	ls, err := img.Layers()
+	if err != nil {
+		return err
+	}
+
+	g, ctx := errgroup.WithContext(pctx)
+	g.SetLimit(rw.o.jobs)
+
+	for _, l := range ls {
+		l := l
+
+		g.Go(func() error {
+			return rw.writeLayer(ctx, l)
+		})
+	}
+
+	mt, err := img.MediaType()
+	if err != nil {
+		return err
+	}
+
+	if mt.IsSchema1() {
+		return g.Wait()
+	}
+
+	cl, err := partial.ConfigLayer(img)
+	if errors.Is(err, stream.ErrNotComputed) {
+		if err := g.Wait(); err != nil {
+			return err
+		}
+
+		cl, err := partial.ConfigLayer(img)
+		if err != nil {
+			return err
+		}
+
+		return rw.writeLayer(pctx, cl)
+	} else if err != nil {
+		return err
+	}
+
+	g.Go(func() error {
+		return rw.writeLayer(ctx, cl)
+	})
+
+	return g.Wait()
+}
+
+func (rw *repoWriter) writeLayer(ctx context.Context, l v1.Layer) error {
+	// Skip any non-distributable things.
+	mt, err := l.MediaType()
+	if err != nil {
+		return err
+	}
+	if !mt.IsDistributable() && !rw.o.allowNondistributableArtifacts {
+		return nil
+	}
+
+	digest, err := l.Digest()
+	if err != nil {
+		if errors.Is(err, stream.ErrNotComputed) {
+			return rw.lazyWriteLayer(ctx, l)
+		}
+		return err
+	}
+
+	return rw.work.Do(digest, func() error {
+		if rw.o.progress != nil {
+			size, err := l.Size()
+			if err != nil {
+				return err
+			}
+			rw.o.progress.total(size)
+		}
+		return rw.w.uploadOne(ctx, l)
+	})
+}
+
+func (rw *repoWriter) lazyWriteLayer(ctx context.Context, l v1.Layer) error {
+	return rw.work.Stream(l, func() error {
+		if err := rw.w.uploadOne(ctx, l); err != nil {
+			return err
+		}
+
+		// Mark this upload completed.
+		digest, err := l.Digest()
+		if err != nil {
+			return err
+		}
+
+		rw.work.Do(digest, nop)
+
+		if rw.o.progress != nil {
+			size, err := l.Size()
+			if err != nil {
+				return err
+			}
+			rw.o.progress.total(size)
+		}
+
+		return nil
+	})
+}
@@ -15,21 +15,103 @@
 package remote

 import (
+	"bytes"
+	"context"
+	"errors"
+	"io"
+	"net/http"
+	"strings"
+
 	"github.com/google/go-containerregistry/pkg/name"
 	v1 "github.com/google/go-containerregistry/pkg/v1"
+	"github.com/google/go-containerregistry/pkg/v1/empty"
+	"github.com/google/go-containerregistry/pkg/v1/mutate"
+	"github.com/google/go-containerregistry/pkg/v1/remote/transport"
+	"github.com/google/go-containerregistry/pkg/v1/types"
 )

 // Referrers returns a list of descriptors that refer to the given manifest digest.
 //
 // The subject manifest doesn't have to exist in the registry for there to be descriptors that refer to it.
-func Referrers(d name.Digest, options ...Option) (*v1.IndexManifest, error) {
-	o, err := makeOptions(d.Context(), options...)
+func Referrers(d name.Digest, options ...Option) (v1.ImageIndex, error) {
+	o, err := makeOptions(options...)
 	if err != nil {
 		return nil, err
 	}
-	f, err := makeFetcher(d, o)
+	return newPuller(o).referrers(o.context, d, o.filter)
+}
+
+// https://github.com/opencontainers/distribution-spec/blob/main/spec.md#referrers-tag-schema
+func fallbackTag(d name.Digest) name.Tag {
+	return d.Context().Tag(strings.Replace(d.DigestStr(), ":", "-", 1))
+}
+
+func (f *fetcher) fetchReferrers(ctx context.Context, filter map[string]string, d name.Digest) (v1.ImageIndex, error) {
+	// Check the Referrers API endpoint first.
+	u := f.url("referrers", d.DigestStr())
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, u.String(), nil)
 	if err != nil {
 		return nil, err
 	}
-	return f.fetchReferrers(o.context, o.filter, d)
+	req.Header.Set("Accept", string(types.OCIImageIndex))
+
+	resp, err := f.client.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	if err := transport.CheckError(resp, http.StatusOK, http.StatusNotFound, http.StatusBadRequest); err != nil {
+		return nil, err
+	}
+
+	var b []byte
+	if resp.StatusCode == http.StatusOK {
+		b, err = io.ReadAll(resp.Body)
+		if err != nil {
+			return nil, err
+		}
+	} else {
+		// The registry doesn't support the Referrers API endpoint, so we'll use the fallback tag scheme.
+		b, _, err = f.fetchManifest(ctx, fallbackTag(d), []types.MediaType{types.OCIImageIndex})
+		var terr *transport.Error
+		if errors.As(err, &terr) && terr.StatusCode == http.StatusNotFound {
+			// Not found just means there are no attachments yet. Start with an empty manifest.
+			return empty.Index, nil
+		} else if err != nil {
+			return nil, err
+		}
+	}
+
+	h, sz, err := v1.SHA256(bytes.NewReader(b))
+	if err != nil {
+		return nil, err
+	}
+	idx := &remoteIndex{
+		fetcher:   *f,
+		ctx:       ctx,
+		manifest:  b,
+		mediaType: types.OCIImageIndex,
+		descriptor: &v1.Descriptor{
+			Digest:    h,
+			MediaType: types.OCIImageIndex,
+			Size:      sz,
+		},
+	}
+	return filterReferrersResponse(filter, idx), nil
+}
+
+// If filter applied, filter out by artifactType.
+// See https://github.com/opencontainers/distribution-spec/blob/main/spec.md#listing-referrers
+func filterReferrersResponse(filter map[string]string, in v1.ImageIndex) v1.ImageIndex {
+	if filter == nil {
+		return in
+	}
+	v, ok := filter["artifactType"]
+	if !ok {
+		return in
+	}
+	return mutate.RemoveManifests(in, func(desc v1.Descriptor) bool {
+		return desc.ArtifactType != v
+	})
 }
@@ -0,0 +1,118 @@
+// Copyright 2023 Google LLC All Rights Reserved.
+//
+// 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 remote
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+
+	"github.com/google/go-containerregistry/pkg/name"
+	v1 "github.com/google/go-containerregistry/pkg/v1"
+	"github.com/google/go-containerregistry/pkg/v1/partial"
+	"github.com/google/go-containerregistry/pkg/v1/types"
+)
+
+type schema1 struct {
+	ref        name.Reference
+	ctx        context.Context
+	fetcher    fetcher
+	manifest   []byte
+	mediaType  types.MediaType
+	descriptor *v1.Descriptor
+}
+
+func (s *schema1) Layers() ([]v1.Layer, error) {
+	m := schema1Manifest{}
+	if err := json.NewDecoder(bytes.NewReader(s.manifest)).Decode(&m); err != nil {
+		return nil, err
+	}
+
+	layers := []v1.Layer{}
+	for i := len(m.FSLayers) - 1; i >= 0; i-- {
+		fsl := m.FSLayers[i]
+
+		h, err := v1.NewHash(fsl.BlobSum)
+		if err != nil {
+			return nil, err
+		}
+		l, err := s.LayerByDigest(h)
+		if err != nil {
+			return nil, err
+		}
+		layers = append(layers, l)
+	}
+
+	return layers, nil
+}
+
+func (s *schema1) MediaType() (types.MediaType, error) {
+	return s.mediaType, nil
+}
+
+func (s *schema1) Size() (int64, error) {
+	return s.descriptor.Size, nil
+}
+
+func (s *schema1) ConfigName() (v1.Hash, error) {
+	return partial.ConfigName(s)
+}
+
+func (s *schema1) ConfigFile() (*v1.ConfigFile, error) {
+	return nil, newErrSchema1(s.mediaType)
+}
+
+func (s *schema1) RawConfigFile() ([]byte, error) {
+	return []byte("{}"), nil
+}
+
+func (s *schema1) Digest() (v1.Hash, error) {
+	return s.descriptor.Digest, nil
+}
+
+func (s *schema1) Manifest() (*v1.Manifest, error) {
+	return nil, newErrSchema1(s.mediaType)
+}
+
+func (s *schema1) RawManifest() ([]byte, error) {
+	return s.manifest, nil
+}
+
+func (s *schema1) LayerByDigest(h v1.Hash) (v1.Layer, error) {
+	l, err := partial.CompressedToLayer(&remoteLayer{
+		fetcher: s.fetcher,
+		ctx:     s.ctx,
+		digest:  h,
+	})
+	if err != nil {
+		return nil, err
+	}
+	return &MountableLayer{
+		Layer:     l,
+		Reference: s.ref.Context().Digest(h.String()),
+	}, nil
+}
+
+func (s *schema1) LayerByDiffID(v1.Hash) (v1.Layer, error) {
+	return nil, newErrSchema1(s.mediaType)
+}
+
+type fslayer struct {
+	BlobSum string `json:"blobSum"`
+}
+
+type schema1Manifest struct {
+	FSLayers []fslayer `json:"fsLayers"`
+}
@@ -34,6 +34,9 @@ type Error struct {
 	Request *http.Request
 	// The raw body if we couldn't understand it.
 	rawBody string
+
+	// Bit of a hack to make it easier to force a retry.
+	temporary bool
 }

 // Check that Error implements error
@@ -72,6 +75,10 @@ func (e *Error) responseErr() string {

 // Temporary returns whether the request that preceded the error is temporary.
 func (e *Error) Temporary() bool {
+	if e.temporary {
+		return true
+	}
+
 	if len(e.Errors) == 0 {
 		_, ok := temporaryStatusCodes[e.StatusCode]
 		return ok
@@ -153,21 +160,37 @@ func CheckError(resp *http.Response, codes ...int) error {
 			return nil
 		}
 	}
+
 	b, err := io.ReadAll(resp.Body)
 	if err != nil {
 		return err
 	}

+	return makeError(resp, b)
+}
+
+func makeError(resp *http.Response, body []byte) *Error {
 	// https://github.com/docker/distribution/blob/master/docs/spec/api.md#errors
 	structuredError := &Error{}

 	// This can fail if e.g. the response body is not valid JSON. That's fine,
 	// we'll construct an appropriate error string from the body and status code.
-	_ = json.Unmarshal(b, structuredError)
+	_ = json.Unmarshal(body, structuredError)

-	structuredError.rawBody = string(b)
+	structuredError.rawBody = string(body)
 	structuredError.StatusCode = resp.StatusCode
 	structuredError.Request = resp.Request

 	return structuredError
 }
+
+func retryError(resp *http.Response) error {
+	b, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return err
+	}
+
+	rerr := makeError(resp, b)
+	rerr.temporary = true
+	return rerr
+}
@@ -100,7 +100,7 @@ func (t *retryTransport) RoundTrip(in *http.Request) (out *http.Response, err er
 		if out != nil {
 			for _, code := range t.codes {
 				if out.StatusCode == code {
-					return CheckError(out)
+					return retryError(out)
 				}
 			}
 		}
@@ -25,17 +25,17 @@ import (
 	"net/url"
 	"sort"
 	"strings"
+	"sync"

 	"github.com/google/go-containerregistry/internal/redact"
 	"github.com/google/go-containerregistry/internal/retry"
+	"github.com/google/go-containerregistry/pkg/authn"
 	"github.com/google/go-containerregistry/pkg/logs"
 	"github.com/google/go-containerregistry/pkg/name"
 	v1 "github.com/google/go-containerregistry/pkg/v1"
-	"github.com/google/go-containerregistry/pkg/v1/partial"
 	"github.com/google/go-containerregistry/pkg/v1/remote/transport"
 	"github.com/google/go-containerregistry/pkg/v1/stream"
 	"github.com/google/go-containerregistry/pkg/v1/types"
-	"golang.org/x/sync/errgroup"
 )

 // Taggable is an interface that enables a manifest PUT (e.g. for tagging).
@@ -45,137 +45,64 @@ type Taggable interface {

 // Write pushes the provided img to the specified image reference.
 func Write(ref name.Reference, img v1.Image, options ...Option) (rerr error) {
-	o, err := makeOptions(ref.Context(), options...)
+	o, err := makeOptions(options...)
 	if err != nil {
 		return err
 	}
-
-	var p *progress
-	if o.updates != nil {
-		p = &progress{updates: o.updates}
-		p.lastUpdate = &v1.Update{}
-		p.lastUpdate.Total, err = countImage(img, o.allowNondistributableArtifacts)
-		if err != nil {
-			return err
-		}
-		defer close(o.updates)
-		defer func() { _ = p.err(rerr) }()
+	if o.progress != nil {
+		defer func() { o.progress.Close(rerr) }()
 	}
-	return writeImage(o.context, ref, img, o, p)
-}
-
-func writeImage(ctx context.Context, ref name.Reference, img v1.Image, o *options, progress *progress) error {
-	ls, err := img.Layers()
-	if err != nil {
-		return err
-	}
-	scopes := scopesForUploadingImage(ref.Context(), ls)
-	tr, err := transport.NewWithContext(o.context, ref.Context().Registry, o.auth, o.transport, scopes)
-	if err != nil {
-		return err
-	}
-	w := writer{
-		repo:      ref.Context(),
-		client:    &http.Client{Transport: tr},
-		progress:  progress,
-		backoff:   o.retryBackoff,
-		predicate: o.retryPredicate,
-	}
-
-	// Upload individual blobs and collect any errors.
-	blobChan := make(chan v1.Layer, 2*o.jobs)
-	g, gctx := errgroup.WithContext(ctx)
-	for i := 0; i < o.jobs; i++ {
-		// Start N workers consuming blobs to upload.
-		g.Go(func() error {
-			for b := range blobChan {
-				if err := w.uploadOne(gctx, b); err != nil {
-					return err
-				}
-			}
-			return nil
-		})
-	}
-
-	// Upload individual layers in goroutines and collect any errors.
-	// If we can dedupe by the layer digest, try to do so. If we can't determine
-	// the digest for whatever reason, we can't dedupe and might re-upload.
-	g.Go(func() error {
-		defer close(blobChan)
-		uploaded := map[v1.Hash]bool{}
-		for _, l := range ls {
-			l := l
-
-			// Handle foreign layers.
-			mt, err := l.MediaType()
-			if err != nil {
-				return err
-			}
-			if !mt.IsDistributable() && !o.allowNondistributableArtifacts {
-				continue
-			}
-
-			// Streaming layers calculate their digests while uploading them. Assume
-			// an error here indicates we need to upload the layer.
-			h, err := l.Digest()
-			if err == nil {
-				// If we can determine the layer's digest ahead of
-				// time, use it to dedupe uploads.
-				if uploaded[h] {
-					continue // Already uploading.
-				}
-				uploaded[h] = true
-			}
-			select {
-			case blobChan <- l:
-			case <-gctx.Done():
-				return gctx.Err()
-			}
-		}
-		return nil
-	})
-
-	if l, err := partial.ConfigLayer(img); err != nil {
-		// We can't read the ConfigLayer, possibly because of streaming layers,
-		// since the layer DiffIDs haven't been calculated yet. Attempt to wait
-		// for the other layers to be uploaded, then try the config again.
-		if err := g.Wait(); err != nil {
-			return err
-		}
-
-		// Now that all the layers are uploaded, try to upload the config file blob.
-		l, err := partial.ConfigLayer(img)
-		if err != nil {
-			return err
-		}
-		if err := w.uploadOne(ctx, l); err != nil {
-			return err
-		}
-	} else {
-		// We *can* read the ConfigLayer, so upload it concurrently with the layers.
-		g.Go(func() error {
-			return w.uploadOne(gctx, l)
-		})
-
-		// Wait for the layers + config.
-		if err := g.Wait(); err != nil {
-			return err
-		}
-	}
-
-	// With all of the constituent elements uploaded, upload the manifest
-	// to commit the image.
-	return w.commitManifest(ctx, img, ref)
+	return newPusher(o).Push(o.context, ref, img)
 }

 // writer writes the elements of an image to a remote image reference.
 type writer struct {
-	repo   name.Repository
+	repo      name.Repository
+	auth      authn.Authenticator
+	transport http.RoundTripper
+
 	client *http.Client

 	progress  *progress
 	backoff   Backoff
 	predicate retry.Predicate
+
+	scopeLock sync.Mutex
+	// Keep track of scopes that we have already requested.
+	scopeSet map[string]struct{}
+	scopes   []string
+}
+
+func makeWriter(ctx context.Context, repo name.Repository, ls []v1.Layer, o *options) (*writer, error) {
+	auth := o.auth
+	if o.keychain != nil {
+		kauth, err := o.keychain.Resolve(repo)
+		if err != nil {
+			return nil, err
+		}
+		auth = kauth
+	}
+	scopes := scopesForUploadingImage(repo, ls)
+	tr, err := transport.NewWithContext(ctx, repo.Registry, auth, o.transport, scopes)
+	if err != nil {
+		return nil, err
+	}
+
+	scopeSet := map[string]struct{}{}
+	for _, scope := range scopes {
+		scopeSet[scope] = struct{}{}
+	}
+	return &writer{
+		repo:      repo,
+		client:    &http.Client{Transport: tr},
+		auth:      auth,
+		transport: o.transport,
+		progress:  o.progress,
+		backoff:   o.retryBackoff,
+		predicate: o.retryPredicate,
+		scopes:    scopes,
+		scopeSet:  scopeSet,
+	}, nil
 }

 // url returns a url.Url for the specified path in the context of this remote image reference.
@@ -187,6 +114,34 @@ func (w *writer) url(path string) url.URL {
 	}
 }

+func (w *writer) maybeUpdateScopes(ctx context.Context, ml *MountableLayer) error {
+	if ml.Reference.Context().String() == w.repo.String() {
+		return nil
+	}
+	if ml.Reference.Context().Registry.String() != w.repo.Registry.String() {
+		return nil
+	}
+
+	scope := ml.Reference.Scope(transport.PullScope)
+
+	w.scopeLock.Lock()
+	defer w.scopeLock.Unlock()
+
+	if _, ok := w.scopeSet[scope]; !ok {
+		w.scopeSet[scope] = struct{}{}
+		w.scopes = append(w.scopes, scope)
+
+		logs.Debug.Printf("Refreshing token to add scope %q", scope)
+		wt, err := transport.NewWithContext(ctx, w.repo.Registry, w.auth, w.transport, w.scopes)
+		if err != nil {
+			return err
+		}
+		w.client = &http.Client{Transport: wt}
+	}
+
+	return nil
+}
+
 // nextLocation extracts the fully-qualified URL to which we should send the next request in an upload sequence.
 func (w *writer) nextLocation(resp *http.Response) (string, error) {
 	loc := resp.Header.Get("Location")
@@ -228,30 +183,6 @@ func (w *writer) checkExistingBlob(ctx context.Context, h v1.Hash) (bool, error)
 	return resp.StatusCode == http.StatusOK, nil
 }

-// checkExistingManifest checks if a manifest exists already in the repository
-// by making a HEAD request to the manifest API.
-func (w *writer) checkExistingManifest(ctx context.Context, h v1.Hash, mt types.MediaType) (bool, error) {
-	u := w.url(fmt.Sprintf("/v2/%s/manifests/%s", w.repo.RepositoryStr(), h.String()))
-
-	req, err := http.NewRequest(http.MethodHead, u.String(), nil)
-	if err != nil {
-		return false, err
-	}
-	req.Header.Set("Accept", string(mt))
-
-	resp, err := w.client.Do(req.WithContext(ctx))
-	if err != nil {
-		return false, err
-	}
-	defer resp.Body.Close()
-
-	if err := transport.CheckError(resp, http.StatusOK, http.StatusNotFound); err != nil {
-		return false, err
-	}
-
-	return resp.StatusCode == http.StatusOK, nil
-}
-
 // initiateUpload initiates the blob upload, which starts with a POST that can
 // optionally include the hash of the layer and a list of repositories from
 // which that layer might be read. On failure, an error is returned.
@@ -279,6 +210,11 @@ func (w *writer) initiateUpload(ctx context.Context, from, mount, origin string)
 	req.Header.Set("Content-Type", "application/json")
 	resp, err := w.client.Do(req.WithContext(ctx))
 	if err != nil {
+		if origin != "" && origin != w.repo.RegistryStr() {
+			// https://github.com/google/go-containerregistry/issues/1679
+			logs.Warn.Printf("retrying without mount: %v", err)
+			return w.initiateUpload(ctx, "", "", "")
+		}
 		return "", false, err
 	}
 	defer resp.Body.Close()
@@ -421,6 +357,9 @@ func (w *writer) uploadOne(ctx context.Context, l v1.Layer) error {
 			mount = h.String()
 		}
 		if ml, ok := l.(*MountableLayer); ok {
+			if err := w.maybeUpdateScopes(ctx, ml); err != nil {
+				return err
+			}
 			from = ml.Reference.Context().RepositoryStr()
 			origin = ml.Reference.Context().RegistryStr()
 		}
@@ -474,69 +413,6 @@ func (w *writer) uploadOne(ctx context.Context, l v1.Layer) error {
 	return retry.Retry(tryUpload, w.predicate, w.backoff)
 }

-type withLayer interface {
-	Layer(v1.Hash) (v1.Layer, error)
-}
-
-func (w *writer) writeIndex(ctx context.Context, ref name.Reference, ii v1.ImageIndex, options ...Option) error {
-	index, err := ii.IndexManifest()
-	if err != nil {
-		return err
-	}
-
-	o, err := makeOptions(ref.Context(), options...)
-	if err != nil {
-		return err
-	}
-
-	// TODO(#803): Pipe through remote.WithJobs and upload these in parallel.
-	for _, desc := range index.Manifests {
-		ref := ref.Context().Digest(desc.Digest.String())
-		exists, err := w.checkExistingManifest(ctx, desc.Digest, desc.MediaType)
-		if err != nil {
-			return err
-		}
-		if exists {
-			logs.Progress.Print("existing manifest: ", desc.Digest)
-			continue
-		}
-
-		switch desc.MediaType {
-		case types.OCIImageIndex, types.DockerManifestList:
-			ii, err := ii.ImageIndex(desc.Digest)
-			if err != nil {
-				return err
-			}
-			if err := w.writeIndex(ctx, ref, ii, options...); err != nil {
-				return err
-			}
-		case types.OCIManifestSchema1, types.DockerManifestSchema2:
-			img, err := ii.Image(desc.Digest)
-			if err != nil {
-				return err
-			}
-			if err := writeImage(ctx, ref, img, o, w.progress); err != nil {
-				return err
-			}
-		default:
-			// Workaround for #819.
-			if wl, ok := ii.(withLayer); ok {
-				layer, err := wl.Layer(desc.Digest)
-				if err != nil {
-					return err
-				}
-				if err := w.uploadOne(ctx, layer); err != nil {
-					return err
-				}
-			}
-		}
-	}
-
-	// With all of the constituent elements uploaded, upload the manifest
-	// to commit the image.
-	return w.commitManifest(ctx, ii, ref)
-}
-
 type withMediaType interface {
 	MediaType() (types.MediaType, error)
 }
@@ -655,10 +531,7 @@ func (w *writer) commitSubjectReferrers(ctx context.Context, sub name.Digest, ad
 		return im.Manifests[i].Digest.String() < im.Manifests[j].Digest.String()
 	})
 	logs.Progress.Printf("updating fallback tag %s with new referrer", t.Identifier())
-	if err := w.commitManifest(ctx, fallbackTaggable{im}, t); err != nil {
-		return err
-	}
-	return nil
+	return w.commitManifest(ctx, fallbackTaggable{im}, t)
 }

 type fallbackTaggable struct {
@@ -770,183 +643,26 @@ func scopesForUploadingImage(repo name.Repository, layers []v1.Layer) []string {
 // WriteIndex will attempt to push all of the referenced manifests before
 // attempting to push the ImageIndex, to retain referential integrity.
 func WriteIndex(ref name.Reference, ii v1.ImageIndex, options ...Option) (rerr error) {
-	o, err := makeOptions(ref.Context(), options...)
+	o, err := makeOptions(options...)
 	if err != nil {
 		return err
 	}
-
-	scopes := []string{ref.Scope(transport.PushScope)}
-	tr, err := transport.NewWithContext(o.context, ref.Context().Registry, o.auth, o.transport, scopes)
-	if err != nil {
-		return err
+	if o.progress != nil {
+		defer func() { o.progress.Close(rerr) }()
 	}
-	w := writer{
-		repo:      ref.Context(),
-		client:    &http.Client{Transport: tr},
-		backoff:   o.retryBackoff,
-		predicate: o.retryPredicate,
-	}
-
-	if o.updates != nil {
-		w.progress = &progress{updates: o.updates}
-		w.progress.lastUpdate = &v1.Update{}
-
-		defer close(o.updates)
-		defer func() { w.progress.err(rerr) }()
-
-		w.progress.lastUpdate.Total, err = countIndex(ii, o.allowNondistributableArtifacts)
-		if err != nil {
-			return err
-		}
-	}
-
-	return w.writeIndex(o.context, ref, ii, options...)
-}
-
-// countImage counts the total size of all layers + config blob + manifest for
-// an image. It de-dupes duplicate layers.
-func countImage(img v1.Image, allowNondistributableArtifacts bool) (int64, error) {
-	var total int64
-	ls, err := img.Layers()
-	if err != nil {
-		return 0, err
-	}
-	seen := map[v1.Hash]bool{}
-	for _, l := range ls {
-		// Handle foreign layers.
-		mt, err := l.MediaType()
-		if err != nil {
-			return 0, err
-		}
-		if !mt.IsDistributable() && !allowNondistributableArtifacts {
-			continue
-		}
-
-		// TODO: support streaming layers which update the total count as they write.
-		if _, ok := l.(*stream.Layer); ok {
-			return 0, errors.New("cannot use stream.Layer and WithProgress")
-		}
-
-		// Dedupe layers.
-		d, err := l.Digest()
-		if err != nil {
-			return 0, err
-		}
-		if seen[d] {
-			continue
-		}
-		seen[d] = true
-
-		size, err := l.Size()
-		if err != nil {
-			return 0, err
-		}
-		total += size
-	}
-	b, err := img.RawConfigFile()
-	if err != nil {
-		return 0, err
-	}
-	total += int64(len(b))
-	size, err := img.Size()
-	if err != nil {
-		return 0, err
-	}
-	total += size
-	return total, nil
-}
-
-// countIndex counts the total size of all images + sub-indexes for an index.
-// It does not attempt to de-dupe duplicate images, etc.
-func countIndex(idx v1.ImageIndex, allowNondistributableArtifacts bool) (int64, error) {
-	var total int64
-	mf, err := idx.IndexManifest()
-	if err != nil {
-		return 0, err
-	}
-
-	for _, desc := range mf.Manifests {
-		switch desc.MediaType {
-		case types.OCIImageIndex, types.DockerManifestList:
-			sidx, err := idx.ImageIndex(desc.Digest)
-			if err != nil {
-				return 0, err
-			}
-			size, err := countIndex(sidx, allowNondistributableArtifacts)
-			if err != nil {
-				return 0, err
-			}
-			total += size
-		case types.OCIManifestSchema1, types.DockerManifestSchema2:
-			simg, err := idx.Image(desc.Digest)
-			if err != nil {
-				return 0, err
-			}
-			size, err := countImage(simg, allowNondistributableArtifacts)
-			if err != nil {
-				return 0, err
-			}
-			total += size
-		default:
-			// Workaround for #819.
-			if wl, ok := idx.(withLayer); ok {
-				layer, err := wl.Layer(desc.Digest)
-				if err != nil {
-					return 0, err
-				}
-				size, err := layer.Size()
-				if err != nil {
-					return 0, err
-				}
-				total += size
-			}
-		}
-	}
-
-	size, err := idx.Size()
-	if err != nil {
-		return 0, err
-	}
-	total += size
-	return total, nil
+	return newPusher(o).Push(o.context, ref, ii)
 }

 // WriteLayer uploads the provided Layer to the specified repo.
 func WriteLayer(repo name.Repository, layer v1.Layer, options ...Option) (rerr error) {
-	o, err := makeOptions(repo, options...)
+	o, err := makeOptions(options...)
 	if err != nil {
 		return err
 	}
-	scopes := scopesForUploadingImage(repo, []v1.Layer{layer})
-	tr, err := transport.NewWithContext(o.context, repo.Registry, o.auth, o.transport, scopes)
-	if err != nil {
-		return err
+	if o.progress != nil {
+		defer func() { o.progress.Close(rerr) }()
 	}
-	w := writer{
-		repo:      repo,
-		client:    &http.Client{Transport: tr},
-		backoff:   o.retryBackoff,
-		predicate: o.retryPredicate,
-	}
-
-	if o.updates != nil {
-		w.progress = &progress{updates: o.updates}
-		w.progress.lastUpdate = &v1.Update{}
-
-		defer close(o.updates)
-		defer func() { w.progress.err(rerr) }()
-
-		// TODO: support streaming layers which update the total count as they write.
-		if _, ok := layer.(*stream.Layer); ok {
-			return errors.New("cannot use stream.Layer and WithProgress")
-		}
-		size, err := layer.Size()
-		if err != nil {
-			return err
-		}
-		w.progress.total(size)
-	}
-	return w.uploadOne(o.context, layer)
+	return newPusher(o).Upload(o.context, repo, layer)
 }

 // Tag adds a tag to the given Taggable via PUT /v2/.../manifests/<tag>
@@ -976,28 +692,9 @@ func Tag(tag name.Tag, t Taggable, options ...Option) error {
 // should ensure that all blobs or manifests that are referenced by t exist
 // in the target registry.
 func Put(ref name.Reference, t Taggable, options ...Option) error {
-	o, err := makeOptions(ref.Context(), options...)
+	o, err := makeOptions(options...)
 	if err != nil {
 		return err
 	}
-	scopes := []string{ref.Scope(transport.PushScope)}
-
-	// TODO: This *always* does a token exchange. For some registries,
-	// that's pretty slow. Some ideas;
-	// * Tag could take a list of tags.
-	// * Allow callers to pass in a transport.Transport, typecheck
-	//   it to allow them to reuse the transport across multiple calls.
-	// * WithTag option to do multiple manifest PUTs in commitManifest.
-	tr, err := transport.NewWithContext(o.context, ref.Context().Registry, o.auth, o.transport, scopes)
-	if err != nil {
-		return err
-	}
-	w := writer{
-		repo:      ref.Context(),
-		client:    &http.Client{Transport: tr},
-		backoff:   o.retryBackoff,
-		predicate: o.retryPredicate,
-	}
-
-	return w.commitManifest(o.context, t, ref)
+	return newPusher(o).Push(o.context, ref, t)
 }
@@ -130,6 +130,8 @@ func (l *Layer) Uncompressed() (io.ReadCloser, error) {

 // Compressed implements v1.Layer.
 func (l *Layer) Compressed() (io.ReadCloser, error) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
 	if l.consumed {
 		return nil, ErrConsumed
 	}
@@ -223,9 +223,9 @@ func extractFileFromTar(opener Opener, filePath string) (io.ReadCloser, error) {
 	if err != nil {
 		return nil, err
 	}
-	close := true
+	needClose := true
 	defer func() {
-		if close {
+		if needClose {
 			f.Close()
 		}
 	}()
@@ -244,7 +244,7 @@ func extractFileFromTar(opener Opener, filePath string) (io.ReadCloser, error) {
 				currentDir := filepath.Dir(filePath)
 				return extractFileFromTar(opener, path.Join(currentDir, path.Clean(hdr.Linkname)))
 			}
-			close = false
+			needClose = false
 			return tarFile{
 				Reader: tf,
 				Closer: f,
@@ -160,6 +160,8 @@ func WithCompressedCaching(l *layer) {
 // WithEstargzOptions is a functional option that allow the caller to pass
 // through estargz.Options to the underlying compression layer.  This is
 // only meaningful when estargz is enabled.
+//
+// Deprecated: WithEstargz is deprecated, and will be removed in a future release.
 func WithEstargzOptions(opts ...estargz.Option) LayerOption {
 	return func(l *layer) {
 		l.estgzopts = opts
@@ -167,6 +169,8 @@ func WithEstargzOptions(opts ...estargz.Option) LayerOption {
 }

 // WithEstargz is a functional option that explicitly enables estargz support.
+//
+// Deprecated: WithEstargz is deprecated, and will be removed in a future release.
 func WithEstargz(l *layer) {
 	oguncompressed := l.uncompressedopener
 	estargz := func() (io.ReadCloser, error) {
@@ -238,6 +242,7 @@ func LayerFromOpener(opener Opener, opts ...LayerOption) (v1.Layer, error) {
 	}

 	if estgz := os.Getenv("GGCR_EXPERIMENT_ESTARGZ"); estgz == "1" {
+		logs.Warn.Println("GGCR_EXPERIMENT_ESTARGZ is deprecated, and will be removed in a future release.")
 		opts = append([]LayerOption{WithEstargz}, opts...)
 	}

@@ -80,3 +80,19 @@ func (m MediaType) IsConfig() bool {
 	}
 	return false
 }
+
+func (m MediaType) IsSchema1() bool {
+	switch m {
+	case DockerManifestSchema1, DockerManifestSchema1Signed:
+		return true
+	}
+	return false
+}
+
+func (m MediaType) IsLayer() bool {
+	switch m {
+	case DockerLayer, DockerUncompressedLayer, OCILayer, OCILayerZStd, OCIUncompressedLayer, DockerForeignLayer, OCIRestrictedLayer, OCIUncompressedRestrictedLayer:
+		return true
+	}
+	return false
+}