diff --git a/internal/imagecache/README.md b/internal/imagecache/README.md index e02d7f27da..ce87433c2e 100644 --- a/internal/imagecache/README.md +++ b/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. diff --git a/internal/imagecache/gc.go b/internal/imagecache/gc.go index bdc1d6d192..8fca8e5c1f 100644 --- a/internal/imagecache/gc.go +++ b/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. diff --git a/internal/imagecache/imagecache.go b/internal/imagecache/imagecache.go index e0a0302347..220a41a74d 100644 --- a/internal/imagecache/imagecache.go +++ b/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) diff --git a/internal/imagecache/retire.go b/internal/imagecache/retire.go index 5d909dc9e5..3a83540bd0 100644 --- a/internal/imagecache/retire.go +++ b/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 } diff --git a/internal/imagecache/retire_test.go b/internal/imagecache/retire_test.go index fe45c5a3cc..f06b2555ac 100644 --- a/internal/imagecache/retire_test.go +++ b/internal/imagecache/retire_test.go @@ -17,8 +17,10 @@ package imagecache import ( "archive/tar" "context" + "errors" "os" "path/filepath" + "runtime" "strings" "sync" "testing" @@ -27,8 +29,8 @@ import ( v1 "github.com/google/go-containerregistry/pkg/v1" ) -// The retire/reuse interlock depends on retireLayer and ensureLayer using -// the same singleflight key; pin the key format to diffID.String()'s. +// The flight key is ensureLayer's dedup key; pin its format to +// diffID.String()'s so log/debug output lines up with recorded diffIDs. func TestLayerFlightKeyMatchesDiffIDString(t *testing.T) { hex := strings.Repeat("ab", 32) want := v1.Hash{Algorithm: "sha256", Hex: hex}.String() @@ -143,8 +145,8 @@ func TestSweepLeavesNonPrefixedEntries(t *testing.T) { } // TestRetireLayerVsEnsureImageRace races retirement against pulls of an -// image using the same layer. The singleflight serializes the retire -// rename against unpack, and the in-flight mtime touch turns concurrent +// image using the same layer. The layer interlock serializes the retire +// rename against unpack, and the mtime touch it covers turns concurrent // reuse into a veto — so neither side may ever error. Run with -race. func TestRetireLayerVsEnsureImageRace(t *testing.T) { _, host := newTestRegistry(t) @@ -201,3 +203,207 @@ func TestRetireLayerVsEnsureImageRace(t *testing.T) { t.Errorf("final layer dir missing: %v", err) } } + +// seedLayer pushes a one-layer image and pulls it, returning the layer, its +// diffID and its dir in the pool. +func seedLayer(t *testing.T, store *Store, ref string) (v1.Layer, v1.Hash, string) { + t.Helper() + layer := layerFromEntries(t, []tarEntry{ + {name: "f", typeflag: tar.TypeReg, mode: 0o644, body: strings.Repeat("j", 512)}, + }) + pushImage(t, ref, v1.Config{}, layer) + if _, err := store.EnsureImage(context.Background(), ref); err != nil { + t.Fatalf("EnsureImage: %v", err) + } + diffID, err := layer.DiffID() + if err != nil { + t.Fatalf("layer diffID: %v", err) + } + return layer, diffID, layerDirOf(t, store, layer) +} + +// waitForBlockedEnsure blocks until an ensureLayer goroutine is parked on +// frame. Observing the block is what makes these tests deterministic: +// releasing on a timer would let a loaded machine free the layer early, so +// ensureLayer would sail through and the contended path would go untested +// while the test still passed. +func waitForBlockedEnsure(t *testing.T, frame string) { + t.Helper() + buf := make([]byte, 1<<20) + waitFor(t, "ensureLayer to block on "+frame, func() bool { + n := runtime.Stack(buf, true) + for n == len(buf) { // runtime.Stack truncates silently at cap + buf = make([]byte, 2*len(buf)) + n = runtime.Stack(buf, true) + } + for _, g := range strings.Split(string(buf[:n]), "\n\ngoroutine ") { + if strings.Contains(g, "ensureLayer") && strings.Contains(g, frame) { + return true + } + } + return false + }) +} + +// ensureLayerAsync runs ensureLayer on its own goroutine. +func ensureLayerAsync(store *Store, diffID v1.Hash, layer v1.Layer) <-chan struct { + dir string + err error +} { + done := make(chan struct { + dir string + err error + }, 1) + go func() { + dir, err := store.ensureLayer(context.Background(), diffID, layer) + done <- struct { + dir string + err error + }{dir, err} + }() + return done +} + +// A GC pass must never block behind the pull path: a layer whose interlock +// is held is reused or being unpacked right now, which is a veto. +func TestRetireVetoedWhileLayerInterlockHeld(t *testing.T) { + _, host := newTestRegistry(t) + store := newTestStore(t) + _, diffID, dir := seedLayer(t, store, host+"/test/held:latest") + backdate(t, dir, 3*time.Hour) + + lock := store.layerLock(diffID.Hex) + if err := lock.Acquire(context.Background(), 1); err != nil { + t.Fatal(err) + } + defer lock.Release(1) + + _, st, err := store.retireLayer(diffID.Hex, time.Now()) + if err != nil || st != retireVetoed { + t.Fatalf("retireLayer while interlock held = %v, %v; want retireVetoed, nil", st, err) + } + if _, err := os.Stat(filepath.Join(dir, layerFSDirName)); err != nil { + t.Errorf("vetoed layer was disturbed: %v", err) + } +} + +// ensureLayer must never hand back a dir a retirement renamed away. It waits +// out the retirement on the interlock and then unpacks the layer again. +func TestEnsureLayerWaitsOutRetirementThenRepacks(t *testing.T) { + _, host := newTestRegistry(t) + store := newTestStore(t) + layer, diffID, dir := seedLayer(t, store, host+"/test/retire-wait:latest") + + // Hold the interlock across a rename, exactly as retireLayer does. + lock := store.layerLock(diffID.Hex) + if err := lock.Acquire(context.Background(), 1); err != nil { + t.Fatal(err) + } + if err := os.Rename(dir, dir+"-retired"); err != nil { + t.Fatal(err) + } + + done := ensureLayerAsync(store, diffID, layer) + waitForBlockedEnsure(t, "semaphore.(*Weighted).Acquire") + lock.Release(1) + + got := <-done + if got.err != nil { + t.Fatalf("ensureLayer: %v", got.err) + } + if _, err := os.Stat(filepath.Join(got.dir, layerFSDirName)); err != nil { + t.Fatalf("ensureLayer returned a retired dir: %v", err) + } +} + +// A retirement whose rename fails leaves the layer live and usable, so the +// ensureLayer waiting behind it must reuse that layer, not fail the pull. +func TestEnsureLayerWaitsOutFailedRetirementAndReusesLayer(t *testing.T) { + _, host := newTestRegistry(t) + store := newTestStore(t) + layer, diffID, dir := seedLayer(t, store, host+"/test/retire-failed:latest") + + // Rename "fails": the interlock is held but the dir is left in place. + lock := store.layerLock(diffID.Hex) + if err := lock.Acquire(context.Background(), 1); err != nil { + t.Fatal(err) + } + + done := ensureLayerAsync(store, diffID, layer) + waitForBlockedEnsure(t, "semaphore.(*Weighted).Acquire") + lock.Release(1) + + got := <-done + if got.err != nil { + t.Fatalf("ensureLayer: %v", got.err) + } + if got.dir != dir { + t.Errorf("ensureLayer = %q, want the live layer dir %q", got.dir, dir) + } +} + +// holdEnsureFlight occupies the layer's dedup flight, finishing with err, so +// an ensureLayer called meanwhile joins rather than leads. +func holdEnsureFlight(t *testing.T, store *Store, diffID v1.Hash, body func() error) (release func()) { + t.Helper() + held, releaseCh := make(chan struct{}), make(chan struct{}) + go func() { + _, _, _ = store.layerSF.Do(layerFlightKey(diffID.Hex), func() (any, error) { + err := body() + close(held) + <-releaseCh + return nil, err + }) + }() + <-held + var once sync.Once + release = func() { once.Do(func() { close(releaseCh) }) } + t.Cleanup(release) + return release +} + +// Joining an ensure flight is the dedup working: one download shared by the +// herd, failure included. Retrying per waiter would multiply full-size +// downloads under a persistent failure. +func TestEnsureLayerJoiningFailedEnsureSharesError(t *testing.T) { + _, host := newTestRegistry(t) + store := newTestStore(t) + layer, diffID, dir := seedLayer(t, store, host+"/test/join-pullfail:latest") + wantErr := errors.New("while unpacking layer: connection refused") + + release := holdEnsureFlight(t, store, diffID, func() error { + return errors.Join(os.RemoveAll(dir), wantErr) + }) + done := ensureLayerAsync(store, diffID, layer) + waitForBlockedEnsure(t, "sync.(*WaitGroup).Wait") + release() + + if got := <-done; !errors.Is(got.err, wantErr) { + t.Fatalf("ensureLayer = %v, want the joined flight's error %v", got.err, wantErr) + } +} + +// Distinct layers must never share an interlock. It is held across a whole +// unpack, so sharing would serialize unrelated multi-GiB downloads and make +// a retirement veto — and so re-date for a whole min-age window — a layer it +// could have taken. +func TestLayerLocksAreIndependentPerLayer(t *testing.T) { + store := newTestStore(t) + a, b := strings.Repeat("ab", 32), strings.Repeat("cd", 32) + + lockA, lockAgain, lockB := store.layerLock(a), store.layerLock(a), store.layerLock(b) + if lockA != lockAgain { + t.Fatal("one layer resolved to two different interlocks") + } + if lockA == lockB { + t.Fatal("two distinct layers share one interlock") + } + if !store.layerLock(a).TryAcquire(1) { + t.Fatal("could not take the first layer's interlock") + } + defer store.layerLock(a).Release(1) + if !store.layerLock(b).TryAcquire(1) { + t.Fatal("holding one layer's interlock blocked an unrelated layer") + } + store.layerLock(b).Release(1) +}