diff --git a/core/mount/mount.go b/core/mount/mount.go
index 58c3c57b78a59..4caecfc451984 100644
--- a/core/mount/mount.go
+++ b/core/mount/mount.go
@@ -93,10 +93,35 @@ func CanonicalizePath(path string) (string, error) {
 	return filepath.EvalSymlinks(path)
 }
 
-// ReadOnly returns a boolean value indicating whether this mount has the "ro"
-// option set.
+// ReadOnly reports whether this mount is read-only, deriving it from the mount
+// type where the options alone don't say so.
 func (m *Mount) ReadOnly() bool {
-	return slices.Contains(m.Options, "ro")
+	typ := m.Type
+	// The mount type may carry "/"-separated modifiers meaningful only to the
+	// mount manager (e.g. "format/mkdir/overlay"), so only its last segment is
+	// considered.
+	if i := strings.LastIndex(typ, "/"); i >= 0 {
+		typ = typ[i+1:]
+	}
+	switch typ {
+	case "erofs":
+		// Read-only by construction, whatever the options say.
+		return true
+	case "overlay":
+		// Writable only through an upperdir, which a snapshotter signals by
+		// setting it rather than by setting "rw". An element may be a
+		// comma-joined fragment ("lowerdir=a,upperdir=b"), so split first.
+		options := strings.Split(strings.Join(m.Options, ","), ",")
+		// An explicit "ro" wins over an upperdir.
+		if slices.Contains(options, "ro") {
+			return true
+		}
+		return !slices.ContainsFunc(options, func(o string) bool {
+			return strings.HasPrefix(o, "upperdir=")
+		})
+	default:
+		return slices.Contains(m.Options, "ro")
+	}
 }
 
 // Mount to the provided target path.
diff --git a/core/mount/mount_test.go b/core/mount/mount_test.go
index 74694b1227e31..995d8a6a4c029 100644
--- a/core/mount/mount_test.go
+++ b/core/mount/mount_test.go
@@ -149,6 +149,71 @@ func TestReadonlyMounts(t *testing.T) {
 	}
 }
 
+func TestMountReadOnly(t *testing.T) {
+	testCases := []struct {
+		desc     string
+		mount    Mount
+		expected bool
+	}{
+		{
+			desc:     "erofs is always read-only",
+			mount:    Mount{Type: "erofs", Source: "/path/to/layer.erofs", Options: []string{"loop"}},
+			expected: true,
+		},
+		{
+			desc:     "overlay without upperdir is read-only",
+			mount:    Mount{Type: "overlay", Source: "overlay", Options: []string{"lowerdir=/lower"}},
+			expected: true,
+		},
+		{
+			desc:     "overlay with upperdir is writable",
+			mount:    Mount{Type: "overlay", Source: "overlay", Options: []string{"lowerdir=/lower", "upperdir=/upper"}},
+			expected: false,
+		},
+		{
+			desc:     "overlay with upperdir packed into a comma-joined options string is writable",
+			mount:    Mount{Type: "overlay", Source: "overlay", Options: []string{"lowerdir=/lower,upperdir=/upper,workdir=/work"}},
+			expected: false,
+		},
+		{
+			desc:     "type modifiers are stripped before matching overlay",
+			mount:    Mount{Type: "format/mkdir/overlay", Source: "overlay", Options: []string{"lowerdir=/lower"}},
+			expected: true,
+		},
+		{
+			desc:     "type modifiers are stripped, overlay with upperdir still writable",
+			mount:    Mount{Type: "format/mkdir/overlay", Source: "overlay", Options: []string{"upperdir=/upper"}},
+			expected: false,
+		},
+		{
+			desc:     "overlay with an explicit `ro` option is read-only despite upperdir",
+			mount:    Mount{Type: "overlay", Source: "overlay", Options: []string{"lowerdir=/lower", "upperdir=/upper", "ro"}},
+			expected: true,
+		},
+		{
+			desc:     "overlay `ro` packed into a comma-joined options string is read-only",
+			mount:    Mount{Type: "overlay", Source: "overlay", Options: []string{"lowerdir=/lower,upperdir=/upper,ro"}},
+			expected: true,
+		},
+		{
+			desc:     "other types are read-only only with the `ro` option",
+			mount:    Mount{Type: "bind", Source: "/path", Options: []string{"ro", "rbind"}},
+			expected: true,
+		},
+		{
+			desc:     "other types are writable without the `ro` option",
+			mount:    Mount{Type: "bind", Source: "/path", Options: []string{"rbind"}},
+			expected: false,
+		},
+	}
+
+	for _, tc := range testCases {
+		if got := tc.mount.ReadOnly(); got != tc.expected {
+			t.Errorf("%s: ReadOnly() = %v, want %v", tc.desc, got, tc.expected)
+		}
+	}
+}
+
 func TestRemoveVolatileTempMount(t *testing.T) {
 	testCases := []struct {
 		desc     string
diff --git a/core/snapshots/snapshotter.go b/core/snapshots/snapshotter.go
index 9140a0644e0e2..c6a7101e6e795 100644
--- a/core/snapshots/snapshotter.go
+++ b/core/snapshots/snapshotter.go
@@ -58,6 +58,12 @@ const (
 	// Ignoring is not a failure — callers that require enforcement must
 	// pick a snapshotter that supports it.
 	LabelSnapshotMaxSize = "containerd.io/snapshot/max-size"
+
+	// RebaseCap is a snapshotter capability (advertised via the plugin's metadata)
+	// indicating that an active snapshot may be committed with a parent supplied at
+	// Commit time (via WithParent). It lets the unpacker prepare and apply layers in
+	// parallel and rebase the chain into place at commit.
+	RebaseCap = "rebase"
 )
 
 // Kind identifies the kind of snapshot.
diff --git a/core/unpack/unpacker.go b/core/unpack/unpacker.go
index ec55e65317179..d0313e80c821e 100644
--- a/core/unpack/unpacker.go
+++ b/core/unpack/unpacker.go
@@ -404,7 +404,11 @@ func (u *Unpacker) unpack(
 		var (
 			key    string
 			mounts []mount.Mount
-			opts   = append(unpack.SnapshotOpts, snapshots.WithLabels(snapshotLabels))
+			// Clone before appending: topHalf runs concurrently per layer in
+			// parallel mode, and appending directly to unpack.SnapshotOpts could
+			// write into its shared backing array from multiple goroutines.
+			opts   = append(slices.Clone(unpack.SnapshotOpts), snapshots.WithLabels(snapshotLabels))
+			staged bool
 		)
 
 		for try := 1; try <= 3; try++ {
@@ -435,6 +439,13 @@ func (u *Unpacker) unpack(
 			return nil, fmt.Errorf("unable to prepare extraction snapshot: %w", err)
 		}
 
+		if isStaged(mounts) {
+			// The snapshotter staged the layer content into the active snapshot
+			// as read-only (e.g. a layer content cache hit). Skip fetch+apply,
+			// but still commit it below (which applies the parent).
+			staged = true
+		}
+
 		// Abort the snapshot if commit does not happen
 		abort := func(ctx context.Context) {
 			if err := sn.Remove(ctx, key); err != nil {
@@ -442,6 +453,61 @@ func (u *Unpacker) unpack(
 			}
 		}
 
+		// commitF is the bottom half shared by normal and staged layers: it rebases
+		// in the real parent (parallel mode) and commits the snapshot. Staged layers
+		// have no fetched content, so they skip the post-apply uncompressed label.
+		commitF := func(shouldAbort bool) error {
+			defer unlock()
+			if shouldAbort {
+				cleanup.Do(ctx, abort)
+				return nil
+			}
+
+			if i > 0 && parallel {
+				opts = append(opts, snapshots.WithParent(chainIDs[i-1].String()))
+			}
+			if err := sn.Commit(ctx, chainID, key, opts...); err != nil {
+				cleanup.Do(ctx, abort)
+				if errdefs.IsAlreadyExists(err) {
+					return nil
+				}
+				return fmt.Errorf("failed to commit snapshot %s: %w", key, err)
+			}
+
+			if staged {
+				// No layer was fetched, so there is no content to label.
+				return nil
+			}
+
+			// Set the uncompressed label after the uncompressed
+			// digest has been verified through apply.
+			cinfo := content.Info{
+				Digest: desc.Digest,
+				Labels: map[string]string{
+					labels.LabelUncompressed: diffIDs[i].String(),
+				},
+			}
+			if _, err := cs.Update(ctx, cinfo, "labels."+labels.LabelUncompressed); err != nil {
+				return err
+			}
+			return nil
+		}
+
+		if staged {
+			// Content is already staged in the active snapshot; there is nothing to
+			// fetch or apply. Emit a status that runs commitF in the (serialized)
+			// bottom half so the parent is rebased in and the chain is linked.
+			resCh := make(chan *unpackStatus, 1)
+			resCh <- &unpackStatus{
+				desc:    desc,
+				span:    span,
+				startAt: startAt,
+				bottomF: commitF,
+			}
+			close(resCh)
+			return resCh, nil
+		}
+
 		if fetchErr == nil {
 			fetchOffset = i
 			n := len(layers) - fetchOffset
@@ -478,38 +544,7 @@ func (u *Unpacker) unpack(
 				desc:    desc,
 				span:    span,
 				startAt: startAt,
-				bottomF: func(shouldAbort bool) error {
-					defer unlock()
-					if shouldAbort {
-						cleanup.Do(ctx, abort)
-						return nil
-					}
-
-					if i > 0 && parallel {
-						parent = chainIDs[i-1].String()
-						opts = append(opts, snapshots.WithParent(parent))
-					}
-					if err = sn.Commit(ctx, chainID, key, opts...); err != nil {
-						cleanup.Do(ctx, abort)
-						if errdefs.IsAlreadyExists(err) {
-							return nil
-						}
-						return fmt.Errorf("failed to commit snapshot %s: %w", key, err)
-					}
-
-					// Set the uncompressed label after the uncompressed
-					// digest has been verified through apply.
-					cinfo := content.Info{
-						Digest: desc.Digest,
-						Labels: map[string]string{
-							labels.LabelUncompressed: diffIDs[i].String(),
-						},
-					}
-					if _, err := cs.Update(ctx, cinfo, "labels."+labels.LabelUncompressed); err != nil {
-						return err
-					}
-					return nil
-				},
+				bottomF: commitF,
 			}
 
 			select {
@@ -746,7 +781,7 @@ func (u *Unpacker) supportParallel(unpack *Platform) bool {
 	if u.unpackLimiter == nil {
 		return false
 	}
-	if !slices.Contains(unpack.SnapshotterCapabilities, "rebase") {
+	if !slices.Contains(unpack.SnapshotterCapabilities, snapshots.RebaseCap) {
 		log.L.Infof("snapshotter does not support rebase capability, unpacking will be sequential")
 		return false
 	}
@@ -761,6 +796,24 @@ func uniquePart() string {
 	return fmt.Sprintf("%d-%s", t.Nanosecond(), base64.URLEncoding.EncodeToString(b[:]))
 }
 
+// isStaged reports whether a successful Prepare has already staged the
+// layer's content into the active snapshot instead of returning a normal,
+// writable active snapshot (e.g. a snapshotter serving the layer from a local
+// content cache). There is nothing to write into a staged snapshot, so the
+// caller should skip fetching and applying the layer, and just Commit the
+// snapshot as-is (applying the real parent at Commit time).
+//
+// Only the last mount in the slice is inspected: earlier entries are inputs
+// consumed by mount templating (e.g. "{{ mount 0 }}" in an overlay's
+// lowerdir) rather than the mount that is actually stacked on top, so they
+// carry no information about writability.
+func isStaged(mounts []mount.Mount) bool {
+	if len(mounts) == 0 {
+		return false
+	}
+	return mounts[len(mounts)-1].ReadOnly()
+}
+
 // TODO: this is a temporary workaround until #13053 lands.
 func bindToOverlay(mounts []mount.Mount) []mount.Mount {
 	if len(mounts) != 1 || mounts[0].Type != "bind" {
diff --git a/core/unpack/unpacker_test.go b/core/unpack/unpacker_test.go
index bdd08348f56f7..cce256a9b8446 100644
--- a/core/unpack/unpacker_test.go
+++ b/core/unpack/unpacker_test.go
@@ -17,14 +17,25 @@
 package unpack
 
 import (
+	"context"
 	"crypto/rand"
 	"fmt"
 	"reflect"
 	"testing"
 
-	"github.com/containerd/containerd/v2/core/mount"
 	"github.com/opencontainers/go-digest"
 	"github.com/opencontainers/image-spec/identity"
+	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"golang.org/x/sync/semaphore"
+
+	"github.com/containerd/containerd/v2/core/diff"
+	"github.com/containerd/containerd/v2/core/images"
+	"github.com/containerd/containerd/v2/core/images/imagetest"
+	"github.com/containerd/containerd/v2/core/mount"
+	"github.com/containerd/containerd/v2/core/snapshots"
+	"github.com/containerd/platforms"
 )
 
 func generateRandomDiffIDs(t testing.TB, num int) []digest.Digest {
@@ -177,3 +188,137 @@ func TestBindToOverlay(t *testing.T) {
 		})
 	}
 }
+
+func TestIsStaged(t *testing.T) {
+	testCases := []struct {
+		name   string
+		mounts []mount.Mount
+		expect bool
+	}{
+		{
+			name:   "no mounts",
+			mounts: nil,
+			expect: false,
+		},
+		{
+			name: "read-only mount",
+			mounts: []mount.Mount{
+				{Type: "erofs", Source: "/path/to/layer.erofs", Options: []string{"loop"}},
+			},
+			expect: true,
+		},
+		{
+			name: "writable mount",
+			mounts: []mount.Mount{
+				{Type: "bind", Source: "/path", Options: []string{"rbind"}},
+			},
+			expect: false,
+		},
+		{
+			name: "only the last mount is inspected",
+			mounts: []mount.Mount{
+				{Type: "bind", Source: "/lower", Options: []string{"rbind"}},
+				{Type: "erofs", Source: "/path/to/layer.erofs", Options: []string{"loop"}},
+			},
+			expect: true,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			assert.Equal(t, tc.expect, isStaged(tc.mounts))
+		})
+	}
+}
+
+// stagedSnapshotter reports every layer as staged (read-only mounts) and
+// records the Prepare/Commit calls. Only Prepare and Commit are exercised on
+// the staged path, so the embedded (nil) Snapshotter covers the rest of the
+// interface.
+type stagedSnapshotter struct {
+	snapshots.Snapshotter
+	prepares []call
+	commits  []call
+}
+
+type call struct{ name, key, parent string }
+
+func (s *stagedSnapshotter) Prepare(_ context.Context, key, parent string, _ ...snapshots.Opt) ([]mount.Mount, error) {
+	s.prepares = append(s.prepares, call{key: key, parent: parent})
+	return []mount.Mount{{Type: "erofs", Source: "/staged/layer.erofs", Options: []string{"ro"}}}, nil
+}
+
+func (s *stagedSnapshotter) Commit(_ context.Context, name, key string, opts ...snapshots.Opt) error {
+	var info snapshots.Info
+	for _, o := range opts {
+		_ = o(&info)
+	}
+	s.commits = append(s.commits, call{name: name, key: key, parent: info.Parent})
+	return nil
+}
+
+// failApplier fails the test if Apply is called; a staged layer is never applied.
+type failApplier struct{ t *testing.T }
+
+func (a failApplier) Apply(_ context.Context, desc ocispec.Descriptor, _ []mount.Mount, _ ...diff.ApplyOpt) (ocispec.Descriptor, error) {
+	a.t.Errorf("Apply must not be called for a staged layer (%s)", desc.Digest)
+	return ocispec.Descriptor{}, nil
+}
+
+// TestUnpackStagedLayers verifies that when the snapshotter reports layers as
+// staged (read-only mounts from Prepare) in parallel mode, the unpacker skips
+// fetch+apply but still commits each layer, rebasing the real parent in at
+// Commit time.
+func TestUnpackStagedLayers(t *testing.T) {
+	ctx := context.Background()
+
+	diffIDs := generateRandomDiffIDs(t, 2)
+	chainIDs := identity.ChainIDs(append([]digest.Digest{}, diffIDs...))
+	layers := []ocispec.Descriptor{
+		{MediaType: ocispec.MediaTypeImageLayerGzip, Digest: digest.FromString("layer-0"), Size: 1},
+		{MediaType: ocispec.MediaTypeImageLayerGzip, Digest: digest.FromString("layer-1"), Size: 1},
+	}
+
+	cs := imagetest.NewContentStore(ctx, t)
+
+	// Minimal image config carrying the layer diffIDs.
+	config := cs.JSONObject(ocispec.MediaTypeImageConfig, struct {
+		ocispec.Platform
+		RootFS ocispec.RootFS `json:"rootfs"`
+	}{
+		Platform: ocispec.Platform{OS: "linux", Architecture: "amd64"},
+		RootFS:   ocispec.RootFS{Type: "layers", DiffIDs: diffIDs},
+	}).Descriptor
+
+	sn := &stagedSnapshotter{}
+	u, err := NewUnpacker(ctx, cs.Store,
+		WithUnpackLimiter(semaphore.NewWeighted(4)),
+		WithUnpackPlatform(Platform{
+			Platform:                platforms.All,
+			Snapshotter:             sn,
+			Applier:                 failApplier{t},
+			SnapshotterCapabilities: []string{snapshots.RebaseCap},
+		}),
+	)
+	require.NoError(t, err)
+
+	// A staged layer must never be fetched.
+	fetch := images.HandlerFunc(func(_ context.Context, desc ocispec.Descriptor) ([]ocispec.Descriptor, error) {
+		t.Errorf("fetch must not happen for a staged layer (%s)", desc.Digest)
+		return nil, nil
+	})
+
+	require.NoError(t, u.unpack(fetch, config, layers))
+
+	// Parallel mode: Prepare gets no parent...
+	require.Len(t, sn.prepares, 2)
+	assert.Equal(t, "", sn.prepares[0].parent)
+	assert.Equal(t, "", sn.prepares[1].parent)
+
+	// ...and the parent is rebased in at Commit.
+	require.Len(t, sn.commits, 2)
+	assert.Equal(t, chainIDs[0].String(), sn.commits[0].name)
+	assert.Equal(t, "", sn.commits[0].parent)
+	assert.Equal(t, chainIDs[1].String(), sn.commits[1].name)
+	assert.Equal(t, chainIDs[0].String(), sn.commits[1].parent)
+}
diff --git a/plugins/snapshots/erofs/erofs.go b/plugins/snapshots/erofs/erofs.go
index 9224f8b5c21b0..4892bb420f2cc 100644
--- a/plugins/snapshots/erofs/erofs.go
+++ b/plugins/snapshots/erofs/erofs.go
@@ -54,9 +54,9 @@ type SnapshotterConfig struct {
 	dmverityMode string
 	// layerContentCache is a directory of pre-converted, diffID-keyed erofs
 	// layer blobs. When set and an unpacked layer's blob is present, the
-	// snapshotter commits the layer immediately (symlinking the blob) and
-	// returns ErrAlreadyExists, skipping the download and tar->erofs
-	// conversion. Empty disables the feature.
+	// snapshotter stages the blob (as a symlink) into the active snapshot,
+	// skipping the download and tar->erofs conversion. Only parentless Prepares
+	// can be served. Empty disables the feature.
 	layerContentCache string
 }
 
@@ -173,11 +173,10 @@ func NewSnapshotter(root string, opts ...Opt) (snapshots.Snapshotter, error) {
 		}
 	}
 
-	// Cache blobs may live on a read-only mount the snapshotter can't modify, so
-	// fsverity and IMMUTABLE_FL can't be applied to them. Be explicit about this to
-	// the user instead of ignoring them silently (they're bypassed because cache
-	// hits commit during Prepare and skip Commit); dm-verity is the cache's
-	// integrity mechanism.
+	// 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.
 	if config.layerContentCache != "" {
 		if config.enableFsverity {
 			return nil, fmt.Errorf("enable_fsverity is incompatible with layer_content_cache; use dm-verity for cache integrity")
@@ -298,7 +297,7 @@ func (s *snapshotter) lowerPath(id string) (string, error) {
 	return layerBlob, nil
 }
 
-func (s *snapshotter) prepareDirectory(ctx context.Context, snapshotDir string, kind snapshots.Kind, entry *cacheEntry) (string, error) {
+func (s *snapshotter) prepareDirectory(ctx context.Context, snapshotDir string, kind snapshots.Kind, cacheBlob string) (string, error) {
 	td, err := os.MkdirTemp(snapshotDir, "new-")
 	if err != nil {
 		return "", fmt.Errorf("failed to create temp dir: %w", err)
@@ -320,11 +319,11 @@ func (s *snapshotter) prepareDirectory(ctx context.Context, snapshotDir string,
 		}
 	}
 
-	// Layer content cache hit: stage the pre-converted blob as a symlink so the
-	// caller's rename publishes a ready committed layer.
-	if entry != nil {
+	// 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(entry.blob, layerBlob); err != nil {
+		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,
@@ -332,7 +331,7 @@ func (s *snapshotter) prepareDirectory(ctx context.Context, snapshotDir string,
 		// 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(entry.blob)); err != nil {
+			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)
 				}
@@ -422,9 +421,6 @@ func (s *snapshotter) mounts(snap storage.Snapshot, info snapshots.Info) ([]moun
 
 	if len(snap.ParentIDs) == 0 {
 		if layerBlob, err := s.lowerPath(snap.ID); err == nil {
-			if snap.Kind != snapshots.KindView {
-				return nil, fmt.Errorf("only works for snapshots.KindView on a committed snapshot: %w", err)
-			}
 			if s.enableFsverity {
 				if err := s.verifyFsverity(layerBlob); err != nil {
 					return nil, err
@@ -578,11 +574,12 @@ func (s *snapshotter) mounts(snap storage.Snapshot, info snapshots.Info) ([]moun
 }
 
 // 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 instead stages the cached blob and commits the snapshot as the target
-// chainID in the same transaction, then returns ErrAlreadyExists (the
-// remote-snapshot signal that makes the unpacker skip the layer download and
-// conversion).
+// On a parentless image-layer extraction whose diffID blob is in the layer content cache,
+// it stages the cached blob into the active snapshot (without committing) and
+// returns it as a read-only mount, so the unpacker can detect the fast path
+// (skip the layer download and conversion) while still committing the
+// snapshot normally — applying the parent at Commit time, which keeps the
+// cache compatible with 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
@@ -590,21 +587,21 @@ func (s *snapshotter) createSnapshot(ctx context.Context, kind snapshots.Kind, k
 		info     snapshots.Info
 	)
 
-	// Only image-layer extractions (active snapshots) can be served from the
-	// layer content cache; View and container-rootfs Prepares get a nil entry and
-	// fall through to the normal path.
-	var entry *cacheEntry
-	if kind == snapshots.KindActive {
-		entry = s.lookupCache(ctx, opts...)
+	// Only parentless extractions can be served: s.mounts picks a staged blob up
+	// only when there are no parents, so with a parent the differ would write
+	// through the staged symlink into the shared cache blob.
+	var cacheBlob string
+	if kind == snapshots.KindActive && parent == "" {
+		if cacheBlob = s.lookupCache(ctx, opts...); cacheBlob != "" {
+			log.G(ctx).WithFields(log.Fields{
+				"key":  key,
+				"blob": cacheBlob,
+			}).Debug("layer content cache hit, staged cached erofs blob")
+		}
 	}
 
-	// committed is set only once the cached layer is committed and we deliberately
-	// return ErrAlreadyExists; the committed dir must then be kept. Any real error
-	// (including an unexpected AlreadyExists from CreateSnapshot) leaves it false
-	// so the staged td/path is reclaimed.
-	var committed bool
 	defer func() {
-		if err != nil && !committed {
+		if err != nil {
 			if td != "" {
 				if err1 := os.RemoveAll(td); err1 != nil {
 					log.G(ctx).WithError(err1).Warn("failed to cleanup temp snapshot directory")
@@ -620,7 +617,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, entry)
+	td, err = s.prepareDirectory(ctx, snapshotDir, kind, cacheBlob)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create prepare snapshot dir: %w", err)
 	}
@@ -691,34 +688,11 @@ func (s *snapshotter) createSnapshot(ctx context.Context, kind snapshots.Kind, k
 			return fmt.Errorf("failed to rename: %w", err)
 		}
 		td = ""
-
-		// Commit the cached layer straight away as the target chainID. CommitActive
-		// replaces labels with those from opts (which carry snapshot.ref), which the
-		// metadata layer's Walk filter needs to resolve the backend target.
-		if entry != nil {
-			if _, err = storage.CommitActive(ctx, key, entry.target, snapshots.Usage{}, opts...); err != nil {
-				return fmt.Errorf("unable to commit active snapshot: %w", err)
-			}
-		}
 		return nil
 	}); err != nil {
 		return nil, err
 	}
 
-	// Cache hit committed successfully: signal the unpacker via ErrAlreadyExists to
-	// skip the layer download and conversion. (A concurrent pull that already
-	// committed the same target returned a plain AlreadyExists error above, which
-	// the metadata layer resolves the same way.)
-	if entry != nil {
-		log.G(ctx).WithFields(log.Fields{
-			"key":     key,
-			"chainID": entry.target,
-			"blob":    entry.blob,
-		}).Debug("layer content cache hit, committed cached erofs blob")
-		committed = true
-		return nil, errdefs.ErrAlreadyExists
-	}
-
 	return s.mounts(snap, info)
 }
 
@@ -726,61 +700,55 @@ func (s *snapshotter) Prepare(ctx context.Context, key, parent string, opts ...s
 	return s.createSnapshot(ctx, snapshots.KindActive, key, parent, opts)
 }
 
-// cacheEntry describes a resolved layer content cache entry: the target chainID
-// to commit as and the absolute path of the cached blob to symlink (any
-// dm-verity sidecar is derived from blob via dmverity.MetadataPath).
-type cacheEntry struct {
-	target string
-	blob   string
-}
-
 // cacheBlobPath returns the expected path of the cached erofs blob for a diffID.
 func (s *snapshotter) cacheBlobPath(diffID digest.Digest) string {
 	return erofsutils.CacheBlobPath(s.layerContentCache, diffID)
 }
 
-// lookupCache resolves the layer content cache entry that can serve the layer
-// being prepared, or nil on a miss. It gates on: the cache being configured, the
-// Prepare being an image-layer extraction (carries the snapshot.ref and diff-id
-// labels), and the diffID blob being present. Misses (cache disabled,
-// non-extraction Prepare, missing entries, unreadable cache dirs such as a FUSE
-// mount being down, malformed labels) all return nil so pulls keep working. The
-// dm-verity sidecar is handled when the blob is materialized (prepareDirectory).
-func (s *snapshotter) lookupCache(ctx context.Context, opts ...snapshots.Opt) *cacheEntry {
+// lookupCache returns the absolute path of the cached erofs blob that can serve
+// the layer being prepared, or "" on a miss. It gates on: the cache being
+// configured, the Prepare being an image-layer extraction (carries the
+// snapshot.ref and diff-id labels), and the diffID blob being present. Misses
+// (cache disabled, non-extraction Prepare, missing entries, unreadable cache dirs
+// such as a FUSE mount being down, malformed labels) all return "" so pulls keep
+// working. Any dm-verity sidecar is derived from the blob path (via
+// dmverity.MetadataPath) when the blob is materialized (prepareDirectory).
+func (s *snapshotter) lookupCache(ctx context.Context, opts ...snapshots.Opt) string {
 	if s.layerContentCache == "" {
-		return nil
+		return ""
 	}
 
 	var base snapshots.Info
 	for _, opt := range opts {
 		if err := opt(&base); err != nil {
-			return nil
+			return ""
 		}
 	}
 
-	target := base.Labels[snapshots.LabelSnapshotRef]
 	diffIDStr := base.Labels[snapshots.LabelSnapshotDiffID]
-	if target == "" || diffIDStr == "" {
+	if base.Labels[snapshots.LabelSnapshotRef] == "" || diffIDStr == "" {
 		// Not an image-layer extraction, or no diffID to key on.
-		return nil
+		return ""
 	}
 	diffID, err := digest.Parse(diffIDStr)
 	if err != nil {
 		log.G(ctx).WithError(err).WithField("diffID", diffIDStr).
 			Warn("erofs layer cache: invalid diff-id label, treating as cache miss")
-		return nil
+		return ""
 	}
 
 	blob := s.cacheBlobPath(diffID)
 	if _, err := os.Stat(blob); err != nil {
-		if !os.IsNotExist(err) {
+		if os.IsNotExist(err) {
+			log.G(ctx).WithField("blob", blob).Trace("erofs layer cache miss")
+		} else {
 			log.G(ctx).WithError(err).WithField("blob", blob).
 				Warn("erofs layer cache: failed to stat cache blob, treating as cache miss")
 		}
-		return nil
+		return ""
 	}
 
-	return &cacheEntry{target: target, blob: blob}
+	return blob
 }
 
 func (s *snapshotter) View(ctx context.Context, key, parent string, opts ...snapshots.Opt) ([]mount.Mount, error) {
diff --git a/plugins/snapshots/erofs/erofs_linux_test.go b/plugins/snapshots/erofs/erofs_linux_test.go
index c8edcaf0fa10e..fa08c2a1287bc 100644
--- a/plugins/snapshots/erofs/erofs_linux_test.go
+++ b/plugins/snapshots/erofs/erofs_linux_test.go
@@ -26,7 +26,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/containerd/errdefs"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	bolt "go.etcd.io/bbolt"
@@ -964,18 +963,26 @@ func newCacheSnapshotter(t *testing.T, opts ...Opt) *snapshotter {
 	return sn.(*snapshotter)
 }
 
-// prepareCacheHit runs an extraction Prepare for target/diffID and asserts it was
-// served from the cache (committed and signaled via ErrAlreadyExists, no mounts).
-func prepareCacheHit(t *testing.T, ctx context.Context, s *snapshotter, target string, diffID digest.Digest) {
+// stageCacheHit runs an extraction Prepare for target/diffID and asserts the
+// cache staged the blob into the active snapshot: a read-only mount, with no
+// error (so the unpacker skips fetch+apply but still commits). It returns the
+// extraction key so the caller can Commit it as the target chainID.
+func stageCacheHit(t *testing.T, ctx context.Context, s *snapshotter, target string, diffID digest.Digest) string {
 	t.Helper()
-	mounts, err := s.Prepare(ctx, "extract-1 "+target, "", extractionOpt(target, diffID))
-	require.ErrorIs(t, err, errdefs.ErrAlreadyExists, "cache hit must signal the remote-snapshot protocol")
-	assert.Nil(t, mounts, "a cache hit returns no mounts")
+	key := "extract-1 " + target
+	mounts, err := s.Prepare(ctx, key, "", extractionOpt(target, diffID))
+	require.NoError(t, err, "cache hit must stage without an error")
+	require.NotEmpty(t, mounts, "a staged cache hit returns mounts")
+	for _, m := range mounts {
+		assert.True(t, m.ReadOnly(), "a staged cache hit returns read-only mounts")
+	}
+	return key
 }
 
 // TestCacheHit covers the happy path: an extraction Prepare whose diffID blob is
-// in the cache commits the target chainID (right kind, parent, and snapshot.ref
-// label), symlinks the blob, and returns ErrAlreadyExists.
+// 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.
 func TestCacheHit(t *testing.T) {
 	ctx := namespaces.WithNamespace(context.Background(), "test")
 
@@ -986,18 +993,15 @@ func TestCacheHit(t *testing.T) {
 	s := newCacheSnapshotter(t, WithLayerContentCache(cacheDir))
 
 	target := cacheTestChainID
-	prepareCacheHit(t, ctx, s, target, diffID)
+	key := stageCacheHit(t, ctx, s, target, diffID)
 
-	// The target chainID is committed, with the parent and snapshot.ref label the
-	// metadata layer's Walk filter needs to resolve the backend target.
-	info, err := s.Stat(ctx, target)
-	require.NoError(t, err, "committed snapshot must exist under the target chainID")
-	assert.Equal(t, snapshots.KindCommitted, info.Kind)
-	assert.Equal(t, "", info.Parent)
-	assert.Equal(t, target, info.Labels[snapshots.LabelSnapshotRef])
+	// The blob is staged into the active snapshot but 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.
-	link := s.layerBlobPath(snapshotID(t, ctx, s, target))
+	link := s.layerBlobPath(snapshotID(t, ctx, s, key))
 	fi, err := os.Lstat(link)
 	require.NoError(t, err)
 	assert.NotZero(t, fi.Mode()&os.ModeSymlink, "layer.erofs should be a symlink")
@@ -1005,6 +1009,14 @@ func TestCacheHit(t *testing.T) {
 	require.NoError(t, err)
 	assert.True(t, filepath.IsAbs(dst), "symlink target should be absolute")
 	assert.Equal(t, blob, dst)
+
+	// Commit finalizes the staged 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)
+	require.NoError(t, err, "committed snapshot must exist under the target chainID")
+	assert.Equal(t, snapshots.KindCommitted, info.Kind)
+	assert.Equal(t, "", info.Parent)
 }
 
 // TestCacheSidecar covers a hit in the default "auto" dm-verity mode where the
@@ -1022,11 +1034,11 @@ func TestCacheSidecar(t *testing.T) {
 	s := newCacheSnapshotter(t, WithLayerContentCache(cacheDir))
 
 	target := cacheTestChainID
-	prepareCacheHit(t, ctx, s, target, diffID)
+	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, target)))
+	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")
@@ -1045,11 +1057,16 @@ func TestCacheMiss(t *testing.T) {
 	target := cacheTestChainID
 
 	// Each case must leave the extraction as a normal active snapshot: mounts are
-	// returned and the target chainID is not committed.
-	assertFellThrough := func(t *testing.T, s *snapshotter, mounts []mount.Mount, err error) {
+	// 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).
+	assertFellThrough := func(t *testing.T, s *snapshotter, mounts []mount.Mount, err error, wantRO bool) {
 		t.Helper()
 		require.NoError(t, err)
 		assert.NotEmpty(t, mounts, "a miss must return normal active-snapshot mounts")
+		for _, m := range mounts {
+			assert.Equal(t, wantRO, m.ReadOnly())
+		}
 		_, err = s.Stat(ctx, target)
 		assert.Error(t, err, "target chainID must not be committed on a miss")
 	}
@@ -1057,13 +1074,13 @@ func TestCacheMiss(t *testing.T) {
 	t.Run("cache disabled", func(t *testing.T) {
 		s := newCacheSnapshotter(t) // no cache configured
 		mounts, err := s.Prepare(ctx, "extract-1 "+target, "", extractionOpt(target, diffID))
-		assertFellThrough(t, s, mounts, err)
+		assertFellThrough(t, s, mounts, err, false)
 	})
 
 	t.Run("blob absent", func(t *testing.T) {
 		s := newCacheSnapshotter(t, WithLayerContentCache(t.TempDir()))
 		mounts, err := s.Prepare(ctx, "extract-1 "+target, "", extractionOpt(target, diffID))
-		assertFellThrough(t, s, mounts, err)
+		assertFellThrough(t, s, mounts, err, false)
 	})
 
 	t.Run("no extraction labels", func(t *testing.T) {
@@ -1081,11 +1098,48 @@ func TestCacheMiss(t *testing.T) {
 		writeCacheBlob(t, cacheDir, diffID, []byte("blob"))
 		s := newCacheSnapshotter(t, WithLayerContentCache(cacheDir))
 		// Even with matching labels and a cached blob, a View must not commit.
+		// It's read-only, but via the KindView roFlag in mounts(), not the cache.
 		mounts, err := s.View(ctx, "view-1", "", extractionOpt(target, diffID))
-		assertFellThrough(t, s, mounts, err)
+		assertFellThrough(t, s, mounts, err, true)
 	})
 }
 
+// TestCacheParentedPrepare covers a cached blob whose extraction Prepare carries
+// a parent (the sequential unpack path): it must not be staged. mounts() only
+// picks a staged blob up when the snapshot has no parents, so otherwise the
+// unpacker would see a writable overlay, apply the layer, and let the differ
+// write through the symlink into the shared cache blob.
+func TestCacheParentedPrepare(t *testing.T) {
+	ctx := namespaces.WithNamespace(context.Background(), "test")
+
+	var (
+		cacheDir     = t.TempDir()
+		parentDiffID = digest.Digest(cacheTestDiffID)
+		parentChain  = cacheTestChainID
+		childDiffID  = digest.Digest("sha256:0000000000000000000000000000000000000000000000000000000000000003")
+		childChain   = "sha256:0000000000000000000000000000000000000000000000000000000000000004"
+	)
+	writeCacheBlob(t, cacheDir, parentDiffID, []byte("fake parent blob"))
+	writeCacheBlob(t, cacheDir, childDiffID, []byte("fake child blob"))
+
+	s := newCacheSnapshotter(t, WithLayerContentCache(cacheDir))
+
+	// The first layer has no parent, so it is served from the cache as usual.
+	require.NoError(t, s.Commit(ctx, parentChain, stageCacheHit(t, ctx, s, parentChain, parentDiffID)))
+
+	key := "extract-1 " + childChain
+	mounts, err := s.Prepare(ctx, key, parentChain, extractionOpt(childChain, childDiffID))
+	require.NoError(t, err)
+	require.NotEmpty(t, mounts)
+	// The unpacker only inspects the last mount, which must be the writable
+	// overlay so the layer gets applied (the parents' lowers are read-only).
+	assert.False(t, mounts[len(mounts)-1].ReadOnly(), "a parented Prepare must expose a writable overlay")
+
+	// Nothing was staged, so the differ has no symlink to write through.
+	_, err = os.Lstat(s.layerBlobPath(snapshotID(t, ctx, s, key)))
+	assert.ErrorIs(t, err, os.ErrNotExist, "no cached blob may be staged into a parented snapshot")
+}
+
 // 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.
@@ -1101,7 +1155,8 @@ func TestCacheRemove(t *testing.T) {
 	s := newCacheSnapshotter(t, WithLayerContentCache(cacheDir))
 
 	target := cacheTestChainID
-	prepareCacheHit(t, ctx, s, target, diffID)
+	key := stageCacheHit(t, ctx, s, target, diffID)
+	require.NoError(t, s.Commit(ctx, target, key))
 
 	snapDir := filepath.Dir(s.layerBlobPath(snapshotID(t, ctx, s, target)))
 
@@ -1119,8 +1174,8 @@ func TestCacheRemove(t *testing.T) {
 }
 
 // TestCacheDmverity covers dmverity_mode="on": a cache entry with a sidecar is
-// committed (and the sidecar copied), while an entry missing its required
-// sidecar is a hard error (not a hit, nothing committed).
+// staged (and the sidecar copied), while an entry missing its required sidecar is
+// a hard error (not staged, nothing committed).
 func TestCacheDmverity(t *testing.T) {
 	if supported, err := dmverity.IsSupported(); err != nil || !supported {
 		t.Skip("dm-verity is not supported on this system")
@@ -1129,15 +1184,15 @@ func TestCacheDmverity(t *testing.T) {
 	diffID := digest.Digest(cacheTestDiffID)
 	target := cacheTestChainID
 
-	t.Run("with sidecar commits and copies it", func(t *testing.T) {
+	t.Run("with sidecar stages and copies 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))
 
 		s := newCacheSnapshotter(t, WithLayerContentCache(cacheDir), WithDmverityMode("on"))
-		prepareCacheHit(t, ctx, s, target, diffID)
+		key := stageCacheHit(t, ctx, s, target, diffID)
 
-		_, err := os.Stat(dmverity.MetadataPath(s.layerBlobPath(snapshotID(t, ctx, s, target))))
+		_, 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")
 	})
 
@@ -1149,9 +1204,9 @@ func TestCacheDmverity(t *testing.T) {
 
 		// dmverity_mode=on requires a sidecar; a cache entry without one is a hard
 		// error rather than a silent fallback.
-		_, err := s.Prepare(ctx, "extract-1 "+target, "", extractionOpt(target, diffID))
+		mounts, err := s.Prepare(ctx, "extract-1 "+target, "", extractionOpt(target, diffID))
 		require.Error(t, err)
-		assert.False(t, errdefs.IsAlreadyExists(err), "missing sidecar must not be treated as a hit")
+		assert.Nil(t, mounts, "missing sidecar must not be treated as a hit")
 		_, err = s.Stat(ctx, target)
 		assert.Error(t, err, "no snapshot should be committed on failure")
 	})
diff --git a/plugins/snapshots/erofs/plugin/plugin.go b/plugins/snapshots/erofs/plugin/plugin.go
index adf387b12fad8..0315e45564adc 100644
--- a/plugins/snapshots/erofs/plugin/plugin.go
+++ b/plugins/snapshots/erofs/plugin/plugin.go
@@ -24,6 +24,7 @@ import (
 	"github.com/containerd/plugin"
 	"github.com/containerd/plugin/registry"
 
+	"github.com/containerd/containerd/v2/core/snapshots"
 	"github.com/containerd/containerd/v2/plugins"
 	"github.com/containerd/containerd/v2/plugins/snapshots/erofs"
 	"github.com/docker/go-units"
@@ -59,6 +60,10 @@ type Config struct {
 	// LayerContentCache is a directory of pre-converted, diffID-keyed erofs
 	// layer blobs. When set, layers already present in the cache are committed
 	// without being downloaded or converted. Empty disables the feature.
+	//
+	// Only layers prepared without a parent can be served from the cache. With
+	// sequential unpacking that is the first layer alone, so getting hits for a
+	// whole image needs max_concurrent_unpacks > 1, which is not the default.
 	LayerContentCache string `toml:"layer_content_cache"`
 }
 
@@ -117,19 +122,7 @@ func init() {
 			}
 
 			ic.Meta.Exports[plugins.SnapshotterRootDir] = root
-			// The "rebase" capability lets the unpacker unpack layers in parallel
-			// via a deferred commit: Prepare receives no parent and the real parent
-			// is applied at Commit time. The layer content cache is incompatible
-			// with that — it commits the layer during Prepare (returning
-			// ErrAlreadyExists), when the parent is not yet known in parallel mode,
-			// so the committed layer would be parentless and the chain would break.
-			// With the cache enabled we therefore unpack sequentially. Cache hits
-			// skip the download and conversion anyway, but a cache *miss* is then
-			// slower than a cold pull on an uncached node.
-			// TODO: keep "rebase" and defer the cache commit so misses stay parallel.
-			if config.LayerContentCache == "" {
-				ic.Meta.Capabilities = append(ic.Meta.Capabilities, "rebase")
-			}
+			ic.Meta.Capabilities = append(ic.Meta.Capabilities, snapshots.RebaseCap)
 			return erofs.NewSnapshotter(root, opts...)
 		},
 	})
diff --git a/plugins/snapshots/overlay/plugin/plugin.go b/plugins/snapshots/overlay/plugin/plugin.go
index 5b9ae9480c47f..aafb590080727 100644
--- a/plugins/snapshots/overlay/plugin/plugin.go
+++ b/plugins/snapshots/overlay/plugin/plugin.go
@@ -23,6 +23,7 @@ import (
 
 	"github.com/moby/sys/userns"
 
+	"github.com/containerd/containerd/v2/core/snapshots"
 	"github.com/containerd/containerd/v2/plugins"
 	"github.com/containerd/containerd/v2/plugins/snapshots/overlay"
 	"github.com/containerd/containerd/v2/plugins/snapshots/overlay/overlayutils"
@@ -34,7 +35,6 @@ import (
 const (
 	capaRemapIDs     = "remap-ids"
 	capaOnlyRemapIDs = "only-remap-ids"
-	capaRebase       = "rebase"
 )
 
 // Config represents configuration for the overlay plugin.
@@ -99,7 +99,7 @@ func init() {
 				// "rebase" capability depends on `mknod c 0 0` via OverlayConvertWhiteout,
 				// so it does not work when running in UserNS.
 				// https://github.com/containerd/containerd/issues/13388
-				ic.Meta.Capabilities = append(ic.Meta.Capabilities, capaRebase)
+				ic.Meta.Capabilities = append(ic.Meta.Capabilities, snapshots.RebaseCap)
 			}
 
 			ic.Meta.Exports[plugins.SnapshotterRootDir] = root