diff --git a/core/metadata/snapshot.go b/core/metadata/snapshot.go
index 5880df57ec..132de60ec7 100644
--- a/core/metadata/snapshot.go
+++ b/core/metadata/snapshot.go
@@ -39,7 +39,6 @@ import (
 
 const (
 	inheritedLabelsPrefix = "containerd.io/snapshot/"
-	labelSnapshotRef      = "containerd.io/snapshot.ref"
 )
 
 type snapshotter struct {
@@ -316,7 +315,7 @@ func (s *snapshotter) createSnapshot(ctx context.Context, key, parent string, re
 	}
 
 	var (
-		target  = base.Labels[labelSnapshotRef]
+		target  = base.Labels[snapshots.LabelSnapshotRef]
 		bparent string
 		bkey    string
 		bopts   = []snapshots.Opt{
@@ -388,10 +387,10 @@ func (s *snapshotter) createSnapshot(ctx context.Context, key, parent string, re
 	if errdefs.IsAlreadyExists(err) {
 		if target != "" {
 			var tinfo *snapshots.Info
-			filter := fmt.Sprintf(`labels."containerd.io/snapshot.ref"==%s,parent==%q`, target, bparent)
+			filter := fmt.Sprintf(`labels.%q==%s,parent==%q`, snapshots.LabelSnapshotRef, target, bparent)
 			if err := s.Snapshotter.Walk(ctx, func(ctx context.Context, i snapshots.Info) error {
 				if tinfo == nil && i.Kind == snapshots.KindCommitted {
-					if i.Labels["containerd.io/snapshot.ref"] != target {
+					if i.Labels[snapshots.LabelSnapshotRef] != target {
 						// Walk did not respect filter
 						return nil
 					}
diff --git a/core/metadata/snapshot_test.go b/core/metadata/snapshot_test.go
index e005cb898d..ae818aebed 100644
--- a/core/metadata/snapshot_test.go
+++ b/core/metadata/snapshot_test.go
@@ -70,7 +70,7 @@ func TestSnapshotterWithRef(t *testing.T) {
 	key1 := "test1"
 	test1opt := snapshots.WithLabels(
 		map[string]string{
-			labelSnapshotRef: key1,
+			snapshots.LabelSnapshotRef: key1,
 		},
 	)
 
@@ -113,7 +113,7 @@ func TestSnapshotterWithRef(t *testing.T) {
 	key2 := "test2"
 	test2opt := snapshots.WithLabels(
 		map[string]string{
-			labelSnapshotRef: key2,
+			snapshots.LabelSnapshotRef: key2,
 		},
 	)
 
@@ -340,7 +340,7 @@ func (s *tmpSnapshotter) create(ctx context.Context, key, parent string, kind sn
 	base.Name = key
 	base.Kind = kind
 
-	target := base.Labels[labelSnapshotRef]
+	target := base.Labels[snapshots.LabelSnapshotRef]
 	if target != "" {
 		for _, name := range s.targets[target] {
 			if s.snapshots[name].Parent == parent {
@@ -399,7 +399,7 @@ func (s *tmpSnapshotter) Commit(ctx context.Context, name, key string, opts ...s
 	s.snapshots[name] = base
 	delete(s.snapshots, key)
 
-	if target := base.Labels[labelSnapshotRef]; target != "" {
+	if target := base.Labels[snapshots.LabelSnapshotRef]; target != "" {
 		s.targets[target] = append(s.targets[target], name)
 	}
 
diff --git a/core/snapshots/snapshotter.go b/core/snapshots/snapshotter.go
index cb94e12110..9140a0644e 100644
--- a/core/snapshots/snapshotter.go
+++ b/core/snapshots/snapshotter.go
@@ -33,7 +33,17 @@ const (
 	// UnpackKeyFormat is the format for the snapshotter keys used for extraction
 	UnpackKeyFormat       = UnpackKeyPrefix + "-%s %s"
 	inheritedLabelsPrefix = "containerd.io/snapshot/"
-	labelSnapshotRef      = "containerd.io/snapshot.ref"
+
+	// LabelSnapshotRef is set by the unpacker on the extraction Prepare to
+	// the target chainID. A snapshotter that already has the layer commits a
+	// snapshot named after this value and returns ErrAlreadyExists, which
+	// makes the unpacker skip fetching and applying the layer (the remote
+	// snapshot protocol). It is inherited by FilterInheritedLabels.
+	LabelSnapshotRef = "containerd.io/snapshot.ref"
+
+	// LabelSnapshotDiffID is set by the unpacker on the extraction Prepare to
+	// the uncompressed digest (diffID) of the layer being unpacked.
+	LabelSnapshotDiffID = "containerd.io/snapshot/diff-id"
 
 	// LabelSnapshotUIDMapping is the label used for UID mappings
 	LabelSnapshotUIDMapping = "containerd.io/snapshot/uidmapping"
@@ -397,7 +407,7 @@ func FilterInheritedLabels(labels map[string]string) map[string]string {
 
 	filtered := make(map[string]string)
 	for k, v := range labels {
-		if k == labelSnapshotRef || strings.HasPrefix(k, inheritedLabelsPrefix) {
+		if k == LabelSnapshotRef || strings.HasPrefix(k, inheritedLabelsPrefix) {
 			filtered[k] = v
 		}
 	}
diff --git a/core/unpack/unpacker.go b/core/unpack/unpacker.go
index 67f2dcb7c4..ec55e65317 100644
--- a/core/unpack/unpacker.go
+++ b/core/unpack/unpacker.go
@@ -49,9 +49,7 @@ import (
 )
 
 const (
-	labelSnapshotRef    = "containerd.io/snapshot.ref"
 	labelSnapshotParent = "containerd.io/snapshot/parent-chain-id"
-	labelSnapshotDiffID = "containerd.io/snapshot/diff-id"
 	unpackSpanPrefix    = "pkg.unpack.unpacker"
 )
 
@@ -397,8 +395,8 @@ func (u *Unpacker) unpack(
 		if snapshotLabels == nil {
 			snapshotLabels = make(map[string]string)
 		}
-		snapshotLabels[labelSnapshotRef] = chainID
-		snapshotLabels[labelSnapshotDiffID] = diffIDs[i].String()
+		snapshotLabels[snapshots.LabelSnapshotRef] = chainID
+		snapshotLabels[snapshots.LabelSnapshotDiffID] = diffIDs[i].String()
 		if i > 0 {
 			snapshotLabels[labelSnapshotParent] = chainIDs[i-1].String()
 		}
diff --git a/internal/erofsutils/mount.go b/internal/erofsutils/mount.go
index cc9a4a25f5..6c3ef9f648 100644
--- a/internal/erofsutils/mount.go
+++ b/internal/erofsutils/mount.go
@@ -29,10 +29,22 @@ import (
 
 	"github.com/containerd/errdefs"
 	"github.com/containerd/log"
+	"github.com/opencontainers/go-digest"
 
 	"github.com/containerd/containerd/v2/core/mount"
 )
 
+// CacheBlobPath returns the layer content cache path for a diffID, following the
+// <dir>/<algo>/<xx>/<hex>.erofs layout, where <xx> is the first two characters of
+// the encoded digest. The extra prefix directory shards blobs so no single
+// directory grows unwieldy for a large cache. It is the single source of the
+// layout, shared by the cache producer (ctr build-erofs-cache) and the erofs
+// snapshotter that reads it, so the two never drift.
+func CacheBlobPath(dir string, diffID digest.Digest) string {
+	enc := diffID.Encoded()
+	return filepath.Join(dir, diffID.Algorithm().String(), enc[:2], enc+".erofs")
+}
+
 // IsErofsMediaType returns true if the media type is an EROFS layer type.
 func IsErofsMediaType(mt string) bool {
 	return strings.HasPrefix(mt, "application/vnd.erofs.layer")
diff --git a/internal/erofsutils/mount_test.go b/internal/erofsutils/mount_test.go
new file mode 100644
index 0000000000..4314fb9aca
--- /dev/null
+++ b/internal/erofsutils/mount_test.go
@@ -0,0 +1,37 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package erofsutils
+
+import (
+	"path/filepath"
+	"testing"
+
+	"github.com/opencontainers/go-digest"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestCacheBlobPath(t *testing.T) {
+	diffID := digest.FromString("hello")
+	enc := diffID.Encoded()
+
+	got := CacheBlobPath("/cache", diffID)
+
+	// <dir>/<algo>/<xx>/<hex>.erofs, where <xx> is the first two hex characters.
+	want := filepath.Join("/cache", "sha256", enc[:2], enc+".erofs")
+	assert.Equal(t, want, got)
+	assert.Equal(t, enc[:2], filepath.Base(filepath.Dir(got)), "blob must live under a 2-char prefix dir")
+}
diff --git a/plugins/snapshots/erofs/erofs.go b/plugins/snapshots/erofs/erofs.go
index 2f038706f2..4af0f034e7 100644
--- a/plugins/snapshots/erofs/erofs.go
+++ b/plugins/snapshots/erofs/erofs.go
@@ -18,6 +18,7 @@ package erofs
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"os"
 	"path/filepath"
@@ -27,11 +28,13 @@ import (
 	"github.com/containerd/continuity/fs"
 	"github.com/containerd/errdefs"
 	"github.com/containerd/log"
+	"github.com/opencontainers/go-digest"
 
 	"github.com/containerd/containerd/v2/core/mount"
 	"github.com/containerd/containerd/v2/core/snapshots"
 	"github.com/containerd/containerd/v2/core/snapshots/storage"
 	"github.com/containerd/containerd/v2/internal/dmverity"
+	"github.com/containerd/containerd/v2/internal/erofsutils"
 	"github.com/containerd/containerd/v2/internal/fsverity"
 	"github.com/containerd/containerd/v2/internal/userns"
 )
@@ -49,6 +52,12 @@ type SnapshotterConfig struct {
 	remapIDs    bool
 	// dmverityMode controls dm-verity behavior: "auto" (use if .dmverity exists), "on" (require .dmverity), "off" (disable)
 	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.
+	layerContentCache string
 }
 
 // Opt is an option to configure the erofs snapshotter
@@ -96,6 +105,16 @@ func WithRemapIDs() Opt {
 	}
 }
 
+// WithLayerContentCache configures a read-only directory of pre-converted,
+// diffID-keyed erofs layer blobs that the snapshotter sources layers from on
+// pull instead of downloading and converting them. See the layerContentCache
+// field for details.
+func WithLayerContentCache(path string) Opt {
+	return func(config *SnapshotterConfig) {
+		config.layerContentCache = path
+	}
+}
+
 type MetaStore interface {
 	TransactionContext(ctx context.Context, writable bool) (context.Context, storage.Transactor, error)
 	WithTransaction(ctx context.Context, writable bool, fn storage.TransactionCallback) error
@@ -103,15 +122,16 @@ type MetaStore interface {
 }
 
 type snapshotter struct {
-	root            string
-	ms              *storage.MetaStore
-	ovlOptions      []string
-	enableFsverity  bool
-	setImmutable    bool
-	defaultWritable int64
-	blockMode       bool
-	remapIDs        bool
-	dmverityMode    string
+	root              string
+	ms                MetaStore
+	ovlOptions        []string
+	enableFsverity    bool
+	setImmutable      bool
+	defaultWritable   int64
+	blockMode         bool
+	remapIDs          bool
+	dmverityMode      string
+	layerContentCache string
 }
 
 // NewSnapshotter returns a Snapshotter which uses EROFS+OverlayFS. The layers
@@ -153,6 +173,20 @@ 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.
+	if config.layerContentCache != "" {
+		if config.enableFsverity {
+			return nil, fmt.Errorf("enable_fsverity is incompatible with layer_content_cache; use dm-verity for cache integrity")
+		}
+		if config.setImmutable {
+			return nil, fmt.Errorf("set_immutable is incompatible with layer_content_cache")
+		}
+	}
+
 	// Check fsverity support if enabled
 	if config.enableFsverity {
 		// TODO: Call specific function here
@@ -169,6 +203,24 @@ func NewSnapshotter(root string, opts ...Opt) (snapshots.Snapshotter, error) {
 		return nil, fmt.Errorf("setting IMMUTABLE_FL is only supported on Linux")
 	}
 
+	// Resolve the cache dir to an absolute path so materialized layer blobs are
+	// absolute symlinks, independent of the process working directory, and verify
+	// it exists and is a directory so misconfiguration fails fast at startup.
+	if config.layerContentCache != "" {
+		abs, err := filepath.Abs(config.layerContentCache)
+		if err != nil {
+			return nil, fmt.Errorf("failed to resolve layer_content_cache path %q: %w", config.layerContentCache, err)
+		}
+		fi, err := os.Stat(abs)
+		if err != nil {
+			return nil, fmt.Errorf("failed to access layer_content_cache %q: %w", abs, err)
+		}
+		if !fi.IsDir() {
+			return nil, fmt.Errorf("layer_content_cache %q is not a directory", abs)
+		}
+		config.layerContentCache = abs
+	}
+
 	ms, err := storage.NewMetaStore(filepath.Join(root, "metadata.db"))
 	if err != nil {
 		return nil, err
@@ -179,15 +231,16 @@ func NewSnapshotter(root string, opts ...Opt) (snapshots.Snapshotter, error) {
 	}
 
 	return &snapshotter{
-		root:            root,
-		ms:              ms,
-		ovlOptions:      config.ovlOptions,
-		enableFsverity:  config.enableFsverity,
-		setImmutable:    config.setImmutable,
-		defaultWritable: config.defaultSize,
-		blockMode:       config.defaultSize > 0,
-		remapIDs:        config.remapIDs,
-		dmverityMode:    config.dmverityMode,
+		root:              root,
+		ms:                ms,
+		ovlOptions:        config.ovlOptions,
+		enableFsverity:    config.enableFsverity,
+		setImmutable:      config.setImmutable,
+		defaultWritable:   config.defaultSize,
+		blockMode:         config.defaultSize > 0,
+		remapIDs:          config.remapIDs,
+		dmverityMode:      config.dmverityMode,
+		layerContentCache: config.layerContentCache,
 	}, nil
 }
 
@@ -245,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) (string, error) {
+func (s *snapshotter) prepareDirectory(ctx context.Context, snapshotDir string, kind snapshots.Kind, entry *cacheEntry) (string, error) {
 	td, err := os.MkdirTemp(snapshotDir, "new-")
 	if err != nil {
 		return "", fmt.Errorf("failed to create temp dir: %w", err)
@@ -267,6 +320,26 @@ 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 {
+		layerBlob := filepath.Join(td, "layer.erofs")
+		if err := os.Symlink(entry.blob, layerBlob); err != nil {
+			return td, fmt.Errorf("failed to symlink cached layer blob: %w", err)
+		}
+		// Copy the dm-verity sidecar alongside the blob (unless dm-verity is off,
+		// when it's never consumed) so mount-time metadata resolution and the
+		// pinned root hash match locally-converted layers. A missing sidecar is
+		// fine except with dmverity_mode "on", which requires it.
+		if s.dmverityMode != "off" {
+			if err := fs.CopyFile(dmverity.MetadataPath(layerBlob), dmverity.MetadataPath(entry.blob)); err != nil {
+				if s.dmverityMode == "on" || !errors.Is(err, os.ErrNotExist) {
+					return td, fmt.Errorf("failed to copy dm-verity sidecar: %w", err)
+				}
+			}
+		}
+	}
+
 	return td, nil
 }
 
@@ -503,6 +576,12 @@ func (s *snapshotter) mounts(snap storage.Snapshot, info snapshots.Info) ([]moun
 	}), nil
 }
 
+// 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).
 func (s *snapshotter) createSnapshot(ctx context.Context, kind snapshots.Kind, key, parent string, opts []snapshots.Opt) (_ []mount.Mount, err error) {
 	var (
 		snap     storage.Snapshot
@@ -510,8 +589,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...)
+	}
+
+	// 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 {
+		if err != nil && !committed {
 			if td != "" {
 				if err1 := os.RemoveAll(td); err1 != nil {
 					log.G(ctx).WithError(err1).Warn("failed to cleanup temp snapshot directory")
@@ -527,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)
+	td, err = s.prepareDirectory(ctx, snapshotDir, kind, entry)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create prepare snapshot dir: %w", err)
 	}
@@ -598,11 +690,34 @@ 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)
 }
 
@@ -610,6 +725,63 @@ 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 {
+	if s.layerContentCache == "" {
+		return nil
+	}
+
+	var base snapshots.Info
+	for _, opt := range opts {
+		if err := opt(&base); err != nil {
+			return nil
+		}
+	}
+
+	target := base.Labels[snapshots.LabelSnapshotRef]
+	diffIDStr := base.Labels[snapshots.LabelSnapshotDiffID]
+	if target == "" || diffIDStr == "" {
+		// Not an image-layer extraction, or no diffID to key on.
+		return nil
+	}
+	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
+	}
+
+	blob := s.cacheBlobPath(diffID)
+	if _, err := os.Stat(blob); err != nil {
+		if !os.IsNotExist(err) {
+			log.G(ctx).WithError(err).WithField("blob", blob).
+				Warn("erofs layer cache: failed to stat cache blob, treating as cache miss")
+		}
+		return nil
+	}
+
+	return &cacheEntry{target: target, blob: blob}
+}
+
 func (s *snapshotter) View(ctx context.Context, key, parent string, opts ...snapshots.Opt) ([]mount.Mount, error) {
 	return s.createSnapshot(ctx, snapshots.KindView, key, parent, opts)
 }
@@ -798,10 +970,20 @@ func (s *snapshotter) Remove(ctx context.Context, key string) (err error) {
 
 		// The layer blob is only persisted for committed snapshots.
 		if info.Kind == snapshots.KindCommitted {
-			// Clear IMMUTABLE_FL before removal, since this flag avoids it.
-			err = setImmutable(s.layerBlobPath(id), false)
-			if err != nil && !errdefs.IsNotImplemented(err) {
-				return fmt.Errorf("failed to clear IMMUTABLE_FL: %w", err)
+			layerBlob := s.layerBlobPath(id)
+			// A cache-hit snapshot's blob is a symlink into the operator-owned
+			// cache dir. Skip clearing IMMUTABLE_FL: setImmutable's os.Open would
+			// follow the link and ioctl the cache entry (which we don't own), and
+			// cache blobs were never made immutable by us in the first place.
+			// os.RemoveAll below unlinks the symlink without following it.
+			if fi, lerr := os.Lstat(layerBlob); lerr == nil && fi.Mode()&os.ModeSymlink != 0 {
+				log.G(ctx).WithField("id", id).Trace("erofs layer cache: skipping IMMUTABLE_FL clear for symlinked cache blob")
+			} else {
+				// Clear IMMUTABLE_FL before removal, since this flag avoids it.
+				err = setImmutable(layerBlob, false)
+				if err != nil && !errdefs.IsNotImplemented(err) {
+					return fmt.Errorf("failed to clear IMMUTABLE_FL: %w", err)
+				}
 			}
 		}
 		_, _, err = storage.Remove(ctx, key)
diff --git a/plugins/snapshots/erofs/erofs_linux_test.go b/plugins/snapshots/erofs/erofs_linux_test.go
index 3c182d800d..6b6ff9b13e 100644
--- a/plugins/snapshots/erofs/erofs_linux_test.go
+++ b/plugins/snapshots/erofs/erofs_linux_test.go
@@ -26,6 +26,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/containerd/errdefs"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	bolt "go.etcd.io/bbolt"
@@ -818,3 +819,258 @@ func TestMountFsMeta(t *testing.T) {
 		}, m.Options)
 	})
 }
+
+// --- layer content cache tests ---
+
+const (
+	cacheTestDiffID  = "sha256:0000000000000000000000000000000000000000000000000000000000000001"
+	cacheTestChainID = "sha256:0000000000000000000000000000000000000000000000000000000000000002"
+)
+
+// requireErofs skips a test unless the erofs kernel filesystem is available,
+// which NewSnapshotter requires. The layer content cache tests don't mount, so
+// they need neither root nor mkfs.erofs.
+func requireErofs(t *testing.T) {
+	t.Helper()
+	if !FindErofs() {
+		t.Skip("check for erofs kernel support failed, skipping test")
+	}
+}
+
+// writeCacheBlob writes a fake erofs blob into cacheDir keyed by diffID and
+// returns its absolute path. The bytes need not be a valid erofs image: the
+// snapshotter only symlinks the blob, so these tests exercise the
+// Prepare/Commit/Remove logic rather than mounting.
+func writeCacheBlob(t *testing.T, cacheDir string, diffID digest.Digest, data []byte) string {
+	t.Helper()
+	blob := erofsutils.CacheBlobPath(cacheDir, diffID)
+	require.NoError(t, os.MkdirAll(filepath.Dir(blob), 0755))
+	require.NoError(t, os.WriteFile(blob, data, 0644))
+	return blob
+}
+
+// extractionOpt builds the snapshot options the unpacker attaches to an
+// image-layer extraction Prepare (the snapshot.ref target and the diffID).
+func extractionOpt(target string, diffID digest.Digest) snapshots.Opt {
+	return snapshots.WithLabels(map[string]string{
+		snapshots.LabelSnapshotRef:    target,
+		snapshots.LabelSnapshotDiffID: diffID.String(),
+	})
+}
+
+// snapshotID returns the backend snapshot ID for key.
+func snapshotID(t *testing.T, ctx context.Context, s *snapshotter, key string) string {
+	t.Helper()
+	var id string
+	require.NoError(t, s.ms.WithTransaction(ctx, false, func(ctx context.Context) error {
+		var err error
+		id, _, _, err = storage.GetInfo(ctx, key)
+		return err
+	}))
+	return id
+}
+
+// newCacheSnapshotter creates an erofs snapshotter rooted in a temp dir with the
+// given options, skipping the test if erofs is unavailable and registering the
+// snapshotter's cleanup.
+func newCacheSnapshotter(t *testing.T, opts ...Opt) *snapshotter {
+	t.Helper()
+	requireErofs(t)
+	sn, err := NewSnapshotter(t.TempDir(), opts...)
+	require.NoError(t, err)
+	t.Cleanup(func() { sn.Close() })
+	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) {
+	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")
+}
+
+// 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.
+func TestCacheHit(t *testing.T) {
+	ctx := namespaces.WithNamespace(context.Background(), "test")
+
+	cacheDir := t.TempDir()
+	diffID := digest.Digest(cacheTestDiffID)
+	blob := writeCacheBlob(t, cacheDir, diffID, []byte("fake erofs blob"))
+
+	s := newCacheSnapshotter(t, WithLayerContentCache(cacheDir))
+
+	target := cacheTestChainID
+	prepareCacheHit(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])
+
+	// layer.erofs is an absolute symlink into the operator-owned cache blob.
+	link := s.layerBlobPath(snapshotID(t, ctx, s, target))
+	fi, err := os.Lstat(link)
+	require.NoError(t, err)
+	assert.NotZero(t, fi.Mode()&os.ModeSymlink, "layer.erofs should be a symlink")
+	dst, err := os.Readlink(link)
+	require.NoError(t, err)
+	assert.True(t, filepath.IsAbs(dst), "symlink target should be absolute")
+	assert.Equal(t, blob, dst)
+}
+
+// TestCacheSidecar covers a hit in the default "auto" dm-verity mode where the
+// cache entry has a sidecar: it must be copied into the snapshot dir as a plain
+// regular file (not symlinked).
+func TestCacheSidecar(t *testing.T) {
+	ctx := namespaces.WithNamespace(context.Background(), "test")
+
+	cacheDir := t.TempDir()
+	diffID := digest.Digest(cacheTestDiffID)
+	blob := writeCacheBlob(t, cacheDir, diffID, []byte("fake erofs blob"))
+	require.NoError(t, os.WriteFile(dmverity.MetadataPath(blob), []byte(testDmverityMetadata), 0644))
+
+	// dmverity_mode defaults to "auto": use the sidecar if present.
+	s := newCacheSnapshotter(t, WithLayerContentCache(cacheDir))
+
+	target := cacheTestChainID
+	prepareCacheHit(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)))
+	fi, err := os.Lstat(sidecar)
+	require.NoError(t, err, "sidecar should be copied into the snapshot dir")
+	assert.Zero(t, fi.Mode()&os.ModeSymlink, "sidecar should be a regular file, not a symlink")
+	data, err := os.ReadFile(sidecar)
+	require.NoError(t, err)
+	assert.Equal(t, testDmverityMetadata, string(data))
+}
+
+// TestCacheMiss covers the cases that must NOT be served from the cache and
+// instead create a normal active snapshot: cache disabled, blob absent, a
+// label-less (container-rootfs) Prepare, and a View (which the KindActive gate
+// excludes even with matching labels and a cached blob).
+func TestCacheMiss(t *testing.T) {
+	ctx := namespaces.WithNamespace(context.Background(), "test")
+	diffID := digest.Digest(cacheTestDiffID)
+	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) {
+		t.Helper()
+		require.NoError(t, err)
+		assert.NotEmpty(t, mounts, "a miss must return normal active-snapshot mounts")
+		_, err = s.Stat(ctx, target)
+		assert.Error(t, err, "target chainID must not be committed on a miss")
+	}
+
+	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)
+	})
+
+	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)
+	})
+
+	t.Run("no extraction labels", func(t *testing.T) {
+		cacheDir := t.TempDir()
+		writeCacheBlob(t, cacheDir, diffID, []byte("blob"))
+		s := newCacheSnapshotter(t, WithLayerContentCache(cacheDir))
+		// A container-rootfs Prepare carries no snapshot.ref/diff-id labels.
+		mounts, err := s.Prepare(ctx, "container-rootfs", "")
+		require.NoError(t, err)
+		assert.NotEmpty(t, mounts)
+	})
+
+	t.Run("view is never short-circuited", func(t *testing.T) {
+		cacheDir := t.TempDir()
+		writeCacheBlob(t, cacheDir, diffID, []byte("blob"))
+		s := newCacheSnapshotter(t, WithLayerContentCache(cacheDir))
+		// Even with matching labels and a cached blob, a View must not commit.
+		mounts, err := s.View(ctx, "view-1", "", extractionOpt(target, diffID))
+		assertFellThrough(t, s, mounts, err)
+	})
+}
+
+// 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.
+func TestCacheRemove(t *testing.T) {
+	ctx := namespaces.WithNamespace(context.Background(), "test")
+
+	cacheDir := t.TempDir()
+	diffID := digest.Digest(cacheTestDiffID)
+	blob := writeCacheBlob(t, cacheDir, diffID, []byte("fake erofs blob"))
+	sidecar := dmverity.MetadataPath(blob)
+	require.NoError(t, os.WriteFile(sidecar, []byte(testDmverityMetadata), 0644))
+
+	s := newCacheSnapshotter(t, WithLayerContentCache(cacheDir))
+
+	target := cacheTestChainID
+	prepareCacheHit(t, ctx, s, target, diffID)
+
+	snapDir := filepath.Dir(s.layerBlobPath(snapshotID(t, ctx, s, target)))
+
+	// Remove must succeed (the symlinked blob is skipped by the setImmutable guard,
+	// which would otherwise follow the link and ioctl the operator-owned blob).
+	require.NoError(t, s.Remove(ctx, target))
+
+	// The snapshot dir (and its symlink) is gone, but the cache is untouched.
+	_, err := os.Stat(snapDir)
+	assert.True(t, os.IsNotExist(err), "snapshot dir should be removed")
+	_, err = os.Stat(blob)
+	require.NoError(t, err, "cache blob must be untouched by Remove")
+	_, err = os.Stat(sidecar)
+	require.NoError(t, err, "cache sidecar must be untouched by Remove")
+}
+
+// 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).
+func TestCacheDmverity(t *testing.T) {
+	if supported, err := dmverity.IsSupported(); err != nil || !supported {
+		t.Skip("dm-verity is not supported on this system")
+	}
+	ctx := namespaces.WithNamespace(context.Background(), "test")
+	diffID := digest.Digest(cacheTestDiffID)
+	target := cacheTestChainID
+
+	t.Run("with sidecar commits 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)
+
+		_, err := os.Stat(dmverity.MetadataPath(s.layerBlobPath(snapshotID(t, ctx, s, target))))
+		require.NoError(t, err, "sidecar must be present for a dmverity_mode=on hit")
+	})
+
+	t.Run("without sidecar fails the pull", func(t *testing.T) {
+		cacheDir := t.TempDir()
+		writeCacheBlob(t, cacheDir, diffID, []byte("fake erofs blob")) // no sidecar
+
+		s := newCacheSnapshotter(t, WithLayerContentCache(cacheDir), WithDmverityMode("on"))
+
+		// 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))
+		require.Error(t, err)
+		assert.False(t, errdefs.IsAlreadyExists(err), "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 834767f714..adf387b12f 100644
--- a/plugins/snapshots/erofs/plugin/plugin.go
+++ b/plugins/snapshots/erofs/plugin/plugin.go
@@ -55,6 +55,11 @@ type Config struct {
 	// DmverityMode controls dm-verity behavior: "auto" (use if available), "on" (require), "off" (disable)
 	// Linux only
 	DmverityMode string `toml:"dmverity_mode"`
+
+	// 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.
+	LayerContentCache string `toml:"layer_content_cache"`
 }
 
 func init() {
@@ -100,6 +105,10 @@ func init() {
 				opts = append(opts, erofs.WithDmverityMode(config.DmverityMode))
 			}
 
+			if config.LayerContentCache != "" {
+				opts = append(opts, erofs.WithLayerContentCache(config.LayerContentCache))
+			}
+
 			// Don't bother supporting overlay's slow_chown, only RemapIDs
 			ic.Meta.Capabilities = append(ic.Meta.Capabilities, capaOnlyRemapIDs)
 			if ok, err := supportsIDMappedMounts(); err == nil && ok {
@@ -108,7 +117,19 @@ func init() {
 			}
 
 			ic.Meta.Exports[plugins.SnapshotterRootDir] = root
-			ic.Meta.Capabilities = append(ic.Meta.Capabilities, "rebase")
+			// 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")
+			}
 			return erofs.NewSnapshotter(root, opts...)
 		},
 	})