diff --git a/internal/erofsutils/mount_test.go b/internal/erofsutils/mount_test.go
index 67bf90c0a3..8ae6064421 100644
--- a/internal/erofsutils/mount_test.go
+++ b/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)
diff --git a/plugins/snapshots/erofs/blobsource.go b/plugins/snapshots/erofs/blobsource.go
new file mode 100644
index 0000000000..b4af5638c4
--- /dev/null
+++ b/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
+}
diff --git a/plugins/snapshots/erofs/blobsource_linux_test.go b/plugins/snapshots/erofs/blobsource_linux_test.go
new file mode 100644
index 0000000000..a248767870
--- /dev/null
+++ b/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")
+}
diff --git a/plugins/snapshots/erofs/erofs.go b/plugins/snapshots/erofs/erofs.go
index db551ecba5..c9fcc84ef2 100644
--- a/plugins/snapshots/erofs/erofs.go
+++ b/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
 	})
diff --git a/plugins/snapshots/erofs/erofs_linux_test.go b/plugins/snapshots/erofs/erofs_linux_test.go
index a02ce24e68..751f76ab55 100644
--- a/plugins/snapshots/erofs/erofs_linux_test.go
+++ b/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)
 
diff --git a/plugins/snapshots/erofs/plugin/plugin.go b/plugins/snapshots/erofs/plugin/plugin.go
index d1f99118f4..960f0c3bfe 100644
--- a/plugins/snapshots/erofs/plugin/plugin.go
+++ b/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"`
 }