core/unpack: fetch layers of every config-sharing manifest

An index can carry more than one manifest that shares a config. A config lists uncompressed layer digests, so packaging the same layers with different compression (which `podman manifest push --add-compression` does) produces manifests with identical configs and the same config digest. Right now, when the unpacker visits a manifest, it stores its layers in a map keyed by its config's digest, so they can be unpacked later once the config descriptor is visited. Because that map holds only one manifest's layers per digest, when more than one manifest shares a config, only the last visited manifest's layers are stored (and unpacked), and the others' layers are never fetched. Image pulls still report success, but a later export or push of the image fails with NotFound. Fix this by changing the map to store the layers for all manifests pointing at a specific config (instead of just one manifest). This means manifests sharing the same config result in more than one set of layers being stored for the same config digest. Because manifests that share a config unpack to the same snapshot chain, we only need to unpack the first manifest's layers, and fetch the others (so their blobs reach the content store). Unpack skips fetching a layer whose snapshot already exists, so once a second manifest shares the config, the first manifest's layers are also fetched explicitly; a config named by a single manifest keeps that skip. Note that for indexes with more than one config-sharing manifest, this will now result in additional download time and storage for those layers. Don't store anything for a manifest with no layers; an empty slice reaching the unpack would fail the pull. Assisted-by: Claude Code Signed-off-by: Chris Henzie <chrishenzie@gmail.com>
2026-11-06 05:17:02 +00:00 · 2026-09-08 13:28:18 -07:00
1 parent 54a25ed633
commit 3c5d9fefd8
@@ -193,8 +193,18 @@ func NewUnpacker(ctx context.Context, cs content.Store, opts ...UnpackerOpt) (*U
 // process will be started in a goroutine.
 func (u *Unpacker) Unpack(h images.Handler) images.Handler {
 	var (
-		lock   sync.Mutex
-		layers = map[digest.Digest][]ocispec.Descriptor{}
+		lock sync.Mutex
+		// Maps a config's digest to the layer descriptors of each manifest
+		// that names it, one slice per manifest. Layers cannot be unpacked
+		// until the diffIDs are known from the config. Manifests can share a
+		// config (compression variants of one image have identical diffIDs
+		// and so identical configs), which is why one digest can hold the
+		// layers of several manifests.
+		queuedLayers = map[digest.Digest][][]ocispec.Descriptor{}
+		// Maps a config's digest to the layer descriptors selected for unpack.
+		// Once these layers are queued for fetch, the value is set to nil. The key
+		// remains to prevent scheduling another unpack.
+		unpackedLayers = map[digest.Digest][]ocispec.Descriptor{}
 	)

 	var layerTypes map[string]bool
@@ -248,20 +258,53 @@ func (u *Unpacker) Unpack(h images.Handler) images.Handler {
 				}
 			}

+			if len(manifestLayers) == 0 {
+				return nonLayers, nil
+			}
+
 			lock.Lock()
 			for _, nl := range nonLayers {
-				layers[nl.Digest] = manifestLayers
+				if images.IsConfigType(nl.MediaType) || configTypes[nl.MediaType] {
+					queuedLayers[nl.Digest] = append(queuedLayers[nl.Digest], manifestLayers)
+				}
 			}
 			lock.Unlock()

 			children = nonLayers
 		} else if images.IsConfigType(desc.MediaType) || configTypes[desc.MediaType] {
 			lock.Lock()
-			l := layers[desc.Digest]
+			queued := queuedLayers[desc.Digest]
+			delete(queuedLayers, desc.Digest)
+			// Because manifests that share the same config unpack to the same
+			// snapshot chain, we only need to unpack the layers for one
+			// manifest. The layers for any remaining manifests sharing that
+			// config are still fetched to ensure they make it in the content
+			// store.
+			var first []ocispec.Descriptor
+			unpacked, ok := unpackedLayers[desc.Digest]
+			if !ok && len(queued) > 0 {
+				first, queued = queued[0], queued[1:]
+				unpacked = first
+				unpackedLayers[desc.Digest] = first
+			}
+			// Unpack may skip fetching layers whose snapshots already exist.
+			// If another manifest shares this config, explicitly fetch the first
+			// manifest's layers too so every manifest's blobs reach the content
+			// store. For a config used by only one manifest, leave fetching to
+			// unpack.
+			if len(queued) > 0 && len(unpacked) > 0 {
+				queued = append(queued, unpacked)
+				unpackedLayers[desc.Digest] = nil
+			}
 			lock.Unlock()
-			if len(l) > 0 {
+			if len(first) > 0 {
 				u.eg.Go(func() error {
-					return u.unpack(h, desc, l)
+					return u.unpack(h, desc, first)
+				})
+			}
+			for _, layers := range queued {
+				u.eg.Go(func() error {
+					return u.fetch(u.ctx, h, layers, nil)
 				})
 			}
 		}
@@ -19,9 +19,12 @@ package unpack
 import (
 	"context"
 	"crypto/rand"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"reflect"
+	"slices"
+	"sync"
 	"testing"

 	"github.com/opencontainers/go-digest"
@@ -35,7 +38,11 @@ import (
 	"github.com/containerd/containerd/v2/core/images"
 	"github.com/containerd/containerd/v2/core/images/imagetest"
 	"github.com/containerd/containerd/v2/core/mount"
+	"github.com/containerd/containerd/v2/core/remotes"
+	"github.com/containerd/containerd/v2/core/remotes/docker"
 	"github.com/containerd/containerd/v2/core/snapshots"
+	"github.com/containerd/containerd/v2/pkg/testutil"
+	"github.com/containerd/errdefs"
 	"github.com/containerd/platforms"
 )

@@ -258,11 +265,11 @@ func (s *stagedSnapshotter) Commit(_ context.Context, name, key string, opts ...
 	return nil
 }

-// failApplier fails the test if Apply is called; a staged layer is never applied.
+// failApplier fails the test if Apply is called.
 type failApplier struct{ t *testing.T }

 func (a failApplier) Apply(_ context.Context, desc ocispec.Descriptor, _ []mount.Mount, _ ...diff.ApplyOpt) (ocispec.Descriptor, error) {
-	a.t.Errorf("Apply must not be called for a staged layer (%s)", desc.Digest)
+	a.t.Errorf("Apply must not be called for layer %s", desc.Digest)
 	return ocispec.Descriptor{}, nil
 }

@@ -399,3 +406,213 @@ func TestUnpackParallelPrepareError(t *testing.T) {
 	require.Len(t, sn.commits, 1)
 	assert.Equal(t, chainIDs[0].String(), sn.commits[0].name)
 }
+
+func layerDesc(id string) ocispec.Descriptor {
+	return ocispec.Descriptor{MediaType: ocispec.MediaTypeImageLayerGzip, Digest: digest.FromString(id), Size: 1}
+}
+
+func manifestDesc(id string) ocispec.Descriptor {
+	return ocispec.Descriptor{MediaType: ocispec.MediaTypeImageManifest, Digest: digest.FromString(id), Size: 1}
+}
+
+// alreadyExistsSnapshotter answers every Prepare with AlreadyExists and finds
+// the chain in Stat, so unpack skips each layer without fetching it.
+type alreadyExistsSnapshotter struct{ snapshots.Snapshotter }
+
+func (alreadyExistsSnapshotter) Prepare(context.Context, string, string, ...snapshots.Opt) ([]mount.Mount, error) {
+	return nil, errdefs.ErrAlreadyExists
+}
+
+func (alreadyExistsSnapshotter) Stat(_ context.Context, key string) (snapshots.Info, error) {
+	return snapshots.Info{Name: key}, nil
+}
+
+// TestUnpackConfigSharingManifests verifies that when several manifests share a
+// config, every manifest's layers are fetched even when the snapshots exist.
+// Only one unpack is scheduled per config. A lone manifest skips fetching
+// layers whose snapshots exist, and a manifest without layers is ignored.
+func TestUnpackConfigSharingManifests(t *testing.T) {
+	ctx := context.Background()
+	cs := imagetest.NewContentStore(ctx, t)
+
+	config := cs.JSONObject(ocispec.MediaTypeImageConfig, ocispec.Image{
+		Platform: ocispec.Platform{OS: "linux", Architecture: "amd64"},
+		RootFS:   ocispec.RootFS{Type: "layers", DiffIDs: []digest.Digest{digest.FromString("diff-0"), digest.FromString("diff-1")}},
+	}).Descriptor
+	manifestA, manifestB := manifestDesc("manifest-a"), manifestDesc("manifest-b")
+	layersA := []ocispec.Descriptor{layerDesc("a-0"), layerDesc("a-1")}
+	layersB := []ocispec.Descriptor{layerDesc("b-0"), layerDesc("b-1")}
+	sharedConfig := map[digest.Digest][]ocispec.Descriptor{
+		manifestA.Digest: append([]ocispec.Descriptor{config}, layersA...),
+		manifestB.Digest: append([]ocispec.Descriptor{config}, layersB...),
+	}
+
+	for _, tc := range []struct {
+		name        string
+		children    map[digest.Digest][]ocispec.Descriptor
+		order       []ocispec.Descriptor
+		wantFetched []ocispec.Descriptor
+		wantUnpacks int
+	}{
+		{
+			name:        "config after both manifests",
+			children:    sharedConfig,
+			order:       []ocispec.Descriptor{manifestA, manifestB, config, config},
+			wantFetched: slices.Concat([]ocispec.Descriptor{manifestA, manifestB, config}, layersA, layersB),
+			wantUnpacks: 1,
+		},
+		{
+			name:        "config between the manifests",
+			children:    sharedConfig,
+			order:       []ocispec.Descriptor{manifestA, config, manifestB, config},
+			wantFetched: slices.Concat([]ocispec.Descriptor{manifestA, manifestB, config}, layersA, layersB),
+			wantUnpacks: 1,
+		},
+		{
+			name:        "lone manifest with existing snapshots",
+			children:    sharedConfig,
+			order:       []ocispec.Descriptor{manifestA, config},
+			wantFetched: []ocispec.Descriptor{manifestA, config},
+			wantUnpacks: 1,
+		},
+		{
+			name:        "manifest without layers",
+			children:    map[digest.Digest][]ocispec.Descriptor{manifestA.Digest: {config}},
+			order:       []ocispec.Descriptor{manifestA, config},
+			wantFetched: []ocispec.Descriptor{manifestA, config},
+			wantUnpacks: 0,
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			var mu sync.Mutex
+			fetched := map[digest.Digest]struct{}{}
+			h := images.HandlerFunc(func(ctx context.Context, desc ocispec.Descriptor) ([]ocispec.Descriptor, error) {
+				mu.Lock()
+				fetched[desc.Digest] = struct{}{}
+				mu.Unlock()
+				return tc.children[desc.Digest], nil
+			})
+
+			u, err := NewUnpacker(ctx, cs.Store, WithUnpackPlatform(Platform{
+				Snapshotter: alreadyExistsSnapshotter{},
+				Applier:     failApplier{t},
+			}))
+			require.NoError(t, err)
+
+			wrapped := u.Unpack(h)
+			for _, desc := range tc.order {
+				_, err := wrapped.Handle(ctx, desc)
+				require.NoError(t, err)
+			}
+			res, err := u.Wait()
+			require.NoError(t, err)
+
+			want := map[digest.Digest]struct{}{}
+			for _, desc := range tc.wantFetched {
+				want[desc.Digest] = struct{}{}
+			}
+			assert.Equal(t, want, fetched)
+			assert.Equal(t, tc.wantUnpacks, res.Unpacks)
+		})
+	}
+}
+
+// TestPullFetchesLayersOfEveryConfigSharingManifest verifies that a pull
+// through the real handler chain, from a local registry, fetches the layers
+// of every config-sharing manifest into the content store.
+func TestPullFetchesLayersOfEveryConfigSharingManifest(t *testing.T) {
+	ctx := context.Background()
+
+	imagePlatform := ocispec.Platform{OS: "linux", Architecture: "amd64"}
+
+	// One config, shared by both manifests. Both can share it because a config
+	// records each layer's uncompressed digest (diffID), which compression does
+	// not change.
+	config := mustMarshal(t, ocispec.Image{
+		Platform: imagePlatform,
+		RootFS:   ocispec.RootFS{Type: "layers", DiffIDs: []digest.Digest{digest.FromString("diff-0")}},
+	})
+	configDesc := blobDesc(ocispec.MediaTypeImageConfig, config)
+
+	layerA := []byte("layer-a-compressed-bytes")
+	layerB := []byte("layer-b-compressed-bytes")
+	layerADesc := blobDesc(ocispec.MediaTypeImageLayerGzip, layerA)
+	layerBDesc := blobDesc(ocispec.MediaTypeImageLayerGzip, layerB)
+
+	manifestA := mustMarshal(t, ocispec.Manifest{
+		MediaType: ocispec.MediaTypeImageManifest,
+		Config:    configDesc,
+		Layers:    []ocispec.Descriptor{layerADesc},
+	})
+	manifestB := mustMarshal(t, ocispec.Manifest{
+		MediaType: ocispec.MediaTypeImageManifest,
+		Config:    configDesc,
+		Layers:    []ocispec.Descriptor{layerBDesc},
+	})
+	manifestADesc := blobDesc(ocispec.MediaTypeImageManifest, manifestA)
+	manifestADesc.Platform = &imagePlatform
+	manifestBDesc := blobDesc(ocispec.MediaTypeImageManifest, manifestB)
+	manifestBDesc.Platform = &imagePlatform
+
+	index := mustMarshal(t, ocispec.Index{
+		MediaType: ocispec.MediaTypeImageIndex,
+		Manifests: []ocispec.Descriptor{manifestADesc, manifestBDesc},
+	})
+	indexDesc := blobDesc(ocispec.MediaTypeImageIndex, index)
+
+	ref := testutil.ServeImage(t, "img", "latest", indexDesc, map[digest.Digest][]byte{
+		indexDesc.Digest:     index,
+		manifestADesc.Digest: manifestA,
+		manifestBDesc.Digest: manifestB,
+		configDesc.Digest:    config,
+		layerADesc.Digest:    layerA,
+		layerBDesc.Digest:    layerB,
+	})
+
+	store := imagetest.NewContentStore(ctx, t).Store
+
+	// NewResolver uses plain HTTP for localhost, where the registry listens,
+	// so no hosts configuration is needed.
+	resolver := docker.NewResolver(docker.ResolverOptions{})
+	_, resolved, err := resolver.Resolve(ctx, ref)
+	require.NoError(t, err)
+	fetcher, err := resolver.Fetcher(ctx, ref)
+	require.NoError(t, err)
+
+	// Assemble the same handler chain a transfer-service pull uses.
+	children := images.FilterPlatforms(images.ChildrenHandler(store), platforms.Only(imagePlatform))
+	handler := images.Handlers(remotes.FetchHandler(store, fetcher), children)
+
+	u, err := NewUnpacker(ctx, store, WithUnpackPlatform(Platform{
+		// No image matches, so unpack fetches every layer instead of applying
+		// it, and no real snapshotter is needed.
+		Platform:    platforms.OnlyStrict(platforms.MustParse("linux/arm64")),
+		Snapshotter: alreadyExistsSnapshotter{},
+		Applier:     failApplier{t},
+	}))
+	require.NoError(t, err)
+
+	require.NoError(t, images.Dispatch(ctx, u.Unpack(handler), nil, resolved))
+	_, err = u.Wait()
+	require.NoError(t, err)
+
+	_, err = store.Info(ctx, layerADesc.Digest)
+	require.NoError(t, err, "layer of the first config-sharing manifest should be present, but is not")
+	_, err = store.Info(ctx, layerBDesc.Digest)
+	require.NoError(t, err, "layer of the second config-sharing manifest should be present, but is not")
+}
+
+func mustMarshal(t *testing.T, v any) []byte {
+	t.Helper()
+	b, err := json.Marshal(v)
+	require.NoError(t, err)
+	return b
+}
+
+func blobDesc(mediaType string, body []byte) ocispec.Descriptor {
+	return ocispec.Descriptor{
+		MediaType: mediaType,
+		Digest:    digest.FromBytes(body),
+		Size:      int64(len(body)),
+	}
+}