imagecache: Re-unpack a layer retired while joining its flight

internal/imagecache/README.md

@@ -244,13 +244,14 @@ skips the scan entirely, conservatively, if any record or bundle spec
 fails to read. There is no online whole-pool scan (ext4's split: bounded
 recovery at mount, fsck offline).
 
-**Deletion is two-phase.** A layer is atomically renamed to `.rm-*`
-inside the layer's singleflight (one `rename(2)` — eviction can never
-stall a pull), then removed asynchronously; a crash in between leaves
-the dir for the startup sweep. This matters because the kernel offers no
-protection here: deleting a directory that is a live overlay lowerdir in
-another mount namespace succeeds silently, leaves the overlay's behavior
-undefined, and doesn't even free the space until the mount goes away.
+**Deletion is two-phase.** A layer is atomically renamed to `.rm-*` under
+the layer's interlock (one `rename(2)`, taken only if uncontended — so
+eviction can never stall a pull), then removed asynchronously; a crash in
+between leaves the dir for the startup sweep. This matters because the
+kernel offers no protection here: deleting a directory that is a live
+overlay lowerdir in another mount namespace succeeds silently, leaves the
+overlay's behavior undefined, and doesn't even free the space until the
+mount goes away.
 
 Deleting the cache root by hand (while no actors are starting) remains
 safe — the store re-pulls whatever is missing.

internal/imagecache/gc.go

@@ -38,7 +38,7 @@ package imagecache
 //
 // Deletion is two-phase: the only steps that contend with the pull path
 // are one os.Remove of a record and one rename of a layer dir to a ".rm-*"
-// name inside the layer's singleflight (see retireLayer); the slow
+// name under the layer's interlock (see retireLayer); the slow
 // RemoveAll of multi-GB trees happens afterwards, on dirs nothing can
 // reach by diffID. A crash in between leaves a ".rm-*" dir for the
 // startup sweep.

internal/imagecache/imagecache.go

@@ -215,6 +215,11 @@ type Store struct {
 	imageSF singleflight.Group
 	layerSF singleflight.Group
 
+	// layerLocks holds the retire/reuse interlock per layer (see layerLock).
+	// Entries are added on first use and never removed, so this is the one
+	// piece of per-layer bookkeeping eviction does not reclaim.
+	layerLocks sync.Map
+
 	// evictMu serializes EvictUnused passes (concurrent passes would fight
 	// over the same candidates for no benefit).
 	evictMu sync.Mutex
@@ -517,7 +522,7 @@ func (s *Store) cachedImage(digest v1.Hash) (*Image, error) {
 			return nil, fmt.Errorf("invalid diffID %q in image record for %s: %w", d, digest, err)
 		}
 		dir := s.layerDir(diffID)
-		if _, err := os.Stat(filepath.Join(dir, layerFSDirName)); err != nil {
+		if !layerFSPresent(dir) {
 			return nil, nil
 		}
 		layerDirs[i] = dir
@@ -648,16 +653,34 @@ func (s *Store) pull(ctx context.Context, parsedRef name.Reference, digest v1.Ha
 	return &Image{Digest: digest, Config: cfgFile.Config, LayerDirs: layerDirs}, nil
 }
 
+// layerFSPresent reports whether a layer's unpacked tree is in the pool.
+// Any stat failure counts as absent, so an unreadable layer is re-pulled
+// rather than handed to a caller that cannot use it.
+func layerFSPresent(dir string) bool {
+	_, err := os.Stat(filepath.Join(dir, layerFSDirName))
+	return err == nil
+}
+
 // ensureLayer makes the unpacked tree for diffID present in the pool,
 // collapsing concurrent requests for the same layer across images.
+//
+// A layer pull that joins the flight instead of leading it shares the
+// leader's outcome: one download per herd, failures and the leader's own
+// cancellation included.
 func (s *Store) ensureLayer(ctx context.Context, diffID v1.Hash, layer v1.Layer) (string, error) {
 	dir := s.layerDir(diffID)
 	_, err, _ := s.layerSF.Do(layerFlightKey(diffID.Hex), func() (any, error) {
-		if _, err := os.Stat(filepath.Join(dir, layerFSDirName)); err == nil {
-			// Refresh the dir mtime inside the flight: retireLayer re-checks
-			// the mtime in this same flight, so a layer reused here can
-			// never be renamed away between this stat and the image record
-			// that will re-reference it.
+		lock := s.layerLock(diffID.Hex)
+		if err := lock.Acquire(ctx, 1); err != nil {
+			return nil, err
+		}
+		defer lock.Release(1)
+
+		if layerFSPresent(dir) {
+			// Refresh the mtime under the interlock: retireLayer re-checks it
+			// under the same lock, so a layer reused here can never be renamed
+			// away between this stat and the image record that will
+			// re-reference it.
 			now := time.Now()
 			if err := os.Chtimes(dir, now, now); err != nil {
 				slog.WarnContext(ctx, "Failed to refresh layer mtime on reuse", slog.String("diffid", diffID.String()), slog.Any("err", err))
@@ -731,7 +754,7 @@ func (s *Store) unpackLayerToPool(ctx context.Context, diffID v1.Hash, layer v1.
 	if err := os.Rename(tmp, s.layerDir(diffID)); err != nil {
 		// A concurrent unpack (another process sharing the pool) may have won;
 		// its layer is as good as ours.
-		if _, statErr := os.Stat(filepath.Join(s.layerDir(diffID), layerFSDirName)); statErr == nil {
+		if layerFSPresent(s.layerDir(diffID)) {
 			return nil
 		}
 		return fmt.Errorf("while moving layer into pool: %w", err)

internal/imagecache/retire.go

@@ -15,7 +15,7 @@
 package imagecache
 
 // Two-phase layer deletion: eviction renames a layer dir aside (one
-// rename(2) inside the layer singleflight — the only step that contends
+// rename(2) under the layer interlock — the only step that contends
 // with the pull path) and the slow RemoveAll of the renamed-aside tree
 // happens afterwards, outside all locks. A crash in between leaves a
 // ".rm-*" dir for the startup sweep. Nothing here needs privileges:
@@ -29,6 +29,8 @@ import (
 	"os"
 	"path/filepath"
 	"time"
+
+	"golang.org/x/sync/semaphore"
 )
 
 // retiredPrefix marks a layer dir that eviction has renamed aside and that
@@ -56,10 +58,27 @@ const (
 	retireRetired
 )
 
-// layerFlightKey is the singleflight key shared by ensureLayer and
-// retireLayer; the retire/reuse interlock depends on both using it.
+// layerFlightKey is ensureLayer's singleflight key: it collapses concurrent
+// requests for one layer onto a single download. Retirement deliberately
+// does not use it — see layerLock.
 func layerFlightKey(hex string) string { return "sha256:" + hex }
 
+// layerLock returns one layer's interlock, held by ensureLayer across its
+// stat and refresh-or-unpack and by retireLayer across its stat and rename,
+// so a reuse and a retirement never interleave. A lock rather than the
+// shared flight key, because a caller that joins a flight cannot tell
+// whether it joined a reuse or a retirement that renamed the dir away.
+// Never share one between layers: it is held for the length of an unpack,
+// so sharing would serialize unrelated downloads and let a retirement veto
+// a layer it could have taken.
+func (s *Store) layerLock(hex string) *semaphore.Weighted {
+	if lk, ok := s.layerLocks.Load(hex); ok {
+		return lk.(*semaphore.Weighted)
+	}
+	lk, _ := s.layerLocks.LoadOrStore(hex, semaphore.NewWeighted(1))
+	return lk.(*semaphore.Weighted)
+}
+
 // isLayerDirName reports whether name is a well-formed sha256 layer
 // directory name. Callers enumerate directories and read hexes out of
 // records, so they can encounter anything an operator (or a corrupt
@@ -80,68 +99,52 @@ func isLayerDirName(name string) bool {
 // returns the renamed path; the caller deletes it afterwards. A layer
 // with an mtime after cutoff is vetoed and left in place.
 //
-// The mtime check and the rename run inside the layer singleflight — the
-// same flight in which ensureLayer refreshes the mtime — so a retirement
-// and a reuse cannot interleave: whichever runs second sees the first.
+// The mtime check and the rename run under the layer's interlock — the same
+// lock ensureLayer holds while it refreshes the mtime — so a retirement and
+// a reuse cannot interleave: whichever runs second sees the first.
 func (s *Store) retireLayer(hex string, cutoff time.Time) (string, retireStatus, error) {
 	if !isLayerDirName(hex) {
 		return "", retireVetoed, fmt.Errorf("not a layer dir name: %q", hex)
 	}
 	dir := filepath.Join(s.layersDir(), hex)
 
-	// Pre-flight, outside the singleflight: a missing dir or a fresh mtime
-	// means nothing to do, and no reason to enter the flight and block
-	// behind an in-progress download. Both checks are re-run inside the
-	// flight before the rename, so this is a fast path, not the
+	// Pre-flight, outside the interlock: a missing dir or a fresh mtime means
+	// nothing to do and no reason to contend with the pull path at all. Both
+	// checks are re-run under the lock, so this is a fast path, not the
 	// correctness path.
+	if fi, err := os.Stat(dir); errors.Is(err, os.ErrNotExist) {
+		return "", retireGone, nil
+	} else if err != nil {
+		return "", retireVetoed, err
+	} else if fi.ModTime().After(cutoff) {
+		slog.Info(logMsgLayerRetireVetoed, slog.String("diffid", hex), slog.Time("last_used", fi.ModTime()))
+		return "", retireVetoed, nil
+	}
+
+	// If we cannot acquire the lock it's being held by ensureLayer, which is
+	// using the layer.
+	lock := s.layerLock(hex)
+	if !lock.TryAcquire(1) {
+		return "", retireVetoed, nil
+	}
+	defer lock.Release(1)
+
 	fi, err := os.Stat(dir)
 	if errors.Is(err, os.ErrNotExist) {
 		return "", retireGone, nil
 	} else if err != nil {
 		return "", retireVetoed, err
 	}
+	// A concurrent ensureLayer touched the dir if it reused this layer since
+	// the pre-flight stat.
 	if fi.ModTime().After(cutoff) {
 		slog.Info(logMsgLayerRetireVetoed, slog.String("diffid", hex), slog.Time("last_used", fi.ModTime()))
 		return "", retireVetoed, nil
 	}
 
-	var retired string
-	status := retireGone
-	ran := false
-	_, err, _ = s.layerSF.Do(layerFlightKey(hex), func() (any, error) {
-		ran = true
-		fi, err := os.Stat(dir)
-		if errors.Is(err, os.ErrNotExist) {
-			status = retireGone
-			return nil, nil
-		} else if err != nil {
-			status = retireVetoed
-			return nil, err
-		}
-		// A concurrent ensureLayer touched the dir if it reused this layer
-		// since the pre-flight stat.
-		if fi.ModTime().After(cutoff) {
-			slog.Info(logMsgLayerRetireVetoed, slog.String("diffid", hex), slog.Time("last_used", fi.ModTime()))
-			status = retireVetoed
-			return nil, nil
-		}
-		dst := filepath.Join(s.layersDir(), fmt.Sprintf("%s%s-%d", retiredPrefix, hex[:12], time.Now().UnixNano()))
-		if err := os.Rename(dir, dst); err != nil {
-			status = retireVetoed
-			return nil, fmt.Errorf("while retiring layer %s: %w", hex, err)
-		}
-		retired = dst
-		status = retireRetired
-		return nil, nil
-	})
-	if !ran {
-		// Our closure never executed: Do joined a flight already in progress
-		// (an ensureLayer reuse or another retire), so status/retired are
-		// stale zero values. Concurrent activity on the layer is a veto.
-		return "", retireVetoed, nil
-	}
-	if err != nil {
-		return "", status, err
+	dst := filepath.Join(s.layersDir(), fmt.Sprintf("%s%s-%d", retiredPrefix, hex[:12], time.Now().UnixNano()))
+	if err := os.Rename(dir, dst); err != nil {
+		return "", retireVetoed, fmt.Errorf("while retiring layer %s: %w", hex, err)
 	}
-	return retired, status, nil
+	return dst, retireRetired, nil
 }