new file mode 100644

@@ -0,0 +1,30 @@

+package daemon

+

+import (

+	"os"

+	"path/filepath"

+

+	"github.com/containerd/containerd/content"

+	"github.com/containerd/containerd/content/local"

+	"github.com/containerd/containerd/leases"

+	"github.com/containerd/containerd/metadata"

+	"github.com/pkg/errors"

+	"go.etcd.io/bbolt"

+)

+

+func (d *Daemon) configureLocalContentStore() (content.Store, leases.Manager, error) {

+	if err := os.MkdirAll(filepath.Join(d.root, "content"), 0700); err != nil {

+		return nil, nil, errors.Wrap(err, "error creating dir for content store")

+	}

+	db, err := bbolt.Open(filepath.Join(d.root, "content", "metadata.db"), 0600, nil)

+	if err != nil {

+		return nil, nil, errors.Wrap(err, "error opening bolt db for content metadata store")

+	}

+	cs, err := local.NewStore(filepath.Join(d.root, "content", "data"))

+	if err != nil {

+		return nil, nil, errors.Wrap(err, "error setting up content store")

+	}

+	md := metadata.NewDB(db, cs, nil)

+	d.mdDB = db

+	return md.ContentStore(), metadata.NewLeaseManager(md), nil

+}

@@ -20,6 +20,7 @@ import (

 	"time"

 	"github.com/docker/docker/pkg/fileutils"

+	"go.etcd.io/bbolt"

 	"google.golang.org/grpc"

 	"google.golang.org/grpc/backoff"

@@ -129,6 +130,11 @@ type Daemon struct {

 	attachmentStore       network.AttachmentStore

 	attachableNetworkLock *locker.Locker

+

+	// This is used for Windows which doesn't currently support running on containerd

+	// It stores metadata for the content store (used for manifest caching)

+	// This needs to be closed on daemon exit

+	mdDB *bbolt.DB

 }

 // StoreHosts stores the addresses the daemon is listening on

@@ -1066,10 +1072,7 @@ func NewDaemon(ctx context.Context, config *config.Config, pluginStore *plugin.S

 	d.linkIndex = newLinkIndex()

-	// TODO: imageStore, distributionMetadataStore, and ReferenceStore are only

-	// used above to run migration. They could be initialized in ImageService

-	// if migration is called from daemon/images. layerStore might move as well.

-	d.imageService = images.NewImageService(images.ImageServiceConfig{

+	imgSvcConfig := images.ImageServiceConfig{

 		ContainerStore:            d.containers,

 		DistributionMetadataStore: distributionMetadataStore,

 		EventsService:             d.EventsService,

@@ -1081,7 +1084,28 @@ func NewDaemon(ctx context.Context, config *config.Config, pluginStore *plugin.S

 		ReferenceStore:            rs,

 		RegistryService:           registryService,

 		TrustKey:                  trustKey,

-	})

+		ContentNamespace:          config.ContainerdNamespace,

+	}

+

+	// containerd is not currently supported with Windows.

+	// So sometimes d.containerdCli will be nil

+	// In that case we'll create a local content store... but otherwise we'll use containerd

+	if d.containerdCli != nil {

+		imgSvcConfig.Leases = d.containerdCli.LeasesService()

+		imgSvcConfig.ContentStore = d.containerdCli.ContentStore()

+	} else {

+		cs, lm, err := d.configureLocalContentStore()

+		if err != nil {

+			return nil, err

+		}

+		imgSvcConfig.ContentStore = cs

+		imgSvcConfig.Leases = lm

+	}

+

+	// TODO: imageStore, distributionMetadataStore, and ReferenceStore are only

+	// used above to run migration. They could be initialized in ImageService

+	// if migration is called from daemon/images. layerStore might move as well.

+	d.imageService = images.NewImageService(imgSvcConfig)

 	go d.execCommandGC()

@@ -1246,6 +1270,10 @@ func (daemon *Daemon) Shutdown() error {

 		daemon.containerdCli.Close()

 	}

+	if daemon.mdDB != nil {

+		daemon.mdDB.Close()

+	}

+

 	return daemon.cleanupMounts()

 }

@@ -6,6 +6,8 @@ import (

 	"strings"

 	"time"

+	"github.com/containerd/containerd/leases"

+	"github.com/containerd/containerd/namespaces"

 	dist "github.com/docker/distribution"

 	"github.com/docker/distribution/reference"

 	"github.com/docker/docker/api/types"

@@ -16,6 +18,7 @@ import (

 	"github.com/docker/docker/registry"

 	digest "github.com/opencontainers/go-digest"

 	specs "github.com/opencontainers/image-spec/specs-go/v1"

+	"github.com/pkg/errors"

 )

 // PullImage initiates a pull operation. image is the repository name to pull, and

@@ -65,6 +68,25 @@ func (i *ImageService) pullImageWithReference(ctx context.Context, ref reference

 		close(writesDone)

 	}()

+	ctx = namespaces.WithNamespace(ctx, i.contentNamespace)

+	// Take out a temporary lease for everything that gets persisted to the content store.

+	// Before the lease is cancelled, any content we want to keep should have it's own lease applied.

+	ctx, done, err := tempLease(ctx, i.leases)

+	if err != nil {

+		return err

+	}

+	defer done(ctx)

+

+	cs := &contentStoreForPull{

+		ContentStore: i.content,

+		leases:       i.leases,

+	}

+	imageStore := &imageStoreForPull{

+		ImageConfigStore: distribution.NewImageConfigStoreFromStore(i.imageStore),

+		ingested:         cs,

+		leases:           i.leases,

+	}

+

 	imagePullConfig := &distribution.ImagePullConfig{

 		Config: distribution.Config{

 			MetaHeaders:      metaHeaders,

@@ -73,7 +95,7 @@ func (i *ImageService) pullImageWithReference(ctx context.Context, ref reference

 			RegistryService:  i.registryService,

 			ImageEventLogger: i.LogImageEvent,

 			MetadataStore:    i.distributionMetadataStore,

-			ImageStore:       distribution.NewImageConfigStoreFromStore(i.imageStore),

+			ImageStore:       imageStore,

 			ReferenceStore:   i.referenceStore,

 		},

 		DownloadManager: i.downloadManager,

@@ -81,7 +103,7 @@ func (i *ImageService) pullImageWithReference(ctx context.Context, ref reference

 		Platform:        platform,

 	}

-	err := distribution.Pull(ctx, ref, imagePullConfig)

+	err = distribution.Pull(ctx, ref, imagePullConfig, cs)

 	close(progressChan)

 	<-writesDone

 	return err

@@ -124,3 +146,29 @@ func (i *ImageService) GetRepository(ctx context.Context, ref reference.Named, a

 	}

 	return repository, confirmedV2, lastError

 }

+

+func tempLease(ctx context.Context, mgr leases.Manager) (context.Context, func(context.Context) error, error) {

+	nop := func(context.Context) error { return nil }

+	_, ok := leases.FromContext(ctx)

+	if ok {

+		return ctx, nop, nil

+	}

+

+	// Use an expiration that ensures the lease is cleaned up at some point if there is a crash, SIGKILL, etc.

+	opts := []leases.Opt{

+		leases.WithRandomID(),

+		leases.WithExpiration(24 * time.Hour),

+		leases.WithLabels(map[string]string{

+			"moby.lease/temporary": time.Now().UTC().Format(time.RFC3339Nano),

+		}),

+	}

+	l, err := mgr.Create(ctx, opts...)

+	if err != nil {

+		return ctx, nop, errors.Wrap(err, "error creating temporary lease")

+	}

+

+	ctx = leases.WithLease(ctx, l.ID)

+	return ctx, func(ctx context.Context) error {

+		return mgr.Delete(ctx, l)

+	}, nil

+}

@@ -5,6 +5,8 @@ import (

 	"os"

 	"runtime"

+	"github.com/containerd/containerd/content"

+	"github.com/containerd/containerd/leases"

 	"github.com/docker/docker/container"

 	daemonevents "github.com/docker/docker/daemon/events"

 	"github.com/docker/docker/distribution"

@@ -42,6 +44,9 @@ type ImageServiceConfig struct {

 	ReferenceStore            dockerreference.Store

 	RegistryService           registry.Service

 	TrustKey                  libtrust.PrivateKey

+	ContentStore              content.Store

+	Leases                    leases.Manager

+	ContentNamespace          string

 }

 // NewImageService returns a new ImageService from a configuration

@@ -54,12 +59,15 @@ func NewImageService(config ImageServiceConfig) *ImageService {

 		distributionMetadataStore: config.DistributionMetadataStore,

 		downloadManager:           xfer.NewLayerDownloadManager(config.LayerStores, config.MaxConcurrentDownloads, xfer.WithMaxDownloadAttempts(config.MaxDownloadAttempts)),

 		eventsService:             config.EventsService,

-		imageStore:                config.ImageStore,

+		imageStore:                &imageStoreWithLease{Store: config.ImageStore, leases: config.Leases, ns: config.ContentNamespace},

 		layerStores:               config.LayerStores,

 		referenceStore:            config.ReferenceStore,

 		registryService:           config.RegistryService,

 		trustKey:                  config.TrustKey,

 		uploadManager:             xfer.NewLayerUploadManager(config.MaxConcurrentUploads),

+		leases:                    config.Leases,

+		content:                   config.ContentStore,

+		contentNamespace:          config.ContentNamespace,

 	}

 }

@@ -76,6 +84,9 @@ type ImageService struct {

 	registryService           registry.Service

 	trustKey                  libtrust.PrivateKey

 	uploadManager             *xfer.LayerUploadManager

+	leases                    leases.Manager

+	content                   content.Store

+	contentNamespace          string

 }

 // DistributionServices provides daemon image storage services

new file mode 100644

@@ -0,0 +1,155 @@

+package images

+

+import (

+	"context"

+	"sync"

+

+	"github.com/containerd/containerd/content"

+	c8derrdefs "github.com/containerd/containerd/errdefs"

+	"github.com/containerd/containerd/leases"

+	"github.com/containerd/containerd/log"

+	"github.com/containerd/containerd/namespaces"

+	"github.com/docker/docker/distribution"

+	"github.com/docker/docker/image"

+	"github.com/docker/docker/layer"

+	digest "github.com/opencontainers/go-digest"

+	"github.com/pkg/errors"

+	"github.com/sirupsen/logrus"

+)

+

+func imageKey(dgst digest.Digest) string {

+	return "moby-image-" + dgst.String()

+}

+

+// imageStoreWithLease wraps the configured image store with one that deletes the lease

+// reigstered for a given image ID, if one exists

+//

+// This is used by the main image service to wrap delete calls to the real image store.

+type imageStoreWithLease struct {

+	image.Store

+	leases leases.Manager

+

+	// Normally we'd pass namespace down through a context.Context, however...

+	// The interface for image store doesn't allow this, so we store it here.

+	ns string

+}

+

+func (s *imageStoreWithLease) Delete(id image.ID) ([]layer.Metadata, error) {

+	ctx := namespaces.WithNamespace(context.TODO(), s.ns)

+	if err := s.leases.Delete(ctx, leases.Lease{ID: imageKey(digest.Digest(id))}); err != nil && !c8derrdefs.IsNotFound(err) {

+		return nil, errors.Wrap(err, "error deleting lease")

+	}

+	return s.Store.Delete(id)

+}

+

+// iamgeStoreForPull is created for each pull It wraps an underlying image store

+// to handle registering leases for content fetched in a single image pull.

+type imageStoreForPull struct {

+	distribution.ImageConfigStore

+	leases   leases.Manager

+	ingested *contentStoreForPull

+}

+

+func (s *imageStoreForPull) Put(ctx context.Context, config []byte) (digest.Digest, error) {

+	id, err := s.ImageConfigStore.Put(ctx, config)

+	if err != nil {

+		return "", err

+	}

+	return id, s.updateLease(ctx, id)

+}

+

+func (s *imageStoreForPull) Get(ctx context.Context, dgst digest.Digest) ([]byte, error) {

+	id, err := s.ImageConfigStore.Get(ctx, dgst)

+	if err != nil {

+		return nil, err

+	}

+	return id, s.updateLease(ctx, dgst)

+}

+

+func (s *imageStoreForPull) updateLease(ctx context.Context, dgst digest.Digest) error {

+	leaseID := imageKey(dgst)

+	lease, err := s.leases.Create(ctx, leases.WithID(leaseID))

+	if err != nil {

+		if !c8derrdefs.IsAlreadyExists(err) {

+			return errors.Wrap(err, "error creating lease")

+		}

+		lease = leases.Lease{ID: leaseID}

+	}

+

+	digested := s.ingested.getDigested()

+	resource := leases.Resource{

+		Type: "content",

+	}

+	for _, dgst := range digested {

+		log.G(ctx).WithFields(logrus.Fields{

+			"digest": dgst,

+			"lease":  lease.ID,

+		}).Debug("Adding content digest to lease")

+

+		resource.ID = dgst.String()

+		if err := s.leases.AddResource(ctx, lease, resource); err != nil {

+			return errors.Wrapf(err, "error adding content digest to lease: %s", dgst)

+		}

+	}

+	return nil

+}

+

+// contentStoreForPull is used to wrap the configured content store to

+// add lease management for a single `pull`

+// It stores all committed digests so that `imageStoreForPull` can add

+// the digsted resources to the lease for an image.

+type contentStoreForPull struct {

+	distribution.ContentStore

+	leases leases.Manager

+

+	mu       sync.Mutex

+	digested []digest.Digest

+}

+

+func (c *contentStoreForPull) addDigested(dgst digest.Digest) {

+	c.mu.Lock()

+	c.digested = append(c.digested, dgst)

+	c.mu.Unlock()

+}

+

+func (c *contentStoreForPull) getDigested() []digest.Digest {

+	c.mu.Lock()

+	digested := make([]digest.Digest, len(c.digested))

+	copy(digested, c.digested)

+	c.mu.Unlock()

+	return digested

+}

+

+func (c *contentStoreForPull) Writer(ctx context.Context, opts ...content.WriterOpt) (content.Writer, error) {

+	w, err := c.ContentStore.Writer(ctx, opts...)

+	if err != nil {

+		if c8derrdefs.IsAlreadyExists(err) {

+			var cfg content.WriterOpts

+			for _, o := range opts {

+				if err := o(&cfg); err != nil {

+					return nil, err

+				}

+

+			}

+			c.addDigested(cfg.Desc.Digest)

+		}

+		return nil, err

+	}

+	return &contentWriter{

+		cs:     c,

+		Writer: w,

+	}, nil

+}

+

+type contentWriter struct {

+	cs *contentStoreForPull

+	content.Writer

+}

+

+func (w *contentWriter) Commit(ctx context.Context, size int64, expected digest.Digest, opts ...content.Opt) error {

+	err := w.Writer.Commit(ctx, size, expected, opts...)

+	if err == nil || c8derrdefs.IsAlreadyExists(err) {

+		w.cs.addDigested(expected)

+	}

+	return err

+}

new file mode 100644

@@ -0,0 +1,124 @@

+package images

+

+import (

+	"context"

+	"io/ioutil"

+	"os"

+	"path/filepath"

+	"testing"

+

+	"github.com/containerd/containerd/content"

+	"github.com/containerd/containerd/content/local"

+	c8derrdefs "github.com/containerd/containerd/errdefs"

+	"github.com/containerd/containerd/leases"

+	"github.com/containerd/containerd/metadata"

+	"github.com/containerd/containerd/namespaces"

+	"github.com/docker/docker/image"

+	digest "github.com/opencontainers/go-digest"

+	v1 "github.com/opencontainers/image-spec/specs-go/v1"

+	"go.etcd.io/bbolt"

+	"gotest.tools/v3/assert"

+	"gotest.tools/v3/assert/cmp"

+)

+

+func setupTestStores(t *testing.T) (context.Context, content.Store, *imageStoreWithLease, func(t *testing.T)) {

+	dir, err := ioutil.TempDir("", t.Name())

+	assert.NilError(t, err)

+

+	backend, err := image.NewFSStoreBackend(filepath.Join(dir, "images"))

+	assert.NilError(t, err)

+	is, err := image.NewImageStore(backend, nil)

+	assert.NilError(t, err)

+

+	db, err := bbolt.Open(filepath.Join(dir, "metadata.db"), 0600, nil)

+	assert.NilError(t, err)

+

+	cs, err := local.NewStore(filepath.Join(dir, "content"))

+	assert.NilError(t, err)

+	mdb := metadata.NewDB(db, cs, nil)

+

+	cleanup := func(t *testing.T) {

+		assert.Check(t, db.Close())

+		assert.Check(t, os.RemoveAll(dir))

+	}

+	ctx := namespaces.WithNamespace(context.Background(), t.Name())

+	images := &imageStoreWithLease{Store: is, ns: t.Name(), leases: metadata.NewLeaseManager(mdb)}

+

+	return ctx, cs, images, cleanup

+}

+

+func TestImageDelete(t *testing.T) {

+	ctx, _, images, cleanup := setupTestStores(t)

+	defer cleanup(t)

+

+	t.Run("no lease", func(t *testing.T) {

+		id, err := images.Create([]byte(`{"rootFS": {}}`))

+		assert.NilError(t, err)

+		defer images.Delete(id)

+

+		ls, err := images.leases.List(ctx)

+		assert.NilError(t, err)

+		assert.Equal(t, len(ls), 0, ls)

+

+		_, err = images.Delete(id)

+		assert.NilError(t, err, "should not error when there is no lease")

+	})

+

+	t.Run("lease exists", func(t *testing.T) {

+		id, err := images.Create([]byte(`{"rootFS": {}}`))

+		assert.NilError(t, err)

+		defer images.Delete(id)

+

+		leaseID := imageKey(digest.Digest(id))

+		_, err = images.leases.Create(ctx, leases.WithID(leaseID))

+		assert.NilError(t, err)

+		defer images.leases.Delete(ctx, leases.Lease{ID: leaseID})

+

+		ls, err := images.leases.List(ctx)

+		assert.NilError(t, err)

+		assert.Check(t, cmp.Equal(len(ls), 1), ls)

+

+		_, err = images.Delete(id)

+		assert.NilError(t, err)

+

+		ls, err = images.leases.List(ctx)

+		assert.NilError(t, err)

+		assert.Check(t, cmp.Equal(len(ls), 0), ls)

+	})

+}

+

+func TestContentStoreForPull(t *testing.T) {

+	ctx, cs, is, cleanup := setupTestStores(t)

+	defer cleanup(t)

+

+	csP := &contentStoreForPull{

+		ContentStore: cs,

+		leases:       is.leases,

+	}

+

+	data := []byte(`{}`)

+	desc := v1.Descriptor{

+		Digest: digest.Canonical.FromBytes(data),

+		Size:   int64(len(data)),

+	}

+

+	w, err := csP.Writer(ctx, content.WithRef(t.Name()), content.WithDescriptor(desc))

+	assert.NilError(t, err)

+

+	_, err = w.Write(data)

+	assert.NilError(t, err)

+	defer w.Close()

+

+	err = w.Commit(ctx, desc.Size, desc.Digest)

+	assert.NilError(t, err)

+

+	assert.Equal(t, len(csP.digested), 1)

+	assert.Check(t, cmp.Equal(csP.digested[0], desc.Digest))

+

+	// Test already exists

+	csP.digested = nil

+	_, err = csP.Writer(ctx, content.WithRef(t.Name()), content.WithDescriptor(desc))

+	assert.Check(t, c8derrdefs.IsAlreadyExists(err))

+	assert.Equal(t, len(csP.digested), 1)

+	assert.Check(t, cmp.Equal(csP.digested[0], desc.Digest))

+}

@@ -84,8 +84,8 @@ type ImagePushConfig struct {

 // by digest. Allows getting an image configurations rootfs from the

 // configuration.

 type ImageConfigStore interface {

-	Put([]byte) (digest.Digest, error)

-	Get(digest.Digest) ([]byte, error)

+	Put(context.Context, []byte) (digest.Digest, error)

+	Get(context.Context, digest.Digest) ([]byte, error)

 	RootFSFromConfig([]byte) (*image.RootFS, error)

 	PlatformFromConfig([]byte) (*specs.Platform, error)

 }

@@ -128,12 +128,12 @@ func NewImageConfigStoreFromStore(is image.Store) ImageConfigStore {

 	}

 }

-func (s *imageConfigStore) Put(c []byte) (digest.Digest, error) {

+func (s *imageConfigStore) Put(_ context.Context, c []byte) (digest.Digest, error) {

 	id, err := s.Store.Create(c)

 	return digest.Digest(id), err

 }

-func (s *imageConfigStore) Get(d digest.Digest) ([]byte, error) {

+func (s *imageConfigStore) Get(_ context.Context, d digest.Digest) ([]byte, error) {

 	img, err := s.Store.Get(image.IDFromDigest(d))

 	if err != nil {

 		return nil, err

new file mode 100644

@@ -0,0 +1,195 @@

+package distribution

+

+import (

+	"context"

+	"encoding/json"

+	"io"

+	"io/ioutil"

+

+	"github.com/containerd/containerd/content"

+	"github.com/containerd/containerd/errdefs"

+	"github.com/containerd/containerd/log"

+	"github.com/containerd/containerd/remotes"

+	"github.com/docker/distribution"

+	"github.com/docker/distribution/manifest/schema1"

+	digest "github.com/opencontainers/go-digest"

+	specs "github.com/opencontainers/image-spec/specs-go/v1"

+	"github.com/pkg/errors"

+)

+

+// This is used by manifestStore to pare down the requirements to implement a

+// full distribution.ManifestService, since `Get` is all we use here.

+type manifestGetter interface {

+	Get(ctx context.Context, dgst digest.Digest, options ...distribution.ManifestServiceOption) (distribution.Manifest, error)

+}

+

+type manifestStore struct {

+	local  ContentStore

+	remote manifestGetter

+}

+

+// ContentStore is the interface used to persist registry blobs

+//

+// Currently this is only used to persist manifests and manifest lists.

+// It is exported because `distribution.Pull` takes one as an argument.

+type ContentStore interface {

+	content.Ingester

+	content.Provider

+	Info(ctx context.Context, dgst digest.Digest) (content.Info, error)

+	Abort(ctx context.Context, ref string) error

+}

+

+func (m *manifestStore) getLocal(ctx context.Context, desc specs.Descriptor) (distribution.Manifest, error) {

+	ra, err := m.local.ReaderAt(ctx, desc)

+	if err != nil {

+		return nil, errors.Wrap(err, "error getting content store reader")

+	}

+	defer ra.Close()

+

+	r := io.NewSectionReader(ra, 0, ra.Size())

+	data, err := ioutil.ReadAll(r)

+	if err != nil {

+		return nil, errors.Wrap(err, "error reading manifest from content store")

+	}

+

+	manifest, _, err := distribution.UnmarshalManifest(desc.MediaType, data)

+	if err != nil {

+		return nil, errors.Wrap(err, "error unmarshaling manifest from content store")

+	}

+	return manifest, nil

+}

+

+func (m *manifestStore) getMediaType(ctx context.Context, desc specs.Descriptor) (string, error) {

+	ra, err := m.local.ReaderAt(ctx, desc)

+	if err != nil {

+		return "", errors.Wrap(err, "error getting reader to detect media type")

+	}

+	defer ra.Close()

+

+	mt, err := detectManifestMediaType(ra)

+	if err != nil {

+		return "", errors.Wrap(err, "error detecting media type")

+	}

+	return mt, nil

+}

+

+func (m *manifestStore) Get(ctx context.Context, desc specs.Descriptor) (distribution.Manifest, error) {

+	l := log.G(ctx)

+

+	if desc.MediaType == "" {

+		// When pulling by digest we will not have the media type on the

+		// descriptor since we have not made a request to the registry yet

+		//

+		// We already have the digest, so we only lookup locally... by digest.

+		//

+		// Let's try to detect the media type so we can have a good ref key

+		// here. We may not even have the content locally, and this is fine, but

+		// if we do we should determine that.

+		mt, err := m.getMediaType(ctx, desc)

+		if err != nil && !errdefs.IsNotFound(err) {

+			l.WithError(err).Warn("Error looking up media type of content")

+		}

+		desc.MediaType = mt

+	}

+

+	key := remotes.MakeRefKey(ctx, desc)

+

+	// Here we open a writer to the requested content. This both gives us a

+	// reference to write to if indeed we need to persist it and increments the

+	// ref count on the content.

+	w, err := m.local.Writer(ctx, content.WithDescriptor(desc), content.WithRef(key))

+	if err != nil {

+		if errdefs.IsAlreadyExists(err) {

+			var manifest distribution.Manifest

+			if manifest, err = m.getLocal(ctx, desc); err == nil {

+				return manifest, nil

+			}

+		}

+		// always fallback to the remote if there is an error with the local store

+	}

+	if w != nil {

+		defer w.Close()

+	}

+

+	l.WithError(err).Debug("Fetching manifest from remote")

+

+	manifest, err := m.remote.Get(ctx, desc.Digest)

+	if err != nil {

+		if err := m.local.Abort(ctx, key); err != nil {

+			l.WithError(err).Warn("Error while attempting to abort content ingest")

+		}

+		return nil, err

+	}

+

+	if w != nil {

+		// if `w` is nil here, something happened with the content store, so don't bother trying to persist.

+		if err := m.Put(ctx, manifest, desc, w); err != nil {

+			if err := m.local.Abort(ctx, key); err != nil {

+				l.WithError(err).Warn("error aborting content ingest")

+			}

+			l.WithError(err).Warn("Error persisting manifest")

+		}

+	}

+	return manifest, nil

+}

+

+func (m *manifestStore) Put(ctx context.Context, manifest distribution.Manifest, desc specs.Descriptor, w content.Writer) error {

+	mt, payload, err := manifest.Payload()

+	if err != nil {

+		return err

+	}

+	desc.Size = int64(len(payload))

+	desc.MediaType = mt

+

+	if _, err = w.Write(payload); err != nil {

+		return errors.Wrap(err, "error writing manifest to content store")

+	}

+

+	if err := w.Commit(ctx, desc.Size, desc.Digest); err != nil {

+		return errors.Wrap(err, "error committing manifest to content store")

+	}

+	return nil

+}

+

+func detectManifestMediaType(ra content.ReaderAt) (string, error) {

+	dt := make([]byte, ra.Size())

+	if _, err := ra.ReadAt(dt, 0); err != nil {

+		return "", err

+	}

+

+	return detectManifestBlobMediaType(dt)

+}

+

+// This is used when the manifest store does not know the media type of a sha it

+// was told to get. This would currently only happen when pulling by digest.

+// The media type is needed so the blob can be unmarshalled properly.

+func detectManifestBlobMediaType(dt []byte) (string, error) {

+	var mfst struct {

+		MediaType string          `json:"mediaType"`

+		Config    json.RawMessage `json:"config"`   // schema2 Manifest

+		FSLayers  json.RawMessage `json:"fsLayers"` // schema1 Manifest

+	}

+

+	if err := json.Unmarshal(dt, &mfst); err != nil {

+		return "", err

+	}

+

+	// We may have a media type specified in the json, in which case that should be used.

+	// Docker types should generally have a media type set.

+	// OCI (golang) types do not have a `mediaType` defined, and it is optional in the spec.

+	//

+	// `distrubtion.UnmarshalManifest`, which is used to unmarshal this for real, checks these media type values.

+	// If the specified media type does not match it will error, and in some cases (docker media types) it is required.

+	// So pretty much if we don't have a media type we can fall back to OCI.

+	// This does have a special fallback for schema1 manifests just because it is easy to detect.

+	switch {

+	case mfst.MediaType != "":

+		return mfst.MediaType, nil

+	case mfst.FSLayers != nil:

+		return schema1.MediaTypeManifest, nil

+	case mfst.Config != nil:

+		return specs.MediaTypeImageManifest, nil

+	default:

+		return specs.MediaTypeImageIndex, nil

+	}

+}

new file mode 100644

@@ -0,0 +1,351 @@

+package distribution

+

+import (

+	"context"

+	"encoding/json"

+	"io/ioutil"

+	"os"

+	"strings"

+	"sync"

+	"testing"

+

+	"github.com/containerd/containerd/content"

+	"github.com/containerd/containerd/content/local"

+	"github.com/containerd/containerd/errdefs"

+	"github.com/containerd/containerd/remotes"

+	"github.com/docker/distribution"

+	"github.com/docker/distribution/manifest/ocischema"

+	"github.com/docker/distribution/manifest/schema1"

+	"github.com/google/go-cmp/cmp/cmpopts"

+	digest "github.com/opencontainers/go-digest"

+	specs "github.com/opencontainers/image-spec/specs-go/v1"

+	"github.com/pkg/errors"

+	"gotest.tools/v3/assert"

+	"gotest.tools/v3/assert/cmp"

+)

+

+type mockManifestGetter struct {

+	manifests map[digest.Digest]distribution.Manifest

+	gets      int

+}

+

+func (m *mockManifestGetter) Get(ctx context.Context, dgst digest.Digest, options ...distribution.ManifestServiceOption) (distribution.Manifest, error) {

+	m.gets++

+	manifest, ok := m.manifests[dgst]

+	if !ok {

+		return nil, distribution.ErrManifestUnknown{Tag: dgst.String()}

+	}

+	return manifest, nil

+}

+

+type memoryLabelStore struct {

+	mu     sync.Mutex

+	labels map[digest.Digest]map[string]string

+}

+

+// Get returns all the labels for the given digest

+func (s *memoryLabelStore) Get(dgst digest.Digest) (map[string]string, error) {

+	s.mu.Lock()

+	labels := s.labels[dgst]

+	s.mu.Unlock()

+	return labels, nil

+}

+

+// Set sets all the labels for a given digest

+func (s *memoryLabelStore) Set(dgst digest.Digest, labels map[string]string) error {

+	s.mu.Lock()

+	if s.labels == nil {

+		s.labels = make(map[digest.Digest]map[string]string)

+	}

+	s.labels[dgst] = labels

+	s.mu.Unlock()

+	return nil

+}

+

+// Update replaces the given labels for a digest,

+// a key with an empty value removes a label.

+func (s *memoryLabelStore) Update(dgst digest.Digest, update map[string]string) (map[string]string, error) {

+	s.mu.Lock()

+	defer s.mu.Unlock()

+

+	labels, ok := s.labels[dgst]

+	if !ok {

+		labels = map[string]string{}

+	}

+	for k, v := range update {

+		labels[k] = v

+	}

+

+	s.labels[dgst] = labels

+

+	return labels, nil

+}

+

+type testingContentStoreWrapper struct {

+	ContentStore

+	errorOnWriter error

+	errorOnCommit error

+}

+

+func (s *testingContentStoreWrapper) Writer(ctx context.Context, opts ...content.WriterOpt) (content.Writer, error) {

+	if s.errorOnWriter != nil {

+		return nil, s.errorOnWriter

+	}

+

+	w, err := s.ContentStore.Writer(ctx, opts...)

+	if err != nil {

+		return nil, err

+	}

+

+	if s.errorOnCommit != nil {

+		w = &testingContentWriterWrapper{w, s.errorOnCommit}

+	}

+	return w, nil

+}

+

+type testingContentWriterWrapper struct {

+	content.Writer

+	err error

+}

+

+func (w *testingContentWriterWrapper) Commit(ctx context.Context, size int64, dgst digest.Digest, opts ...content.Opt) error {

+	if w.err != nil {

+		// The contract for `Commit` is to always close.

+		// Since this is returning early before hitting the real `Commit`, we should close it here.

+		w.Close()

+		return w.err

+	}

+	return w.Writer.Commit(ctx, size, dgst, opts...)

+}

+

+func TestManifestStore(t *testing.T) {

+	ociManifest := &specs.Manifest{}

+	serialized, err := json.Marshal(ociManifest)

+	assert.NilError(t, err)