From 1e001e6dfe711e3d81582e94e30162bbd889ba48 Mon Sep 17 00:00:00 2001
From: Maksym Pavlenko <pavlenko.maksym@gmail.com>
Date: Tue, 21 Jul 2026 18:57:45 -0700
Subject: [PATCH 1/2] erofs: make the layer content cache work with parallel
 unpack

The cache used to serve a hit by committing the layer during Prepare and
returning ErrAlreadyExists, which is incompatible with parallel unpacking:
the "rebase" capability defers the parent to Commit time, so a
commit-at-Prepare layer ends up parentless. The erofs snapshotter therefore
had to disable rebase whenever the cache was enabled, making every cache
*miss* fall back to a fully sequential (slower-than-cold) pull.

Introduce snapshots.ErrAlreadyStaged, returned from Prepare to mean "the
active snapshot's content is staged; skip the layer download and apply, but
still Commit it" (where the parent is applied). Unlike ErrAlreadyExists it
does not end the layer's lifecycle. The unpacker handles it by emitting a
status whose bottom half runs the normal serialized commit (rebasing the
parent in), and the metadata snapshotter threads it through Prepare like a
normal active snapshot. The erofs cache hit now stages the blob and returns
ErrAlreadyStaged instead of committing, so it can advertise "rebase"
unconditionally: hits skip download+conversion and misses stay parallel.

Also promote the "rebase" capability string to snapshots.RebaseCap (shared by
the unpacker, erofs, and overlay).

Follow up:  fetch is range-based, so only a fully cached contiguous
prefix skips downloads; a miss in a lower layer still pulls everything above
it. Per-layer fetch-skip is left as a follow-up.

Signed-off-by: Maksym Pavlenko <pavlenko.maksym@gmail.com>
---
 core/metadata/snapshot.go                   |  18 ++-
 core/snapshots/snapshotter.go               |  17 +++
 core/unpack/unpacker.go                     |  99 +++++++++++------
 core/unpack/unpacker_test.go                | 103 +++++++++++++++++-
 plugins/snapshots/erofs/erofs.go            | 115 +++++++++-----------
 plugins/snapshots/erofs/erofs_linux_test.go |  64 ++++++-----
 plugins/snapshots/erofs/plugin/plugin.go    |  15 +--
 plugins/snapshots/overlay/plugin/plugin.go  |   4 +-
 8 files changed, 291 insertions(+), 144 deletions(-)

diff --git a/core/metadata/snapshot.go b/core/metadata/snapshot.go
index 132de60ec7999..0557632d66ad6 100644
--- a/core/metadata/snapshot.go
+++ b/core/metadata/snapshot.go
@@ -18,6 +18,7 @@ package metadata
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"maps"
 	"strings"
@@ -321,7 +322,8 @@ func (s *snapshotter) createSnapshot(ctx context.Context, key, parent string, re
 		bopts   = []snapshots.Opt{
 			snapshots.WithLabels(snapshots.FilterInheritedLabels(base.Labels)),
 		}
-		rerr error
+		rerr   error
+		staged bool
 	)
 
 	if err := update(ctx, s.db, func(tx *bolt.Tx) error {
@@ -433,9 +435,15 @@ func (s *snapshotter) createSnapshot(ctx context.Context, key, parent string, re
 			// to avoid confusing callers handling already exists.
 			return nil, fmt.Errorf("unexpected error from snapshotter: %v: %w", err, errdefs.ErrUnknown)
 		}
-	} else if err != nil {
+	} else if err != nil && !errors.Is(err, snapshots.ErrAlreadyStaged) {
 		return nil, err
 	} else {
+		// A normal Prepare, or a backend that staged content into the active
+		// snapshot without committing (ErrAlreadyStaged, e.g. a layer content cache
+		// hit). Either way record the active snapshot; for the staged case
+		// propagate the sentinel after the txn so the caller skips fetch/apply but
+		// still commits it (applying the parent).
+		staged = errors.Is(err, snapshots.ErrAlreadyStaged)
 		ts := time.Now().UTC()
 		base.Created = ts
 		base.Updated = ts
@@ -514,6 +522,12 @@ func (s *snapshotter) createSnapshot(ctx context.Context, key, parent string, re
 		return nil, rerr
 	}
 
+	// The active snapshot was recorded; propagate the staged signal so the caller
+	// skips fetch/apply but still commits it.
+	if staged {
+		return m, snapshots.ErrAlreadyStaged
+	}
+
 	return m, nil
 }
 
diff --git a/core/snapshots/snapshotter.go b/core/snapshots/snapshotter.go
index 9140a0644e0e2..f6ac35e79ae31 100644
--- a/core/snapshots/snapshotter.go
+++ b/core/snapshots/snapshotter.go
@@ -19,6 +19,7 @@ package snapshots
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"maps"
 	"strings"
 	"time"
@@ -26,6 +27,16 @@ import (
 	"github.com/containerd/containerd/v2/core/mount"
 )
 
+// ErrAlreadyStaged is returned from Prepare when the snapshotter has staged a
+// layer's content into the active snapshot (e.g. from a layer content cache) so
+// the caller should skip fetching and applying the layer. Unlike
+// ErrAlreadyExists — which reports an already-committed snapshot and ends the
+// layer's lifecycle — the snapshot is NOT committed: the caller must still Commit
+// it, which is where the parent is applied. This keeps the cache compatible with
+// parallel unpacking (the "rebase" capability), where the parent is only known
+// at Commit time.
+var ErrAlreadyStaged = errors.New("snapshot already staged")
+
 const (
 	// UnpackKeyPrefix is the beginning of the key format used for snapshots that will have
 	// image content unpacked into them.
@@ -58,6 +69,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..4a3541068d06f 100644
--- a/core/unpack/unpacker.go
+++ b/core/unpack/unpacker.go
@@ -405,6 +405,7 @@ func (u *Unpacker) unpack(
 			key    string
 			mounts []mount.Mount
 			opts   = append(unpack.SnapshotOpts, snapshots.WithLabels(snapshotLabels))
+			staged bool
 		)
 
 		for try := 1; try <= 3; try++ {
@@ -412,6 +413,14 @@ func (u *Unpacker) unpack(
 			key = fmt.Sprintf(snapshots.UnpackKeyFormat, uniquePart(), chainID)
 			mounts, err = sn.Prepare(ctx, key, parent, opts...)
 			if err != nil {
+				if errors.Is(err, snapshots.ErrAlreadyStaged) {
+					// The snapshotter staged the layer content into the active
+					// snapshot (e.g. a layer content cache hit). Skip fetch+apply,
+					// but still commit it below (which applies the parent).
+					staged = true
+					err = nil
+					break
+				}
 				if errdefs.IsAlreadyExists(err) {
 					if snInfo, err := sn.Stat(ctx, chainID); err != nil {
 						if !errdefs.IsNotFound(err) {
@@ -442,6 +451,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 +542,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 +779,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
 	}
diff --git a/core/unpack/unpacker_test.go b/core/unpack/unpacker_test.go
index bdd08348f56f7..d2af5414f6dbe 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,93 @@ func TestBindToOverlay(t *testing.T) {
 		})
 	}
 }
+
+// stagedSnapshotter reports every layer as staged (ErrAlreadyStaged) 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 nil, snapshots.ErrAlreadyStaged
+}
+
+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 (ErrAlreadyStaged) 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 4af0f034e7a3e..5b2e626fa0ee8 100644
--- a/plugins/snapshots/erofs/erofs.go
+++ b/plugins/snapshots/erofs/erofs.go
@@ -298,7 +298,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 +320,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 +332,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)
 				}
@@ -578,10 +578,10 @@ 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).
+// it stages the cached blob into the active snapshot (without committing) and
+// returns ErrAlreadyStaged, so the unpacker skips the layer download and
+// conversion but still commits 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,20 +590,20 @@ func (s *snapshotter) createSnapshot(ctx context.Context, kind snapshots.Kind, k
 	)
 
 	// 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
+	// layer content cache; View and container-rootfs Prepares get an empty path
+	// and fall through to the normal path.
+	var cacheBlob string
 	if kind == snapshots.KindActive {
-		entry = s.lookupCache(ctx, opts...)
+		cacheBlob = s.lookupCache(ctx, opts...)
 	}
 
-	// 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
+	// staged is set once the cached blob has been staged into the active snapshot
+	// and we deliberately return ErrAlreadyStaged; the snapshot dir must then be
+	// kept (the caller commits it later). Any real error leaves it false so the
+	// staged td/path is reclaimed.
+	var staged bool
 	defer func() {
-		if err != nil && !committed {
+		if err != nil && !staged {
 			if td != "" {
 				if err1 := os.RemoveAll(td); err1 != nil {
 					log.G(ctx).WithError(err1).Warn("failed to cleanup temp snapshot directory")
@@ -619,7 +619,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)
 	}
@@ -690,32 +690,21 @@ 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 {
+	// Cache hit: the blob is staged into the active snapshot but not committed.
+	// Signal the unpacker via ErrAlreadyStaged so it skips the layer download and
+	// conversion but still commits the snapshot (applying the parent at Commit).
+	if cacheBlob != "" {
 		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
+			"key":  key,
+			"blob": cacheBlob,
+		}).Debug("layer content cache hit, staged cached erofs blob")
+		staged = true
+		return nil, snapshots.ErrAlreadyStaged
 	}
 
 	return s.mounts(snap, info)
@@ -725,61 +714,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 6b6ff9b13e279..3f98e0ab113a7 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"
@@ -882,18 +881,23 @@ 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: no mounts, and ErrAlreadyStaged
+// (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.ErrorIs(t, err, snapshots.ErrAlreadyStaged, "cache hit must stage and signal ErrAlreadyStaged")
+	assert.Nil(t, mounts, "a staged cache hit returns no 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
+// ErrAlreadyStaged without committing; a subsequent Commit finalizes the target
+// chainID without re-converting.
 func TestCacheHit(t *testing.T) {
 	ctx := namespaces.WithNamespace(context.Background(), "test")
 
@@ -904,18 +908,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")
@@ -923,6 +924,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
@@ -940,11 +949,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")
@@ -1019,7 +1028,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)))
 
@@ -1037,8 +1047,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")
@@ -1047,15 +1057,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")
 	})
 
@@ -1069,7 +1079,7 @@ func TestCacheDmverity(t *testing.T) {
 		// error rather than a silent fallback.
 		_, 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.NotErrorIs(t, err, snapshots.ErrAlreadyStaged, "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..36068f341a921 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"
@@ -117,19 +118,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

From 257a5900b054c157ef87e85f889cd33887b342f9 Mon Sep 17 00:00:00 2001
From: Maksym Pavlenko <pavlenko.maksym@gmail.com>
Date: Mon, 27 Jul 2026 14:18:16 -0700
Subject: [PATCH 2/2] core/unpack: detect staged layers via read-only mounts

Replace the snapshots.ErrAlreadyStaged sentinel error with a
read-only-mounts signal: a snapshotter (e.g. erofs serving a layer
content cache hit) now returns Prepare mounts normally, with no
error, when the layer content is already staged into the active
snapshot. The unpacker's isStaged helper checks the last mount's
ReadOnly() to decide whether to skip fetch+apply and just commit.

Signed-off-by: Maksym Pavlenko <pavlenko.maksym@gmail.com>
---
 core/metadata/snapshot.go                   | 18 +-----
 core/mount/mount.go                         | 31 ++++++++-
 core/mount/mount_test.go                    | 65 +++++++++++++++++++
 core/snapshots/snapshotter.go               | 11 ----
 core/unpack/unpacker.go                     | 38 ++++++++---
 core/unpack/unpacker_test.go                | 56 ++++++++++++++--
 plugins/snapshots/erofs/erofs.go            | 61 +++++++-----------
 plugins/snapshots/erofs/erofs_linux_test.go | 71 +++++++++++++++++----
 plugins/snapshots/erofs/plugin/plugin.go    |  4 ++
 9 files changed, 259 insertions(+), 96 deletions(-)

diff --git a/core/metadata/snapshot.go b/core/metadata/snapshot.go
index 0557632d66ad6..132de60ec7999 100644
--- a/core/metadata/snapshot.go
+++ b/core/metadata/snapshot.go
@@ -18,7 +18,6 @@ package metadata
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"maps"
 	"strings"
@@ -322,8 +321,7 @@ func (s *snapshotter) createSnapshot(ctx context.Context, key, parent string, re
 		bopts   = []snapshots.Opt{
 			snapshots.WithLabels(snapshots.FilterInheritedLabels(base.Labels)),
 		}
-		rerr   error
-		staged bool
+		rerr error
 	)
 
 	if err := update(ctx, s.db, func(tx *bolt.Tx) error {
@@ -435,15 +433,9 @@ func (s *snapshotter) createSnapshot(ctx context.Context, key, parent string, re
 			// to avoid confusing callers handling already exists.
 			return nil, fmt.Errorf("unexpected error from snapshotter: %v: %w", err, errdefs.ErrUnknown)
 		}
-	} else if err != nil && !errors.Is(err, snapshots.ErrAlreadyStaged) {
+	} else if err != nil {
 		return nil, err
 	} else {
-		// A normal Prepare, or a backend that staged content into the active
-		// snapshot without committing (ErrAlreadyStaged, e.g. a layer content cache
-		// hit). Either way record the active snapshot; for the staged case
-		// propagate the sentinel after the txn so the caller skips fetch/apply but
-		// still commits it (applying the parent).
-		staged = errors.Is(err, snapshots.ErrAlreadyStaged)
 		ts := time.Now().UTC()
 		base.Created = ts
 		base.Updated = ts
@@ -522,12 +514,6 @@ func (s *snapshotter) createSnapshot(ctx context.Context, key, parent string, re
 		return nil, rerr
 	}
 
-	// The active snapshot was recorded; propagate the staged signal so the caller
-	// skips fetch/apply but still commits it.
-	if staged {
-		return m, snapshots.ErrAlreadyStaged
-	}
-
 	return m, nil
 }
 
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 f6ac35e79ae31..c6a7101e6e795 100644
--- a/core/snapshots/snapshotter.go
+++ b/core/snapshots/snapshotter.go
@@ -19,7 +19,6 @@ package snapshots
 import (
 	"context"
 	"encoding/json"
-	"errors"
 	"maps"
 	"strings"
 	"time"
@@ -27,16 +26,6 @@ import (
 	"github.com/containerd/containerd/v2/core/mount"
 )
 
-// ErrAlreadyStaged is returned from Prepare when the snapshotter has staged a
-// layer's content into the active snapshot (e.g. from a layer content cache) so
-// the caller should skip fetching and applying the layer. Unlike
-// ErrAlreadyExists — which reports an already-committed snapshot and ends the
-// layer's lifecycle — the snapshot is NOT committed: the caller must still Commit
-// it, which is where the parent is applied. This keeps the cache compatible with
-// parallel unpacking (the "rebase" capability), where the parent is only known
-// at Commit time.
-var ErrAlreadyStaged = errors.New("snapshot already staged")
-
 const (
 	// UnpackKeyPrefix is the beginning of the key format used for snapshots that will have
 	// image content unpacked into them.
diff --git a/core/unpack/unpacker.go b/core/unpack/unpacker.go
index 4a3541068d06f..d0313e80c821e 100644
--- a/core/unpack/unpacker.go
+++ b/core/unpack/unpacker.go
@@ -404,7 +404,10 @@ 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
 		)
 
@@ -413,14 +416,6 @@ func (u *Unpacker) unpack(
 			key = fmt.Sprintf(snapshots.UnpackKeyFormat, uniquePart(), chainID)
 			mounts, err = sn.Prepare(ctx, key, parent, opts...)
 			if err != nil {
-				if errors.Is(err, snapshots.ErrAlreadyStaged) {
-					// The snapshotter staged the layer content into the active
-					// snapshot (e.g. a layer content cache hit). Skip fetch+apply,
-					// but still commit it below (which applies the parent).
-					staged = true
-					err = nil
-					break
-				}
 				if errdefs.IsAlreadyExists(err) {
 					if snInfo, err := sn.Stat(ctx, chainID); err != nil {
 						if !errdefs.IsNotFound(err) {
@@ -444,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 {
@@ -794,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 d2af5414f6dbe..cce256a9b8446 100644
--- a/core/unpack/unpacker_test.go
+++ b/core/unpack/unpacker_test.go
@@ -189,9 +189,52 @@ func TestBindToOverlay(t *testing.T) {
 	}
 }
 
-// stagedSnapshotter reports every layer as staged (ErrAlreadyStaged) 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.
+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
@@ -202,7 +245,7 @@ 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 nil, snapshots.ErrAlreadyStaged
+	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 {
@@ -223,8 +266,9 @@ func (a failApplier) Apply(_ context.Context, desc ocispec.Descriptor, _ []mount
 }
 
 // TestUnpackStagedLayers verifies that when the snapshotter reports layers as
-// staged (ErrAlreadyStaged) in parallel mode, the unpacker skips fetch+apply but
-// still commits each layer, rebasing the real parent in at Commit time.
+// 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()
 
diff --git a/plugins/snapshots/erofs/erofs.go b/plugins/snapshots/erofs/erofs.go
index 5b2e626fa0ee8..54c616c24c2a7 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")
@@ -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
@@ -577,11 +573,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,
+// 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 ErrAlreadyStaged, so the unpacker skips the layer download and
-// conversion but still commits the snapshot normally — applying the parent at
-// Commit time, which keeps the cache compatible with parallel unpacking.
+// 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
@@ -589,21 +586,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 an empty path
-	// and fall through to the normal path.
+	// 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 {
-		cacheBlob = s.lookupCache(ctx, opts...)
+	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")
+		}
 	}
 
-	// staged is set once the cached blob has been staged into the active snapshot
-	// and we deliberately return ErrAlreadyStaged; the snapshot dir must then be
-	// kept (the caller commits it later). Any real error leaves it false so the
-	// staged td/path is reclaimed.
-	var staged bool
 	defer func() {
-		if err != nil && !staged {
+		if err != nil {
 			if td != "" {
 				if err1 := os.RemoveAll(td); err1 != nil {
 					log.G(ctx).WithError(err1).Warn("failed to cleanup temp snapshot directory")
@@ -695,18 +692,6 @@ func (s *snapshotter) createSnapshot(ctx context.Context, kind snapshots.Kind, k
 		return nil, err
 	}
 
-	// Cache hit: the blob is staged into the active snapshot but not committed.
-	// Signal the unpacker via ErrAlreadyStaged so it skips the layer download and
-	// conversion but still commits the snapshot (applying the parent at Commit).
-	if cacheBlob != "" {
-		log.G(ctx).WithFields(log.Fields{
-			"key":  key,
-			"blob": cacheBlob,
-		}).Debug("layer content cache hit, staged cached erofs blob")
-		staged = true
-		return nil, snapshots.ErrAlreadyStaged
-	}
-
 	return s.mounts(snap, info)
 }
 
diff --git a/plugins/snapshots/erofs/erofs_linux_test.go b/plugins/snapshots/erofs/erofs_linux_test.go
index 3f98e0ab113a7..6ac0514da469f 100644
--- a/plugins/snapshots/erofs/erofs_linux_test.go
+++ b/plugins/snapshots/erofs/erofs_linux_test.go
@@ -882,21 +882,24 @@ func newCacheSnapshotter(t *testing.T, opts ...Opt) *snapshotter {
 }
 
 // stageCacheHit runs an extraction Prepare for target/diffID and asserts the
-// cache staged the blob into the active snapshot: no mounts, and ErrAlreadyStaged
-// (so the unpacker skips fetch+apply but still commits). It returns the extraction
-// key so the caller can Commit it as the target chainID.
+// 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()
 	key := "extract-1 " + target
 	mounts, err := s.Prepare(ctx, key, "", extractionOpt(target, diffID))
-	require.ErrorIs(t, err, snapshots.ErrAlreadyStaged, "cache hit must stage and signal ErrAlreadyStaged")
-	assert.Nil(t, mounts, "a staged cache hit returns no mounts")
+	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 stages the blob into the active snapshot (symlinked) and returns
-// ErrAlreadyStaged without committing; a subsequent Commit finalizes the target
+// 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")
@@ -972,11 +975,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")
 	}
@@ -984,13 +992,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) {
@@ -1008,11 +1016,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.
@@ -1077,9 +1122,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.NotErrorIs(t, err, snapshots.ErrAlreadyStaged, "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 36068f341a921..0315e45564adc 100644
--- a/plugins/snapshots/erofs/plugin/plugin.go
+++ b/plugins/snapshots/erofs/plugin/plugin.go
@@ -60,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"`
 }