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, err error) {
188 if !validKey(key) {
189 return false, ErrNotFound
190 }
191 for {
192 if c.verify(key) == nil {
193 c.touch(key, true)
194 return true, nil
195 }
196 c.mu.Lock()
197 if f, ok := c.flights[key]; ok {
198 c.mu.Unlock()
199 select {
200 case <-f.done:
201 case <-ctx.Done():
202 return false, ctx.Err()
203 }
204 if f.err != nil {
205 return false, f.err
206 }
207 continue
208 }
209 f := &flight{done: make(chan struct{})}
210 c.flights[key] = f
211 c.mu.Unlock()
212 f.err = c.fill(ctx, key, fetch)
213 c.mu.Lock()
214 delete(c.flights, key)
215 c.mu.Unlock()
216 close(f.done)
217 if f.err != nil {
218 return false, f.err
219 }
220 c.touch(key, true)
221 return false, nil
222 }
223}
224
225// fill downloads into a temporary file, then renames it into place: a reader
226// sees a whole archive with its sum or nothing.
227func (c *Cache) fill(ctx context.Context, key string, fetch Fetch) error {
228 f, err := os.CreateTemp(c.dir, tmpPrefix+"*")
229 if err != nil {
230 return err
231 }
232 tmp := f.Name()
233 defer func() { _ = os.Remove(tmp) }()
234 h := sha256.New()
235 cw := &countWriter{w: io.MultiWriter(f, h), c: c}
236 // The bytes of the temporary file are charged while they are written and
237 // released when it is gone (renamed into an entry, or removed).
238 defer func() {
239 c.mu.Lock()
240 c.pending -= cw.n
241 c.mu.Unlock()
242 }()
243 err = fetch(ctx, cw)
244 if cerr := f.Close(); err == nil {
245 err = cerr
246 }
247 if err != nil {
248 return err
249 }
250 sum := hex.EncodeToString(h.Sum(nil)) + " " + strconv.FormatInt(cw.n, 10)
251 // A corrupt older entry of this key is replaced.
252 c.mu.Lock()
253 defer c.mu.Unlock()
254 if old, ok := c.entries[key]; ok {
255 c.total -= old.size
256 delete(c.entries, key)
257 }
258 c.remove(key)
259 if err := os.Rename(tmp, c.archivePath(key)); err != nil {
260 return err
261 }
262 stmp := c.sumPath(key) + ".new"
263 if err := os.WriteFile(stmp, []byte(sum), 0o600); err != nil {
264 _ = os.Remove(c.archivePath(key))
265 return err
266 }
267 if err := os.Rename(stmp, c.sumPath(key)); err != nil {
268 _ = os.Remove(stmp)
269 _ = os.Remove(c.archivePath(key))
270 return err
271 }
272 c.entries[key] = &entry{size: cw.n, used: c.now()}
273 c.total += cw.n
274 c.pending -= cw.n // now counted in total
275 cw.n = 0
276 return nil
277}
278
279// countWriter charges every byte it writes to the cache before it reaches the
280// disk (the size is what is written, never a declared length): when the entries
281// and the downloads in progress would exceed the cap, the least recently used
282// unpinned entries make room, and if that is not enough the write fails with
283// ErrTooLarge, so an archive bigger than the cap is never stored and the
284// cache, temporary files included, never holds more than the cap.
285type countWriter struct {
286 w io.Writer
287 c *Cache
288 n int64
289}
290
291func (w *countWriter) Write(p []byte) (int, error) {
292 c := w.c
293 c.mu.Lock()
294 need := int64(len(p))
295 if c.total+c.pending+need > c.max {
296 c.evictLocked("", need)
297 }
298 if c.total+c.pending+need > c.max {
299 c.mu.Unlock()
300 return 0, ErrTooLarge
301 }
302 c.pending += need
303 w.n += need
304 c.mu.Unlock()
305 n, err := w.w.Write(p)
306 if n < len(p) { // short write: give back what did not reach the file
307 c.mu.Lock()
308 c.pending -= int64(len(p) - n)
309 w.n -= int64(len(p) - n)
310 c.mu.Unlock()
311 }
312 return n, err
313}
314
315// verify re-hashes the archive of key against its sum. Any mismatch (or a
316// missing file) drops the entry and returns ErrNotFound.
317func (c *Cache) verify(key string) error {
318 sb, err := os.ReadFile(c.sumPath(key))
319 if err != nil {
320 c.drop(key)
321 return ErrNotFound
322 }
323 want, sizeS, _ := strings.Cut(strings.TrimSpace(string(sb)), " ")
324 size, perr := strconv.ParseInt(sizeS, 10, 64)
325 f, err := os.Open(c.archivePath(key))
326 if err != nil || perr != nil {
327 if f != nil {
328 _ = f.Close()
329 }
330 c.drop(key)
331 return ErrNotFound
332 }
333 defer func() { _ = f.Close() }()
334 h := sha256.New()
335 n, err := io.Copy(h, f)
336 if err != nil || n != size || hex.EncodeToString(h.Sum(nil)) != want {
337 c.log.Warn().Str("key", key).Msg("action archive failed its integrity check; dropped")
338 c.drop(key)
339 return ErrNotFound
340 }
341 return nil
342}
343
344func (c *Cache) drop(key string) {
345 c.mu.Lock()
346 defer c.mu.Unlock()
347 if e, ok := c.entries[key]; ok {
348 c.total -= e.size
349 delete(c.entries, key)
350 }
351 if _, filling := c.flights[key]; !filling {
352 c.remove(key)
353 }
354}
355
356// touch records a use (and a pin) of key.
357func (c *Cache) touch(key string, pin bool) {
358 now := c.now()
359 c.mu.Lock()
360 if e, ok := c.entries[key]; ok {
361 e.used = now
362 if pin {
363 e.pinUntil = now.Add(PinFor)
364 }
365 }
366 c.mu.Unlock()
367 _ = os.Chtimes(c.sumPath(key), now, now)
368}
369
370// evictLocked removes the least recently used entries until the cache fits,
371// leaving keep and the pinned ones alone.
372func (c *Cache) evictLocked(keep string, extra int64) {
373 if c.total+c.pending+extra <= c.max {
374 return
375 }
376 now := c.now()
377 var order []string
378 for k, e := range c.entries {
379 if k != keep && !now.Before(e.pinUntil) {
380 order = append(order, k)
381 }
382 }
383 sort.Slice(order, func(i, j int) bool { return c.entries[order[i]].used.Before(c.entries[order[j]].used) })
384 for _, k := range order {
385 if c.total+c.pending+extra <= c.max {
386 return
387 }
388 c.total -= c.entries[k].size
389 delete(c.entries, k)
390 c.remove(k)
391 }
392}
393
394// Open returns the verified archive of key. The caller closes it.
395func (c *Cache) Open(key string) (*os.File, error) {
396 if !validKey(key) {
397 return nil, ErrNotFound
398 }
399 if err := c.verify(key); err != nil {
400 return nil, err
401 }
402 f, err := os.Open(c.archivePath(key))
403 if err != nil {
404 c.drop(key)
405 return nil, ErrNotFound
406 }
407 c.touch(key, false)
408 return f, nil
409}
410
411// Size is the bytes held (tests, logs).
412func (c *Cache) Size() int64 {
413 c.mu.Lock()
414 defer c.mu.Unlock()
415 return c.total
416}
417
418// ParseMax reads ERMITE_RUNNER_ACTIONS_CACHE_MAX: empty is DefaultMax, 0 is off
419// (returned as 0), otherwise a number of bytes or a size with a K, M, G or T
420// suffix (optionally followed by iB or B).
421func ParseMax(v string) (int64, error) {
422 v = strings.TrimSpace(v)
423 if v == "" {
424 return DefaultMax, nil
425 }
426 s := strings.TrimRight(strings.ToUpper(v), "IB")
427 mult := int64(1)
428 if n := len(s); n > 0 {
429 switch s[n-1] {
430 case 'K':
431 mult, s = 1<<10, s[:n-1]
432 case 'M':
433 mult, s = 1<<20, s[:n-1]
434 case 'G':
435 mult, s = 1<<30, s[:n-1]
436 case 'T':
437 mult, s = 1<<40, s[:n-1]
438 }
439 }
440 n, err := strconv.ParseInt(s, 10, 64)
441 if err != nil || n < 0 || n > (1<<62)/mult {
442 return 0, fmt.Errorf("%s: %q is not a size (examples: 2G, 500M, 0)", EnvMax, v)
443 }
444 return n * mult, nil
445}
446
447// Dir is the directory of the cache (tests).
448func (c *Cache) Dir() string { return c.dir }