From f394887a48b7142d289dc301527fe550bc5ef237 Mon Sep 17 00:00:00 2001 From: Jonathan Ogilvie Date: Sat, 7 Nov 2026 15:39:32 -0500 Subject: [PATCH 1/2] fix(project): make the base-image layer cache write atomically go-containerregistry's filesystem cache creates a layer's final file up front and tees the download into it, so a concurrent build sharing the cache directory could Get a half-written file as a hit and take the layer's size from the partial bytes. Writing the image to the daemon then failed with "archive/tar: write too long". Add atomicCache, which writes each layer to a temp file in the cache directory and renames it into place only after the stream was read to EOF and closed cleanly. The on-disk layout is unchanged. Co-Authored-By: Claude Opus 5.5 (1M context) Signed-off-by: Jonathan Ogilvie --- internal/project/functions/basecache.go | 126 ++++++++++++++ internal/project/functions/basecache_test.go | 168 ++++++++++++++++++- internal/project/functions/kcl.go | 2 +- 3 files changed, 292 insertions(+), 4 deletions(-) diff --git a/internal/project/functions/basecache.go b/internal/project/functions/basecache.go index ca3202e4..3624e093 100644 --- a/internal/project/functions/basecache.go +++ b/internal/project/functions/basecache.go @@ -17,9 +17,11 @@ limitations under the License. package functions import ( + "fmt" "io" "os" "path/filepath" + "runtime" v1 "github.com/google/go-containerregistry/pkg/v1" "github.com/google/go-containerregistry/pkg/v1/cache" @@ -54,6 +56,130 @@ func DefaultBaseImageCacheDir() string { return filepath.Join(base, "crossplane", "base-images") } +// atomicCache is a cache.Cache whose entries only become visible once complete. +// +// go-containerregistry's filesystem cache creates a layer's final file up front +// and tees the download into it, so a concurrent Get (another build, possibly +// another process, sharing the directory) can find a half-written file and +// treat it as a hit, taking the layer's size from the partial bytes. Writing +// that layer to a tarball then fails with "archive/tar: write too long". +// +// atomicCache instead writes each layer to a temporary file in the cache +// directory and renames it to its final path only after the whole stream has +// been read to EOF and closed without error. Any other outcome removes the +// temporary file. Concurrent writers of one digest are harmless: rename is +// atomic and the content is identical. Get and Delete use the filesystem +// cache, and the final paths are the same, so the on-disk layout is unchanged +// and existing caches stay valid. A process killed mid-download can leave a +// ".tmp-*" file behind; nothing reads those. +type atomicCache struct { + cache.Cache + + dir string +} + +func newAtomicCache(dir string) atomicCache { + return atomicCache{Cache: cache.NewFilesystemCache(dir), dir: dir} +} + +// Put returns a layer that populates the cache as it is read. +func (c atomicCache) Put(l v1.Layer) (v1.Layer, error) { + digest, err := l.Digest() + if err != nil { + return nil, err + } + diffID, err := l.DiffID() + if err != nil { + return nil, err + } + return &atomicLayer{Layer: l, dir: c.dir, digest: digest, diffID: diffID}, nil +} + +type atomicLayer struct { + v1.Layer + + dir string + digest, diffID v1.Hash +} + +func (l *atomicLayer) Compressed() (io.ReadCloser, error) { + return l.tee(l.digest, l.Layer.Compressed) +} + +func (l *atomicLayer) Uncompressed() (io.ReadCloser, error) { + return l.tee(l.diffID, l.Layer.Uncompressed) +} + +func (l *atomicLayer) tee(h v1.Hash, open func() (io.ReadCloser, error)) (io.ReadCloser, error) { + if err := os.MkdirAll(l.dir, 0o700); err != nil { + return nil, err + } + f, err := os.CreateTemp(l.dir, ".tmp-*") + if err != nil { + return nil, err + } + rc, err := open() + if err != nil { + _ = f.Close() + _ = os.Remove(f.Name()) + return nil, err + } + return &atomicReader{rc: rc, f: f, final: cachePath(l.dir, h)}, nil +} + +// atomicReader copies what it reads into a temporary file, and publishes the +// file only if the stream was read to EOF without error. +type atomicReader struct { + rc io.ReadCloser + f *os.File + final string + + complete bool + failed bool +} + +func (r *atomicReader) Read(b []byte) (int, error) { + n, err := r.rc.Read(b) + if n > 0 { + if _, werr := r.f.Write(b[:n]); werr != nil { + r.failed = true + return n, werr + } + } + switch { + case errors.Is(err, io.EOF): + r.complete = true + case err != nil: + r.failed = true + } + return n, err +} + +func (r *atomicReader) Close() error { + rerr := r.rc.Close() + ferr := r.f.Close() + if r.failed || !r.complete || rerr != nil || ferr != nil { + _ = os.Remove(r.f.Name()) + if rerr != nil { + return rerr + } + return ferr + } + if err := os.Rename(r.f.Name(), r.final); err != nil { + _ = os.Remove(r.f.Name()) + return err + } + return nil +} + +// cachePath mirrors go-containerregistry's filesystem cache layout. +func cachePath(dir string, h v1.Hash) string { + if runtime.GOOS == "windows" { + return filepath.Join(dir, fmt.Sprintf("%s-%s", h.Algorithm, h.Hex)) + } + return filepath.Join(dir, h.String()) +} + // tolerantCache degrades to the registry instead of failing the build when the // cache itself cannot be used. // diff --git a/internal/project/functions/basecache_test.go b/internal/project/functions/basecache_test.go index 9d94961f..74e26aaf 100644 --- a/internal/project/functions/basecache_test.go +++ b/internal/project/functions/basecache_test.go @@ -17,10 +17,15 @@ limitations under the License. package functions import ( + "bytes" + "compress/gzip" + "crypto/rand" "io" "os" "path/filepath" + "sync" "testing" + "time" "github.com/google/go-cmp/cmp" v1 "github.com/google/go-containerregistry/pkg/v1" @@ -94,7 +99,7 @@ func TestTolerantCacheUnwritableDirStillReads(t *testing.T) { t.Fatalf("creating read-only cache dir: %v", err) } - plain := cache.NewFilesystemCache(dir) + plain := newAtomicCache(dir) l, err := plain.Put(testLayer()) if err != nil { t.Fatalf("Put(): unexpected error: %v", err) @@ -104,7 +109,7 @@ func TestTolerantCacheUnwritableDirStillReads(t *testing.T) { } // Same directory, wrapped: the layer must read regardless. - tolerant := tolerantCache{cache.NewFilesystemCache(dir)} + tolerant := tolerantCache{newAtomicCache(dir)} wrapped, err := tolerant.Put(testLayer()) if err != nil { t.Fatalf("Put(): unexpected error: %v", err) @@ -117,7 +122,7 @@ func TestTolerantCacheUnwritableDirStillReads(t *testing.T) { func TestTolerantCachePassesThroughOnSuccess(t *testing.T) { // A working cache must behave exactly as it would unwrapped: a miss is // still a miss, and a stored layer still reads back. - c := tolerantCache{cache.NewFilesystemCache(t.TempDir())} + c := tolerantCache{newAtomicCache(t.TempDir())} if _, err := c.Get(v1.Hash{Algorithm: "sha256", Hex: "cafe"}); !errors.Is(err, cache.ErrNotFound) { t.Errorf("Get() on empty cache: want cache.ErrNotFound, got %v", err) @@ -140,3 +145,160 @@ func TestTolerantCachePassesThroughOnSuccess(t *testing.T) { t.Errorf("Get() after populating cache: unexpected error: %v", err) } } + +// gzipLayer returns a layer whose compressed form is real gzip, as registry +// layers are, so the filesystem cache reads it back with the same digest. +func gzipLayer(t *testing.T, payload []byte) v1.Layer { + t.Helper() + var buf bytes.Buffer + zw := gzip.NewWriter(&buf) + if _, err := zw.Write(payload); err != nil { + t.Fatalf("gzip: %v", err) + } + if err := zw.Close(); err != nil { + t.Fatalf("gzip: %v", err) + } + return static.NewLayer(buf.Bytes(), types.DockerLayer) +} + +// slowLayer delivers its compressed bytes in small chunks with pauses, so +// concurrent readers reliably overlap the download. +type slowLayer struct{ v1.Layer } + +func (l slowLayer) Compressed() (io.ReadCloser, error) { + rc, err := l.Layer.Compressed() + if err != nil { + return nil, err + } + return slowReader{rc}, nil +} + +type slowReader struct{ io.ReadCloser } + +func (r slowReader) Read(b []byte) (int, error) { + time.Sleep(time.Millisecond) + return r.ReadCloser.Read(b[:min(len(b), 1024)]) +} + +func TestAtomicCachePartialReadLeavesNoEntry(t *testing.T) { + noise := make([]byte, 1<<20) + _, _ = rand.Read(noise) + big := gzipLayer(t, noise) + d, err := big.Digest() + if err != nil { + t.Fatalf("Digest(): %v", err) + } + + tests := map[string]func(rc io.ReadCloser){ + "ClosedEarly": func(rc io.ReadCloser) { + _, _ = io.CopyN(io.Discard, rc, 100) + }, + "NeverRead": func(io.ReadCloser) {}, + } + for name, consume := range tests { + t.Run(name, func(t *testing.T) { + c := newAtomicCache(t.TempDir()) + l, err := c.Put(big) + if err != nil { + t.Fatalf("Put(): %v", err) + } + rc, err := l.Compressed() + if err != nil { + t.Fatalf("Compressed(): %v", err) + } + consume(rc) + _ = rc.Close() + + if _, err := c.Get(d); !errors.Is(err, cache.ErrNotFound) { + t.Errorf("Get() after partial read: want cache.ErrNotFound, got %v", err) + } + if ents, _ := os.ReadDir(c.dir); len(ents) != 0 { + t.Errorf("cache dir should be empty after partial read, got %d entries", len(ents)) + } + }) + } +} + +func TestAtomicCacheFullReadIsHit(t *testing.T) { + c := newAtomicCache(t.TempDir()) + src := gzipLayer(t, []byte("hello from a layer")) + l, err := c.Put(src) + if err != nil { + t.Fatalf("Put(): %v", err) + } + readAll(t, l) + + d, _ := src.Digest() + wantSize, _ := src.Size() + got, err := c.Get(d) + if err != nil { + t.Fatalf("Get(): %v", err) + } + gotDigest, _ := got.Digest() + gotSize, _ := got.Size() + if gotDigest != d || gotSize != wantSize { + t.Errorf("Get(): want %v/%d, got %v/%d", d, wantSize, gotDigest, gotSize) + } +} + +func TestAtomicCacheConcurrentReadersNeverSeePartialLayers(t *testing.T) { + payload := make([]byte, 256<<10) + _, _ = rand.Read(payload) + src := slowLayer{gzipLayer(t, payload)} + d, _ := src.Digest() + wantSize, _ := src.Size() + + c := newAtomicCache(t.TempDir()) + var wg sync.WaitGroup + errs := make(chan error, 64) + + for range 8 { + wg.Go(func() { + l, err := c.Put(src) + if err != nil { + errs <- err + return + } + rc, err := l.Compressed() + if err != nil { + errs <- err + return + } + // Small reads keep the download slow enough to overlap Gets. + buf := make([]byte, 4096) + for { + if _, err := rc.Read(buf); err != nil { + break + } + } + if err := rc.Close(); err != nil { + errs <- err + } + }) + } + for range 8 { + wg.Go(func() { + for range 200 { + l, err := c.Get(d) + if errors.Is(err, cache.ErrNotFound) { + continue + } + if err != nil { + errs <- err + return + } + gotSize, _ := l.Size() + gotDigest, _ := l.Digest() + if gotSize != wantSize || gotDigest != d { + errs <- errors.Errorf("Get() returned %v/%d, want %v/%d", gotDigest, gotSize, d, wantSize) + return + } + } + }) + } + wg.Wait() + close(errs) + for err := range errs { + t.Error(err) + } +} diff --git a/internal/project/functions/kcl.go b/internal/project/functions/kcl.go index 257827ce..a6e268b3 100644 --- a/internal/project/functions/kcl.go +++ b/internal/project/functions/kcl.go @@ -154,7 +154,7 @@ func baseImageForArch(ref name.Reference, arch string, transport http.RoundTripp // tolerantCache keeps a cache that cannot be read or written from // failing the build; see its doc comment for the one case it cannot // recover from. - img = cache.Image(img, tolerantCache{cache.NewFilesystemCache(cacheDir)}) + img = cache.Image(img, tolerantCache{newAtomicCache(cacheDir)}) } cfg, err := img.ConfigFile() From 3e1bdebfa72a1a38c65a2265599def65c95eda8d Mon Sep 17 00:00:00 2001 From: Jonathan Ogilvie Date: Sat, 7 Nov 2026 17:00:55 -0500 Subject: [PATCH 2/2] fix(project): keep cache-only failures from failing a layer read Co-Authored-By: Claude Opus 5.5 (1M context) Signed-off-by: Jonathan Ogilvie --- internal/project/functions/basecache.go | 60 ++-- internal/project/functions/basecache_test.go | 328 ++++++++++++------- 2 files changed, 255 insertions(+), 133 deletions(-) diff --git a/internal/project/functions/basecache.go b/internal/project/functions/basecache.go index 3624e093..87021f21 100644 --- a/internal/project/functions/basecache.go +++ b/internal/project/functions/basecache.go @@ -67,11 +67,12 @@ func DefaultBaseImageCacheDir() string { // atomicCache instead writes each layer to a temporary file in the cache // directory and renames it to its final path only after the whole stream has // been read to EOF and closed without error. Any other outcome removes the -// temporary file. Concurrent writers of one digest are harmless: rename is -// atomic and the content is identical. Get and Delete use the filesystem -// cache, and the final paths are the same, so the on-disk layout is unchanged -// and existing caches stay valid. A process killed mid-download can leave a -// ".tmp-*" file behind; nothing reads those. +// temporary file, and a failure to write or publish it is absorbed, so a full +// or unwritable disk degrades to "not cached". Concurrent writers of one +// digest are harmless: rename is atomic and the content is identical. Get and +// Delete use the filesystem cache, and the final paths are the same, so the +// on-disk layout is unchanged and existing caches stay valid. A process killed +// mid-download can leave a ".tmp-*" file behind; nothing reads those. type atomicCache struct { cache.Cache @@ -128,7 +129,8 @@ func (l *atomicLayer) tee(h v1.Hash, open func() (io.ReadCloser, error)) (io.Rea } // atomicReader copies what it reads into a temporary file, and publishes the -// file only if the stream was read to EOF without error. +// file only if the stream was read to EOF without error. Failures that only +// affect the cache never change what the caller reads. type atomicReader struct { rc io.ReadCloser f *os.File @@ -140,10 +142,11 @@ type atomicReader struct { func (r *atomicReader) Read(b []byte) (int, error) { n, err := r.rc.Read(b) - if n > 0 { + if n > 0 && r.f != nil { if _, werr := r.f.Write(b[:n]); werr != nil { - r.failed = true - return n, werr + // The caller still gets the source's bytes; only the cache entry + // is lost. + r.abandon() } } switch { @@ -155,20 +158,33 @@ func (r *atomicReader) Read(b []byte) (int, error) { return n, err } +// abandon drops the temporary file and stops teeing into it. +func (r *atomicReader) abandon() { + if r.f == nil { + return + } + _ = r.f.Close() + _ = os.Remove(r.f.Name()) + r.f = nil +} + +// Close publishes the entry if the whole stream was read. Once the source has +// been read to EOF and closed cleanly the caller has every byte, so a failure +// to finish the cache entry is absorbed rather than returned. func (r *atomicReader) Close() error { rerr := r.rc.Close() - ferr := r.f.Close() - if r.failed || !r.complete || rerr != nil || ferr != nil { - _ = os.Remove(r.f.Name()) - if rerr != nil { - return rerr - } - return ferr + if r.f == nil { + return rerr + } + if r.failed || !r.complete || rerr != nil { + r.abandon() + return rerr } - if err := os.Rename(r.f.Name(), r.final); err != nil { - _ = os.Remove(r.f.Name()) - return err + name := r.f.Name() + if err := r.f.Close(); err != nil || os.Rename(name, r.final) != nil { + _ = os.Remove(name) } + r.f = nil return nil } @@ -191,10 +207,8 @@ func cachePath(dir string, h v1.Hash) string { // nothing prunes this cache, which makes a full disk a plausible way to reach // it. See DefaultBaseImageCacheDir. // -// One case is not recoverable here: if the directory is writable but fills up -// partway through a layer, the write error surfaces mid-stream, after the -// reader has been handed to the caller. Restarting from the remote layer at -// that point would mean re-reading bytes the caller already consumed. +// A directory that fills up partway through a layer is absorbed by atomicCache, +// which stops caching and keeps serving the remote bytes. type tolerantCache struct { cache.Cache } diff --git a/internal/project/functions/basecache_test.go b/internal/project/functions/basecache_test.go index 74e26aaf..03a1bcda 100644 --- a/internal/project/functions/basecache_test.go +++ b/internal/project/functions/basecache_test.go @@ -23,6 +23,7 @@ import ( "io" "os" "path/filepath" + "strings" "sync" "testing" "time" @@ -62,87 +63,99 @@ func (erroringCache) Get(v1.Hash) (v1.Layer, error) { return nil, errors.New("b func (erroringCache) Put(v1.Layer) (v1.Layer, error) { return nil, errors.New("boom") } func (erroringCache) Delete(v1.Hash) error { return errors.New("boom") } -func TestTolerantCacheGetReportsFailuresAsMisses(t *testing.T) { - // A read failure that is not a genuine miss must still look like a miss, - // so the caller falls through to the registry instead of erroring out. - c := tolerantCache{erroringCache{}} - - _, err := c.Get(v1.Hash{Algorithm: "sha256", Hex: "cafe"}) - if !errors.Is(err, cache.ErrNotFound) { - t.Errorf("Get(): want cache.ErrNotFound, got %v", err) - } -} - -func TestTolerantCachePutFailureReturnsUsableLayer(t *testing.T) { - // Put failing outright must hand back a layer that still reads. - c := tolerantCache{erroringCache{}} - - l, err := c.Put(testLayer()) - if err != nil { - t.Fatalf("Put(): unexpected error: %v", err) +func TestTolerantCacheGet(t *testing.T) { + cases := map[string]struct { + reason string + cache cache.Cache + }{ + "FailureIsMiss": { + reason: "A read failure that is not a genuine miss must still look like a miss, so the caller falls through to the registry.", + cache: erroringCache{}, + }, + "EmptyCacheIsMiss": { + reason: "A working cache must report a genuine miss as a miss.", + cache: newAtomicCache(t.TempDir()), + }, } - if diff := cmp.Diff("hello from a layer", readAll(t, l)); diff != "" { - t.Errorf("layer contents (-want +got):\n%s", diff) + for name, tc := range cases { + t.Run(name, func(t *testing.T) { + _, err := tolerantCache{tc.cache}.Get(v1.Hash{Algorithm: "sha256", Hex: "cafe"}) + if !errors.Is(err, cache.ErrNotFound) { + t.Errorf("\n%s\nGet(...): want cache.ErrNotFound, got %v", tc.reason, err) + } + }) } } -func TestTolerantCacheUnwritableDirStillReads(t *testing.T) { - // The regression this guards: an unwritable cache directory used to fail - // the build, because the filesystem cache creates its backing file lazily - // inside Compressed(). - if os.Geteuid() == 0 { - t.Skip("running as root, which can write to a read-only directory") - } - - dir := filepath.Join(t.TempDir(), "cache") - if err := os.Mkdir(dir, 0o500); err != nil { - t.Fatalf("creating read-only cache dir: %v", err) - } - - plain := newAtomicCache(dir) - l, err := plain.Put(testLayer()) - if err != nil { - t.Fatalf("Put(): unexpected error: %v", err) - } - if _, err := l.Compressed(); err == nil { - t.Skip("cache directory turned out to be writable; nothing to assert") - } - - // Same directory, wrapped: the layer must read regardless. - tolerant := tolerantCache{newAtomicCache(dir)} - wrapped, err := tolerant.Put(testLayer()) - if err != nil { - t.Fatalf("Put(): unexpected error: %v", err) - } - if diff := cmp.Diff("hello from a layer", readAll(t, wrapped)); diff != "" { - t.Errorf("layer contents (-want +got):\n%s", diff) +func TestTolerantCachePut(t *testing.T) { + type want struct { + contents string + cached bool } -} - -func TestTolerantCachePassesThroughOnSuccess(t *testing.T) { - // A working cache must behave exactly as it would unwrapped: a miss is - // still a miss, and a stored layer still reads back. - c := tolerantCache{newAtomicCache(t.TempDir())} - - if _, err := c.Get(v1.Hash{Algorithm: "sha256", Hex: "cafe"}); !errors.Is(err, cache.ErrNotFound) { - t.Errorf("Get() on empty cache: want cache.ErrNotFound, got %v", err) - } - - l, err := c.Put(testLayer()) - if err != nil { - t.Fatalf("Put(): unexpected error: %v", err) - } - if diff := cmp.Diff("hello from a layer", readAll(t, l)); diff != "" { - t.Errorf("layer contents (-want +got):\n%s", diff) - } - - // Reading the layer populates the cache, so the digest is now a hit. - d, err := l.Digest() - if err != nil { - t.Fatalf("Digest(): unexpected error: %v", err) + cases := map[string]struct { + reason string + cache func(t *testing.T) cache.Cache + want want + }{ + "PutFailureReturnsUsableLayer": { + reason: "Put failing outright must hand back a layer that still reads.", + cache: func(*testing.T) cache.Cache { return erroringCache{} }, + want: want{contents: "hello from a layer"}, + }, + "UnwritableDirStillReads": { + // An unwritable cache directory used to fail the build, because the + // filesystem cache creates its backing file lazily inside Compressed(). + reason: "A layer must read even when the cache directory is unwritable.", + cache: func(t *testing.T) cache.Cache { + t.Helper() + if os.Geteuid() == 0 { + t.Skip("running as root, which can write to a read-only directory") + } + dir := filepath.Join(t.TempDir(), "cache") + if err := os.Mkdir(dir, 0o500); err != nil { + t.Fatalf("creating read-only cache dir: %v", err) + } + l, err := newAtomicCache(dir).Put(testLayer()) + if err != nil { + t.Fatalf("Put(): unexpected error: %v", err) + } + if _, err := l.Compressed(); err == nil { + t.Skip("cache directory turned out to be writable; nothing to assert") + } + return newAtomicCache(dir) + }, + want: want{contents: "hello from a layer"}, + }, + "PassesThroughOnSuccess": { + reason: "A working cache must behave as it would unwrapped: reading the layer populates the cache.", + cache: func(t *testing.T) cache.Cache { + t.Helper() + return newAtomicCache(t.TempDir()) + }, + want: want{contents: "hello from a layer", cached: true}, + }, } - if _, err := c.Get(d); err != nil { - t.Errorf("Get() after populating cache: unexpected error: %v", err) + for name, tc := range cases { + t.Run(name, func(t *testing.T) { + c := tolerantCache{tc.cache(t)} + l, err := c.Put(testLayer()) + if err != nil { + t.Fatalf("\n%s\nPut(...): unexpected error: %v", tc.reason, err) + } + if diff := cmp.Diff(tc.want.contents, readAll(t, l)); diff != "" { + t.Errorf("\n%s\nPut(...): layer contents -want, +got:\n%s", tc.reason, diff) + } + if !tc.want.cached { + return + } + d, err := l.Digest() + if err != nil { + t.Fatalf("Digest(): unexpected error: %v", err) + } + if _, err := c.Get(d); err != nil { + t.Errorf("\n%s\nGet(...) after populating cache: unexpected error: %v", tc.reason, err) + } + }) } } @@ -180,68 +193,150 @@ func (r slowReader) Read(b []byte) (int, error) { return r.ReadCloser.Read(b[:min(len(b), 1024)]) } -func TestAtomicCachePartialReadLeavesNoEntry(t *testing.T) { +// closeErrLayer is a layer whose streams fail on Close. +type closeErrLayer struct{ v1.Layer } + +func (l closeErrLayer) Compressed() (io.ReadCloser, error) { + rc, err := l.Layer.Compressed() + if err != nil { + return nil, err + } + return closeErrReader{rc}, nil +} + +type closeErrReader struct{ io.ReadCloser } + +func (closeErrReader) Close() error { return errors.New("boom") } + +func TestAtomicCacheReadOutcomes(t *testing.T) { noise := make([]byte, 1<<20) _, _ = rand.Read(noise) big := gzipLayer(t, noise) - d, err := big.Digest() - if err != nil { - t.Fatalf("Digest(): %v", err) + small := gzipLayer(t, []byte("hello from a layer")) + + type args struct { + layer v1.Layer + // blockFinalPath puts a non-empty directory at the entry's final path, + // so publishing the entry fails. + blockFinalPath bool + consume func(rc io.ReadCloser) } - - tests := map[string]func(rc io.ReadCloser){ - "ClosedEarly": func(rc io.ReadCloser) { - _, _ = io.CopyN(io.Discard, rc, 100) + type want struct { + closeErr bool + hit bool + } + readAllOf := func(rc io.ReadCloser) { _, _ = io.Copy(io.Discard, rc) } + + cases := map[string]struct { + reason string + args args + want want + }{ + "ClosedEarly": { + reason: "A stream closed before EOF must not leave a cache entry.", + args: args{layer: big, consume: func(rc io.ReadCloser) { _, _ = io.CopyN(io.Discard, rc, 100) }}, + }, + "NeverRead": { + reason: "A stream never read must not leave a cache entry.", + args: args{layer: big, consume: func(io.ReadCloser) {}}, + }, + "FullReadIsHit": { + reason: "A stream read to EOF and closed must be published as a hit.", + args: args{layer: small, consume: readAllOf}, + want: want{hit: true}, + }, + "SourceCloseFailsAfterEOF": { + reason: "A failing source Close must be returned and must not publish an entry.", + args: args{layer: closeErrLayer{small}, consume: readAllOf}, + want: want{closeErr: true}, + }, + "RenameFailsAfterFullRead": { + reason: "A cache-only failure after a full read must not fail Close, and must not leave a temp file.", + args: args{layer: small, blockFinalPath: true, consume: readAllOf}, }, - "NeverRead": func(io.ReadCloser) {}, } - for name, consume := range tests { + for name, tc := range cases { t.Run(name, func(t *testing.T) { c := newAtomicCache(t.TempDir()) - l, err := c.Put(big) + d, _ := tc.args.layer.Digest() + if tc.args.blockFinalPath { + if err := os.MkdirAll(filepath.Join(cachePath(c.dir, d), "x"), 0o700); err != nil { + t.Fatal(err) + } + } + l, err := c.Put(tc.args.layer) if err != nil { - t.Fatalf("Put(): %v", err) + t.Fatalf("\n%s\nPut(...): %v", tc.reason, err) } rc, err := l.Compressed() if err != nil { - t.Fatalf("Compressed(): %v", err) + t.Fatalf("\n%s\nCompressed(): %v", tc.reason, err) + } + tc.args.consume(rc) + if gotErr := rc.Close() != nil; gotErr != tc.want.closeErr { + t.Errorf("\n%s\nClose(): want error %t, got error %t", tc.reason, tc.want.closeErr, gotErr) } - consume(rc) - _ = rc.Close() - if _, err := c.Get(d); !errors.Is(err, cache.ErrNotFound) { - t.Errorf("Get() after partial read: want cache.ErrNotFound, got %v", err) + ents, _ := os.ReadDir(c.dir) + for _, e := range ents { + if strings.HasPrefix(e.Name(), ".tmp-") { + t.Errorf("\n%s\ntemp file %s left behind", tc.reason, e.Name()) + } + } + if tc.args.blockFinalPath { + return + } + got, err := c.Get(d) + if !tc.want.hit { + if !errors.Is(err, cache.ErrNotFound) { + t.Errorf("\n%s\nGet(...): want cache.ErrNotFound, got %v", tc.reason, err) + } + if len(ents) != 0 { + t.Errorf("\n%s\ncache dir should be empty, got %d entries", tc.reason, len(ents)) + } + return + } + if err != nil { + t.Fatalf("\n%s\nGet(...): %v", tc.reason, err) } - if ents, _ := os.ReadDir(c.dir); len(ents) != 0 { - t.Errorf("cache dir should be empty after partial read, got %d entries", len(ents)) + gotDigest, _ := got.Digest() + gotSize, _ := got.Size() + wantSize, _ := tc.args.layer.Size() + if gotDigest != d || gotSize != wantSize { + t.Errorf("\n%s\nGet(...): want %v/%d, got %v/%d", tc.reason, d, wantSize, gotDigest, gotSize) } }) } } -func TestAtomicCacheFullReadIsHit(t *testing.T) { - c := newAtomicCache(t.TempDir()) - src := gzipLayer(t, []byte("hello from a layer")) - l, err := c.Put(src) +func TestAtomicReaderWriteFailureStillReturnsSourceBytes(t *testing.T) { + reason := "A cache write failure mid-stream must not change what the caller reads." + dir := t.TempDir() + f, err := os.CreateTemp(dir, ".tmp-*") if err != nil { - t.Fatalf("Put(): %v", err) + t.Fatal(err) } - readAll(t, l) - - d, _ := src.Digest() - wantSize, _ := src.Size() - got, err := c.Get(d) + // Closing the file up front makes every write to it fail. + _ = f.Close() + want := []byte("hello from a layer") + r := &atomicReader{rc: io.NopCloser(bytes.NewReader(want)), f: f, final: filepath.Join(dir, "final")} + got, err := io.ReadAll(r) if err != nil { - t.Fatalf("Get(): %v", err) + t.Fatalf("\n%s\nReadAll(...): %v", reason, err) } - gotDigest, _ := got.Digest() - gotSize, _ := got.Size() - if gotDigest != d || gotSize != wantSize { - t.Errorf("Get(): want %v/%d, got %v/%d", d, wantSize, gotDigest, gotSize) + if diff := cmp.Diff(want, got); diff != "" { + t.Errorf("\n%s\nReadAll(...): -want, +got:\n%s", reason, diff) + } + if err := r.Close(); err != nil { + t.Errorf("\n%s\nClose(): %v", reason, err) + } + if ents, _ := os.ReadDir(dir); len(ents) != 0 { + t.Errorf("\n%s\ndir should be empty, got %d entries", reason, len(ents)) } } func TestAtomicCacheConcurrentReadersNeverSeePartialLayers(t *testing.T) { + reason := "Readers must never see a partial layer, and the finished layer must be a hit." payload := make([]byte, 256<<10) _, _ = rand.Read(payload) src := slowLayer{gzipLayer(t, payload)} @@ -268,6 +363,9 @@ func TestAtomicCacheConcurrentReadersNeverSeePartialLayers(t *testing.T) { buf := make([]byte, 4096) for { if _, err := rc.Read(buf); err != nil { + if !errors.Is(err, io.EOF) { + errs <- err + } break } } @@ -299,6 +397,16 @@ func TestAtomicCacheConcurrentReadersNeverSeePartialLayers(t *testing.T) { wg.Wait() close(errs) for err := range errs { - t.Error(err) + t.Errorf("\n%s\nconcurrent reader: %v", reason, err) + } + + got, err := c.Get(d) + if err != nil { + t.Fatalf("\n%s\nGet() after all writers finished: %v", reason, err) + } + gotSize, _ := got.Size() + gotDigest, _ := got.Digest() + if gotSize != wantSize || gotDigest != d { + t.Errorf("\n%s\nGet() after all writers finished: want %v/%d, got %v/%d", reason, d, wantSize, gotDigest, gotSize) } }