diff --git a/cmd/atelet/internal/filecache/filecache.go b/cmd/atelet/internal/filecache/filecache.go new file mode 100644 index 0000000000..8521cb43b5 --- /dev/null +++ b/cmd/atelet/internal/filecache/filecache.go @@ -0,0 +1,265 @@ +// Copyright 2026 Google LLC +// +// 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 filecache is a node-local disk cache of downloaded artifacts, +// keyed by opaque identities (see Key). Its contract: an artifact wanted by +// N concurrent callers is fetched once, publication into the cache is atomic +// and crash-safe, and cached bytes are evicted under byte-budget pressure +// without ever breaking a consumer. +// +// On-disk layout, under a store's root (which the store owns exclusively): +// +// entries// +// data # the cached file (or directory tree) +// meta.json # canonical key + creation time; debugging only +// tmp/ # in-flight fetches; same filesystem as entries/, so +// # publication is one atomic rename +// .rm-* # retired entries awaiting removal +// +// An entry directory's mtime is its last-use clock. Nothing on a correctness +// path reads meta.json: entries are matched to keys by hashing the keys. +// +// A crash can leave debris in tmp/ (a fetch that never finished) or .rm-* +// dirs (an eviction that renamed but never removed); SweepDebris reaps both +// and runs once at startup, before the store serves requests. Everything +// under entries/ is a complete, published entry. +package filecache + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io/fs" + "os" + "path/filepath" + "strings" + "sync" + "time" + + "golang.org/x/sync/singleflight" +) + +const ( + entriesDirName = "entries" + tmpDirName = "tmp" + // rmPrefix marks a retired entry awaiting removal, at the store root (not + // under entries/, so a retired entry is invisible to lookups and GC + // listings the moment it is renamed). + rmPrefix = ".rm-" + + dataName = "data" + metaName = "meta.json" + + // defaultMinAge is the default eviction minimum age (see WithMinAge). + defaultMinAge = 10 * time.Minute + // defaultFetchTimeout is the default per-fetch bound (see + // WithFetchTimeout). Generous enough for multi-GiB artifacts on a busy + // node. + defaultFetchTimeout = 10 * time.Minute +) + +// Store is one on-disk cache. It is safe for concurrent use and assumes it +// is the only writer under its root (one atelet per node). +type Store struct { + root string + + // minAge vetoes eviction of any entry younger than this, covering the + // window between publication and the consumer's use becoming visible to + // GC (a hardlink's Nlink, or a root-set record). + minAge time.Duration + + // fetchTimeout bounds each fetch. Fetches run detached from the contexts + // of the callers waiting on them, so this is the only bound on how long + // one can run. + fetchTimeout time.Duration + + // sf collapses concurrent fetches of the same key into one flight. + // Eviction will retire entries inside the same flight, so a retire can + // never race a fetch of the key it is removing. + sf singleflight.Group + + // hitMu closes the hit-vs-evict window: held shared by the link-out path + // (stat, link, and last-use touch), exclusive by eviction's final + // re-check and retire rename, so an entry can never vanish between a + // hit's stat and its link. Uncontended except during an eviction pass. + hitMu sync.RWMutex +} + +// Option configures a Store. +type Option func(*Store) + +// WithMinAge sets the eviction minimum age. +func WithMinAge(d time.Duration) Option { + return func(s *Store) { s.minAge = d } +} + +// WithFetchTimeout sets the per-fetch bound. +func WithFetchTimeout(d time.Duration) Option { + return func(s *Store) { s.fetchTimeout = d } +} + +// New opens (creating if needed) the store rooted at root. +func New(root string, opts ...Option) (*Store, error) { + s := &Store{ + root: root, + minAge: defaultMinAge, + fetchTimeout: defaultFetchTimeout, + } + for _, opt := range opts { + opt(s) + } + // Entries are world-readable (their files get hard-linked into consumer + // dirs and, later, consumed in place); tmp holds unpublished fetches and + // stays private. + if err := os.MkdirAll(s.entriesDir(), 0o755); err != nil { + return nil, fmt.Errorf("while creating entries dir: %w", err) + } + if err := os.MkdirAll(s.tmpDir(), 0o700); err != nil { + return nil, fmt.Errorf("while creating tmp dir: %w", err) + } + return s, nil +} + +func (s *Store) entriesDir() string { return filepath.Join(s.root, entriesDirName) } +func (s *Store) tmpDir() string { return filepath.Join(s.root, tmpDirName) } + +// entryDir is the published location of k's entry. +func (s *Store) entryDir(k Key) string { return filepath.Join(s.entriesDir(), k.dir) } + +// dataPath is the published location of k's cached file (or tree). +func (s *Store) dataPath(k Key) string { return filepath.Join(s.entryDir(k), dataName) } + +// entryMeta is the debugging sidecar written next to an entry's data. It is +// never read on a correctness path; a missing or corrupt one affects +// nothing. +type entryMeta struct { + // Key is the canonical key string, so an operator staring at du output + // can tell what an entry holds. + Key string `json:"key"` + CreatedAt time.Time `json:"createdAt"` +} + +// writeEntryMeta writes the meta.json sidecar into an (unpublished) entry +// dir. +func writeEntryMeta(entryDir string, k Key, createdAt time.Time) error { + data, err := json.Marshal(entryMeta{Key: k.String(), CreatedAt: createdAt}) + if err != nil { + return fmt.Errorf("while marshaling entry meta: %w", err) + } + if err := os.WriteFile(filepath.Join(entryDir, metaName), data, 0o644); err != nil { + return fmt.Errorf("while writing entry meta: %w", err) + } + return nil +} + +// readEntryMeta loads an entry dir's meta.json sidecar. +func readEntryMeta(entryDir string) (entryMeta, error) { + data, err := os.ReadFile(filepath.Join(entryDir, metaName)) + if err != nil { + return entryMeta{}, fmt.Errorf("while reading entry meta: %w", err) + } + var m entryMeta + if err := json.Unmarshal(data, &m); err != nil { + return entryMeta{}, fmt.Errorf("while parsing entry meta: %w", err) + } + return m, nil +} + +// SweepStats reports what a SweepDebris pass removed. +type SweepStats struct { + // TmpRemoved counts removed tmp/ children (unfinished fetches). + TmpRemoved int + // RetiredRemoved counts removed .rm-* dirs (interrupted evictions). + RetiredRemoved int +} + +// SweepDebris removes crash debris: everything under tmp/ and every .rm-* +// dir at the store root. It runs once at startup, before the store serves +// requests; published entries are never touched. Removal failures are +// joined and reported after the sweep visits everything, so one bad path +// does not shadow the rest. +func (s *Store) SweepDebris(ctx context.Context) (SweepStats, error) { + var stats SweepStats + var errs []error + + tmpChildren, err := os.ReadDir(s.tmpDir()) + if err != nil { + errs = append(errs, fmt.Errorf("while listing tmp dir: %w", err)) + } + for _, child := range tmpChildren { + if err := ctx.Err(); err != nil { + return stats, err + } + if err := os.RemoveAll(filepath.Join(s.tmpDir(), child.Name())); err != nil { + errs = append(errs, fmt.Errorf("while removing tmp debris %q: %w", child.Name(), err)) + continue + } + stats.TmpRemoved++ + } + + rootChildren, err := os.ReadDir(s.root) + if err != nil { + errs = append(errs, fmt.Errorf("while listing store root: %w", err)) + } + for _, child := range rootChildren { + if err := ctx.Err(); err != nil { + return stats, err + } + if !strings.HasPrefix(child.Name(), rmPrefix) { + continue + } + if err := os.RemoveAll(filepath.Join(s.root, child.Name())); err != nil { + errs = append(errs, fmt.Errorf("while removing retired entry %q: %w", child.Name(), err)) + continue + } + stats.RetiredRemoved++ + } + + return stats, errors.Join(errs...) +} + +// TotalBytes sums the sizes of all published entries' regular files. It is +// the GC driver's usage measure against the store's byte budget. +func (s *Store) TotalBytes(ctx context.Context) (int64, error) { + var total int64 + err := filepath.WalkDir(s.entriesDir(), func(path string, d fs.DirEntry, err error) error { + if err != nil { + // An entry retired mid-walk is not an error; skip what vanished. + if errors.Is(err, fs.ErrNotExist) { + return nil + } + return err + } + if err := ctx.Err(); err != nil { + return err + } + if !d.Type().IsRegular() { + return nil + } + info, err := d.Info() + if err != nil { + if errors.Is(err, fs.ErrNotExist) { + return nil + } + return err + } + total += info.Size() + return nil + }) + if err != nil { + return 0, fmt.Errorf("while sizing entries: %w", err) + } + return total, nil +} diff --git a/cmd/atelet/internal/filecache/filecache_test.go b/cmd/atelet/internal/filecache/filecache_test.go new file mode 100644 index 0000000000..7576b184ea --- /dev/null +++ b/cmd/atelet/internal/filecache/filecache_test.go @@ -0,0 +1,213 @@ +// Copyright 2026 Google LLC +// +// 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 filecache + +import ( + "context" + "os" + "path/filepath" + "testing" + "time" +) + +func newTestStore(t *testing.T, opts ...Option) *Store { + t.Helper() + s, err := New(filepath.Join(t.TempDir(), "cache"), opts...) + if err != nil { + t.Fatalf("New: %v", err) + } + return s +} + +// publishTestEntry plants a published entry with the given data files +// (relative name -> content), bypassing retrieval, and returns its key. +func publishTestEntry(t *testing.T, s *Store, name string, files map[string]string) Key { + t.Helper() + k := URIKey("test://" + name) + if len(files) == 0 { + if err := os.MkdirAll(s.entryDir(k), 0o755); err != nil { + t.Fatal(err) + } + } + for rel, content := range files { + path := filepath.Join(s.entryDir(k), rel) + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(path, []byte(content), 0o444); err != nil { + t.Fatal(err) + } + } + if err := writeEntryMeta(s.entryDir(k), k, time.Now()); err != nil { + t.Fatalf("writeEntryMeta: %v", err) + } + return k +} + +func TestNewCreatesLayout(t *testing.T) { + root := filepath.Join(t.TempDir(), "cache") + s, err := New(root) + if err != nil { + t.Fatalf("New: %v", err) + } + for _, dir := range []string{s.entriesDir(), s.tmpDir()} { + fi, err := os.Stat(dir) + if err != nil || !fi.IsDir() { + t.Errorf("stat %q: err=%v, isDir=%v", dir, err, err == nil && fi.IsDir()) + } + } + + // Reopening an existing root keeps published entries. + k := publishTestEntry(t, s, "survivor", map[string]string{dataName: "x"}) + if _, err := New(root); err != nil { + t.Fatalf("New (reopen): %v", err) + } + if _, err := os.Stat(s.dataPath(k)); err != nil { + t.Errorf("entry lost across reopen: %v", err) + } +} + +func TestNewDefaultsAndOptions(t *testing.T) { + s := newTestStore(t) + if s.minAge != defaultMinAge { + t.Errorf("minAge = %v, want default %v", s.minAge, defaultMinAge) + } + if s.fetchTimeout != defaultFetchTimeout { + t.Errorf("fetchTimeout = %v, want default %v", s.fetchTimeout, defaultFetchTimeout) + } + + s = newTestStore(t, WithMinAge(time.Second), WithFetchTimeout(time.Minute)) + if s.minAge != time.Second { + t.Errorf("minAge = %v, want %v", s.minAge, time.Second) + } + if s.fetchTimeout != time.Minute { + t.Errorf("fetchTimeout = %v, want %v", s.fetchTimeout, time.Minute) + } +} + +func TestEntryMetaRoundTrip(t *testing.T) { + s := newTestStore(t) + k := publishTestEntry(t, s, "meta", map[string]string{dataName: "x"}) + + m, err := readEntryMeta(s.entryDir(k)) + if err != nil { + t.Fatalf("readEntryMeta: %v", err) + } + if m.Key != k.String() { + t.Errorf("meta key = %q, want %q", m.Key, k.String()) + } + if m.CreatedAt.IsZero() { + t.Error("meta createdAt is zero") + } +} + +func TestSweepDebris(t *testing.T) { + s := newTestStore(t) + kept := publishTestEntry(t, s, "kept", map[string]string{dataName: "x"}) + + // Unfinished fetches: a bare temp file and a temp extraction dir. + if err := os.WriteFile(filepath.Join(s.tmpDir(), "dl-1"), []byte("partial"), 0o600); err != nil { + t.Fatal(err) + } + if err := os.MkdirAll(filepath.Join(s.tmpDir(), "dl-2", "nested"), 0o700); err != nil { + t.Fatal(err) + } + // An eviction that renamed but never removed. + if err := os.MkdirAll(filepath.Join(s.root, rmPrefix+"deadbeef", "nested"), 0o755); err != nil { + t.Fatal(err) + } + + stats, err := s.SweepDebris(context.Background()) + if err != nil { + t.Fatalf("SweepDebris: %v", err) + } + if stats.TmpRemoved != 2 || stats.RetiredRemoved != 1 { + t.Errorf("stats = %+v, want TmpRemoved=2 RetiredRemoved=1", stats) + } + + tmpChildren, err := os.ReadDir(s.tmpDir()) + if err != nil { + t.Fatal(err) + } + if len(tmpChildren) != 0 { + t.Errorf("tmp dir not empty after sweep: %d children", len(tmpChildren)) + } + if _, err := os.Stat(filepath.Join(s.root, rmPrefix+"deadbeef")); !os.IsNotExist(err) { + t.Errorf("retired dir survived sweep: err=%v", err) + } + if _, err := os.Stat(s.dataPath(kept)); err != nil { + t.Errorf("published entry removed by sweep: %v", err) + } + + // A clean store sweeps to zero. + stats, err = s.SweepDebris(context.Background()) + if err != nil { + t.Fatalf("SweepDebris (clean): %v", err) + } + if stats != (SweepStats{}) { + t.Errorf("stats on clean store = %+v, want zero", stats) + } +} + +func TestSweepDebrisCanceled(t *testing.T) { + s := newTestStore(t) + if err := os.WriteFile(filepath.Join(s.tmpDir(), "dl-1"), nil, 0o600); err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if _, err := s.SweepDebris(ctx); err == nil { + t.Error("SweepDebris with canceled ctx succeeded, want error") + } +} + +func TestTotalBytes(t *testing.T) { + s := newTestStore(t) + + total, err := s.TotalBytes(context.Background()) + if err != nil { + t.Fatalf("TotalBytes (empty): %v", err) + } + if total != 0 { + t.Errorf("TotalBytes on empty store = %d, want 0", total) + } + + publishTestEntry(t, s, "file", map[string]string{dataName: "12345"}) + publishTestEntry(t, s, "tree", map[string]string{ + filepath.Join(dataName, "runsc"): "1234567890", + filepath.Join(dataName, "gvisor-bin", "help"): "123", + }) + // Unpublished bytes in tmp/ must not count. + if err := os.WriteFile(filepath.Join(s.tmpDir(), "dl-1"), []byte("zzzz"), 0o600); err != nil { + t.Fatal(err) + } + + total, err = s.TotalBytes(context.Background()) + if err != nil { + t.Fatalf("TotalBytes: %v", err) + } + // 5 + 10 + 3 data bytes plus the two entries' meta.json sidecars. + var metaBytes int64 + for _, name := range []string{"file", "tree"} { + fi, err := os.Stat(filepath.Join(s.entryDir(URIKey("test://"+name)), metaName)) + if err != nil { + t.Fatal(err) + } + metaBytes += fi.Size() + } + if want := int64(18) + metaBytes; total != want { + t.Errorf("TotalBytes = %d, want %d", total, want) + } +} diff --git a/cmd/atelet/internal/filecache/getfileto.go b/cmd/atelet/internal/filecache/getfileto.go new file mode 100644 index 0000000000..d14f1f3865 --- /dev/null +++ b/cmd/atelet/internal/filecache/getfileto.go @@ -0,0 +1,191 @@ +// Copyright 2026 Google LLC +// +// 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 filecache + +import ( + "context" + "errors" + "fmt" + "io/fs" + "os" + "path/filepath" + "syscall" + "time" +) + +// FileFetcher produces a single-file artifact at dstPath (creating the file +// itself, so it can seek and truncate for sparse output). It runs at most +// once per key across concurrent GetFileTo callers, detached from their +// contexts and bounded by the store's fetch timeout. Its error is delivered +// to every caller waiting on the flight, wrapped with %w so error +// classification (errors.Is) sees through the store. +type FileFetcher func(ctx context.Context, dstPath string) error + +// linkRetries bounds the publish-then-link loop in GetFileTo. An entry can +// be evicted between a flight completing and this caller's link only if it +// sat unlinked past the store's min age, so a single retry is already an +// anomaly; more than a few means something is deleting entries out from +// under the store. +const linkRetries = 3 + +// GetFileTo materializes the artifact identified by key at dst, fetching it +// with fetch if it is not cached. dst must be an absolute path that does not +// exist yet and lives on the cache's filesystem (the same mount, not just +// the same disk): on success it is a hard link to the read-only cache +// copy, so the caller keeps a valid file regardless of later eviction, and +// the entry is published mode 0444 so an in-place write fails loudly rather +// than corrupting the shared bytes. +// +// Concurrent calls for one key share a single fetch. A caller whose ctx is +// canceled returns early with ctx.Err() while the fetch keeps running for +// the others; there is no negative caching, so after a failed fetch the +// next call starts fresh. +func (s *Store) GetFileTo(ctx context.Context, key Key, dst string, fetch FileFetcher) error { + if key.isZero() { + return errors.New("filecache: zero Key (use a Key constructor)") + } + if !filepath.IsAbs(dst) { + return fmt.Errorf("filecache: destination %q is not an absolute path", dst) + } + if fetch == nil { + return errors.New("filecache: nil FileFetcher") + } + for attempt := 0; ; attempt++ { + linked, err := s.linkOut(key, dst) + if err != nil { + return err + } + if linked { + return nil + } + if attempt >= linkRetries { + return fmt.Errorf("entry for %v vanished after %d fetches; is something else deleting under the store root?", key, attempt) + } + if err := s.fetchFlight(ctx, key, fetch); err != nil { + return err + } + } +} + +// linkOut links key's published data file to dst and touches the entry's +// last-use clock, reporting false (and no error) on a cache miss. The +// shared hitMu holds eviction's final re-check out of the stat-to-link +// window, so a published entry cannot be retired mid-hit. +func (s *Store) linkOut(key Key, dst string) (bool, error) { + s.hitMu.RLock() + defer s.hitMu.RUnlock() + + if _, err := os.Stat(s.dataPath(key)); err != nil { + if errors.Is(err, fs.ErrNotExist) { + return false, nil + } + return false, fmt.Errorf("while checking cache for %v: %w", key, err) + } + if err := os.Link(s.dataPath(key), dst); err != nil { + if errors.Is(err, fs.ErrExist) { + return false, fmt.Errorf("destination %s already exists: %w", dst, err) + } + if errors.Is(err, syscall.EXDEV) { + return false, fmt.Errorf("destination %s is not on the cache's filesystem (link-out requires one filesystem): %w", dst, err) + } + return false, fmt.Errorf("while linking %v to %s: %w", key, dst, err) + } + // Last-use touch for eviction's LRU ordering; best-effort, a miss only + // ages the entry. + now := time.Now() + _ = os.Chtimes(s.entryDir(key), now, now) + return true, nil +} + +// fetchFlight runs (or joins) the singleflight fetch for key and waits for +// it or for ctx. The flight itself runs on a context detached from the +// callers' — bounded only by the store's fetch timeout — so one canceled +// caller never aborts a download other callers are waiting on. +func (s *Store) fetchFlight(ctx context.Context, key Key, fetch FileFetcher) error { + ch := s.sf.DoChan(key.dir, func() (any, error) { + return nil, s.fetchAndPublish(context.WithoutCancel(ctx), key, fetch) + }) + select { + case res := <-ch: + if res.Err != nil { + return fmt.Errorf("while fetching %v: %w", key, res.Err) + } + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + +// fetchAndPublish runs one fetch into tmp/ and atomically publishes the +// result under entries/. On any failure the temp dir is removed, so a bad +// fetch is never visible in the cache; a crash instead leaves it for +// SweepDebris. +func (s *Store) fetchAndPublish(ctx context.Context, key Key, fetch FileFetcher) error { + ctx, cancel := context.WithTimeout(ctx, s.fetchTimeout) + defer cancel() + + // A flight that completed between this caller's miss and this flight + // starting may have published already. + if _, err := os.Stat(s.dataPath(key)); err == nil { + return nil + } + + tmpDir, err := os.MkdirTemp(s.tmpDir(), key.dir+"-") + if err != nil { + return fmt.Errorf("while creating fetch temp dir: %w", err) + } + published := false + defer func() { + if !published { + _ = os.RemoveAll(tmpDir) + } + }() + + dataPath := filepath.Join(tmpDir, dataName) + if err := fetch(ctx, dataPath); err != nil { + return err + } + fi, err := os.Stat(dataPath) + if err != nil { + return fmt.Errorf("fetcher succeeded but produced no file: %w", err) + } + if !fi.Mode().IsRegular() { + return fmt.Errorf("fetcher produced %v, want a regular file", fi.Mode()) + } + + // Read-only before publication: cached bytes are shared through hard + // links, so an in-place write by any consumer must fail rather than + // poison the copy every later caller links. + if err := os.Chmod(dataPath, 0o444); err != nil { + return fmt.Errorf("while making fetched file read-only: %w", err) + } + if err := writeEntryMeta(tmpDir, key, time.Now()); err != nil { + return err + } + if err := os.Chmod(tmpDir, 0o755); err != nil { // MkdirTemp created it 0700 + return fmt.Errorf("while setting entry dir mode: %w", err) + } + if err := os.Rename(tmpDir, s.entryDir(key)); err != nil { + // Within one process the singleflight makes a rename race + // unreachable; tolerating a loss anyway keeps overlap with a + // crashed predecessor's published entry safe. + if errors.Is(err, syscall.EEXIST) || errors.Is(err, syscall.ENOTEMPTY) { + return nil + } + return fmt.Errorf("while publishing entry: %w", err) + } + published = true + return nil +} diff --git a/cmd/atelet/internal/filecache/getfileto_test.go b/cmd/atelet/internal/filecache/getfileto_test.go new file mode 100644 index 0000000000..574a5d6fc3 --- /dev/null +++ b/cmd/atelet/internal/filecache/getfileto_test.go @@ -0,0 +1,328 @@ +// Copyright 2026 Google LLC +// +// 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 filecache + +import ( + "context" + "errors" + "fmt" + "os" + "path/filepath" + "sync" + "sync/atomic" + "testing" + "time" +) + +// countingFetcher returns a FileFetcher writing content, and a counter of +// how many times it ran. +func countingFetcher(content string) (FileFetcher, *atomic.Int32) { + var calls atomic.Int32 + return func(ctx context.Context, dstPath string) error { + calls.Add(1) + return os.WriteFile(dstPath, []byte(content), 0o600) + }, &calls +} + +// dstPath returns a fresh destination path (which must not exist yet) in a +// per-test consumer dir on the same filesystem as the store. +func dstPath(t *testing.T, s *Store, name string) string { + t.Helper() + dir := filepath.Join(s.root, "..", "consumer") + if err := os.MkdirAll(dir, 0o755); err != nil { + t.Fatal(err) + } + return filepath.Join(dir, name) +} + +func TestGetFileToMissThenHit(t *testing.T) { + s := newTestStore(t) + key := URIKey("gs://bucket/golden", "mem.img") + fetch, calls := countingFetcher("golden bytes") + + dst1 := dstPath(t, s, "d1") + if err := s.GetFileTo(context.Background(), key, dst1, fetch); err != nil { + t.Fatalf("GetFileTo (miss): %v", err) + } + got, err := os.ReadFile(dst1) + if err != nil || string(got) != "golden bytes" { + t.Fatalf("dst content = %q, %v; want %q", got, err, "golden bytes") + } + fi, err := os.Stat(dst1) + if err != nil { + t.Fatal(err) + } + if fi.Mode().Perm() != 0o444 { + t.Errorf("dst mode = %v, want 0444", fi.Mode().Perm()) + } + + dst2 := dstPath(t, s, "d2") + if err := s.GetFileTo(context.Background(), key, dst2, fetch); err != nil { + t.Fatalf("GetFileTo (hit): %v", err) + } + if calls.Load() != 1 { + t.Errorf("fetch ran %d times, want 1", calls.Load()) + } + + // Both destinations and the cache copy share one inode. + fi2, err := os.Stat(dst2) + if err != nil { + t.Fatal(err) + } + cfi, err := os.Stat(s.dataPath(key)) + if err != nil { + t.Fatal(err) + } + if !os.SameFile(fi, fi2) || !os.SameFile(fi, cfi) { + t.Error("dst1, dst2, and cache copy are not one inode") + } + + // Nothing left in flight. + tmpChildren, err := os.ReadDir(s.tmpDir()) + if err != nil { + t.Fatal(err) + } + if len(tmpChildren) != 0 { + t.Errorf("tmp dir has %d leftover children", len(tmpChildren)) + } +} + +func TestGetFileToConcurrentCallersShareOneFetch(t *testing.T) { + s := newTestStore(t) + key := URIKey("gs://bucket/golden", "mem.img") + + var calls atomic.Int32 + release := make(chan struct{}) + fetch := func(ctx context.Context, dstPath string) error { + calls.Add(1) + <-release // hold every caller in one flight + return os.WriteFile(dstPath, []byte("x"), 0o600) + } + + const n = 16 + errs := make([]error, n) + var started, done sync.WaitGroup + for i := range n { + started.Add(1) + done.Go(func() { + started.Done() + errs[i] = s.GetFileTo(context.Background(), key, dstPath(t, s, fmt.Sprintf("d%d", i)), fetch) + }) + } + started.Wait() + close(release) + done.Wait() + + for i, err := range errs { + if err != nil { + t.Errorf("caller %d: %v", i, err) + } + } + if calls.Load() != 1 { + t.Errorf("fetch ran %d times, want 1", calls.Load()) + } +} + +func TestGetFileToCanceledWaiterDoesNotAbortFlight(t *testing.T) { + s := newTestStore(t) + key := URIKey("gs://bucket/golden", "mem.img") + + var calls atomic.Int32 + entered := make(chan struct{}) + release := make(chan struct{}) + fetch := func(ctx context.Context, dstPath string) error { + calls.Add(1) + close(entered) + select { + case <-release: + case <-ctx.Done(): // must NOT fire on the caller's cancel + return ctx.Err() + } + return os.WriteFile(dstPath, []byte("x"), 0o600) + } + + ctx, cancel := context.WithCancel(context.Background()) + callerErr := make(chan error, 1) + go func() { + callerErr <- s.GetFileTo(ctx, key, dstPath(t, s, "canceled"), fetch) + }() + <-entered + cancel() + if err := <-callerErr; !errors.Is(err, context.Canceled) { + t.Fatalf("canceled caller returned %v, want context.Canceled", err) + } + + // The flight is still running detached; releasing it publishes the + // entry, and a later caller hits the cache with no second fetch. + close(release) + if err := s.GetFileTo(context.Background(), key, dstPath(t, s, "later"), fetch); err != nil { + t.Fatalf("GetFileTo after canceled waiter: %v", err) + } + if calls.Load() != 1 { + t.Errorf("fetch ran %d times, want 1", calls.Load()) + } +} + +func TestGetFileToFetchErrorReachesAllWaitersThenRetries(t *testing.T) { + s := newTestStore(t) + key := URIKey("gs://bucket/golden", "mem.img") + fetchErr := errors.New("bucket unreachable") + + var calls atomic.Int32 + release := make(chan struct{}) + failing := func(ctx context.Context, dstPath string) error { + calls.Add(1) + <-release + return fetchErr + } + + const n = 4 + errs := make([]error, n) + var started, done sync.WaitGroup + for i := range n { + started.Add(1) + done.Go(func() { + started.Done() + errs[i] = s.GetFileTo(context.Background(), key, dstPath(t, s, fmt.Sprintf("f%d", i)), failing) + }) + } + started.Wait() + close(release) + done.Wait() + + for i, err := range errs { + if !errors.Is(err, fetchErr) { + t.Errorf("caller %d: %v, want the fetch error", i, err) + } + } + if calls.Load() != 1 { + t.Fatalf("failing fetch ran %d times, want 1", calls.Load()) + } + + // No debris and no published entry after the failure. + tmpChildren, err := os.ReadDir(s.tmpDir()) + if err != nil { + t.Fatal(err) + } + if len(tmpChildren) != 0 { + t.Errorf("tmp dir has %d children after failed fetch", len(tmpChildren)) + } + if _, err := os.Stat(s.entryDir(key)); !os.IsNotExist(err) { + t.Errorf("entry published despite failed fetch: err=%v", err) + } + + // No negative caching: the next call fetches fresh and succeeds. + ok, okCalls := countingFetcher("recovered") + if err := s.GetFileTo(context.Background(), key, dstPath(t, s, "retry"), ok); err != nil { + t.Fatalf("GetFileTo (retry): %v", err) + } + if okCalls.Load() != 1 { + t.Errorf("retry fetch ran %d times, want 1", okCalls.Load()) + } +} + +func TestGetFileToRejectsExistingDst(t *testing.T) { + s := newTestStore(t) + key := URIKey("gs://bucket/golden", "mem.img") + fetch, _ := countingFetcher("x") + + dst := dstPath(t, s, "occupied") + if err := os.WriteFile(dst, []byte("previous"), 0o600); err != nil { + t.Fatal(err) + } + if err := s.GetFileTo(context.Background(), key, dst, fetch); err == nil { + t.Fatal("GetFileTo to existing dst succeeded, want error") + } + got, err := os.ReadFile(dst) + if err != nil || string(got) != "previous" { + t.Errorf("existing dst clobbered: %q, %v", got, err) + } +} + +func TestGetFileToRejectsBadArguments(t *testing.T) { + s := newTestStore(t) + key := URIKey("gs://bucket/golden", "mem.img") + fetch, calls := countingFetcher("x") + + if err := s.GetFileTo(context.Background(), Key{}, dstPath(t, s, "zero"), fetch); err == nil { + t.Error("GetFileTo with zero key succeeded, want error") + } + if err := s.GetFileTo(context.Background(), key, "", fetch); err == nil { + t.Error("GetFileTo with empty dst succeeded, want error") + } + if err := s.GetFileTo(context.Background(), key, "relative/dst", fetch); err == nil { + t.Error("GetFileTo with relative dst succeeded, want error") + } + if err := s.GetFileTo(context.Background(), key, dstPath(t, s, "nilfetch"), nil); err == nil { + t.Error("GetFileTo with nil fetcher succeeded, want error") + } + if calls.Load() != 0 { + t.Errorf("fetch ran %d times for rejected arguments, want 0", calls.Load()) + } +} + +func TestGetFileToFetcherProducingNoFileFails(t *testing.T) { + s := newTestStore(t) + key := URIKey("gs://bucket/golden", "mem.img") + noop := func(ctx context.Context, dstPath string) error { return nil } + + if err := s.GetFileTo(context.Background(), key, dstPath(t, s, "empty"), noop); err == nil { + t.Fatal("GetFileTo with file-less fetcher succeeded, want error") + } + if _, err := os.Stat(s.entryDir(key)); !os.IsNotExist(err) { + t.Errorf("entry published despite file-less fetcher: err=%v", err) + } +} + +func TestGetFileToHitTouchesLastUse(t *testing.T) { + s := newTestStore(t) + key := URIKey("gs://bucket/golden", "mem.img") + fetch, _ := countingFetcher("x") + + if err := s.GetFileTo(context.Background(), key, dstPath(t, s, "first"), fetch); err != nil { + t.Fatal(err) + } + stale := time.Now().Add(-time.Hour) + if err := os.Chtimes(s.entryDir(key), stale, stale); err != nil { + t.Fatal(err) + } + + if err := s.GetFileTo(context.Background(), key, dstPath(t, s, "second"), fetch); err != nil { + t.Fatal(err) + } + fi, err := os.Stat(s.entryDir(key)) + if err != nil { + t.Fatal(err) + } + if !fi.ModTime().After(stale.Add(time.Minute)) { + t.Errorf("entry mtime %v not refreshed by hit (stale mark %v)", fi.ModTime(), stale) + } +} + +func TestGetFileToPublishedCopyIsWriteProtected(t *testing.T) { + s := newTestStore(t) + key := URIKey("gs://bucket/golden", "mem.img") + fetch, _ := countingFetcher("precious") + + dst := dstPath(t, s, "d") + if err := s.GetFileTo(context.Background(), key, dst, fetch); err != nil { + t.Fatal(err) + } + // The mutation tripwire: writing through the consumer's link must fail, + // not silently poison the shared copy. + if err := os.WriteFile(dst, []byte("mutated"), 0o444); err == nil { + t.Error("write through consumer link succeeded, want permission error") + } +} diff --git a/cmd/atelet/internal/filecache/key.go b/cmd/atelet/internal/filecache/key.go new file mode 100644 index 0000000000..72102f8e71 --- /dev/null +++ b/cmd/atelet/internal/filecache/key.go @@ -0,0 +1,77 @@ +// Copyright 2026 Google LLC +// +// 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 filecache + +import ( + "crypto/sha256" + "encoding/hex" + "fmt" + "strings" +) + +// Key identifies a cache entry. It is an identity, not an address: it says +// nothing about where the bytes come from (the fetcher does), only what they +// are, so two sources serving the same identity share one entry. +// +// Keys are built via constructors only. The zero Key is invalid. +type Key struct { + // canonical is the unambiguous string form, recorded in the entry's + // meta.json for debugging. A constructor-specific prefix keeps the key + // spaces disjoint (a sha256 digest can never equal a URI key). + canonical string + // dir is hex(sha256(canonical)): the entry's directory name under + // entries/. Hashing (rather than escaping the canonical form) yields a + // fixed-length, filesystem-safe name for arbitrarily long keys, and lets + // a GC root set be matched against entry dirs by hashing its keys. + dir string +} + +// keyPartSeparator joins URIKey parts unambiguously: NUL cannot appear in a +// URI or file name, so ("a/b","c") and ("a","b/c") canonicalize differently. +const keyPartSeparator = "\x00" + +// SHA256Key returns the key for a content-addressed artifact, identified by +// the lowercase hex sha256 of its bytes. Uppercase digits are normalized so +// equal digests always yield equal keys. +func SHA256Key(hexDigest string) (Key, error) { + d := strings.ToLower(hexDigest) + if len(d) != sha256.Size*2 { + return Key{}, fmt.Errorf("sha256 key: digest %q has length %d, want %d", hexDigest, len(d), sha256.Size*2) + } + if _, err := hex.DecodeString(d); err != nil { + return Key{}, fmt.Errorf("sha256 key: digest %q is not hex: %w", hexDigest, err) + } + return newKey("sha256:" + d), nil +} + +// URIKey returns the key for an artifact identified by an immutable source, +// e.g. URIKey(goldenSnapshotURI, fileName). The parts must identify content +// that never changes underneath them; the cache has no invalidation, so a +// republished URI would serve stale bytes forever. Callers pass at least one +// non-empty part. +func URIKey(parts ...string) Key { + return newKey("uri:" + strings.Join(parts, keyPartSeparator)) +} + +func newKey(canonical string) Key { + sum := sha256.Sum256([]byte(canonical)) + return Key{canonical: canonical, dir: hex.EncodeToString(sum[:])} +} + +// String returns the canonical form, for logs and meta.json. +func (k Key) String() string { return k.canonical } + +// isZero reports whether k was not built by a constructor. +func (k Key) isZero() bool { return k.dir == "" } diff --git a/cmd/atelet/internal/filecache/key_test.go b/cmd/atelet/internal/filecache/key_test.go new file mode 100644 index 0000000000..ad27d9647c --- /dev/null +++ b/cmd/atelet/internal/filecache/key_test.go @@ -0,0 +1,97 @@ +// Copyright 2026 Google LLC +// +// 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 filecache + +import ( + "regexp" + "strings" + "testing" +) + +const testDigest = "af2a7458c2c05df1a01d0b2f335f4849a2de84e83160fefdc31a6266015642d4" + +var entryDirNameRE = regexp.MustCompile(`^[0-9a-f]{64}$`) + +func TestSHA256Key(t *testing.T) { + k, err := SHA256Key(testDigest) + if err != nil { + t.Fatalf("SHA256Key(%q): %v", testDigest, err) + } + if want := "sha256:" + testDigest; k.String() != want { + t.Errorf("String() = %q, want %q", k.String(), want) + } + if !entryDirNameRE.MatchString(k.dir) { + t.Errorf("dir = %q, want 64 lowercase hex chars", k.dir) + } + + upper, err := SHA256Key(strings.ToUpper(testDigest)) + if err != nil { + t.Fatalf("SHA256Key(upper): %v", err) + } + if upper != k { + t.Errorf("uppercase digest yielded a different key: %q vs %q", upper.dir, k.dir) + } +} + +func TestSHA256KeyRejectsBadDigests(t *testing.T) { + for _, digest := range []string{ + "", + "abc123", // too short + testDigest + "00", // too long + testDigest[:63] + "g", // not hex + } { + if _, err := SHA256Key(digest); err == nil { + t.Errorf("SHA256Key(%q) succeeded, want error", digest) + } + } +} + +func TestURIKey(t *testing.T) { + base := URIKey("gs://bucket/golden-v3", "mem.img") + if base != URIKey("gs://bucket/golden-v3", "mem.img") { + t.Error("equal parts yielded different keys") + } + if !entryDirNameRE.MatchString(base.dir) { + t.Errorf("dir = %q, want 64 lowercase hex chars", base.dir) + } + + // Part boundaries must be unambiguous: shifting a separator across a + // part boundary, or adding an empty part, is a different identity. + distinct := []Key{ + base, + URIKey("gs://bucket/golden-v3/mem.img"), + URIKey("gs://bucket/golden-v3", "mem.img", ""), + URIKey("gs://bucket", "golden-v3/mem.img"), + } + for i, a := range distinct { + for j, b := range distinct { + if i != j && a == b { + t.Errorf("keys %d and %d collide: %q", i, j, a.dir) + } + } + } +} + +func TestKeySpacesAreDisjoint(t *testing.T) { + // A URI that happens to spell a digest must not collide with the + // content-addressed key for that digest. + sha, err := SHA256Key(testDigest) + if err != nil { + t.Fatal(err) + } + if uri := URIKey(testDigest); uri == sha { + t.Errorf("URIKey and SHA256Key collide for %q", testDigest) + } +}