Move content to core/content

Signed-off-by: Derek McGowan <derek@mcg.dev>

core/content/adaptor.go

@@ -0,0 +1,52 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package content
+
+import (
+	"strings"
+
+	"github.com/containerd/containerd/v2/filters"
+)
+
+// AdaptInfo returns `filters.Adaptor` that handles `content.Info`.
+func AdaptInfo(info Info) filters.Adaptor {
+	return filters.AdapterFunc(func(fieldpath []string) (string, bool) {
+		if len(fieldpath) == 0 {
+			return "", false
+		}
+
+		switch fieldpath[0] {
+		case "digest":
+			return info.Digest.String(), true
+		case "size":
+			// TODO: support size based filtering
+		case "labels":
+			return checkMap(fieldpath[1:], info.Labels)
+		}
+
+		return "", false
+	})
+}
+
+func checkMap(fieldpath []string, m map[string]string) (string, bool) {
+	if len(m) == 0 {
+		return "", false
+	}
+
+	value, ok := m[strings.Join(fieldpath, ".")]
+	return value, ok
+}

core/content/adaptor_test.go

@@ -0,0 +1,89 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package content
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestAdaptInfo(t *testing.T) {
+	tests := []struct {
+		name        string
+		info        Info
+		fieldpath   []string
+		wantValue   string
+		wantPresent bool
+	}{
+		{
+			"empty fieldpath",
+			Info{},
+			[]string{},
+			"",
+			false,
+		},
+		{
+			"digest fieldpath",
+			Info{
+				Digest: "foo",
+			},
+			[]string{"digest"},
+			"foo",
+			true,
+		},
+		{
+			"size fieldpath",
+			Info{},
+			[]string{"size"},
+			"",
+			false,
+		},
+		{
+			"labels fieldpath",
+			Info{
+				Labels: map[string]string{
+					"foo": "bar",
+				},
+			},
+			[]string{"labels", "foo"},
+			"bar",
+			true,
+		},
+		{
+			"labels join fieldpath",
+			Info{
+				Labels: map[string]string{
+					"foo.bar.qux": "quux",
+				},
+			},
+			[]string{"labels", "foo", "bar", "qux"},
+			"quux",
+			true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			adaptInfo := AdaptInfo(tt.info)
+
+			value, present := adaptInfo.Field(tt.fieldpath)
+
+			assert.Equal(t, tt.wantValue, value)
+			assert.Equal(t, tt.wantPresent, present)
+		})
+	}
+}

core/content/content.go

@@ -0,0 +1,207 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package content
+
+import (
+	"context"
+	"io"
+	"time"
+
+	"github.com/opencontainers/go-digest"
+	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
+)
+
+// Store combines the methods of content-oriented interfaces into a set that
+// are commonly provided by complete implementations.
+//
+// Overall content lifecycle:
+//   - Ingester is used to initiate a write operation (aka ingestion)
+//   - IngestManager is used to manage (e.g. list, abort) active ingestions
+//   - Once an ingestion is complete (see Writer.Commit), Provider is used to
+//     query a single piece of content by its digest
+//   - Manager is used to manage (e.g. list, delete) previously committed content
+//
+// Note that until ingestion is complete, its content is not visible through
+// Provider or Manager. Once ingestion is complete, it is no longer exposed
+// through IngestManager.
+type Store interface {
+	Manager
+	Provider
+	IngestManager
+	Ingester
+}
+
+// ReaderAt extends the standard io.ReaderAt interface with reporting of Size and io.Closer
+type ReaderAt interface {
+	io.ReaderAt
+	io.Closer
+	Size() int64
+}
+
+// Provider provides a reader interface for specific content
+type Provider interface {
+	// ReaderAt only requires desc.Digest to be set.
+	// Other fields in the descriptor may be used internally for resolving
+	// the location of the actual data.
+	ReaderAt(ctx context.Context, desc ocispec.Descriptor) (ReaderAt, error)
+}
+
+// Ingester writes content
+type Ingester interface {
+	// Writer initiates a writing operation (aka ingestion). A single ingestion
+	// is uniquely identified by its ref, provided using a WithRef option.
+	// Writer can be called multiple times with the same ref to access the same
+	// ingestion.
+	// Once all the data is written, use Writer.Commit to complete the ingestion.
+	Writer(ctx context.Context, opts ...WriterOpt) (Writer, error)
+}
+
+// IngestManager provides methods for managing ingestions. An ingestion is a
+// not-yet-complete writing operation initiated using Ingester and identified
+// by a ref string.
+type IngestManager interface {
+	// Status returns the status of the provided ref.
+	Status(ctx context.Context, ref string) (Status, error)
+
+	// ListStatuses returns the status of any active ingestions whose ref match
+	// the provided regular expression. If empty, all active ingestions will be
+	// returned.
+	ListStatuses(ctx context.Context, filters ...string) ([]Status, error)
+
+	// Abort completely cancels the ingest operation targeted by ref.
+	Abort(ctx context.Context, ref string) error
+}
+
+// Info holds content specific information
+type Info struct {
+	Digest    digest.Digest
+	Size      int64
+	CreatedAt time.Time
+	UpdatedAt time.Time
+	Labels    map[string]string
+}
+
+// Status of a content operation (i.e. an ingestion)
+type Status struct {
+	Ref       string
+	Offset    int64
+	Total     int64
+	Expected  digest.Digest
+	StartedAt time.Time
+	UpdatedAt time.Time
+}
+
+// WalkFunc defines the callback for a blob walk.
+type WalkFunc func(Info) error
+
+// InfoReaderProvider provides both info and reader for the specific content.
+type InfoReaderProvider interface {
+	InfoProvider
+	Provider
+}
+
+// InfoProvider provides info for content inspection.
+type InfoProvider interface {
+	// Info will return metadata about content available in the content store.
+	//
+	// If the content is not present, ErrNotFound will be returned.
+	Info(ctx context.Context, dgst digest.Digest) (Info, error)
+}
+
+// Manager provides methods for inspecting, listing and removing content.
+type Manager interface {
+	InfoProvider
+
+	// Update updates mutable information related to content.
+	// If one or more fieldpaths are provided, only those
+	// fields will be updated.
+	// Mutable fields:
+	//  labels.*
+	Update(ctx context.Context, info Info, fieldpaths ...string) (Info, error)
+
+	// Walk will call fn for each item in the content store which
+	// match the provided filters. If no filters are given all
+	// items will be walked.
+	Walk(ctx context.Context, fn WalkFunc, filters ...string) error
+
+	// Delete removes the content from the store.
+	Delete(ctx context.Context, dgst digest.Digest) error
+}
+
+// Writer handles writing of content into a content store
+type Writer interface {
+	// Close closes the writer, if the writer has not been
+	// committed this allows resuming or aborting.
+	// Calling Close on a closed writer will not error.
+	io.WriteCloser
+
+	// Digest may return empty digest or panics until committed.
+	Digest() digest.Digest
+
+	// Commit commits the blob (but no roll-back is guaranteed on an error).
+	// size and expected can be zero-value when unknown.
+	// Commit always closes the writer, even on error.
+	// ErrAlreadyExists aborts the writer.
+	Commit(ctx context.Context, size int64, expected digest.Digest, opts ...Opt) error
+
+	// Status returns the current state of write
+	Status() (Status, error)
+
+	// Truncate updates the size of the target blob
+	Truncate(size int64) error
+}
+
+// Opt is used to alter the mutable properties of content
+type Opt func(*Info) error
+
+// WithLabels allows labels to be set on content
+func WithLabels(labels map[string]string) Opt {
+	return func(info *Info) error {
+		info.Labels = labels
+		return nil
+	}
+}
+
+// WriterOpts is internally used by WriterOpt.
+type WriterOpts struct {
+	Ref  string
+	Desc ocispec.Descriptor
+}
+
+// WriterOpt is used for passing options to Ingester.Writer.
+type WriterOpt func(*WriterOpts) error
+
+// WithDescriptor specifies an OCI descriptor.
+// Writer may optionally use the descriptor internally for resolving
+// the location of the actual data.
+// Write does not require any field of desc to be set.
+// If the data size is unknown, desc.Size should be set to 0.
+// Some implementations may also accept negative values as "unknown".
+func WithDescriptor(desc ocispec.Descriptor) WriterOpt {
+	return func(opts *WriterOpts) error {
+		opts.Desc = desc
+		return nil
+	}
+}
+
+// WithRef specifies a ref string.
+func WithRef(ref string) WriterOpt {
+	return func(opts *WriterOpts) error {
+		opts.Ref = ref
+		return nil
+	}
+}

core/content/helpers.go

@@ -0,0 +1,333 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package content
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"sync"
+	"time"
+
+	"github.com/containerd/containerd/v2/errdefs"
+	"github.com/containerd/containerd/v2/pkg/randutil"
+	"github.com/containerd/log"
+	"github.com/opencontainers/go-digest"
+	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
+)
+
+// maxResets is the no.of times the Copy() method can tolerate a reset of the body
+const maxResets = 5
+
+var ErrReset = errors.New("writer has been reset")
+
+var bufPool = sync.Pool{
+	New: func() interface{} {
+		buffer := make([]byte, 1<<20)
+		return &buffer
+	},
+}
+
+type reader interface {
+	Reader() io.Reader
+}
+
+// NewReader returns a io.Reader from a ReaderAt
+func NewReader(ra ReaderAt) io.Reader {
+	if rd, ok := ra.(reader); ok {
+		return rd.Reader()
+	}
+	return io.NewSectionReader(ra, 0, ra.Size())
+}
+
+// ReadBlob retrieves the entire contents of the blob from the provider.
+//
+// Avoid using this for large blobs, such as layers.
+func ReadBlob(ctx context.Context, provider Provider, desc ocispec.Descriptor) ([]byte, error) {
+	if int64(len(desc.Data)) == desc.Size && digest.FromBytes(desc.Data) == desc.Digest {
+		return desc.Data, nil
+	}
+
+	ra, err := provider.ReaderAt(ctx, desc)
+	if err != nil {
+		return nil, err
+	}
+	defer ra.Close()
+
+	p := make([]byte, ra.Size())
+
+	n, err := ra.ReadAt(p, 0)
+	if err == io.EOF {
+		if int64(n) != ra.Size() {
+			err = io.ErrUnexpectedEOF
+		} else {
+			err = nil
+		}
+	}
+	return p, err
+}
+
+// WriteBlob writes data with the expected digest into the content store. If
+// expected already exists, the method returns immediately and the reader will
+// not be consumed.
+//
+// This is useful when the digest and size are known beforehand.
+//
+// Copy is buffered, so no need to wrap reader in buffered io.
+func WriteBlob(ctx context.Context, cs Ingester, ref string, r io.Reader, desc ocispec.Descriptor, opts ...Opt) error {
+	cw, err := OpenWriter(ctx, cs, WithRef(ref), WithDescriptor(desc))
+	if err != nil {
+		if !errdefs.IsAlreadyExists(err) {
+			return fmt.Errorf("failed to open writer: %w", err)
+		}
+
+		return nil // already present
+	}
+	defer cw.Close()
+
+	return Copy(ctx, cw, r, desc.Size, desc.Digest, opts...)
+}
+
+// OpenWriter opens a new writer for the given reference, retrying if the writer
+// is locked until the reference is available or returns an error.
+func OpenWriter(ctx context.Context, cs Ingester, opts ...WriterOpt) (Writer, error) {
+	var (
+		cw    Writer
+		err   error
+		retry = 16
+	)
+	for {
+		cw, err = cs.Writer(ctx, opts...)
+		if err != nil {
+			if !errdefs.IsUnavailable(err) {
+				return nil, err
+			}
+
+			// TODO: Check status to determine if the writer is active,
+			// continue waiting while active, otherwise return lock
+			// error or abort. Requires asserting for an ingest manager
+
+			select {
+			case <-time.After(time.Millisecond * time.Duration(randutil.Intn(retry))):
+				if retry < 2048 {
+					retry = retry << 1
+				}
+				continue
+			case <-ctx.Done():
+				// Propagate lock error
+				return nil, err
+			}
+
+		}
+		break
+	}
+
+	return cw, err
+}
+
+// Copy copies data with the expected digest from the reader into the
+// provided content store writer. This copy commits the writer.
+//
+// This is useful when the digest and size are known beforehand. When
+// the size or digest is unknown, these values may be empty.
+//
+// Copy is buffered, so no need to wrap reader in buffered io.
+func Copy(ctx context.Context, cw Writer, or io.Reader, size int64, expected digest.Digest, opts ...Opt) error {
+	r := or
+	for i := 0; i < maxResets; i++ {
+		if i >= 1 {
+			log.G(ctx).WithField("digest", expected).Debugf("retrying copy due to reset")
+		}
+
+		ws, err := cw.Status()
+		if err != nil {
+			return fmt.Errorf("failed to get status: %w", err)
+		}
+		// Reset the original reader if
+		// 1. there is an offset, or
+		// 2. this is a retry due to Reset error
+		if ws.Offset > 0 || i > 0 {
+			r, err = seekReader(or, ws.Offset, size)
+			if err != nil {
+				return fmt.Errorf("unable to resume write to %v: %w", ws.Ref, err)
+			}
+		}
+
+		copied, err := copyWithBuffer(cw, r)
+		if errors.Is(err, ErrReset) {
+			continue
+		}
+		if err != nil {
+			return fmt.Errorf("failed to copy: %w", err)
+		}
+		if size != 0 && copied < size-ws.Offset {
+			// Short writes would return its own error, this indicates a read failure
+			return fmt.Errorf("failed to read expected number of bytes: %w", io.ErrUnexpectedEOF)
+		}
+		if err := cw.Commit(ctx, size, expected, opts...); err != nil {
+			if errors.Is(err, ErrReset) {
+				continue
+			}
+			if !errdefs.IsAlreadyExists(err) {
+				return fmt.Errorf("failed commit on ref %q: %w", ws.Ref, err)
+			}
+		}
+		return nil
+	}
+
+	log.G(ctx).WithField("digest", expected).Errorf("failed to copy after %d retries", maxResets)
+	return fmt.Errorf("failed to copy after %d retries", maxResets)
+}
+
+// CopyReaderAt copies to a writer from a given reader at for the given
+// number of bytes. This copy does not commit the writer.
+func CopyReaderAt(cw Writer, ra ReaderAt, n int64) error {
+	ws, err := cw.Status()
+	if err != nil {
+		return err
+	}
+
+	copied, err := copyWithBuffer(cw, io.NewSectionReader(ra, ws.Offset, n))
+	if err != nil {
+		return fmt.Errorf("failed to copy: %w", err)
+	}
+	if copied < n {
+		// Short writes would return its own error, this indicates a read failure
+		return fmt.Errorf("failed to read expected number of bytes: %w", io.ErrUnexpectedEOF)
+	}
+	return nil
+}
+
+// CopyReader copies to a writer from a given reader, returning
+// the number of bytes copied.
+// Note: if the writer has a non-zero offset, the total number
+// of bytes read may be greater than those copied if the reader
+// is not an io.Seeker.
+// This copy does not commit the writer.
+func CopyReader(cw Writer, r io.Reader) (int64, error) {
+	ws, err := cw.Status()
+	if err != nil {
+		return 0, fmt.Errorf("failed to get status: %w", err)
+	}
+
+	if ws.Offset > 0 {
+		r, err = seekReader(r, ws.Offset, 0)
+		if err != nil {
+			return 0, fmt.Errorf("unable to resume write to %v: %w", ws.Ref, err)
+		}
+	}
+
+	return copyWithBuffer(cw, r)
+}
+
+// seekReader attempts to seek the reader to the given offset, either by
+// resolving `io.Seeker`, by detecting `io.ReaderAt`, or discarding
+// up to the given offset.
+func seekReader(r io.Reader, offset, size int64) (io.Reader, error) {
+	// attempt to resolve r as a seeker and setup the offset.
+	seeker, ok := r.(io.Seeker)
+	if ok {
+		nn, err := seeker.Seek(offset, io.SeekStart)
+		if nn != offset {
+			if err == nil {
+				err = fmt.Errorf("unexpected seek location without seek error")
+			}
+			return nil, fmt.Errorf("failed to seek to offset %v: %w", offset, err)
+		}
+
+		if err != nil {
+			return nil, err
+		}
+
+		return r, nil
+	}
+
+	// ok, let's try io.ReaderAt!
+	readerAt, ok := r.(io.ReaderAt)
+	if ok && size > offset {
+		sr := io.NewSectionReader(readerAt, offset, size)
+		return sr, nil
+	}
+
+	// well then, let's just discard up to the offset
+	n, err := copyWithBuffer(io.Discard, io.LimitReader(r, offset))
+	if err != nil {
+		return nil, fmt.Errorf("failed to discard to offset: %w", err)
+	}
+	if n != offset {
+		return nil, errors.New("unable to discard to offset")
+	}
+
+	return r, nil
+}
+
+// copyWithBuffer is very similar to  io.CopyBuffer https://golang.org/pkg/io/#CopyBuffer
+// but instead of using Read to read from the src, we use ReadAtLeast to make sure we have
+// a full buffer before we do a write operation to dst to reduce overheads associated
+// with the write operations of small buffers.
+func copyWithBuffer(dst io.Writer, src io.Reader) (written int64, err error) {
+	// If the reader has a WriteTo method, use it to do the copy.
+	// Avoids an allocation and a copy.
+	if wt, ok := src.(io.WriterTo); ok {
+		return wt.WriteTo(dst)
+	}
+	// Similarly, if the writer has a ReadFrom method, use it to do the copy.
+	if rt, ok := dst.(io.ReaderFrom); ok {
+		return rt.ReadFrom(src)
+	}
+	bufRef := bufPool.Get().(*[]byte)
+	defer bufPool.Put(bufRef)
+	buf := *bufRef
+	for {
+		nr, er := io.ReadAtLeast(src, buf, len(buf))
+		if nr > 0 {
+			nw, ew := dst.Write(buf[0:nr])
+			if nw > 0 {
+				written += int64(nw)
+			}
+			if ew != nil {
+				err = ew
+				break
+			}
+			if nr != nw {
+				err = io.ErrShortWrite
+				break
+			}
+		}
+		if er != nil {
+			// If an EOF happens after reading fewer than the requested bytes,
+			// ReadAtLeast returns ErrUnexpectedEOF.
+			if er != io.EOF && er != io.ErrUnexpectedEOF {
+				err = er
+			}
+			break
+		}
+	}
+	return
+}
+
+// Exists returns whether an attempt to access the content would not error out
+// with an ErrNotFound error. It will return an encountered error if it was
+// different than ErrNotFound.
+func Exists(ctx context.Context, provider InfoProvider, desc ocispec.Descriptor) (bool, error) {
+	_, err := provider.Info(ctx, desc.Digest)
+	if errdefs.IsNotFound(err) {
+		return false, nil
+	}
+	return err == nil, err
+}

core/content/helpers_test.go

@@ -0,0 +1,208 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package content
+
+import (
+	"bytes"
+	"context"
+	_ "crypto/sha256" // required by go-digest
+	"fmt"
+	"io"
+	"strings"
+	"testing"
+
+	"github.com/containerd/containerd/v2/errdefs"
+	"github.com/opencontainers/go-digest"
+	"github.com/stretchr/testify/assert"
+)
+
+type copySource struct {
+	reader io.Reader
+	size   int64
+	digest digest.Digest
+}
+
+func TestCopy(t *testing.T) {
+	defaultSource := newCopySource("this is the source to copy")
+
+	cf1 := func(buf *bytes.Buffer, st Status) commitFunction {
+		i := 0
+		return func() error {
+			// function resets the first time
+			if i == 0 {
+				// this is the case where, the pipewriter to which the data was being written has
+				// changed. which means we need to clear the buffer
+				i++
+				buf.Reset()
+				st.Offset = 0
+				return ErrReset
+			}
+			return nil
+		}
+	}
+
+	cf2 := func(buf *bytes.Buffer, st Status) commitFunction {
+		i := 0
+		return func() error {
+			// function resets for more than the maxReset value
+			if i < maxResets+1 {
+				// this is the case where, the pipewriter to which the data was being written has
+				// changed. which means we need to clear the buffer
+				i++
+				buf.Reset()
+				st.Offset = 0
+				return ErrReset
+			}
+			return nil
+		}
+	}
+
+	s1 := Status{}
+	s2 := Status{}
+	b1 := bytes.Buffer{}
+	b2 := bytes.Buffer{}
+
+	var testcases = []struct {
+		name        string
+		source      copySource
+		writer      fakeWriter
+		expected    string
+		expectedErr error
+	}{
+		{
+			name:   "copy no offset",
+			source: defaultSource,
+			writer: fakeWriter{
+				Buffer: &bytes.Buffer{},
+			},
+			expected: "this is the source to copy",
+		},
+		{
+			name:   "copy with offset from seeker",
+			source: defaultSource,
+			writer: fakeWriter{
+				Buffer: &bytes.Buffer{},
+				status: Status{Offset: 8},
+			},
+			expected: "the source to copy",
+		},
+		{
+			name:   "copy with offset from unseekable source",
+			source: copySource{reader: bytes.NewBufferString("foobar"), size: 6},
+			writer: fakeWriter{
+				Buffer: &bytes.Buffer{},
+				status: Status{Offset: 3},
+			},
+			expected: "bar",
+		},
+		{
+			name:   "commit already exists",
+			source: newCopySource("this already exists"),
+			writer: fakeWriter{
+				Buffer: &bytes.Buffer{},
+				commitFunc: func() error {
+					return errdefs.ErrAlreadyExists
+				}},
+			expected: "this already exists",
+		},
+		{
+			name:   "commit fails first time with ErrReset",
+			source: newCopySource("content to copy"),
+			writer: fakeWriter{
+				Buffer:     &b1,
+				status:     s1,
+				commitFunc: cf1(&b1, s1),
+			},
+			expected: "content to copy",
+		},
+		{
+			name:   "write fails more than maxReset times due to reset",
+			source: newCopySource("content to copy"),
+			writer: fakeWriter{
+				Buffer:     &b2,
+				status:     s2,
+				commitFunc: cf2(&b2, s2),
+			},
+			expected:    "",
+			expectedErr: fmt.Errorf("failed to copy after %d retries", maxResets),
+		},
+	}
+
+	for _, testcase := range testcases {
+		testcase := testcase
+		t.Run(testcase.name, func(t *testing.T) {
+			err := Copy(context.Background(),
+				&testcase.writer,
+				testcase.source.reader,
+				testcase.source.size,
+				testcase.source.digest)
+
+			// if an error is expected then further comparisons are not required
+			if testcase.expectedErr != nil {
+				assert.Equal(t, testcase.expectedErr, err)
+				return
+			}
+
+			assert.NoError(t, err)
+			assert.Equal(t, testcase.source.digest, testcase.writer.committedDigest)
+			assert.Equal(t, testcase.expected, testcase.writer.String())
+		})
+	}
+}
+
+func newCopySource(raw string) copySource {
+	return copySource{
+		reader: strings.NewReader(raw),
+		size:   int64(len(raw)),
+		digest: digest.FromBytes([]byte(raw)),
+	}
+}
+
+type commitFunction func() error
+
+type fakeWriter struct {
+	*bytes.Buffer
+	committedDigest digest.Digest
+	status          Status
+	commitFunc      commitFunction
+}
+
+func (f *fakeWriter) Close() error {
+	f.Buffer.Reset()
+	return nil
+}
+
+func (f *fakeWriter) Commit(ctx context.Context, size int64, expected digest.Digest, opts ...Opt) error {
+	f.committedDigest = expected
+	if f.commitFunc == nil {
+		return nil
+	}
+	return f.commitFunc()
+}
+
+func (f *fakeWriter) Digest() digest.Digest {
+	return f.committedDigest
+}
+
+func (f *fakeWriter) Status() (Status, error) {
+	return f.status, nil
+}
+
+func (f *fakeWriter) Truncate(size int64) error {
+	f.Buffer.Truncate(int(size))
+	return nil
+}

core/content/proxy/content_reader.go

@@ -0,0 +1,71 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package proxy
+
+import (
+	"context"
+
+	contentapi "github.com/containerd/containerd/v2/api/services/content/v1"
+	digest "github.com/opencontainers/go-digest"
+)
+
+type remoteReaderAt struct {
+	ctx    context.Context
+	digest digest.Digest
+	size   int64
+	client contentapi.ContentClient
+}
+
+func (ra *remoteReaderAt) Size() int64 {
+	return ra.size
+}
+
+func (ra *remoteReaderAt) ReadAt(p []byte, off int64) (n int, err error) {
+	rr := &contentapi.ReadContentRequest{
+		Digest: ra.digest.String(),
+		Offset: off,
+		Size:   int64(len(p)),
+	}
+	// we need a child context with cancel, or the eventually called
+	// grpc.NewStream will leak the goroutine until the whole thing is cleared.
+	// See comment at https://godoc.org/google.golang.org/grpc#ClientConn.NewStream
+	childCtx, cancel := context.WithCancel(ra.ctx)
+	// we MUST cancel the child context; see comment above
+	defer cancel()
+	rc, err := ra.client.Read(childCtx, rr)
+	if err != nil {
+		return 0, err
+	}
+
+	for len(p) > 0 {
+		var resp *contentapi.ReadContentResponse
+		// fill our buffer up until we can fill p.
+		resp, err = rc.Recv()
+		if err != nil {
+			return n, err
+		}
+
+		copied := copy(p, resp.Data)
+		n += copied
+		p = p[copied:]
+	}
+	return n, nil
+}
+
+func (ra *remoteReaderAt) Close() error {
+	return nil
+}

core/content/proxy/content_store.go

@@ -0,0 +1,235 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package proxy
+
+import (
+	"context"
+	"io"
+
+	contentapi "github.com/containerd/containerd/v2/api/services/content/v1"
+	"github.com/containerd/containerd/v2/core/content"
+	"github.com/containerd/containerd/v2/errdefs"
+	"github.com/containerd/containerd/v2/protobuf"
+	protobuftypes "github.com/containerd/containerd/v2/protobuf/types"
+	digest "github.com/opencontainers/go-digest"
+	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
+)
+
+type proxyContentStore struct {
+	client contentapi.ContentClient
+}
+
+// NewContentStore returns a new content store which communicates over a GRPC
+// connection using the containerd content GRPC API.
+func NewContentStore(client contentapi.ContentClient) content.Store {
+	return &proxyContentStore{
+		client: client,
+	}
+}
+
+func (pcs *proxyContentStore) Info(ctx context.Context, dgst digest.Digest) (content.Info, error) {
+	resp, err := pcs.client.Info(ctx, &contentapi.InfoRequest{
+		Digest: dgst.String(),
+	})
+	if err != nil {
+		return content.Info{}, errdefs.FromGRPC(err)
+	}
+
+	return infoFromGRPC(resp.Info), nil
+}
+
+func (pcs *proxyContentStore) Walk(ctx context.Context, fn content.WalkFunc, filters ...string) error {
+	session, err := pcs.client.List(ctx, &contentapi.ListContentRequest{
+		Filters: filters,
+	})
+	if err != nil {
+		return errdefs.FromGRPC(err)
+	}
+
+	for {
+		msg, err := session.Recv()
+		if err != nil {
+			if err != io.EOF {
+				return errdefs.FromGRPC(err)
+			}
+
+			break
+		}
+
+		for _, info := range msg.Info {
+			if err := fn(infoFromGRPC(info)); err != nil {
+				return err
+			}
+		}
+	}
+
+	return nil
+}
+
+func (pcs *proxyContentStore) Delete(ctx context.Context, dgst digest.Digest) error {
+	if _, err := pcs.client.Delete(ctx, &contentapi.DeleteContentRequest{
+		Digest: dgst.String(),
+	}); err != nil {
+		return errdefs.FromGRPC(err)
+	}
+
+	return nil
+}
+
+// ReaderAt ignores MediaType.
+func (pcs *proxyContentStore) ReaderAt(ctx context.Context, desc ocispec.Descriptor) (content.ReaderAt, error) {
+	i, err := pcs.Info(ctx, desc.Digest)
+	if err != nil {
+		return nil, err
+	}
+
+	return &remoteReaderAt{
+		ctx:    ctx,
+		digest: desc.Digest,
+		size:   i.Size,
+		client: pcs.client,
+	}, nil
+}
+
+func (pcs *proxyContentStore) Status(ctx context.Context, ref string) (content.Status, error) {
+	resp, err := pcs.client.Status(ctx, &contentapi.StatusRequest{
+		Ref: ref,
+	})
+	if err != nil {
+		return content.Status{}, errdefs.FromGRPC(err)
+	}
+
+	status := resp.Status
+	return content.Status{
+		Ref:       status.Ref,
+		StartedAt: protobuf.FromTimestamp(status.StartedAt),
+		UpdatedAt: protobuf.FromTimestamp(status.UpdatedAt),
+		Offset:    status.Offset,
+		Total:     status.Total,
+		Expected:  digest.Digest(status.Expected),
+	}, nil
+}
+
+func (pcs *proxyContentStore) Update(ctx context.Context, info content.Info, fieldpaths ...string) (content.Info, error) {
+	resp, err := pcs.client.Update(ctx, &contentapi.UpdateRequest{
+		Info: infoToGRPC(&info),
+		UpdateMask: &protobuftypes.FieldMask{
+			Paths: fieldpaths,
+		},
+	})
+	if err != nil {
+		return content.Info{}, errdefs.FromGRPC(err)
+	}
+	return infoFromGRPC(resp.Info), nil
+}
+
+func (pcs *proxyContentStore) ListStatuses(ctx context.Context, filters ...string) ([]content.Status, error) {
+	resp, err := pcs.client.ListStatuses(ctx, &contentapi.ListStatusesRequest{
+		Filters: filters,
+	})
+	if err != nil {
+		return nil, errdefs.FromGRPC(err)
+	}
+
+	var statuses []content.Status
+	for _, status := range resp.Statuses {
+		statuses = append(statuses, content.Status{
+			Ref:       status.Ref,
+			StartedAt: protobuf.FromTimestamp(status.StartedAt),
+			UpdatedAt: protobuf.FromTimestamp(status.UpdatedAt),
+			Offset:    status.Offset,
+			Total:     status.Total,
+			Expected:  digest.Digest(status.Expected),
+		})
+	}
+
+	return statuses, nil
+}
+
+// Writer ignores MediaType.
+func (pcs *proxyContentStore) Writer(ctx context.Context, opts ...content.WriterOpt) (content.Writer, error) {
+	var wOpts content.WriterOpts
+	for _, opt := range opts {
+		if err := opt(&wOpts); err != nil {
+			return nil, err
+		}
+	}
+	wrclient, offset, err := pcs.negotiate(ctx, wOpts.Ref, wOpts.Desc.Size, wOpts.Desc.Digest)
+	if err != nil {
+		return nil, errdefs.FromGRPC(err)
+	}
+
+	return &remoteWriter{
+		ref:    wOpts.Ref,
+		client: wrclient,
+		offset: offset,
+	}, nil
+}
+
+// Abort implements asynchronous abort. It starts a new write session on the ref l
+func (pcs *proxyContentStore) Abort(ctx context.Context, ref string) error {
+	if _, err := pcs.client.Abort(ctx, &contentapi.AbortRequest{
+		Ref: ref,
+	}); err != nil {
+		return errdefs.FromGRPC(err)
+	}
+
+	return nil
+}
+
+func (pcs *proxyContentStore) negotiate(ctx context.Context, ref string, size int64, expected digest.Digest) (contentapi.Content_WriteClient, int64, error) {
+	wrclient, err := pcs.client.Write(ctx)
+	if err != nil {
+		return nil, 0, err
+	}
+
+	if err := wrclient.Send(&contentapi.WriteContentRequest{
+		Action:   contentapi.WriteAction_STAT,
+		Ref:      ref,
+		Total:    size,
+		Expected: expected.String(),
+	}); err != nil {
+		return nil, 0, err
+	}
+
+	resp, err := wrclient.Recv()
+	if err != nil {
+		return nil, 0, err
+	}
+
+	return wrclient, resp.Offset, nil
+}
+
+func infoToGRPC(info *content.Info) *contentapi.Info {
+	return &contentapi.Info{
+		Digest:    info.Digest.String(),
+		Size:      info.Size,
+		CreatedAt: protobuf.ToTimestamp(info.CreatedAt),
+		UpdatedAt: protobuf.ToTimestamp(info.UpdatedAt),
+		Labels:    info.Labels,
+	}
+}
+
+func infoFromGRPC(info *contentapi.Info) content.Info {
+	return content.Info{
+		Digest:    digest.Digest(info.Digest),
+		Size:      info.Size,
+		CreatedAt: protobuf.FromTimestamp(info.CreatedAt),
+		UpdatedAt: protobuf.FromTimestamp(info.UpdatedAt),
+		Labels:    info.Labels,
+	}
+}

core/content/proxy/content_writer.go

@@ -0,0 +1,148 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package proxy
+
+import (
+	"context"
+	"fmt"
+	"io"
+
+	contentapi "github.com/containerd/containerd/v2/api/services/content/v1"
+	"github.com/containerd/containerd/v2/core/content"
+	"github.com/containerd/containerd/v2/errdefs"
+	"github.com/containerd/containerd/v2/protobuf"
+	digest "github.com/opencontainers/go-digest"
+)
+
+type remoteWriter struct {
+	ref    string
+	client contentapi.Content_WriteClient
+	offset int64
+	digest digest.Digest
+}
+
+// send performs a synchronous req-resp cycle on the client.
+func (rw *remoteWriter) send(req *contentapi.WriteContentRequest) (*contentapi.WriteContentResponse, error) {
+	if err := rw.client.Send(req); err != nil {
+		return nil, err
+	}
+
+	resp, err := rw.client.Recv()
+
+	if err == nil {
+		// try to keep these in sync
+		if resp.Digest != "" {
+			rw.digest = digest.Digest(resp.Digest)
+		}
+	}
+
+	return resp, err
+}
+
+func (rw *remoteWriter) Status() (content.Status, error) {
+	resp, err := rw.send(&contentapi.WriteContentRequest{
+		Action: contentapi.WriteAction_STAT,
+	})
+	if err != nil {
+		return content.Status{}, fmt.Errorf("error getting writer status: %w", errdefs.FromGRPC(err))
+	}
+
+	return content.Status{
+		Ref:       rw.ref,
+		Offset:    resp.Offset,
+		Total:     resp.Total,
+		StartedAt: protobuf.FromTimestamp(resp.StartedAt),
+		UpdatedAt: protobuf.FromTimestamp(resp.UpdatedAt),
+	}, nil
+}
+
+func (rw *remoteWriter) Digest() digest.Digest {
+	return rw.digest
+}
+
+func (rw *remoteWriter) Write(p []byte) (n int, err error) {
+	offset := rw.offset
+
+	resp, err := rw.send(&contentapi.WriteContentRequest{
+		Action: contentapi.WriteAction_WRITE,
+		Offset: offset,
+		Data:   p,
+	})
+	if err != nil {
+		return 0, fmt.Errorf("failed to send write: %w", errdefs.FromGRPC(err))
+	}
+
+	n = int(resp.Offset - offset)
+	if n < len(p) {
+		err = io.ErrShortWrite
+	}
+
+	rw.offset += int64(n)
+	if resp.Digest != "" {
+		rw.digest = digest.Digest(resp.Digest)
+	}
+	return
+}
+
+func (rw *remoteWriter) Commit(ctx context.Context, size int64, expected digest.Digest, opts ...content.Opt) (err error) {
+	defer func() {
+		err1 := rw.Close()
+		if err == nil {
+			err = err1
+		}
+	}()
+
+	var base content.Info
+	for _, opt := range opts {
+		if err := opt(&base); err != nil {
+			return err
+		}
+	}
+	resp, err := rw.send(&contentapi.WriteContentRequest{
+		Action:   contentapi.WriteAction_COMMIT,
+		Total:    size,
+		Offset:   rw.offset,
+		Expected: expected.String(),
+		Labels:   base.Labels,
+	})
+	if err != nil {
+		return fmt.Errorf("commit failed: %w", errdefs.FromGRPC(err))
+	}
+
+	if size != 0 && resp.Offset != size {
+		return fmt.Errorf("unexpected size: %v != %v", resp.Offset, size)
+	}
+
+	actual := digest.Digest(resp.Digest)
+	if expected != "" && actual != expected {
+		return fmt.Errorf("unexpected digest: %v != %v", resp.Digest, expected)
+	}
+
+	rw.digest = actual
+	rw.offset = resp.Offset
+	return nil
+}
+
+func (rw *remoteWriter) Truncate(size int64) error {
+	// This truncation won't actually be validated until a write is issued.
+	rw.offset = size
+	return nil
+}
+
+func (rw *remoteWriter) Close() error {
+	return rw.client.CloseSend()
+}