diff --git a/cmd/atelet/internal/filecache/filecache.go b/cmd/atelet/internal/filecache/filecache.go new file mode 100644 index 0000000000..79b0b66059 --- /dev/null +++ b/cmd/atelet/internal/filecache/filecache.go @@ -0,0 +1,251 @@ +// 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" + "time" +) + +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 +} + +// 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/key.go b/cmd/atelet/internal/filecache/key.go new file mode 100644 index 0000000000..75542154b3 --- /dev/null +++ b/cmd/atelet/internal/filecache/key.go @@ -0,0 +1,74 @@ +// 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 } 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) + } +}