erofs: record where a layer blob is

A layer served from the layer content cache was symlinked into the
snapshot as its layer.erofs. That put a path inside the snapshot
directory aliasing a blob owned by the operator and shared with every
other snapshot of that layer. Everything written to that path went
straight into the cache entry. Each write site then needed its own
guard: the differ had to refuse to apply, Commit had to refuse to
convert through a dangling link, and Remove had to avoid clearing
IMMUTABLE_FL on a file it did not own. A missed guard corrupted content
shared across images. A pruned entry left a dangling link that read as
an absent blob on one path and as populated content on another.

Record where the blob is on the snapshot and mount it from there.
Nothing in the snapshot directory refers to content it does not own. A
stray write creates a local file, conversion cannot reach the cache, and
removal has nothing to follow. The guards Commit and Remove needed for a
symlink are gone. The differ still refuses, on the read-only mounts the
snapshotter hands out.

resolveBlob is the one place a blob is located. It reports whether the
blob is already populated, meaning there is nothing to apply or convert,
and whether it belongs to this snapshot, meaning it may be written.
Mounts are then assembled without consulting the filesystem to find out
what a layer is. A source that does not resolve is an error on every
path, including the parentless path in mounts(). A flattened parent's
device= is resolved the same way.

fsverity is only measured for a blob the snapshot owns. Measuring a
cache entry would require the operator to enable fsverity on content
this snapshotter never wrote. NewSnapshotter already rejects fsverity
together with the layer content cache.

The record is kept in labels outside the containerd.io/snapshot/
namespace. Labels in that namespace are inherited from image
annotations, which would let an image name any path on the host. The
same filter runs on everything the metadata snapshotter passes down. A
client cannot set these labels, and cannot send them back after reading
them through Stat. Commit and Update carry them across.

The dm-verity sidecar is read from beside the cache entry. Nothing is
copied into the snapshot. The sidecar cannot fall out of step with the
blob it pins. An entry required to have one is rejected at Prepare.

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

internal/erofsutils/mount_test.go

@@ -40,19 +40,18 @@ func TestCacheBlobPath(t *testing.T) {
 	assert.Equal(t, enc[:2], filepath.Base(filepath.Dir(got)), "blob must live under a 2-char prefix dir")
 }
 
-// TestMountsToLayerStagedBlob covers resolving the layer directory of a
-// snapshot whose blob is staged. Compare only reads the blob, so it resolves
-// like any other; the apply path is where a staged blob is refused.
-func TestMountsToLayerStagedBlob(t *testing.T) {
+// TestMountsToLayerErofsMount covers resolving the layer directory from an
+// erofs layer mount. A snapshot holding a single committed layer hands those
+// out.
+func TestMountsToLayerErofsMount(t *testing.T) {
 	layer := t.TempDir()
 	require.NoError(t, os.WriteFile(filepath.Join(layer, ".erofslayer"), nil, 0644))
-	cached := filepath.Join(t.TempDir(), "cached.erofs")
-	require.NoError(t, os.WriteFile(cached, []byte("shared blob"), 0644))
-	require.NoError(t, os.Symlink(cached, filepath.Join(layer, "layer.erofs")))
+	blob := filepath.Join(layer, "layer.erofs")
+	require.NoError(t, os.WriteFile(blob, []byte("layer"), 0644))
 
 	got, err := MountsToLayer([]mount.Mount{{
 		Type:    "erofs",
-		Source:  filepath.Join(layer, "layer.erofs"),
+		Source:  blob,
 		Options: []string{"ro"},
 	}})
 	require.NoError(t, err)

plugins/snapshots/erofs/blobsource.go

@@ -0,0 +1,191 @@
+/*
+   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 erofs
+
+import (
+	"errors"
+	"fmt"
+	"os"
+	"strings"
+
+	"github.com/containerd/errdefs"
+
+	"github.com/containerd/containerd/v2/core/snapshots"
+)
+
+// labelPrefix namespaces the labels the snapshotter records about a snapshot
+// for its own use.
+//
+// The prefix sits outside "containerd.io/snapshot/", whose labels are inherited
+// from image annotations (see snapshots.FilterInheritedLabels): a manifest could
+// otherwise claim its layer lives at any path on the host. The same filter runs
+// on everything the metadata snapshotter passes down. A client cannot set or
+// clear these labels either. They stay readable: Stat and Walk report the
+// backend's labels alongside the metadata store's.
+const labelPrefix = "io.containerd.erofs.v1/"
+
+const (
+	// blobSourceKindLabel records where a snapshot's layer blob comes from.
+	blobSourceKindLabel = labelPrefix + "blob.source"
+	// blobSourceRefLabel records which blob, in the form the kind defines. A
+	// cache ref is an absolute path.
+	blobSourceRefLabel = labelPrefix + "blob.ref"
+)
+
+// blobSourceKind identifies where a snapshot's layer blob comes from, which
+// determines what may be done with it. How a layer is composed into a mount does
+// not depend on its kind.
+type blobSourceKind string
+
+const (
+	// blobSourceLocal is a blob stored in the snapshot directory itself,
+	// written there by a differ. A snapshot that records no source has one.
+	blobSourceLocal blobSourceKind = ""
+
+	// blobSourceCache is a blob in an operator-owned layer content cache,
+	// shared with every other snapshot of that layer. It is pre-converted, so
+	// it is complete before anything is applied. It must never be written to.
+	blobSourceCache blobSourceKind = "cache"
+)
+
+// blobSource records where a snapshot's layer blob comes from. Ref is
+// kind-specific and empty for a local blob; for a cached one it is the absolute
+// path of the cache entry, which is mounted directly.
+type blobSource struct {
+	Kind blobSourceKind
+	Ref  string
+}
+
+// populated reports whether the blob is already complete when the snapshot is
+// prepared. Nothing is fetched or applied into such a snapshot. Committing it
+// does not convert anything.
+func (b blobSource) populated() bool {
+	return b.Kind != blobSourceLocal
+}
+
+// owned reports whether the blob belongs to this snapshot alone. Such a blob may
+// be written, have its attributes changed, and be removed with the snapshot. A
+// blob from any other source is shared and must be left untouched.
+func (b blobSource) owned() bool {
+	return b.Kind == blobSourceLocal
+}
+
+// labels returns the labels recording this source. A local blob does not record
+// one. It returns nil for such a blob.
+func (b blobSource) labels() map[string]string {
+	if !b.populated() {
+		return nil
+	}
+	return map[string]string{
+		blobSourceKindLabel: string(b.Kind),
+		blobSourceRefLabel:  b.Ref,
+	}
+}
+
+// blobSourceFromInfo returns the blob source recorded on a snapshot, or a local
+// one if the snapshot does not record a source.
+func blobSourceFromInfo(info snapshots.Info) (blobSource, error) {
+	src := blobSource{
+		Kind: blobSourceKind(info.Labels[blobSourceKindLabel]),
+		Ref:  info.Labels[blobSourceRefLabel],
+	}
+	switch src.Kind {
+	case blobSourceLocal, blobSourceCache:
+		// A local blob does not name a ref. Every other kind names one.
+		if (src.Ref == "") != (src.Kind == blobSourceLocal) {
+			return blobSource{}, fmt.Errorf("snapshot %q records a %q layer blob source with ref %q", info.Name, src.Kind, src.Ref)
+		}
+		if src.Kind == blobSourceLocal {
+			return blobSource{}, nil
+		}
+		return src, nil
+	default:
+		// A newer version of the snapshotter recorded a source this one does
+		// not implement. Mounting the blob would apply rules this version does
+		// not have.
+		return blobSource{}, fmt.Errorf("snapshot %q records an unknown layer blob source %q", info.Name, src.Kind)
+	}
+}
+
+// privateLabels returns the subset of labels the snapshotter records for
+// itself.
+func privateLabels(labels map[string]string) map[string]string {
+	var private map[string]string
+	for k, v := range labels {
+		if strings.HasPrefix(k, labelPrefix) {
+			if private == nil {
+				private = make(map[string]string, 2)
+			}
+			private[k] = v
+		}
+	}
+	return private
+}
+
+// errNoLayerBlob is returned by resolveBlob for a snapshot that does not hold a
+// layer blob at all, the normal state of an active snapshot before a differ has
+// applied anything into it. A blob that is recorded but unusable returns a
+// different error. errNoLayerBlob wraps os.ErrNotExist for a caller that only
+// needs to know the blob is absent.
+var errNoLayerBlob = fmt.Errorf("no erofs layer blob: %w", os.ErrNotExist)
+
+// resolveBlob returns the path of a snapshot's layer blob together with where
+// it came from. A local blob lives in the snapshot directory. Any other is
+// mounted from its source. Nothing in the snapshot aliases content it does not
+// own.
+//
+// A recorded blob that does not resolve is an error, not a miss. Reporting it
+// as absent would let the caller treat the layer as unapplied. Commit would
+// then convert into a path that does not exist and commit an empty layer.
+func (s *snapshotter) resolveBlob(id string, info snapshots.Info) (string, blobSource, error) {
+	src, err := blobSourceFromInfo(info)
+	if err != nil {
+		return "", blobSource{}, err
+	}
+	if src.populated() {
+		// The ref names content this snapshot does not own. Stat resolves it as
+		// the source publishes it, links included. A ref that names a directory
+		// would otherwise pass as a complete layer: Commit measures it, records
+		// the snapshot, and the mount fails later when it turns out not to be a
+		// layer image.
+		fi, err := os.Stat(src.Ref)
+		if err != nil {
+			return "", src, fmt.Errorf("layer blob %q from the %s is unusable: %w", src.Ref, src.Kind, err)
+		}
+		if !fi.Mode().IsRegular() {
+			return "", src, fmt.Errorf("layer blob %q from the %s is not a regular file: %w", src.Ref, src.Kind, errdefs.ErrFailedPrecondition)
+		}
+		return src.Ref, src, nil
+	}
+
+	// A local blob is a regular file this snapshotter wrote into the snapshot
+	// directory. Lstat answers without following a link. A link would hand back
+	// a path outside the directory as though the snapshot owned it. Conversion
+	// at Commit, fsverity, and IMMUTABLE_FL would then act on its target.
+	layerBlob := s.layerBlobPath(id)
+	fi, err := os.Lstat(layerBlob)
+	if err != nil {
+		if errors.Is(err, os.ErrNotExist) {
+			return "", src, fmt.Errorf("%s: %w", layerBlob, errNoLayerBlob)
+		}
+		return "", src, fmt.Errorf("failed to stat layer blob %q: %w", layerBlob, err)
+	}
+	if !fi.Mode().IsRegular() {
+		return "", src, fmt.Errorf("layer blob %q is not a regular file: %w", layerBlob, errdefs.ErrFailedPrecondition)
+	}
+	return layerBlob, src, nil
+}

plugins/snapshots/erofs/blobsource_linux_test.go

@@ -0,0 +1,205 @@
+/*
+   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 erofs
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/containerd/errdefs"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/containerd/containerd/v2/core/snapshots"
+	"github.com/containerd/containerd/v2/core/snapshots/storage"
+)
+
+func TestBlobSourceRoundTrip(t *testing.T) {
+	// No labels means the blob is the snapshot's own.
+	src, err := blobSourceFromInfo(snapshots.Info{Name: "local"})
+	require.NoError(t, err)
+	assert.False(t, src.populated(), "a local blob is applied into")
+	assert.True(t, src.owned(), "a local blob is the snapshot's to write and remove")
+	assert.Nil(t, src.labels(), "a local blob does not record a source")
+
+	want := blobSource{Kind: blobSourceCache, Ref: "/cache/sha256/ab/abcd.erofs"}
+	src, err = blobSourceFromInfo(snapshots.Info{Name: "cached", Labels: want.labels()})
+	require.NoError(t, err)
+	assert.Equal(t, want, src)
+	assert.True(t, src.populated())
+	assert.False(t, src.owned(), "a blob from another source is shared")
+}
+
+func TestBlobSourceMalformed(t *testing.T) {
+	for _, tc := range []struct {
+		name   string
+		labels map[string]string
+	}{
+		{"ref without a source", map[string]string{blobSourceRefLabel: "/cache/blob.erofs"}},
+		{"source without a ref", map[string]string{blobSourceKindLabel: "cache"}},
+		{"unknown source", map[string]string{blobSourceKindLabel: "wormhole", blobSourceRefLabel: "/x"}},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			// A record that cannot be understood must not read as "no record"
+			// (see errNoLayerBlob).
+			_, err := blobSourceFromInfo(snapshots.Info{Name: "s", Labels: tc.labels})
+			assert.Error(t, err)
+		})
+	}
+}
+
+// TestPrivateLabels covers picking out the labels the snapshotter keeps for
+// itself. Commit and Update carry them across. A committed snapshot still
+// records where its blob is.
+func TestPrivateLabels(t *testing.T) {
+	assert.Nil(t, privateLabels(nil))
+	assert.Nil(t, privateLabels(map[string]string{
+		"containerd.io/snapshot.ref":          "sha256:abcd",
+		"containerd.io/snapshot/diff-id":      "sha256:ef01",
+		"containerd.io/snapshot/erofs/sneaky": "/etc/shadow",
+	}), "only the snapshotter's own namespace is private")
+
+	assert.Equal(t, map[string]string{
+		blobSourceKindLabel: "cache",
+		blobSourceRefLabel:  "/cache/blob.erofs",
+	}, privateLabels(map[string]string{
+		"containerd.io/snapshot.ref": "sha256:abcd",
+		blobSourceKindLabel:          "cache",
+		blobSourceRefLabel:           "/cache/blob.erofs",
+	}))
+}
+
+// TestBlobSourceLabelIsNotInherited covers the reason the snapshotter's labels
+// sit outside the "containerd.io/snapshot/" namespace (see labelPrefix).
+func TestBlobSourceLabelIsNotInherited(t *testing.T) {
+	annotations := map[string]string{
+		blobSourceKindLabel: "cache",
+		blobSourceRefLabel:  "/etc/shadow",
+	}
+	assert.Empty(t, snapshots.FilterInheritedLabels(annotations),
+		"an image must not be able to set the snapshotter's own labels")
+}
+
+func TestResolveBlob(t *testing.T) {
+	s := &snapshotter{root: t.TempDir()}
+	mkSnapshot := func(t *testing.T, id string) {
+		t.Helper()
+		require.NoError(t, os.MkdirAll(filepath.Dir(s.layerBlobPath(id)), 0755))
+	}
+
+	t.Run("no blob", func(t *testing.T) {
+		mkSnapshot(t, "empty")
+		_, _, err := s.resolveBlob("empty", snapshots.Info{Name: "empty"})
+		assert.ErrorIs(t, err, errNoLayerBlob,
+			"an unapplied snapshot must be distinguishable from a broken one")
+		assert.ErrorIs(t, err, os.ErrNotExist,
+			"a caller that only needs to know the blob is absent still sees that")
+	})
+
+	// A local blob is a regular file this snapshotter wrote. Resolving a link
+	// would hand back a path outside the snapshot as though the snapshot owned
+	// it. Commit would then convert into, or set attributes on, whatever it
+	// points at. It must not read as an absent blob either, which is what makes
+	// Commit convert one in its place.
+	t.Run("linked blob is refused", func(t *testing.T) {
+		mkSnapshot(t, "linked")
+		target := filepath.Join(t.TempDir(), "elsewhere.erofs")
+		require.NoError(t, os.WriteFile(target, []byte("not ours"), 0644))
+		require.NoError(t, os.Symlink(target, s.layerBlobPath("linked")))
+
+		_, _, err := s.resolveBlob("linked", snapshots.Info{Name: "linked"})
+		require.Error(t, err)
+		assert.NotErrorIs(t, err, errNoLayerBlob, "a link must not pass as an unapplied layer")
+		assert.ErrorIs(t, err, errdefs.ErrFailedPrecondition)
+	})
+
+	t.Run("local blob", func(t *testing.T) {
+		mkSnapshot(t, "local")
+		require.NoError(t, os.WriteFile(s.layerBlobPath("local"), []byte("layer"), 0644))
+
+		path, src, err := s.resolveBlob("local", snapshots.Info{Name: "local"})
+		require.NoError(t, err)
+		assert.Equal(t, s.layerBlobPath("local"), path)
+		assert.True(t, src.owned())
+	})
+
+	t.Run("recorded blob", func(t *testing.T) {
+		mkSnapshot(t, "cached")
+		blob := filepath.Join(t.TempDir(), "cached.erofs")
+		require.NoError(t, os.WriteFile(blob, []byte("layer"), 0644))
+
+		info := snapshots.Info{Name: "cached", Labels: blobSource{Kind: blobSourceCache, Ref: blob}.labels()}
+		path, src, err := s.resolveBlob("cached", info)
+		require.NoError(t, err)
+		assert.Equal(t, blob, path, "the cache entry is used in place")
+		assert.False(t, src.owned())
+	})
+
+	// A ref that resolves to something other than a file is not a layer. It
+	// must not pass as complete: Commit would measure it and record the
+	// snapshot, and the mount would fail later.
+	t.Run("recorded blob that is not a file", func(t *testing.T) {
+		mkSnapshot(t, "dir")
+		blobDir := filepath.Join(t.TempDir(), "notablob.erofs")
+		require.NoError(t, os.MkdirAll(blobDir, 0755))
+
+		info := snapshots.Info{Name: "dir", Labels: blobSource{Kind: blobSourceCache, Ref: blobDir}.labels()}
+		_, _, err := s.resolveBlob("dir", info)
+		require.Error(t, err)
+		assert.ErrorIs(t, err, errdefs.ErrFailedPrecondition)
+	})
+
+	t.Run("recorded blob that is gone", func(t *testing.T) {
+		mkSnapshot(t, "pruned")
+		blob := filepath.Join(t.TempDir(), "pruned.erofs")
+
+		info := snapshots.Info{Name: "pruned", Labels: blobSource{Kind: blobSourceCache, Ref: blob}.labels()}
+		_, _, err := s.resolveBlob("pruned", info)
+		require.Error(t, err)
+		assert.ErrorIs(t, err, os.ErrNotExist)
+		assert.NotErrorIs(t, err, errNoLayerBlob,
+			"a blob that was recorded and then pruned is broken, not unapplied")
+	})
+}
+
+// TestFsverityOnlyForAnOwnedBlob covers which blob the fsverity check applies
+// to. Commit only enables fsverity on a blob the snapshot owns. Only such a
+// blob can be measured. Measuring a cache entry would require the operator to
+// have enabled fsverity on content this snapshotter never wrote.
+// NewSnapshotter rejects the two together; this is a guard, not a reachable
+// configuration.
+func TestFsverityOnlyForAnOwnedBlob(t *testing.T) {
+	s := &snapshotter{root: t.TempDir(), enableFsverity: true}
+	snap := storage.Snapshot{ID: "1", Kind: snapshots.KindActive}
+	require.NoError(t, os.MkdirAll(filepath.Dir(s.layerBlobPath(snap.ID)), 0755))
+
+	cached := filepath.Join(t.TempDir(), "cached.erofs")
+	require.NoError(t, os.WriteFile(cached, []byte("layer"), 0644))
+
+	info := snapshots.Info{Name: "cached", Labels: blobSource{Kind: blobSourceCache, Ref: cached}.labels()}
+	mounts, err := s.mounts(snap, info, nil)
+	require.NoError(t, err)
+	require.Len(t, mounts, 1)
+	assert.Equal(t, cached, mounts[0].Source, "the cache entry is mounted unmeasured")
+
+	// A blob the snapshot does own is measured. A plain file fails.
+	require.NoError(t, os.WriteFile(s.layerBlobPath(snap.ID), []byte("layer"), 0644))
+	_, err = s.mounts(snap, snapshots.Info{Name: "local"}, nil)
+	require.Error(t, err, "a local blob must still be measured")
+	assert.Contains(t, err.Error(), "fsverity")
+}

plugins/snapshots/erofs/erofs.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"maps"
 	"os"
 	"path/filepath"
 	"runtime"
@@ -56,10 +57,10 @@ type SnapshotterConfig struct {
 	// dmverityMode controls dm-verity behavior: "auto" (use if .dmverity exists), "on" (require .dmverity), "off" (disable)
 	dmverityMode string
 	// layerContentCaches lists directories of pre-converted, diffID-keyed erofs
-	// layer blobs. Each is checked one by one; the first hit is staged into the
-	// snapshot (symlinked) instead of downloading and converting the layer. A
-	// directory that doesn't exist is treated as a cache miss. Layers missing
-	// from all of them are converted normally.
+	// layer blobs. Each is checked one by one; the first hit is recorded as the
+	// snapshot's blob source and mounted from the cache. A directory that
+	// doesn't exist is treated as a cache miss. Layers missing from all of them
+	// are converted normally.
 	layerContentCaches []string
 }
 
@@ -176,10 +177,10 @@ func NewSnapshotter(root string, opts ...Opt) (snapshots.Snapshotter, error) {
 		}
 	}
 
-	// A cache hit merely symlinks a shared, operator-owned blob into the snapshot,
-	// so fsverity and IMMUTABLE_FL can't be applied without mutating that blob out
-	// from under other snapshots. Reject them explicitly instead of silently
-	// skipping; dm-verity is the cache's integrity mechanism.
+	// A cache hit is served straight from a shared, operator-owned blob, so
+	// enabling fsverity or IMMUTABLE_FL would mutate a blob other snapshots are
+	// using. Both are rejected here; dm-verity is the cache's integrity
+	// mechanism.
 	if len(config.layerContentCaches) > 0 {
 		if config.enableFsverity {
 			return nil, fmt.Errorf("enable_fsverity is incompatible with layer_content_caches; use dm-verity for cache integrity")
@@ -188,10 +189,11 @@ func NewSnapshotter(root string, opts ...Opt) (snapshots.Snapshotter, error) {
 			return nil, fmt.Errorf("set_immutable is incompatible with layer_content_caches")
 		}
 
-		// A cache dir is symlinked into snapshots, so a relative one would resolve
-		// against the snapshot dir and dangle. The check is only lexical: dirs are
-		// not required to exist, as a missing one just yields a cache miss and may
-		// well be provisioned after startup.
+		// A cache dir is recorded in a snapshot and mounted later. A relative
+		// path would resolve against the working directory of whichever
+		// process performs the mount. The check is only lexical:
+		// dirs are not required to exist, as a missing one just yields a cache
+		// miss and may well be provisioned after startup.
 		for _, dir := range config.layerContentCaches {
 			if !filepath.IsAbs(dir) {
 				return nil, fmt.Errorf("layer_content_caches %q must be an absolute path", dir)
@@ -285,16 +287,31 @@ func (s *snapshotter) fsMetaPath(id string) string {
 	return filepath.Join(s.root, "snapshots", id, "fsmeta.erofs")
 }
 
-func (s *snapshotter) lowerPath(id string) (string, error) {
-	layerBlob := s.layerBlobPath(id)
-	if _, err := os.Stat(layerBlob); err != nil {
-		return "", fmt.Errorf("failed to find valid erofs layer blob: %w", err)
-	}
-
-	return layerBlob, nil
+// lowerPath returns the path to mount a snapshot's layer blob from, wherever it
+// is stored (see resolveBlob).
+func (s *snapshotter) lowerPath(id string, info snapshots.Info) (string, error) {
+	path, _, err := s.resolveBlob(id, info)
+	return path, err
 }
 
-func (s *snapshotter) prepareDirectory(ctx context.Context, snapshotDir string, kind snapshots.Kind, cacheBlob string) (string, error) {
+// parentInfos returns the info of every ancestor of the snapshot described by
+// info, nearest first, in the same order as ParentIDs on the corresponding
+// storage.Snapshot. Mounts are assembled outside the metadata transaction. A
+// caller reads this inside the transaction that read the snapshot.
+func parentInfos(ctx context.Context, info snapshots.Info) ([]snapshots.Info, error) {
+	var parents []snapshots.Info
+	for name := info.Parent; name != ""; {
+		_, pi, _, err := storage.GetInfo(ctx, name)
+		if err != nil {
+			return nil, fmt.Errorf("failed to get parent snapshot %q info: %w", name, err)
+		}
+		parents = append(parents, pi)
+		name = pi.Parent
+	}
+	return parents, nil
+}
+
+func (s *snapshotter) prepareDirectory(ctx context.Context, snapshotDir string, kind snapshots.Kind) (string, error) {
 	td, err := os.MkdirTemp(snapshotDir, snapshotTempDirPrefix)
 	if err != nil {
 		return "", fmt.Errorf("failed to create temp dir: %w", err)
@@ -316,33 +333,13 @@ func (s *snapshotter) prepareDirectory(ctx context.Context, snapshotDir string,
 		}
 	}
 
-	// Layer content cache hit: stage the pre-converted blob as a symlink into the
-	// active snapshot; the caller commits it later, once the parent is known.
-	if cacheBlob != "" {
-		layerBlob := filepath.Join(td, "layer.erofs")
-		if err := os.Symlink(cacheBlob, layerBlob); err != nil {
-			return td, fmt.Errorf("failed to symlink cached layer blob: %w", err)
-		}
-		// Copy the dm-verity sidecar alongside the blob (unless dm-verity is off,
-		// when it's never consumed) so mount-time metadata resolution and the
-		// pinned root hash match locally-converted layers. A missing sidecar is
-		// fine except with dmverity_mode "on", which requires it.
-		if s.dmverityMode != "off" {
-			if err := fs.CopyFile(dmverity.MetadataPath(layerBlob), dmverity.MetadataPath(cacheBlob)); err != nil {
-				if s.dmverityMode == "on" || !errors.Is(err, os.ErrNotExist) {
-					return td, fmt.Errorf("failed to copy dm-verity sidecar: %w", err)
-				}
-			}
-		}
-	}
-
 	return td, nil
 }
 
-func (s *snapshotter) mountFsMeta(snap storage.Snapshot, id int) (mount.Mount, bool) {
+func (s *snapshotter) mountFsMeta(snap storage.Snapshot, parents []snapshots.Info, id int) (mount.Mount, bool, error) {
 	mergedMeta := s.fsMetaPath(snap.ParentIDs[id])
 	if fi, err := os.Stat(mergedMeta); err != nil || fi.Size() == 0 {
-		return mount.Mount{}, false
+		return mount.Mount{}, false, nil
 	}
 
 	m := mount.Mount{
@@ -351,10 +348,15 @@ func (s *snapshotter) mountFsMeta(snap storage.Snapshot, id int) (mount.Mount, b
 		Options: []string{"ro", "loop"},
 	}
 	for j := len(snap.ParentIDs) - 1; j >= id; j-- {
-		path := s.layerBlobPath(snap.ParentIDs[j])
+		// The flattened layers are addressed by device. Each device path has to
+		// be resolved to where that layer's blob is stored.
+		path, _, err := s.resolveBlob(snap.ParentIDs[j], parents[j])
+		if err != nil {
+			return mount.Mount{}, false, err
+		}
 		m.Options = append(m.Options, "device="+path)
 	}
-	return m, true
+	return m, true, nil
 }
 
 // applyDmverityPolicy validates and applies dm-verity policy for a layer.
@@ -413,62 +415,160 @@ func (s *snapshotter) createErofsMount(layerBlob string) (mount.Mount, error) {
 	}, nil
 }
 
-// stagedBlob reports whether the layer blob at path was staged from the layer
-// content cache. A staged blob is a symlink into the operator-owned cache
-// directory; a blob a differ wrote is a regular file in the snapshot. A missing
-// blob is not staged. Any other stat failure is an error because callers use
-// this to decide whether writing to the blob is safe.
-func stagedBlob(path string) (bool, error) {
-	fi, err := os.Lstat(path)
-	if err != nil {
-		if errors.Is(err, os.ErrNotExist) {
-			return false, nil
+// parentMounts returns the lower mounts for snap's parents, top to bottom. A
+// merged fsmeta stands in for every layer below it. The walk stops there.
+func (s *snapshotter) parentMounts(snap storage.Snapshot, parents []snapshots.Info) ([]mount.Mount, error) {
+	if len(parents) != len(snap.ParentIDs) {
+		return nil, fmt.Errorf("got %d parent infos for %d parents of snapshot %s: %w",
+			len(parents), len(snap.ParentIDs), snap.ID, errdefs.ErrInvalidArgument)
+	}
+	mounts := make([]mount.Mount, 0, len(snap.ParentIDs))
+	for i := range snap.ParentIDs {
+		// If a merged fsmeta is valid for this layer, skip the remaining bottom layers.
+		// Why? Because bottom layers have been flattened with the thin fsmeta.
+		m, ok, err := s.mountFsMeta(snap, parents, i)
+		if err != nil {
+			return nil, err
 		}
-		return false, fmt.Errorf("failed to stat layer blob %q: %w", path, err)
+		if ok {
+			return append(mounts, m), nil
+		}
+
+		layerBlob, err := s.lowerPath(snap.ParentIDs[i], parents[i])
+		if err != nil {
+			return nil, err
+		}
+
+		if m, err = s.createErofsMount(layerBlob); err != nil {
+			return nil, fmt.Errorf("failed to create erofs mount for parent %s: %w", snap.ParentIDs[i], err)
+		}
+
+		mounts = append(mounts, m)
 	}
-	return fi.Mode()&os.ModeSymlink != 0, nil
+	return mounts, nil
 }
 
-// stagedLayerMounts returns the read-only mounts of the cache blob staged into
-// the active snapshot id, or none when the snapshot does not have a staged
-// blob. The caller stacks them on top of the parents. Nothing here verifies
-// fsverity: a staged blob is content the snapshot does not own, and fsverity is
-// rejected together with the cache (see NewSnapshotter).
-func (s *snapshotter) stagedLayerMounts(id string) ([]mount.Mount, error) {
-	// Just a fast path. The erofs differ refuses a symlinked layer blob whatever
-	// the configuration (see erofsutils.StagedLayerBlob), so a blob staged
-	// before caching was turned off stays safe from it.
-	if len(s.layerContentCaches) == 0 {
-		return nil, nil
+// overlayMounts stacks snap's own layer on top of the given parent lowers and
+// assembles the overlay mount over them.
+//
+// An active snapshot with a local blob contributes a writable upper for a
+// differ to apply its layer into. An active snapshot whose blob is already
+// populated (see blobSource) contributes the layer itself as the top lower and
+// no upperdir. A view contributes nothing. Both leave the overlay read-only. A
+// caller reads a read-only overlay as a layer with nothing left to apply.
+//
+// A populated blob is never measured for fsverity. fsverity applies to a blob
+// this snapshotter wrote. NewSnapshotter rejects fsverity together with the
+// layer content cache.
+func (s *snapshotter) overlayMounts(snap storage.Snapshot, info snapshots.Info, parents []mount.Mount) ([]mount.Mount, error) {
+	var (
+		// pre holds mounts that mount templating consumes by index. They sit
+		// outside the lowerdir range. In block mode that is the writable block
+		// image, referenced as "{{ mount 0 }}".
+		pre []mount.Mount
+		// top is this snapshot's own layer, stacked above its parents.
+		top      []mount.Mount
+		options  []string
+		writable bool
+	)
+
+	if snap.Kind == snapshots.KindActive {
+		layerBlob, src, err := s.resolveBlob(snap.ID, info)
+		if err != nil && !errors.Is(err, errNoLayerBlob) {
+			return nil, err
+		}
+		if err == nil && src.populated() {
+			m, err := s.createErofsMount(layerBlob)
+			if err != nil {
+				return nil, fmt.Errorf("failed to create erofs mount: %w", err)
+			}
+			top = append(top, m)
+			// Nothing in this overlay is writable. The "ro" option states
+			// that. A caller does not have to infer it from the missing
+			// upperdir.
+			options = append(options, "ro")
+		} else {
+			writable = true
+			if s.blockMode {
+				pre = append(pre, mount.Mount{
+					Source: s.writablePath(snap.ID),
+					Type:   "mkfs/ext4",
+					Options: []string{
+						"X-containerd.mkfs.fs=ext4",
+						fmt.Sprintf("X-containerd.mkfs.size=%d", s.writableSize(info)),
+						// TODO: Add UUID
+						"rw",
+						"loop",
+					},
+				})
+				options = append(options,
+					"X-containerd.mkdir.path={{ mount 0 }}/upper:0755",
+					"X-containerd.mkdir.path={{ mount 0 }}/work:0755",
+					"workdir={{ mount 0 }}/work",
+					"upperdir={{ mount 0 }}/upper",
+				)
+			} else {
+				options = append(options,
+					fmt.Sprintf("workdir=%s", s.workPath(snap.ID)),
+					fmt.Sprintf("upperdir=%s", s.upperPath(snap.ID)),
+				)
+			}
+		}
 	}
 
-	layerBlob := s.layerBlobPath(id)
-	// A stat failure must not read as "not staged": writable mounts would let
-	// the differ write through a symlink this check failed to rule out.
-	staged, err := stagedBlob(layerBlob)
-	if err != nil || !staged {
-		return nil, err
-	}
-	// The entry may have been pruned since staging. A dangling symlink must not
-	// pass as staged content, or Commit would write an empty blob through it and
-	// leave that empty blob in the cache.
-	if _, err := os.Stat(layerBlob); err != nil {
-		return nil, fmt.Errorf("staged layer content cache blob %q is unusable: %w", layerBlob, err)
+	mounts := make([]mount.Mount, 0, len(pre)+len(top)+len(parents)+1)
+	mounts = append(mounts, pre...)
+	// first marks the start of the lowerdir range. A merged fsmeta ends the
+	// range but never moves its start: lowers stacked above it stay in range.
+	first := len(mounts)
+	mounts = append(mounts, top...)
+	mounts = append(mounts, parents...)
+
+	if s.remapIDs {
+		if v, ok := info.Labels[snapshots.LabelSnapshotUIDMapping]; ok {
+			options = append(options, fmt.Sprintf("uidmap=%s", v))
+		}
+		if v, ok := info.Labels[snapshots.LabelSnapshotGIDMapping]; ok {
+			options = append(options, fmt.Sprintf("gidmap=%s", v))
+		}
 	}
 
-	m, err := s.createErofsMount(layerBlob)
-	if err != nil {
-		return nil, fmt.Errorf("failed to create erofs mount: %w", err)
+	if (len(mounts) - first) == 1 {
+		// End up with one single lowerdir (e.g. fsmerge on):
+		// it's unsupported by overlayfs
+		if !writable {
+			return append(mounts, mount.Mount{
+				Type:    "format/bind",
+				Source:  fmt.Sprintf("{{ mount %d }}", first),
+				Options: append(options, "ro", "rbind"),
+			}), nil
+		}
+		options = append(options, fmt.Sprintf("lowerdir={{ mount %d }}", first))
+	} else {
+		options = append(options, fmt.Sprintf("lowerdir={{ overlay %d %d }}", first, len(mounts)-1))
 	}
-	return []mount.Mount{m}, nil
+	options = append(options, s.ovlOptions...)
+
+	return append(mounts, mount.Mount{
+		Type:    "format/mkdir/overlay",
+		Source:  "overlay",
+		Options: options,
+	}), nil
 }
 
-func (s *snapshotter) mounts(snap storage.Snapshot, info snapshots.Info) ([]mount.Mount, error) {
-	var options []string
-
+func (s *snapshotter) mounts(snap storage.Snapshot, info snapshots.Info, parents []snapshots.Info) ([]mount.Mount, error) {
 	if len(snap.ParentIDs) == 0 {
-		if layerBlob, err := s.lowerPath(snap.ID); err == nil {
-			if s.enableFsverity {
+		// A recorded blob that does not resolve is an error here too (see
+		// resolveBlob): falling through would hand out a writable mount for a
+		// layer the snapshotter has already reported complete.
+		layerBlob, src, err := s.resolveBlob(snap.ID, info)
+		if err != nil && !errors.Is(err, errNoLayerBlob) {
+			return nil, err
+		}
+		if err == nil {
+			// Only a blob this snapshot owns can carry fsverity because only
+			// such a blob is one this snapshotter wrote.
+			if s.enableFsverity && src.owned() {
 				if err := s.verifyFsverity(layerBlob); err != nil {
 					return nil, err
 				}
@@ -501,69 +601,30 @@ func (s *snapshotter) mounts(snap storage.Snapshot, info snapshots.Info) ([]moun
 				{
 					Source: "{{ mount 0 }}/upper",
 					Type:   "format/mkdir/bind",
-					Options: append(options,
+					Options: []string{
 						"X-containerd.mkdir.path={{ mount 0 }}/upper:0755",
 						roFlag,
 						"rbind",
-					),
-				},
-			}, nil
-		} else {
-			return []mount.Mount{
-				{
-					Source: s.upperPath(snap.ID),
-					Type:   "bind",
-					Options: append(options,
-						roFlag,
-						"rbind",
-					),
+					},
 				},
 			}, nil
 		}
+		return []mount.Mount{
+			{
+				Source:  s.upperPath(snap.ID),
+				Type:    "bind",
+				Options: []string{roFlag, "rbind"},
+			},
+		}, nil
 	}
 
-	var (
-		mounts   []mount.Mount
-		staged   []mount.Mount
-		writable bool
-	)
-	if snap.Kind == snapshots.KindActive {
-		// A staged blob already holds this layer, so the snapshot does not get
-		// a writable upperdir.
-		var err error
-		if staged, err = s.stagedLayerMounts(snap.ID); err != nil {
-			return nil, err
+	// A view of a single committed layer is that layer, read-only.
+	if snap.Kind != snapshots.KindActive && len(snap.ParentIDs) == 1 {
+		if len(parents) != 1 {
+			return nil, fmt.Errorf("got %d parent infos for 1 parent of snapshot %s: %w",
+				len(parents), snap.ID, errdefs.ErrInvalidArgument)
 		}
-	}
-	if snap.Kind == snapshots.KindActive && len(staged) == 0 {
-		if s.blockMode {
-			mounts = append(mounts, mount.Mount{
-				Source: s.writablePath(snap.ID),
-				Type:   "mkfs/ext4",
-				Options: []string{
-					"X-containerd.mkfs.fs=ext4",
-					fmt.Sprintf("X-containerd.mkfs.size=%d", s.writableSize(info)),
-					// TODO: Add UUID
-					"rw",
-					"loop",
-				},
-			})
-			options = append(options,
-				"X-containerd.mkdir.path={{ mount 0 }}/upper:0755",
-				"X-containerd.mkdir.path={{ mount 0 }}/work:0755",
-				"workdir={{ mount 0 }}/work",
-				"upperdir={{ mount 0 }}/upper",
-			)
-		} else {
-			options = append(options,
-				fmt.Sprintf("workdir=%s", s.workPath(snap.ID)),
-				fmt.Sprintf("upperdir=%s", s.upperPath(snap.ID)),
-			)
-		}
-		writable = true
-	} else if snap.Kind != snapshots.KindActive && len(snap.ParentIDs) == 1 {
-		// A view of a single committed layer is that layer, read-only.
-		layerBlob, err := s.lowerPath(snap.ParentIDs[0])
+		layerBlob, err := s.lowerPath(snap.ParentIDs[0], parents[0])
 		if err != nil {
 			return nil, err
 		}
@@ -574,97 +635,50 @@ func (s *snapshotter) mounts(snap storage.Snapshot, info snapshots.Info) ([]moun
 		return []mount.Mount{m}, nil
 	}
 
-	// first marks the start of the lowerdir range. A merged fsmeta ends the
-	// range but never moves its start: lowers stacked above it stay in range.
-	first := len(mounts)
-	// A staged blob is this snapshot's own layer, so it stacks above the parents.
-	// The overlay does not get an upperdir, so it is read-only. A caller reads
-	// that as a staged layer.
-	mounts = append(mounts, staged...)
-	for i := range snap.ParentIDs {
-		// If a merged fsmeta is valid for this layer, skip the remaining bottom layers.
-		// Why? Because bottom layers have been flattened with the thin fsmeta.
-		if m, ok := s.mountFsMeta(snap, i); ok {
-			mounts = append(mounts, m)
-			break
-		}
-
-		layerBlob, err := s.lowerPath(snap.ParentIDs[i])
-		if err != nil {
-			return nil, err
-		}
-
-		m, err := s.createErofsMount(layerBlob)
-		if err != nil {
-			return nil, fmt.Errorf("failed to create erofs mount for parent %s: %w", snap.ParentIDs[i], err)
-		}
-
-		mounts = append(mounts, m)
+	lowers, err := s.parentMounts(snap, parents)
+	if err != nil {
+		return nil, err
 	}
-
-	if s.remapIDs {
-		if v, ok := info.Labels[snapshots.LabelSnapshotUIDMapping]; ok {
-			options = append(options, fmt.Sprintf("uidmap=%s", v))
-		}
-		if v, ok := info.Labels[snapshots.LabelSnapshotGIDMapping]; ok {
-			options = append(options, fmt.Sprintf("gidmap=%s", v))
-		}
-	}
-
-	if (len(mounts) - first) == 1 {
-		// End up with one single lowerdir (e.g. fsmerge on):
-		// it's unsupported by overlayfs
-		if !writable {
-			return append(mounts, mount.Mount{
-				Type:    "format/bind",
-				Source:  fmt.Sprintf("{{ mount %d }}", first),
-				Options: append(options, "ro", "rbind"),
-			}), nil
-		}
-		options = append(options, fmt.Sprintf("lowerdir={{ mount %d }}", first))
-	} else {
-		options = append(options, fmt.Sprintf("lowerdir={{ overlay %d %d }}", first, len(mounts)-1))
-	}
-	if len(staged) > 0 {
-		options = append(options, "ro")
-	}
-	options = append(options, s.ovlOptions...)
-
-	return append(mounts, mount.Mount{
-		Type:    "format/mkdir/overlay",
-		Source:  "overlay",
-		Options: options,
-	}), nil
+	return s.overlayMounts(snap, info, lowers)
 }
 
 // createSnapshot creates an active (or view) snapshot and returns its mounts.
 // On an image-layer extraction whose diffID blob is in the layer content cache,
-// it stages the cached blob into the active snapshot and returns it as a
+// it records the cache entry as the snapshot's blob source and returns it as a
 // read-only mount. The unpacker reads that as a staged layer, skips the layer
-// download and conversion, and commits the snapshot normally. The parent is
-// applied either at Prepare (sequential unpacking) or at Commit via WithParent
-// (parallel unpacking).
+// conversion, and commits the snapshot normally. The parent is applied either at
+// Prepare (sequential unpacking) or at Commit via WithParent (parallel
+// unpacking).
 func (s *snapshotter) createSnapshot(ctx context.Context, kind snapshots.Kind, key, parent string, opts []snapshots.Opt) (_ []mount.Mount, err error) {
 	var (
 		snap     storage.Snapshot
 		td, path string
 		info     snapshots.Info
+		parents  []snapshots.Info
 	)
 
-	// Any image-layer extraction can be served from the cache. s.mounts hands a
-	// staged snapshot out read-only, so the differ cannot write the layer
-	// through the staged symlink into the shared cache blob.
+	// Any image-layer extraction can be served from the cache. s.mounts hands
+	// such a snapshot out read-only and its blob stays in the cache. Nothing in
+	// the snapshot can write to the shared entry.
 	//
-	// TODO(2.5): staged layers above the unpacker's first cache miss still have
+	// TODO(2.5): cached layers above the unpacker's first cache miss still have
 	// their blobs downloaded. Skipping those fetches needs the unpacker rework
 	// planned for 2.5.
-	var cacheBlob string
 	if kind == snapshots.KindActive {
-		if cacheBlob = s.lookupCache(ctx, opts...); cacheBlob != "" {
+		if cacheBlob := s.lookupCache(ctx, opts...); cacheBlob != "" {
+			// Mode "on" requires a sidecar. Checking it before the snapshot is
+			// recorded fails Prepare for an unusable cache entry.
+			if s.dmverityMode == "on" {
+				if _, err := os.Stat(dmverity.MetadataPath(cacheBlob)); err != nil {
+					return nil, fmt.Errorf("dm-verity mode is 'on' but the layer content cache entry %q has no .dmverity metadata: %w", cacheBlob, err)
+				}
+			}
 			log.G(ctx).WithFields(log.Fields{
 				"key":  key,
 				"blob": cacheBlob,
-			}).Debug("layer content cache hit, staged cached erofs blob")
+			}).Debug("layer content cache hit, serving the cached erofs blob")
+			opts = append(opts, snapshots.WithLabels(
+				blobSource{Kind: blobSourceCache, Ref: cacheBlob}.labels()))
 		}
 	}
 
@@ -685,7 +699,7 @@ func (s *snapshotter) createSnapshot(ctx context.Context, kind snapshots.Kind, k
 	}()
 
 	snapshotDir := filepath.Join(s.root, "snapshots")
-	td, err = s.prepareDirectory(ctx, snapshotDir, kind, cacheBlob)
+	td, err = s.prepareDirectory(ctx, snapshotDir, kind)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create prepare snapshot dir: %w", err)
 	}
@@ -756,12 +770,15 @@ func (s *snapshotter) createSnapshot(ctx context.Context, kind snapshots.Kind, k
 			return fmt.Errorf("failed to rename: %w", err)
 		}
 		td = ""
-		return nil
+
+		// Read inside the transaction (see parentInfos).
+		parents, err = parentInfos(ctx, info)
+		return err
 	}); err != nil {
 		return nil, err
 	}
 
-	return s.mounts(snap, info)
+	return s.mounts(snap, info, parents)
 }
 
 func (s *snapshotter) Prepare(ctx context.Context, key, parent string, opts ...snapshots.Opt) ([]mount.Mount, error) {
@@ -782,8 +799,7 @@ func cacheID(i int, dir string) string {
 // are checked one by one and the first hit wins. Every miss (cache disabled,
 // non-extraction Prepare, absent or unreadable cache dir, missing entry,
 // malformed labels) returns "" so pulls keep working. Any dm-verity sidecar is
-// derived from the blob path (via dmverity.MetadataPath) when the blob is
-// staged (prepareDirectory).
+// derived from the blob path (via dmverity.MetadataPath).
 func (s *snapshotter) lookupCache(ctx context.Context, opts ...snapshots.Opt) string {
 	if len(s.layerContentCaches) == 0 {
 		return ""
@@ -824,8 +840,9 @@ func (s *snapshotter) lookupCache(ctx context.Context, opts ...snapshots.Opt) st
 		cacheLookups.WithValues("hit").Inc()
 		cacheHits.WithValues(cacheID(i, dir)).Inc()
 		cacheHitBytes.Inc(float64(fi.Size()))
-		// Absolute, since the configured dirs are validated as such: the hit is
-		// symlinked into the snapshot dir, where a relative target would dangle.
+		// The path is absolute because the configured dirs are validated as
+		// absolute. It is recorded in the snapshot and mounted later. It must
+		// not depend on a working directory.
 		return blob
 	}
 
@@ -877,39 +894,44 @@ func (s *snapshotter) commitBlock(ctx context.Context, layerBlob string, id stri
 }
 
 func (s *snapshotter) Commit(ctx context.Context, name, key string, opts ...snapshots.Opt) error {
-	var layerBlob string
-	var id string
+	var (
+		id   string
+		info snapshots.Info
+	)
 
 	// Apply the overlayfs upperdir (generated by non-EROFS differs) into a EROFS blob
 	// in a read transaction first since conversion could be slow.
 	err := s.ms.WithTransaction(ctx, false, func(ctx context.Context) error {
-		sid, _, _, err := storage.GetInfo(ctx, key)
+		sid, si, _, err := storage.GetInfo(ctx, key)
 		if err != nil {
 			return err
 		}
-		id = sid
+		id, info = sid, si
 		return err
 	})
 	if err != nil {
 		return err
 	}
 
-	// If the layer blob doesn't exist, which means this layer wasn't applied by
-	// the EROFS differ (possibly the walking differ), convert the upperdir instead.
-	layerBlob = s.layerBlobPath(id)
-	if _, err := os.Stat(layerBlob); err != nil {
-		// A staged blob whose cache entry has been pruned since is a dangling
-		// symlink, which stats the same way. Converting would run mkfs.erofs
-		// through it, committing an empty layer and leaving that empty blob in
-		// the cache for every future pull of this layer.
-		staged, serr := stagedBlob(layerBlob)
-		if serr != nil {
-			return serr
-		}
-		if staged {
-			return fmt.Errorf("staged layer content cache blob %q is unusable: %w", layerBlob, err)
-		}
+	// storage.CommitActive replaces the snapshot's labels with the ones passed
+	// here. The snapshotter's own labels are added back. The committed layer
+	// keeps the record of where its blob is.
+	if private := privateLabels(info.Labels); len(private) > 0 {
+		opts = append(opts, snapshots.WithLabels(private))
+	}
 
+	// A populated blob is committed as it is: nothing needs conversion and the
+	// blob belongs to its source. resolveBlob fails for a recorded source that
+	// does not resolve. An unapplied layer is never committed as a complete
+	// one.
+	layerBlob, src, err := s.resolveBlob(id, info)
+	if err != nil {
+		if !errors.Is(err, errNoLayerBlob) {
+			return err
+		}
+		// The layer was not applied by the EROFS differ, possibly the walking
+		// differ. Convert the upperdir.
+		layerBlob = s.layerBlobPath(id)
 		if cerr := s.commitBlock(ctx, layerBlob, id); cerr != nil {
 			if errdefs.IsNotImplemented(cerr) {
 				return err
@@ -918,17 +940,24 @@ func (s *snapshotter) Commit(ctx context.Context, name, key string, opts ...snap
 		}
 	}
 
-	// Enable fsverity on the EROFS layer if configured
-	if s.enableFsverity {
-		if err := fsverity.Enable(layerBlob); err != nil {
-			return fmt.Errorf("failed to enable fsverity: %w", err)
+	// Both settings write to the blob, which is only allowed on a blob this
+	// snapshot owns. A shared one would be mutated while every other snapshot
+	// is using it. Neither can be configured together with the layer content
+	// cache (see NewSnapshotter); this states the requirement where it
+	// applies.
+	if src.owned() {
+		// Enable fsverity on the EROFS layer if configured
+		if s.enableFsverity {
+			if err := fsverity.Enable(layerBlob); err != nil {
+				return fmt.Errorf("failed to enable fsverity: %w", err)
+			}
 		}
-	}
 
-	// Set IMMUTABLE_FL on the EROFS layer to avoid artificial data loss
-	if s.setImmutable {
-		if err := setImmutable(layerBlob, true); err != nil {
-			log.G(ctx).WithError(err).Warnf("failed to set IMMUTABLE_FL for %s", layerBlob)
+		// Set IMMUTABLE_FL on the EROFS layer to avoid artificial data loss
+		if s.setImmutable {
+			if err := setImmutable(layerBlob, true); err != nil {
+				log.G(ctx).WithError(err).Warnf("failed to set IMMUTABLE_FL for %s", layerBlob)
+			}
 		}
 	}
 
@@ -939,6 +968,10 @@ func (s *snapshotter) Commit(ctx context.Context, name, key string, opts ...snap
 			return fmt.Errorf("failed to get the converted erofs blob: %w", err)
 		}
 
+		// TODO: a blob this snapshot does not own is counted in full here, even
+		// though removing the snapshot does not reclaim any of it. Usage has
+		// several meanings: bytes on disk, bytes a deletion would free, and
+		// attribution of shared bytes. Choosing one needs its own design.
 		usage, err := fs.DiskUsage(ctx, layerBlob)
 		if err != nil {
 			return err
@@ -951,8 +984,11 @@ func (s *snapshotter) Commit(ctx context.Context, name, key string, opts ...snap
 }
 
 func (s *snapshotter) Mounts(ctx context.Context, key string) (_ []mount.Mount, err error) {
-	var snap storage.Snapshot
-	var info snapshots.Info
+	var (
+		snap    storage.Snapshot
+		info    snapshots.Info
+		parents []snapshots.Info
+	)
 	if err := s.ms.WithTransaction(ctx, false, func(ctx context.Context) error {
 		snap, err = storage.GetSnapshot(ctx, key)
 		if err != nil {
@@ -963,11 +999,14 @@ func (s *snapshotter) Mounts(ctx context.Context, key string) (_ []mount.Mount,
 		if err != nil {
 			return fmt.Errorf("failed to get snapshot info: %w", err)
 		}
-		return nil
+
+		// Read inside the transaction (see parentInfos).
+		parents, err = parentInfos(ctx, info)
+		return err
 	}); err != nil {
 		return nil, err
 	}
-	return s.mounts(snap, info)
+	return s.mounts(snap, info, parents)
 }
 
 func (s *snapshotter) getCleanupDirectories(ctx context.Context) ([]string, error) {
@@ -1039,17 +1078,17 @@ func (s *snapshotter) Remove(ctx context.Context, key string) (err error) {
 			return fmt.Errorf("failed to get snapshot info: %w", err)
 		}
 
-		// The layer blob is only persisted for committed snapshots.
+		// Only a committed snapshot has a layer blob, and only a blob in the
+		// snapshot directory can have had IMMUTABLE_FL set by this snapshotter.
+		// A blob from any other source stays where it is. Removing the
+		// directory does not reach it.
 		if info.Kind == snapshots.KindCommitted {
+			// setImmutable opens the path. A link here would clear the flag on
+			// its target. Lstat and a regular-file check keep this on the
+			// snapshot's own blob. Anything else is skipped: os.RemoveAll
+			// unlinks a link without following it.
 			layerBlob := s.layerBlobPath(id)
-			// A cache-hit snapshot's blob is a symlink into the operator-owned
-			// cache dir. Skip clearing IMMUTABLE_FL: setImmutable's os.Open would
-			// follow the link and ioctl the cache entry (which we don't own), and
-			// cache blobs were never made immutable by us in the first place.
-			// os.RemoveAll below unlinks the symlink without following it.
-			if fi, lerr := os.Lstat(layerBlob); lerr == nil && fi.Mode()&os.ModeSymlink != 0 {
-				log.G(ctx).WithField("id", id).Trace("erofs layer cache: skipping IMMUTABLE_FL clear for symlinked cache blob")
-			} else {
+			if fi, serr := os.Lstat(layerBlob); serr == nil && fi.Mode().IsRegular() {
 				// Clear IMMUTABLE_FL before removal, since this flag avoids it.
 				err = setImmutable(layerBlob, false)
 				if err != nil && !errdefs.IsNotImplemented(err) {
@@ -1084,6 +1123,21 @@ func (s *snapshotter) Stat(ctx context.Context, key string) (info snapshots.Info
 
 func (s *snapshotter) Update(ctx context.Context, info snapshots.Info, fieldpaths ...string) (_ snapshots.Info, err error) {
 	err = s.ms.WithTransaction(ctx, true, func(ctx context.Context) error {
+		// A backend snapshotter only receives inherited labels. A caller cannot
+		// send the snapshotter's own labels back. An Update that
+		// replaces the label set would drop them and leave a snapshot that
+		// cannot find its layer blob. They are merged back in here.
+		_, cur, _, err := storage.GetInfo(ctx, info.Name)
+		if err != nil {
+			return err
+		}
+		if private := privateLabels(cur.Labels); len(private) > 0 {
+			if info.Labels == nil {
+				info.Labels = make(map[string]string, len(private))
+			}
+			maps.Copy(info.Labels, private)
+		}
+
 		info, err = storage.UpdateInfo(ctx, info, fieldpaths...)
 		return err
 	})

plugins/snapshots/erofs/erofs_linux_test.go

@@ -27,6 +27,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/containerd/errdefs"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	bolt "go.etcd.io/bbolt"
@@ -784,6 +785,8 @@ func TestMountFsMeta(t *testing.T) {
 	parents := []string{"p0", "p1", "p2"}
 	for _, id := range parents {
 		require.NoError(t, os.MkdirAll(filepath.Join(root, "snapshots", id), 0755))
+		// The flattened layers are addressed by device. Each needs a blob.
+		require.NoError(t, os.WriteFile(s.layerBlobPath(id), []byte("layer"), 0644))
 	}
 
 	writeMeta := func(t *testing.T, id string, contents []byte) {
@@ -799,12 +802,15 @@ func TestMountFsMeta(t *testing.T) {
 	}
 
 	snap := storage.Snapshot{ParentIDs: parents}
+	// Every parent's blob is local, unless a subtest below records a source.
+	infos := make([]snapshots.Info, len(parents))
 
 	t.Run("missing fsmeta returns false", func(t *testing.T) {
 		for _, id := range parents {
 			removeMeta(t, id)
 		}
-		_, ok := s.mountFsMeta(snap, 0)
+		_, ok, err := s.mountFsMeta(snap, infos, 0)
+		require.NoError(t, err)
 		assert.False(t, ok)
 	})
 
@@ -812,7 +818,8 @@ func TestMountFsMeta(t *testing.T) {
 		writeMeta(t, "p0", nil)
 		t.Cleanup(func() { removeMeta(t, "p0") })
 
-		_, ok := s.mountFsMeta(snap, 0)
+		_, ok, err := s.mountFsMeta(snap, infos, 0)
+		require.NoError(t, err)
 		assert.False(t, ok)
 	})
 
@@ -820,7 +827,8 @@ func TestMountFsMeta(t *testing.T) {
 		writeMeta(t, "p0", []byte("merged"))
 		t.Cleanup(func() { removeMeta(t, "p0") })
 
-		m, ok := s.mountFsMeta(snap, 0)
+		m, ok, err := s.mountFsMeta(snap, infos, 0)
+		require.NoError(t, err)
 		require.True(t, ok)
 		assert.Equal(t, "erofs", m.Type)
 		assert.Equal(t, s.fsMetaPath("p0"), m.Source)
@@ -837,7 +845,8 @@ func TestMountFsMeta(t *testing.T) {
 		writeMeta(t, "p1", []byte("merged"))
 		t.Cleanup(func() { removeMeta(t, "p1") })
 
-		m, ok := s.mountFsMeta(snap, 1)
+		m, ok, err := s.mountFsMeta(snap, infos, 1)
+		require.NoError(t, err)
 		require.True(t, ok)
 		assert.Equal(t, s.fsMetaPath("p1"), m.Source)
 		assert.Equal(t, []string{
@@ -846,6 +855,46 @@ func TestMountFsMeta(t *testing.T) {
 			"device=" + s.layerBlobPath("p1"),
 		}, m.Options)
 	})
+
+	t.Run("a parent's blob is addressed wherever it is stored", func(t *testing.T) {
+		// A flattened layer staged from a layer content cache does not have a
+		// blob in its own directory. The device must point at the cache
+		// entry. A device= under the snapshot directory would not exist.
+		external := filepath.Join(t.TempDir(), "cached.erofs")
+		require.NoError(t, os.WriteFile(external, []byte("cached"), 0644))
+		require.NoError(t, os.Remove(s.layerBlobPath("p1")))
+		infos[1] = snapshots.Info{Labels: blobSource{Kind: blobSourceCache, Ref: external}.labels()}
+		t.Cleanup(func() {
+			infos[1] = snapshots.Info{}
+			require.NoError(t, os.WriteFile(s.layerBlobPath("p1"), []byte("layer"), 0644))
+		})
+
+		writeMeta(t, "p0", []byte("merged"))
+		t.Cleanup(func() { removeMeta(t, "p0") })
+
+		m, ok, err := s.mountFsMeta(snap, infos, 0)
+		require.NoError(t, err)
+		require.True(t, ok)
+		assert.Equal(t, []string{
+			"ro", "loop",
+			"device=" + s.layerBlobPath("p2"),
+			"device=" + external,
+			"device=" + s.layerBlobPath("p0"),
+		}, m.Options)
+	})
+
+	t.Run("a parent whose blob is gone is an error", func(t *testing.T) {
+		require.NoError(t, os.Remove(s.layerBlobPath("p1")))
+		t.Cleanup(func() {
+			require.NoError(t, os.WriteFile(s.layerBlobPath("p1"), []byte("layer"), 0644))
+		})
+
+		writeMeta(t, "p0", []byte("merged"))
+		t.Cleanup(func() { removeMeta(t, "p0") })
+
+		_, _, err := s.mountFsMeta(snap, infos, 0)
+		assert.Error(t, err, "a device must not be silently dropped from the mount")
+	})
 }
 
 // TestMountsWithMergedFsMeta covers s.mounts()'s assembly of the overlay
@@ -868,8 +917,10 @@ func TestMountsWithMergedFsMeta(t *testing.T) {
 
 	snap := storage.Snapshot{Kind: snapshots.KindView, ParentIDs: parents}
 	info := snapshots.Info{}
+	// Every parent's blob is local.
+	parentInfos := make([]snapshots.Info, len(parents))
 
-	mounts, err := s.mounts(snap, info)
+	mounts, err := s.mounts(snap, info, parentInfos)
 	require.NoError(t, err)
 
 	// Expect: [erofs(p0), erofs(p1), erofs(fsmeta p2, device=p3,p2), overlay]
@@ -913,8 +964,10 @@ func TestMountsWithMergedFsMetaOnTopParent(t *testing.T) {
 
 	snap := storage.Snapshot{Kind: snapshots.KindView, ParentIDs: parents}
 	info := snapshots.Info{}
+	// Every parent's blob is local.
+	parentInfos := make([]snapshots.Info, len(parents))
 
-	mounts, err := s.mounts(snap, info)
+	mounts, err := s.mounts(snap, info, parentInfos)
 	require.NoError(t, err)
 
 	require.Len(t, mounts, 2)
@@ -949,8 +1002,8 @@ func requireErofs(t *testing.T) {
 
 // writeCacheBlob writes a fake erofs blob into cacheDir keyed by diffID and
 // returns its absolute path. The bytes need not be a valid erofs image: the
-// snapshotter only symlinks the blob, so these tests exercise the
-// Prepare/Commit/Remove logic rather than mounting.
+// snapshotter only records and references the blob. These tests exercise the
+// Prepare/Commit/Remove logic without mounting.
 func writeCacheBlob(t *testing.T, cacheDir string, diffID digest.Digest, data []byte) string {
 	t.Helper()
 	blob := erofsutils.CacheBlobPath(cacheDir, diffID)
@@ -971,13 +1024,25 @@ func extractionOpt(target string, diffID digest.Digest) snapshots.Opt {
 // snapshotID returns the backend snapshot ID for key.
 func snapshotID(t *testing.T, ctx context.Context, s *snapshotter, key string) string {
 	t.Helper()
-	var id string
+	id, _ := snapshotIDInfo(t, ctx, s, key)
+	return id
+}
+
+// snapshotIDInfo returns a snapshot's id and info, including the labels the
+// snapshotter keeps for itself. The metadata snapshotter filters those out
+// before a client sees them.
+func snapshotIDInfo(t *testing.T, ctx context.Context, s *snapshotter, key string) (string, snapshots.Info) {
+	t.Helper()
+	var (
+		id   string
+		info snapshots.Info
+	)
 	require.NoError(t, s.ms.WithTransaction(ctx, false, func(ctx context.Context) error {
 		var err error
-		id, _, _, err = storage.GetInfo(ctx, key)
+		id, info, _, err = storage.GetInfo(ctx, key)
 		return err
 	}))
-	return id
+	return id, info
 }
 
 // newCacheSnapshotter creates an erofs snapshotter rooted in a temp dir with the
@@ -992,19 +1057,28 @@ func newCacheSnapshotter(t *testing.T, opts ...Opt) *snapshotter {
 	return sn.(*snapshotter)
 }
 
-// requireStagedSymlink asserts that the snapshot behind key holds the cache
-// blob as an absolute symlink, which is what staging leaves behind. A converted
-// layer would be a regular file.
-func requireStagedSymlink(t *testing.T, ctx context.Context, s *snapshotter, key, blob string) {
+// requireCachedBlobSource asserts that the snapshot behind key records blob as
+// its layer's source and does not hold a copy or a link of it. The cache entry
+// is mounted in place. Nothing inside the snapshot aliases it.
+func requireCachedBlobSource(t *testing.T, ctx context.Context, s *snapshotter, key, blob string) {
 	t.Helper()
-	link := s.layerBlobPath(snapshotID(t, ctx, s, key))
-	fi, err := os.Lstat(link)
-	require.NoError(t, err, "the cached blob must be staged as layer.erofs")
-	assert.NotZero(t, fi.Mode()&os.ModeSymlink, "layer.erofs should be a symlink")
-	dst, err := os.Readlink(link)
+	id, info := snapshotIDInfo(t, ctx, s, key)
+
+	src, err := blobSourceFromInfo(info)
+	require.NoError(t, err, "the cache hit must be recorded as a blob source")
+	assert.Equal(t, blobSourceCache, src.Kind)
+	assert.True(t, filepath.IsAbs(src.Ref), "the recorded ref should be absolute")
+	assert.Equal(t, blob, src.Ref)
+
+	_, err = os.Lstat(s.layerBlobPath(id))
+	assert.ErrorIs(t, err, os.ErrNotExist, "a cached blob must not be linked or copied into the snapshot")
+
+	// The record resolves back to the cache entry, which is what gets mounted.
+	path, resolved, err := s.resolveBlob(id, info)
 	require.NoError(t, err)
-	assert.True(t, filepath.IsAbs(dst), "symlink target should be absolute")
-	assert.Equal(t, blob, dst)
+	assert.Equal(t, blob, path)
+	assert.False(t, resolved.owned(), "a cached blob is not this snapshot's to write")
+	assert.True(t, resolved.populated(), "a cached blob needs nothing applied into it")
 }
 
 // stageCacheHit runs a parentless extraction Prepare for target/diffID; see
@@ -1031,10 +1105,10 @@ func stageCacheHitFrom(t *testing.T, ctx context.Context, s *snapshotter, parent
 	return key
 }
 
-// TestCacheHit covers the happy path: an extraction Prepare whose diffID blob is
-// in the cache stages the blob into the active snapshot (symlinked) and returns
-// read-only mounts without committing; a subsequent Commit finalizes the target
-// chainID without re-converting.
+// TestCacheHit covers the happy path. An extraction Prepare whose diffID blob
+// is in the cache records the cache entry as the snapshot's blob source and
+// returns read-only mounts. A subsequent Commit finalizes the target chainID
+// without re-converting the layer.
 func TestCacheHit(t *testing.T) {
 	ctx := namespaces.WithNamespace(context.Background(), "test")
 
@@ -1047,15 +1121,16 @@ func TestCacheHit(t *testing.T) {
 	target := cacheTestChainID
 	key := stageCacheHit(t, ctx, s, target, diffID)
 
-	// The blob is staged into the active snapshot but the target chainID is not
+	// The blob is staged into the active snapshot. The target chainID is not
 	// committed yet.
 	_, err := s.Stat(ctx, target)
 	assert.Error(t, err, "target chainID must not be committed before Commit")
 
-	// layer.erofs is an absolute symlink into the operator-owned cache blob.
-	requireStagedSymlink(t, ctx, s, key, blob)
+	// The operator-owned cache blob is recorded. Nothing in the snapshot links
+	// to it.
+	requireCachedBlobSource(t, ctx, s, key, blob)
 
-	// Commit finalizes the staged snapshot as the target chainID, without any
+	// Commit finalizes the snapshot as the target chainID, without any
 	// re-conversion (the blob is already present).
 	require.NoError(t, s.Commit(ctx, target, key))
 	info, err := s.Stat(ctx, target)
@@ -1065,8 +1140,8 @@ func TestCacheHit(t *testing.T) {
 }
 
 // TestCacheSidecar covers a hit in the default "auto" dm-verity mode where the
-// cache entry has a sidecar: it must be copied into the snapshot dir as a plain
-// regular file (not symlinked).
+// cache entry has a sidecar. The blob is mounted from the cache. The sidecar
+// beside it is read from the cache too. Nothing is copied into the snapshot.
 func TestCacheSidecar(t *testing.T) {
 	ctx := namespaces.WithNamespace(context.Background(), "test")
 
@@ -1081,15 +1156,17 @@ func TestCacheSidecar(t *testing.T) {
 	target := cacheTestChainID
 	key := stageCacheHit(t, ctx, s, target, diffID)
 
-	// The sidecar is copied in as a plain regular file (not a symlink) so mount-time
-	// metadata resolution is independent of the cache filesystem.
-	sidecar := dmverity.MetadataPath(s.layerBlobPath(snapshotID(t, ctx, s, key)))
-	fi, err := os.Lstat(sidecar)
-	require.NoError(t, err, "sidecar should be copied into the snapshot dir")
-	assert.Zero(t, fi.Mode()&os.ModeSymlink, "sidecar should be a regular file, not a symlink")
-	data, err := os.ReadFile(sidecar)
+	mounts, err := s.Mounts(ctx, key)
 	require.NoError(t, err)
-	assert.Equal(t, testDmverityMetadata, string(data))
+	require.NotEmpty(t, mounts)
+	assert.Contains(t, mounts[0].Options, "X-containerd.dmverity="+dmverity.MetadataPath(blob),
+		"the cache entry's own sidecar is used in place")
+
+	// There is no copy in the snapshot that could go stale against the blob it
+	// describes.
+	id := snapshotID(t, ctx, s, key)
+	_, err = os.Lstat(dmverity.MetadataPath(s.layerBlobPath(id)))
+	assert.ErrorIs(t, err, os.ErrNotExist, "no sidecar copy may be left in the snapshot dir")
 }
 
 // TestCacheMiss covers the cases that must NOT be served from the cache and
@@ -1103,8 +1180,9 @@ func TestCacheMiss(t *testing.T) {
 
 	// Each case must leave the extraction as a normal active snapshot: mounts are
 	// returned and the target chainID is not committed. wantRO distinguishes a
-	// KindActive miss (no layer.erofs staged yet, so mounts must be writable)
-	// from the KindView case below (read-only for its own, cache-unrelated reason).
+	// KindActive miss, which does not have a layer blob yet and must return
+	// writable mounts, from the KindView case below, which is read-only for its
+	// own reason unrelated to the cache.
 	assertFellThrough := func(t *testing.T, s *snapshotter, mounts []mount.Mount, err error, wantRO bool) {
 		t.Helper()
 		require.NoError(t, err)
@@ -1152,9 +1230,9 @@ func TestCacheMiss(t *testing.T) {
 		mounts, err := s.Prepare(ctx, key, parentChain)
 		require.NoError(t, err)
 		require.NotEmpty(t, mounts)
-		// assertFellThrough does not apply here, since a parented snapshot also
-		// returns the read-only lowers. The unpacker only inspects the last
-		// mount, which must be the writable overlay.
+		// assertFellThrough does not apply here because a parented snapshot
+		// also returns the read-only lowers. The unpacker only inspects the
+		// last mount, which must be the writable overlay.
 		assert.False(t, mounts[len(mounts)-1].ReadOnly(), "a Prepare without extraction labels must expose a writable overlay")
 
 		_, err = os.Lstat(s.layerBlobPath(snapshotID(t, ctx, s, key)))
@@ -1175,8 +1253,9 @@ func TestCacheMiss(t *testing.T) {
 // TestCacheParentedStage covers a cached blob whose extraction Prepare carries
 // a parent, which is what sequential unpacking does for every layer but the
 // first. It is staged like a parentless one, so an image is served from the
-// cache in either unpack mode. The staged blob is mounted read-only, so the
-// differ can never write through the symlink into the shared cache blob.
+// cache in either unpack mode. The cache entry is mounted in place and
+// read-only, which leaves nothing able to write to a blob shared with every
+// other snapshot of that layer.
 func TestCacheParentedStage(t *testing.T) {
 	ctx := namespaces.WithNamespace(context.Background(), "test")
 
@@ -1193,21 +1272,22 @@ func TestCacheParentedStage(t *testing.T) {
 
 	s := newCacheSnapshotter(t, WithLayerContentCaches(cacheDir))
 
-	// The first layer has no parent, so it is served from the cache as usual.
+	// The first layer does not have a parent. It is served from the cache as
+	// usual.
 	require.NoError(t, s.Commit(ctx, parentChain, stageCacheHit(t, ctx, s, parentChain, parentDiffID)))
 
 	// The second layer is prepared on top of it and is served from the cache too.
 	key := stageCacheHitFrom(t, ctx, s, parentChain, childChain, childDiffID)
 
 	// The Snapshotter contract requires a parented Prepare to hand out the
-	// parent's content, so the staged blob is stacked as the top lower over the
-	// parent. The overlay does not get an upperdir, so it is read-only, and the
+	// parent's content. The cached blob is stacked as the top lower over the
+	// parent. The overlay does not get an upperdir and is read-only. The
 	// unpacker reads that as a staged layer.
 	mounts, err := s.Mounts(ctx, key)
 	require.NoError(t, err)
-	require.Len(t, mounts, 3, "staged blob, the parent lower and the overlay")
-	assert.Equal(t, s.layerBlobPath(snapshotID(t, ctx, s, key)), mounts[0].Source,
-		"the staged blob is the top lower")
+	require.Len(t, mounts, 3, "the cached blob, the parent lower and the overlay")
+	assert.Equal(t, childBlob, mounts[0].Source,
+		"the cache entry is mounted in place, as the top lower")
 	assert.Equal(t, "erofs", mounts[1].Type, "the parent layer is stacked below it")
 	assert.Equal(t, "format/mkdir/overlay", mounts[2].Type)
 	assert.Contains(t, strings.Join(mounts[2].Options, ","), "lowerdir={{ overlay 0 1 }}",
@@ -1224,18 +1304,91 @@ func TestCacheParentedStage(t *testing.T) {
 	assert.Equal(t, snapshots.KindCommitted, info.Kind)
 	assert.Equal(t, parentChain, info.Parent)
 
-	// Commit did not convert anything. The blob is still the symlinked cache
+	// Commit did not convert anything. The blob is still the recorded cache
 	// entry and the operator-owned blob itself is untouched.
-	requireStagedSymlink(t, ctx, s, childChain, childBlob)
+	requireCachedBlobSource(t, ctx, s, childChain, childBlob)
 	data, err := os.ReadFile(childBlob)
 	require.NoError(t, err)
 	assert.Equal(t, childData, data, "the cache blob must not be written through")
 }
 
-// TestCacheStaleBlob covers a cache entry pruned after it was staged. Mounts
-// must not report a dangling symlink as staged content. The caller would skip
-// the layer, and Commit would then write an empty blob through the symlink and
-// leave it in the cache for every image using that layer.
+// TestCacheBlobSourceOutlivesCommit covers that a committed snapshot still
+// records where its blob is. storage.CommitActive replaces an active snapshot's
+// labels with the ones passed to Commit. A record left on the active
+// snapshot would be lost and the layer would then read as one that was never
+// applied.
+func TestCacheBlobSourceOutlivesCommit(t *testing.T) {
+	ctx := namespaces.WithNamespace(context.Background(), "test")
+
+	cacheDir := t.TempDir()
+	diffID := digest.Digest(cacheTestDiffID)
+	blob := writeCacheBlob(t, cacheDir, diffID, []byte("fake erofs blob"))
+
+	s := newCacheSnapshotter(t, WithLayerContentCaches(cacheDir))
+
+	target := cacheTestChainID
+	key := stageCacheHit(t, ctx, s, target, diffID)
+	// Commit with labels of its own, as the unpacker does. They are merged with
+	// the snapshotter's.
+	require.NoError(t, s.Commit(ctx, target, key, snapshots.WithLabels(map[string]string{
+		snapshots.LabelSnapshotRef: target,
+	})))
+
+	requireCachedBlobSource(t, ctx, s, target, blob)
+	_, info := snapshotIDInfo(t, ctx, s, target)
+	assert.Equal(t, target, info.Labels[snapshots.LabelSnapshotRef],
+		"the caller's own labels must survive too")
+
+	// A child stacks it as a lower from the cache, not from a path inside the
+	// parent's snapshot directory.
+	child := "sha256:0000000000000000000000000000000000000000000000000000000000000009"
+	mounts, err := s.Prepare(ctx, child, target)
+	require.NoError(t, err)
+	require.NotEmpty(t, mounts)
+	assert.Equal(t, blob, mounts[0].Source, "the parent is mounted from the cache")
+}
+
+// TestCacheBlobSourceSurvivesUpdate covers an Update that replaces a snapshot's
+// labels wholesale, which is what a client sending no update mask does. A
+// client can read the snapshotter's own labels through Stat, but Update filters
+// them out on the way down. It cannot send them back. Dropping them would
+// leave a committed layer unable to find its blob.
+func TestCacheBlobSourceSurvivesUpdate(t *testing.T) {
+	ctx := namespaces.WithNamespace(context.Background(), "test")
+
+	cacheDir := t.TempDir()
+	diffID := digest.Digest(cacheTestDiffID)
+	blob := writeCacheBlob(t, cacheDir, diffID, []byte("fake erofs blob"))
+
+	s := newCacheSnapshotter(t, WithLayerContentCaches(cacheDir))
+
+	target := cacheTestChainID
+	require.NoError(t, s.Commit(ctx, target, stageCacheHit(t, ctx, s, target, diffID)))
+
+	// A wholesale label replacement, carrying none of the snapshotter's own.
+	_, err := s.Update(ctx, snapshots.Info{
+		Name:   target,
+		Labels: map[string]string{"example.com/mine": "yes"},
+	})
+	require.NoError(t, err)
+
+	requireCachedBlobSource(t, ctx, s, target, blob)
+	_, info := snapshotIDInfo(t, ctx, s, target)
+	assert.Equal(t, "yes", info.Labels["example.com/mine"], "the caller's update must still apply")
+
+	// An attempt to clear them outright is ignored for the same reason.
+	_, err = s.Update(ctx, snapshots.Info{
+		Name:   target,
+		Labels: map[string]string{blobSourceRefLabel: ""},
+	}, "labels."+blobSourceRefLabel)
+	require.NoError(t, err)
+	requireCachedBlobSource(t, ctx, s, target, blob)
+}
+
+// TestCacheStaleBlob covers a cache entry pruned after a snapshot was staged
+// from it. Mounts must not report a source that does not resolve as populated
+// content, because the caller would then skip a layer whose content is not
+// there.
 func TestCacheStaleBlob(t *testing.T) {
 	ctx := namespaces.WithNamespace(context.Background(), "test")
 
@@ -1258,16 +1411,33 @@ func TestCacheStaleBlob(t *testing.T) {
 	require.NoError(t, os.Remove(childBlob))
 
 	_, err := s.Mounts(ctx, key)
-	require.Error(t, err, "a dangling staged blob must not pass as staged content")
+	require.Error(t, err, "an unresolvable blob source must not pass as populated content")
 	assert.ErrorIs(t, err, os.ErrNotExist)
+
+	// The same holds with no parent, where the snapshot's own layer is the
+	// whole mount. A source that does not resolve must not fall through to a
+	// writable mount, which would report the layer as unapplied.
+	t.Run("parentless", func(t *testing.T) {
+		cacheDir := t.TempDir()
+		diffID := digest.Digest(cacheTestDiffID)
+		blob := writeCacheBlob(t, cacheDir, diffID, []byte("fake erofs blob"))
+		s := newCacheSnapshotter(t, WithLayerContentCaches(cacheDir))
+
+		key := stageCacheHit(t, ctx, s, cacheTestChainID, diffID)
+		require.NoError(t, os.Remove(blob))
+
+		mounts, err := s.Mounts(ctx, key)
+		require.Error(t, err, "an unresolvable blob source must not pass as populated content")
+		assert.ErrorIs(t, err, os.ErrNotExist)
+		assert.Nil(t, mounts, "no writable fallback may be handed out")
+	})
 }
 
 // TestCacheStaleBlobCommit covers a cache entry pruned between Prepare and
 // Commit, which is the window the unpacker runs in: it commits the mounts
 // Prepare returned and never calls Mounts. Commit must not fall back to
-// converting the snapshot. mkfs.erofs would write through the dangling symlink,
-// committing an empty layer and leaving that empty blob in the cache for every
-// future pull of the layer.
+// converting the snapshot, which would commit an empty layer for content the
+// snapshotter reported as already there.
 func TestCacheStaleBlobCommit(t *testing.T) {
 	ctx := namespaces.WithNamespace(context.Background(), "test")
 
@@ -1284,19 +1454,19 @@ func TestCacheStaleBlobCommit(t *testing.T) {
 	require.NoError(t, os.Remove(blob))
 
 	err := s.Commit(ctx, target, key)
-	require.Error(t, err, "a dangling staged blob must not be converted")
+	require.Error(t, err, "an unresolvable blob source must not be converted")
 	assert.ErrorIs(t, err, os.ErrNotExist)
 
-	// The symlink target was not written back through.
+	// Nothing was written back to where the entry was.
 	_, err = os.Stat(blob)
 	assert.ErrorIs(t, err, os.ErrNotExist, "the pruned cache entry must not be recreated")
 	_, err = s.Stat(ctx, target)
 	assert.Error(t, err, "the layer must not be committed")
 }
 
-// TestCacheRemove covers removal of a cache-hit snapshot: it succeeds (the
-// setImmutable guard skips the symlink), removes the snapshot dir/symlink, and
-// leaves the operator-owned cache blob and sidecar untouched.
+// TestCacheRemove covers removal of a cache-hit snapshot. Nothing in the
+// snapshot refers to the cache entry. Removal takes the snapshot directory and
+// leaves the operator-owned blob and sidecar untouched.
 func TestCacheRemove(t *testing.T) {
 	ctx := namespaces.WithNamespace(context.Background(), "test")
 
@@ -1314,11 +1484,9 @@ func TestCacheRemove(t *testing.T) {
 
 	snapDir := filepath.Dir(s.layerBlobPath(snapshotID(t, ctx, s, target)))
 
-	// Remove must succeed (the symlinked blob is skipped by the setImmutable guard,
-	// which would otherwise follow the link and ioctl the operator-owned blob).
 	require.NoError(t, s.Remove(ctx, target))
 
-	// The snapshot dir (and its symlink) is gone, but the cache is untouched.
+	// The snapshot dir is gone, but the cache is untouched.
 	_, err := os.Stat(snapDir)
 	assert.True(t, os.IsNotExist(err), "snapshot dir should be removed")
 	_, err = os.Stat(blob)
@@ -1327,9 +1495,47 @@ func TestCacheRemove(t *testing.T) {
 	require.NoError(t, err, "cache sidecar must be untouched by Remove")
 }
 
-// TestCacheDmverity covers dmverity_mode="on": a cache entry with a sidecar is
-// staged (and the sidecar copied), while an entry missing its required sidecar is
-// a hard error (not staged, nothing committed).
+// TestLinkedLayerBlobIsNotOurs covers a layer.erofs that is a link. An earlier
+// version of the layer content cache left these on disk. Such a path names
+// content outside the snapshot. Commit and Remove both leave the link target
+// untouched.
+func TestLinkedLayerBlobIsNotOurs(t *testing.T) {
+	ctx := namespaces.WithNamespace(context.Background(), "test")
+
+	s := newCacheSnapshotter(t, WithImmutable())
+
+	// Stand in for a shared blob owned by somebody else.
+	shared := filepath.Join(t.TempDir(), "shared.erofs")
+	require.NoError(t, os.WriteFile(shared, []byte("not ours to touch"), 0644))
+
+	key := "extract-1 linked"
+	_, err := s.Prepare(ctx, key, "")
+	require.NoError(t, err)
+	id := snapshotID(t, ctx, s, key)
+	require.NoError(t, os.Symlink(shared, s.layerBlobPath(id)))
+
+	// Commit must not convert into it, nor set attributes on it.
+	target := cacheTestChainID
+	err = s.Commit(ctx, target, key)
+	require.Error(t, err, "a linked layer blob must not be committed as the snapshot's own")
+	assert.ErrorIs(t, err, errdefs.ErrFailedPrecondition)
+
+	data, err := os.ReadFile(shared)
+	require.NoError(t, err)
+	assert.Equal(t, "not ours to touch", string(data), "the link target must be untouched")
+
+	// Removal still works, without following the link to clear attributes on
+	// its target.
+	require.NoError(t, s.Remove(ctx, key))
+	_, err = os.Lstat(s.layerBlobPath(id))
+	assert.ErrorIs(t, err, os.ErrNotExist, "the snapshot, and its link, are gone")
+	_, err = os.Stat(shared)
+	require.NoError(t, err, "the link target must survive removal")
+}
+
+// TestCacheDmverity covers dmverity_mode="on". A cache entry with a sidecar is
+// staged and pinned by the sidecar beside it. An entry missing its required
+// sidecar is a hard error: nothing is staged and nothing is committed.
 func TestCacheDmverity(t *testing.T) {
 	if supported, err := dmverity.IsSupported(); err != nil || !supported {
 		t.Skip("dm-verity is not supported on this system")
@@ -1338,7 +1544,7 @@ func TestCacheDmverity(t *testing.T) {
 	diffID := digest.Digest(cacheTestDiffID)
 	target := cacheTestChainID
 
-	t.Run("with sidecar stages and copies it", func(t *testing.T) {
+	t.Run("with sidecar is staged using it", func(t *testing.T) {
 		cacheDir := t.TempDir()
 		blob := writeCacheBlob(t, cacheDir, diffID, []byte("fake erofs blob"))
 		require.NoError(t, os.WriteFile(dmverity.MetadataPath(blob), []byte(testDmverityMetadata), 0644))
@@ -1346,8 +1552,11 @@ func TestCacheDmverity(t *testing.T) {
 		s := newCacheSnapshotter(t, WithLayerContentCaches(cacheDir), WithDmverityMode("on"))
 		key := stageCacheHit(t, ctx, s, target, diffID)
 
-		_, err := os.Stat(dmverity.MetadataPath(s.layerBlobPath(snapshotID(t, ctx, s, key))))
-		require.NoError(t, err, "sidecar must be present for a dmverity_mode=on hit")
+		mounts, err := s.Mounts(ctx, key)
+		require.NoError(t, err)
+		require.NotEmpty(t, mounts)
+		assert.Contains(t, mounts[0].Options, "X-containerd.dmverity="+dmverity.MetadataPath(blob),
+			"the cache entry's sidecar pins the blob it sits beside")
 	})
 
 	t.Run("without sidecar fails the pull", func(t *testing.T) {
@@ -1375,13 +1584,14 @@ func TestCacheMultipleDirs(t *testing.T) {
 	diffID := digest.Digest(cacheTestDiffID)
 	target := cacheTestChainID
 
-	// stagedBlob returns the cache blob the snapshot's layer.erofs symlink points
-	// at, i.e. which of the configured caches actually served the layer.
+	// stagedBlob returns the cache blob recorded as the snapshot's layer source,
+	// which is the cache that served the layer.
 	stagedBlob := func(t *testing.T, s *snapshotter, key string) string {
 		t.Helper()
-		dst, err := os.Readlink(s.layerBlobPath(snapshotID(t, ctx, s, key)))
+		_, info := snapshotIDInfo(t, ctx, s, key)
+		src, err := blobSourceFromInfo(info)
 		require.NoError(t, err)
-		return dst
+		return src.Ref
 	}
 
 	t.Run("first cache with the blob wins", func(t *testing.T) {
@@ -1420,9 +1630,10 @@ func TestCacheMultipleDirs(t *testing.T) {
 }
 
 // TestCacheDirMustBeAbsolute covers the one thing NewSnapshotter checks about a
-// configured cache dir: it must be absolute, since a relative one would be
-// symlinked into the snapshot dir and dangle. The empty string is covered by the
-// same check, which otherwise resolves to the daemon's working directory.
+// configured cache dir. It must be absolute. A relative path is recorded in a
+// snapshot and mounted later, and would resolve against the working directory
+// of whichever process performs the mount. The empty string is covered by the
+// same check.
 func TestCacheDirMustBeAbsolute(t *testing.T) {
 	requireErofs(t)
 

plugins/snapshots/erofs/plugin/plugin.go

@@ -59,10 +59,15 @@ type Config struct {
 	DmverityMode string `toml:"dmverity_mode"`
 
 	// LayerContentCaches lists directories of pre-converted, diffID-keyed erofs
-	// layer blobs. The directories are searched in the order given and the first
-	// one holding the layer wins, in both sequential and parallel unpack modes.
-	// A directory that doesn't exist is treated as a cache miss. A layer found
-	// in none of them is converted normally.
+	// layer blobs. A layer found in one of them is mounted from that directory,
+	// in both sequential and parallel unpack modes. The directories are
+	// searched in the order given and the first one holding the layer wins; one
+	// that doesn't exist is treated as a miss. A layer found in none of them is
+	// converted normally.
+	//
+	// The blobs are read but never written and are expected to outlive the
+	// snapshots using them. Removing one while an image still refers to it
+	// leaves that image unusable.
 	LayerContentCaches []string `toml:"layer_content_caches"`
 }