1// SPDX-License-Identifier: MIT
2
3// Package actionscache is the host-side cache of action archives (plan CI-4,
4// D8). The host downloads the tarball of an action once per (repository,
5// resolved commit) and every job, in any sandbox, is served the same bytes
6// through the cache channel (see cachesvc): the forge is hit once, not once per
7// job (codeload answered 429 during the CI-3 compare).
8//
9// An entry is keyed by Key: a hash of the forge, the repository and the resolved
10// commit sha. A mutable ref (a tag, a branch) is never part of it, so a moved tag
11// cannot serve stale bytes: the server resolves the ref to a sha at each job.
12//
13// The bytes come from the host's own download, never from a sandbox: the guest
14// runs untrusted code and must not be able to plant an archive that other jobs
15// would run. Each entry is two files in Dir, <key>.tgz and <key>.sum (the
16// sha256 and the size of the archive). The sum is written after the archive, so
17// an archive without one is a crash leftover and is removed at start. Every read
18// re-hashes the archive; a mismatch drops the entry and the caller downloads it
19// again.
20//
21// The total size is capped; the least recently used entries go first. An entry
22// handed out is pinned for a while (PinFor), so that it cannot be evicted
23// between the host fetching it and the sandbox reading it.
24package actionscache
25
26import (
27 "context"
28 "crypto/sha256"
29 "encoding/hex"
30 "errors"
31 "fmt"
32 "io"
33 "os"
34 "path/filepath"
35 "sort"
36 "strconv"
37 "strings"
38 "sync"
39 "time"
40
41 "github.com/bnema/zerowrap"
42)
43
44// Defaults and names.
45const (
46 // DefaultMax is the size cap when ERMITE_RUNNER_ACTIONS_CACHE_MAX is unset.
47 DefaultMax = 2 << 30
48 // EnvMax is ERMITE_RUNNER_ACTIONS_CACHE_MAX: a size (2G, 500M…); 0 turns the
49 // cache off.
50 EnvMax = "ERMITE_RUNNER_ACTIONS_CACHE_MAX"
51 // PinFor is how long a served entry is protected from eviction.
52 PinFor = 30 * time.Minute
53
54 archiveExt = ".tgz"
55 sumExt = ".sum"
56 tmpPrefix = "tmp-"
57)
58
59// ErrNotFound is an entry that is absent, corrupt or evicted.
60var ErrNotFound = errors.New("actionscache: no such entry")
61
62// ErrTooLarge is an archive that does not fit under the cap on its own.
63var ErrTooLarge = errors.New("actionscache: archive larger than the cache")
64
65// Config configures a Cache.
66type Config struct {
67 Dir string
68 // Max is the size cap in bytes, > 0.
69 Max int64
70 Now func() time.Time
71 Log zerowrap.Logger
72}
73
74type entry struct {
75 size int64
76 used time.Time
77 pinUntil time.Time
78}
79
80type flight struct {
81 done chan struct{}
82 err error
83}
84
85// Cache is the archive cache.
86type Cache struct {
87 dir string
88 max int64
89 now func() time.Time
90 log zerowrap.Logger
91
92 mu sync.Mutex
93 entries map[string]*entry
94 total int64
95 pending int64 // bytes of downloads in progress, charged as they are written
96 flights map[string]*flight
97}
98
99// New opens (and creates) dir, forgetting what a crash left half written.
100func New(cfg Config) (*Cache, error) {
101 if cfg.Max <= 0 {
102 return nil, errors.New("actionscache: the cap must be positive")
103 }
104 if cfg.Now == nil {
105 cfg.Now = time.Now
106 }
107 if err := os.MkdirAll(cfg.Dir, 0o700); err != nil {
108 return nil, fmt.Errorf("actionscache: %w", err)
109 }
110 c := &Cache{dir: cfg.Dir, max: cfg.Max, now: cfg.Now, log: cfg.Log, entries: map[string]*entry{}, flights: map[string]*flight{}}
111 des, err := os.ReadDir(cfg.Dir)
112 if err != nil {
113 return nil, fmt.Errorf("actionscache: %w", err)
114 }
115 for _, de := range des {
116 n := de.Name()
117 switch {
118 case strings.HasPrefix(n, tmpPrefix):
119 _ = os.Remove(filepath.Join(cfg.Dir, n))
120 case strings.HasSuffix(n, sumExt):
121 key := strings.TrimSuffix(n, sumExt)
122 sfi, err1 := de.Info()
123 afi, err2 := os.Stat(c.archivePath(key))
124 if !validKey(key) || err1 != nil || err2 != nil {
125 c.remove(key)
126 continue
127 }
128 c.entries[key] = &entry{size: afi.Size(), used: sfi.ModTime()}
129 c.total += afi.Size()
130 case strings.HasSuffix(n, archiveExt):
131 if _, err := os.Stat(filepath.Join(cfg.Dir, strings.TrimSuffix(n, archiveExt)+sumExt)); err != nil {
132 _ = os.Remove(filepath.Join(cfg.Dir, n))
133 }
134 }
135 }
136 c.mu.Lock()
137 c.evictLocked("", 0)
138 c.mu.Unlock()
139 return c, nil
140}
141
142// Key is the entry key of an archive: the sha256 of the forge origin, the
143// repository and the resolved commit. ok is false when sha is not a full git
144// object name (40 or 64 hex digits): such an archive is not cached, whatever the
145// ref said.
146func Key(origin, repo, sha string) (key string, ok bool) {
147 if origin == "" || repo == "" || (len(sha) != 40 && len(sha) != 64) {
148 return "", false
149 }
150 if _, err := hex.DecodeString(sha); err != nil {
151 return "", false
152 }
153 h := sha256.New()
154 for _, p := range []string{origin, strings.ToLower(repo), strings.ToLower(sha)} {
155 h.Write([]byte(p))
156 h.Write([]byte{0})
157 }
158 return hex.EncodeToString(h.Sum(nil)), true
159}
160
161func validKey(k string) bool {
162 if len(k) != 64 {
163 return false
164 }
165 _, err := hex.DecodeString(k)
166 return err == nil
167}
168
169// ValidKey reports whether k has the shape of a Key.
170func ValidKey(k string) bool { return validKey(k) }
171
172func (c *Cache) archivePath(key string) string { return filepath.Join(c.dir, key+archiveExt) }
173func (c *Cache) sumPath(key string) string { return filepath.Join(c.dir, key+sumExt) }
174
175func (c *Cache) remove(key string) {
176 _ = os.Remove(c.sumPath(key))
177 _ = os.Remove(c.archivePath(key))
178}
179
180// Fetch writes an archive to w; it is called at most once at a time per key.
181type Fetch func(ctx context.Context, w io.Writer) error
182
183// Ensure makes sure the entry of key exists and is intact, calling fetch to
184// download it when it does not (a miss: hit is false). Concurrent calls for the
185// same key share one download; the others report a hit once it is in place. The
186// entry is pinned for PinFor.
187func (c *Cache) Ensure(ctx context.Context, key string, fetch Fetch) (hit bool, verify time.Duration, err error) {
188 if !validKey(key) {
189 return false, verify, ErrNotFound
190 }
191 for {
192 duration, err := c.verify(key)
193 verify += duration
194 if err == nil {
195 c.touch(key, true)
196 return true, verify, nil
197 }
198 c.mu.Lock()
199 if f, ok := c.flights[key]; ok {
200 c.mu.Unlock()
201 select {
202 case <-f.done:
203 case <-ctx.Done():
204 return false, verify, ctx.Err()
205 }
206 if f.err != nil {
207 return false, verify, f.err
208 }
209 continue
210 }
211 f := &flight{done: make(chan struct{})}
212 c.flights[key] = f
213 c.mu.Unlock()
214 f.err = c.fill(ctx, key, fetch)
215 c.mu.Lock()
216 delete(c.flights, key)
217 c.mu.Unlock()
218 close(f.done)
219 if f.err != nil {
220 return false, verify, f.err
221 }
222 c.touch(key, true)
223 return false, verify, nil
224 }
225}
226
227// fill downloads into a temporary file, then renames it into place: a reader
228// sees a whole archive with its sum or nothing.
229func (c *Cache) fill(ctx context.Context, key string, fetch Fetch) error {
230 f, err := os.CreateTemp(c.dir, tmpPrefix+"*")
231 if err != nil {
232 return err
233 }
234 tmp := f.Name()
235 defer func() { _ = os.Remove(tmp) }()
236 h := sha256.New()
237 cw := &countWriter{w: io.MultiWriter(f, h), c: c}
238 // The bytes of the temporary file are charged while they are written and
239 // released when it is gone (renamed into an entry, or removed).
240 defer func() {
241 c.mu.Lock()
242 c.pending -= cw.n
243 c.mu.Unlock()
244 }()
245 err = fetch(ctx, cw)
246 if cerr := f.Close(); err == nil {
247 err = cerr
248 }
249 if err != nil {
250 return err
251 }
252 sum := hex.EncodeToString(h.Sum(nil)) + " " + strconv.FormatInt(cw.n, 10)
253 // A corrupt older entry of this key is replaced.
254 c.mu.Lock()
255 defer c.mu.Unlock()
256 if old, ok := c.entries[key]; ok {
257 c.total -= old.size
258 delete(c.entries, key)
259 }
260 c.remove(key)
261 if err := os.Rename(tmp, c.archivePath(key)); err != nil {
262 return err
263 }
264 stmp := c.sumPath(key) + ".new"
265 if err := os.WriteFile(stmp, []byte(sum), 0o600); err != nil {
266 _ = os.Remove(c.archivePath(key))
267 return err
268 }
269 if err := os.Rename(stmp, c.sumPath(key)); err != nil {
270 _ = os.Remove(stmp)
271 _ = os.Remove(c.archivePath(key))
272 return err
273 }
274 c.entries[key] = &entry{size: cw.n, used: c.now()}
275 c.total += cw.n
276 c.pending -= cw.n // now counted in total
277 cw.n = 0
278 return nil
279}
280
281// countWriter charges every byte it writes to the cache before it reaches the
282// disk (the size is what is written, never a declared length): when the entries
283// and the downloads in progress would exceed the cap, the least recently used
284// unpinned entries make room, and if that is not enough the write fails with
285// ErrTooLarge, so an archive bigger than the cap is never stored and the
286// cache, temporary files included, never holds more than the cap.
287type countWriter struct {
288 w io.Writer
289 c *Cache
290 n int64
291}
292
293func (w *countWriter) Write(p []byte) (int, error) {
294 c := w.c
295 c.mu.Lock()
296 need := int64(len(p))
297 if c.total+c.pending+need > c.max {
298 c.evictLocked("", need)
299 }
300 if c.total+c.pending+need > c.max {
301 c.mu.Unlock()
302 return 0, ErrTooLarge
303 }
304 c.pending += need
305 w.n += need
306 c.mu.Unlock()
307 n, err := w.w.Write(p)
308 if n < len(p) { // short write: give back what did not reach the file
309 c.mu.Lock()
310 c.pending -= int64(len(p) - n)
311 w.n -= int64(len(p) - n)
312 c.mu.Unlock()
313 }
314 return n, err
315}
316
317// verify re-hashes the archive of key against its sum. Any mismatch (or a
318// missing file) drops the entry and returns ErrNotFound.
319func (c *Cache) verify(key string) (time.Duration, error) {
320 start := time.Now()
321 sb, err := os.ReadFile(c.sumPath(key))
322 if err != nil {
323 c.drop(key)
324 return time.Since(start), ErrNotFound
325 }
326 want, sizeS, _ := strings.Cut(strings.TrimSpace(string(sb)), " ")
327 size, perr := strconv.ParseInt(sizeS, 10, 64)
328 f, err := os.Open(c.archivePath(key))
329 if err != nil || perr != nil {
330 if f != nil {
331 _ = f.Close()
332 }
333 c.drop(key)
334 return time.Since(start), ErrNotFound
335 }
336 defer func() { _ = f.Close() }()
337 h := sha256.New()
338 n, err := io.Copy(h, f)
339 if err != nil || n != size || hex.EncodeToString(h.Sum(nil)) != want {
340 c.log.Warn().Str("key", key).Msg("action archive failed its integrity check; dropped")
341 c.drop(key)
342 return time.Since(start), ErrNotFound
343 }
344 return time.Since(start), nil
345}
346
347func (c *Cache) drop(key string) {
348 c.mu.Lock()
349 defer c.mu.Unlock()
350 if e, ok := c.entries[key]; ok {
351 c.total -= e.size
352 delete(c.entries, key)
353 }
354 if _, filling := c.flights[key]; !filling {
355 c.remove(key)
356 }
357}
358
359// touch records a use (and a pin) of key.
360func (c *Cache) touch(key string, pin bool) {
361 now := c.now()
362 c.mu.Lock()
363 if e, ok := c.entries[key]; ok {
364 e.used = now
365 if pin {
366 e.pinUntil = now.Add(PinFor)
367 }
368 }
369 c.mu.Unlock()
370 _ = os.Chtimes(c.sumPath(key), now, now)
371}
372
373// evictLocked removes the least recently used entries until the cache fits,
374// leaving keep and the pinned ones alone.
375func (c *Cache) evictLocked(keep string, extra int64) {
376 if c.total+c.pending+extra <= c.max {
377 return
378 }
379 now := c.now()
380 var order []string
381 for k, e := range c.entries {
382 if k != keep && !now.Before(e.pinUntil) {
383 order = append(order, k)
384 }
385 }
386 sort.Slice(order, func(i, j int) bool { return c.entries[order[i]].used.Before(c.entries[order[j]].used) })
387 for _, k := range order {
388 if c.total+c.pending+extra <= c.max {
389 return
390 }
391 c.total -= c.entries[k].size
392 delete(c.entries, k)
393 c.remove(k)
394 }
395}
396
397// Open returns the verified archive of key. The caller closes it.
398func (c *Cache) Open(key string) (*os.File, error) {
399 if !validKey(key) {
400 return nil, ErrNotFound
401 }
402 if _, err := c.verify(key); err != nil {
403 return nil, err
404 }
405 f, err := os.Open(c.archivePath(key))
406 if err != nil {
407 c.drop(key)
408 return nil, ErrNotFound
409 }
410 c.touch(key, false)
411 return f, nil
412}
413
414// Size is the bytes held (tests, logs).
415func (c *Cache) Size() int64 {
416 c.mu.Lock()
417 defer c.mu.Unlock()
418 return c.total
419}
420
421// Dir is the directory of the cache (tests).
422func (c *Cache) Dir() string { return c.dir }