1// SPDX-License-Identifier: MIT
2
3package cachesvc
4
5import (
6 "crypto/rand"
7 "encoding/binary"
8 "encoding/json"
9 "errors"
10 "fmt"
11 "os"
12 "path/filepath"
13 "sort"
14 "strconv"
15 "strings"
16 "sync"
17 "time"
18
19 "github.com/bnema/zerowrap"
20)
21
22// Entry is a committed cache entry. Reservations are never persisted: a restart
23// ends every job, so an unfinished upload is dropped.
24type Entry struct {
25 ID int64 `json:"id"`
26 Repository string `json:"repo"`
27 Ref string `json:"ref"`
28 Key string `json:"key"`
29 Version string `json:"version"`
30 Size int64 `json:"size"`
31 Created time.Time `json:"created"`
32 LastUsed time.Time `json:"last_used"`
33}
34
35// reservation is an upload in progress (POST caches, PATCH, POST caches/{id}).
36type reservation struct {
37 id int64
38 job string
39 repo, ref string
40 key, ver string
41 declared int64 // cacheSize of the reservation; 0 when the client gave none
42 // charged is what the reservation holds of the budget: the declared size
43 // (up front), or for a client that gave none the high-water mark of the
44 // ranges it has started to write; never less than the minimum charge. It is
45 // kept until the reservation is committed or dropped.
46 charged int64
47 bound int64 // largest size accepted: declared, or the cache limit
48 touched time.Time
49 // expires is the absolute end of the reservation: the expiry of the token
50 // that made it, or an hour after it was made, whichever comes first. It is
51 // enforced besides the idle ttl; a reservation with a chunk in flight is
52 // reclaimed once that chunk ends (the transfer deadlines are capped at the
53 // token's expiry, so it ends).
54 expires time.Time
55 ranges []span // validated byte ranges: fully written, body checked
56 active []span // ranges being written now; they never overlap each other
57 // committing is set while the commit renames the file.
58 committing bool
59}
60
61// span is the half-open byte range [lo, hi).
62type span struct{ lo, hi int64 }
63
64// add merges s into the sorted, disjoint list.
65func addSpan(list []span, s span) []span {
66 out := make([]span, 0, len(list)+1)
67 placed := false
68 for _, x := range list {
69 switch {
70 case x.hi < s.lo:
71 out = append(out, x)
72 case s.hi < x.lo:
73 if !placed {
74 out, placed = append(out, s), true
75 }
76 out = append(out, x)
77 default: // overlap or touch
78 s = span{min(s.lo, x.lo), max(s.hi, x.hi)}
79 }
80 }
81 if !placed {
82 out = append(out, s)
83 }
84 return out
85}
86
87// subSpan removes s from the sorted, disjoint list.
88func subSpan(list []span, s span) []span {
89 var out []span
90 for _, x := range list {
91 if x.hi <= s.lo || s.hi <= x.lo {
92 out = append(out, x)
93 continue
94 }
95 if x.lo < s.lo {
96 out = append(out, span{x.lo, s.lo})
97 }
98 if s.hi < x.hi {
99 out = append(out, span{s.hi, x.hi})
100 }
101 }
102 return out
103}
104
105func overlaps(list []span, s span) bool {
106 for _, x := range list {
107 if x.lo < s.hi && s.lo < x.hi {
108 return true
109 }
110 }
111 return false
112}
113
114// complete reports whether list covers exactly [0, size).
115func complete(list []span, size int64) bool {
116 return len(list) == 1 && list[0].lo == 0 && list[0].hi == size
117}
118
119// Sentinel errors of the store, mapped to HTTP statuses by the handler.
120var (
121 errConflict = errors.New("a cache entry or reservation with this key and version exists in this ref")
122 errTooBig = errors.New("the cache is larger than the limit of the cache service")
123 errNoRoom = errors.New("no room in the cache: other uploads hold the space")
124 errNotFound = errors.New("no such cache reservation")
125 errDenied = errForbidden
126 errBadSize = errors.New("the uploaded data does not match the size")
127 errState = errors.New("the cache reservation is already being committed")
128 errOverlap = errors.New("this range is being uploaded already")
129 errTooMany = errors.New("too many cache uploads in progress")
130)
131
132// Limits of the store (storeConfig zero values take these).
133const (
134 // DefaultMinCharge is charged against the budget for every committed entry
135 // and reservation however small: a million one-byte entries are not free
136 // (index, inodes). With the 10 GiB default it caps the entries at ~2.6M.
137 DefaultMinCharge = 4 << 10
138 // DefaultMaxPending bounds the uploads in progress, in all and per job.
139 DefaultMaxPending = 256
140 DefaultMaxPendingPerJob = 32
141)
142
143type storeConfig struct {
144 dir string
145 max int64
146 ttl time.Duration // idle life of a reservation that is never committed
147 now func() time.Time
148 minCharge int64
149 maxPending int
150 maxPerJob int
151 // maxLife bounds the life of a reservation from its creation.
152 maxLife time.Duration
153 log zerowrap.Logger
154}
155
156// store keeps the index and the files under dir.
157//
158// Budget: used (the charge of the committed entries) + held (the charge of the
159// reservations) never exceeds max once admitted. Committing moves a
160// reservation's charge to its entry, so it needs no room of its own. Entries
161// with a download in progress (pins) are not evicted: their bytes cannot be
162// reclaimed until the last reader is done, and reservations fail with errNoRoom
163// meanwhile.
164type store struct {
165 storeConfig
166
167 mu sync.Mutex
168 entries map[int64]*Entry
169 pending map[int64]*reservation
170 pins map[int64]int // open downloads per entry
171 used int64
172 held int64
173 dirty bool // lastUsed changed since the index was written
174}
175
176func (s *store) chargeOf(size int64) int64 { return max(size, s.minCharge) }
177
178const indexName = "index.json"
179
180func openStore(c storeConfig) (*store, error) {
181 if c.minCharge <= 0 {
182 c.minCharge = DefaultMinCharge
183 }
184 if c.maxPending <= 0 {
185 c.maxPending = DefaultMaxPending
186 }
187 if c.maxPerJob <= 0 {
188 c.maxPerJob = DefaultMaxPendingPerJob
189 }
190 if c.maxLife <= 0 {
191 c.maxLife = DefaultReservationTTL
192 }
193 dir := c.dir
194 if err := os.MkdirAll(filepath.Join(dir, "data"), 0o700); err != nil {
195 return nil, fmt.Errorf("cache: %w", err)
196 }
197 s := &store{storeConfig: c, entries: map[int64]*Entry{}, pending: map[int64]*reservation{}, pins: map[int64]int{}}
198 b, err := os.ReadFile(filepath.Join(dir, indexName))
199 if err == nil {
200 var list []*Entry
201 if err := json.Unmarshal(b, &list); err != nil {
202 // A corrupt index is not worth refusing to start for: the cache is a
203 // cache. Start empty; the orphan files are removed below.
204 list = nil
205 }
206 for _, e := range list {
207 if st, err := os.Stat(s.file(e.ID)); err == nil && st.Mode().IsRegular() && st.Size() == e.Size && s.entries[e.ID] == nil {
208 s.entries[e.ID] = e
209 s.used += s.chargeOf(e.Size)
210 }
211 }
212 } else if !errors.Is(err, os.ErrNotExist) {
213 return nil, fmt.Errorf("cache: %w", err)
214 }
215 // Anything in data/ that the index does not know (temporary files of
216 // reservations, files of entries lost from the index) goes.
217 des, err := os.ReadDir(filepath.Join(dir, "data"))
218 if err != nil {
219 return nil, fmt.Errorf("cache: %w", err)
220 }
221 for _, de := range des {
222 id, perr := strconv.ParseInt(strings.TrimSuffix(de.Name(), ".tmp"), 16, 64)
223 if perr != nil || strings.HasSuffix(de.Name(), ".tmp") || s.entries[id] == nil {
224 _ = os.Remove(filepath.Join(dir, "data", de.Name()))
225 }
226 }
227 s.evictLocked(0)
228 if err := s.saveLocked(); err != nil {
229 return nil, err
230 }
231 return s, nil
232}
233
234func (s *store) file(id int64) string {
235 return filepath.Join(s.dir, "data", strconv.FormatInt(id, 16))
236}
237func (s *store) tmp(id int64) string { return s.file(id) + ".tmp" }
238
239// saveLocked writes the index atomically and durably: the temporary file is
240// synced before the rename and the directory after it.
241func (s *store) saveLocked() error {
242 list := make([]*Entry, 0, len(s.entries))
243 for _, e := range s.entries {
244 list = append(list, e)
245 }
246 sort.Slice(list, func(i, j int) bool { return list[i].ID < list[j].ID })
247 b, err := json.Marshal(list)
248 if err != nil {
249 return err
250 }
251 tmp := filepath.Join(s.dir, indexName+".tmp")
252 f, err := os.OpenFile(tmp, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o600)
253 if err != nil {
254 return fmt.Errorf("cache index: %w", err)
255 }
256 _, werr := f.Write(b)
257 if werr == nil {
258 werr = f.Sync()
259 }
260 if cerr := f.Close(); werr == nil {
261 werr = cerr
262 }
263 if werr != nil {
264 _ = os.Remove(tmp)
265 return fmt.Errorf("cache index: %w", werr)
266 }
267 if err := os.Rename(tmp, filepath.Join(s.dir, indexName)); err != nil {
268 return fmt.Errorf("cache index: %w", err)
269 }
270 if err := syncDir(s.dir); err != nil {
271 return fmt.Errorf("cache index: %w", err)
272 }
273 s.dirty = false
274 return nil
275}
276
277func syncDir(path string) error {
278 d, err := os.Open(path)
279 if err != nil {
280 return err
281 }
282 err = d.Sync()
283 if cerr := d.Close(); err == nil {
284 err = cerr
285 }
286 return err
287}
288
289// flush writes the index when only usage times changed.
290func (s *store) flush() error {
291 s.mu.Lock()
292 defer s.mu.Unlock()
293 if !s.dirty {
294 return nil
295 }
296 return s.saveLocked()
297}
298
299// newID is a random positive integer below 2^53: the toolkit reads cacheId as a
300// JavaScript number, and ids that cannot be guessed keep one job from writing
301// into another job's reservation.
302func (s *store) newID() int64 {
303 for {
304 var b [8]byte
305 _, _ = rand.Read(b[:])
306 id := int64(binary.BigEndian.Uint64(b[:]) & (1<<53 - 1))
307 if id == 0 || s.entries[id] != nil || s.pending[id] != nil {
308 continue
309 }
310 return id
311 }
312}
313
314// expireLocked drops reservations that saw no activity for the ttl and have no
315// chunk in flight.
316func (s *store) expireLocked() {
317 now := s.now()
318 for _, r := range s.pending {
319 if !r.committing && len(r.active) == 0 && (now.Sub(r.touched) > s.ttl || !now.Before(r.expires)) {
320 s.dropLocked(r)
321 }
322 }
323}
324
325func (s *store) dropLocked(r *reservation) {
326 s.held -= r.charged
327 delete(s.pending, r.id)
328 _ = os.Remove(s.tmp(r.id))
329}
330
331// evictLocked removes least recently used committed entries (not those being
332// downloaded) until need more bytes fit next to what is used and held. It
333// reports whether they fit and how many entries it removed.
334func (s *store) evictLocked(need int64) (fits bool, evicted int) {
335 if s.used+s.held+need <= s.max {
336 return true, 0
337 }
338 list := make([]*Entry, 0, len(s.entries))
339 for _, e := range s.entries {
340 if s.pins[e.ID] == 0 {
341 list = append(list, e)
342 }
343 }
344 sort.Slice(list, func(i, j int) bool {
345 if !list[i].LastUsed.Equal(list[j].LastUsed) {
346 return list[i].LastUsed.Before(list[j].LastUsed)
347 }
348 return list[i].ID < list[j].ID
349 })
350 for _, e := range list {
351 if s.used+s.held+need <= s.max {
352 break
353 }
354 delete(s.entries, e.ID)
355 s.used -= s.chargeOf(e.Size)
356 _ = os.Remove(s.file(e.ID))
357 evicted++
358 // The index still names it until it is saved; a failed save leaves this
359 // set so that the housekeeping tries again.
360 s.dirty = true
361 }
362 return s.used+s.held+need <= s.max, evicted
363}
364
365// expire drops the reservations that were idle for too long.
366func (s *store) expire() {
367 s.mu.Lock()
368 defer s.mu.Unlock()
369 s.expireLocked()
370}
371
372// roomLocked makes need more bytes fit, evicting if it must, and writes the index when
373// it evicted (the files are gone). Timestamps alone wait for the housekeeping.
374func (s *store) roomLocked(need int64) bool {
375 fits, n := s.evictLocked(need)
376 if n > 0 {
377 if err := s.saveLocked(); err != nil {
378 s.log.Error().Err(err).Msg("cache index (after evicting); will retry")
379 }
380 }
381 return fits
382}
383
384// reserve starts an upload for (repo, ref, key, version). declared is the
385// cacheSize of the client, 0 if none.
386func (s *store) reserve(job, repo, ref, key, version string, declared int64, tokenExp time.Time) (int64, error) {
387 s.mu.Lock()
388 defer s.mu.Unlock()
389 s.expireLocked()
390 if declared > s.max {
391 return 0, errTooBig
392 }
393 for _, e := range s.entries {
394 if e.Repository == repo && e.Ref == ref && e.Key == key && e.Version == version {
395 return 0, errConflict
396 }
397 }
398 perJob := 0
399 for _, r := range s.pending {
400 if r.repo == repo && r.ref == ref && r.key == key && r.ver == version {
401 return 0, errConflict
402 }
403 if r.job == job {
404 perJob++
405 }
406 }
407 // Before anything is created.
408 if len(s.pending) >= s.maxPending || perJob >= s.maxPerJob {
409 return 0, errTooMany
410 }
411 // A client that announces its size is charged for it now; one that does not
412 // is charged as its chunks arrive (beginChunk).
413 bound, charge := declared, s.chargeOf(declared)
414 if declared == 0 {
415 bound = s.max
416 }
417 if !s.roomLocked(charge) {
418 return 0, errNoRoom
419 }
420 id := s.newID()
421 f, err := os.OpenFile(s.tmp(id), os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
422 if err != nil {
423 return 0, fmt.Errorf("cache: %w", err)
424 }
425 _ = f.Close()
426 s.pending[id] = &reservation{id: id, job: job, repo: repo, ref: ref, key: key, ver: version, declared: declared, charged: charge, bound: bound, touched: s.now(), expires: s.reservationEnd(tokenExp)}
427 s.held += charge
428 return id, nil
429}
430
431// reservationEnd is the absolute end of a reservation made now with a token that
432// expires at tokenExp.
433func (s *store) reservationEnd(tokenExp time.Time) time.Time {
434 end := s.now().Add(s.maxLife)
435 if tokenExp.Before(end) {
436 end = tokenExp
437 }
438 return end
439}
440
441// owned returns the reservation id if it belongs to this job: same job, same
442// repository, and the ref is the job's write ref.
443func (s *store) ownedLocked(id int64, c Claims) (*reservation, error) {
444 r := s.pending[id]
445 if r == nil {
446 return nil, errNotFound
447 }
448 if r.job != c.JobID || r.repo != c.Repository || c.WriteRef == "" || r.ref != c.WriteRef {
449 return nil, errDenied
450 }
451 if !s.now().Before(r.expires) {
452 // Over: nothing more is accepted. It is dropped as soon as no chunk is in
453 // flight (expireLocked).
454 return nil, errNotFound
455 }
456 return r, nil
457}
458
459// beginChunk admits a chunk [lo, hi) of reservation id and returns the path to
460// write it to. The chunk is charged first: a reservation without a declared size
461// is charged up to hi (evicting to make room, or failing with errNoRoom). The
462// range must not be in flight already, and any coverage it had is removed: a
463// retry that fails halfway must not leave the old bytes counted as valid.
464// endChunk must follow.
465func (s *store) beginChunk(id int64, c Claims, lo, hi int64) (string, error) {
466 s.mu.Lock()
467 defer s.mu.Unlock()
468 s.expireLocked()
469 r, err := s.ownedLocked(id, c)
470 if err != nil {
471 return "", err
472 }
473 if r.committing {
474 return "", errState
475 }
476 if hi > r.bound {
477 return "", errTooBig
478 }
479 sp := span{lo, hi}
480 if overlaps(r.active, sp) {
481 return "", errOverlap
482 }
483 if extra := s.chargeOf(hi) - r.charged; extra > 0 {
484 if !s.roomLocked(extra) {
485 return "", errNoRoom
486 }
487 r.charged += extra
488 s.held += extra
489 }
490 r.ranges = subSpan(r.ranges, sp)
491 r.active = append(r.active, sp)
492 r.touched = s.now()
493 return s.tmp(id), nil
494}
495
496// endChunk ends the chunk begun by beginChunk; only when the whole body was
497// written and checked does [lo, hi) count as received.
498func (s *store) endChunk(id int64, lo, hi int64, written bool) {
499 s.mu.Lock()
500 defer s.mu.Unlock()
501 r := s.pending[id]
502 if r == nil {
503 return
504 }
505 r.active = subSpan(r.active, span{lo, hi})
506 r.touched = s.now()
507 if written {
508 r.ranges = addSpan(r.ranges, span{lo, hi})
509 }
510 // A reservation that ran out while this chunk was in flight is reclaimed now
511 // that its last write is over.
512 s.expireLocked()
513}
514
515// commit makes reservation id an entry once exactly [0, size) was received.
516func (s *store) commit(id int64, c Claims, size int64) error {
517 s.mu.Lock()
518 s.expireLocked()
519 r, err := s.ownedLocked(id, c)
520 if err != nil {
521 s.mu.Unlock()
522 return err
523 }
524 if r.committing || len(r.active) > 0 {
525 s.mu.Unlock()
526 return errState
527 }
528 if size <= 0 || (r.declared != 0 && size != r.declared) || !complete(r.ranges, size) {
529 s.mu.Unlock()
530 return errBadSize
531 }
532 r.committing = true
533 s.mu.Unlock()
534
535 // The data is synced and the file is the right length before it takes its
536 // final name, and the directory is synced before the index names the file:
537 // readers only ever see whole entries, and the index never points at a file
538 // that a crash can lose.
539 err = syncTruncate(s.tmp(id), size)
540 if err == nil {
541 err = os.Rename(s.tmp(id), s.file(id))
542 }
543 if err == nil {
544 err = syncDir(filepath.Join(s.dir, "data"))
545 }
546
547 s.mu.Lock()
548 defer s.mu.Unlock()
549 s.dropPendingKeepFileLocked(r)
550 if err != nil {
551 _ = os.Remove(s.tmp(id))
552 _ = os.Remove(s.file(id))
553 return fmt.Errorf("cache: %w", err)
554 }
555 now := s.now()
556 s.entries[id] = &Entry{ID: id, Repository: r.repo, Ref: r.ref, Key: r.key, Version: r.ver, Size: size, Created: now, LastUsed: now}
557 // The reservation's charge becomes the entry's: size == hwm of the chunks.
558 s.used += s.chargeOf(size)
559 s.dirty = true // stays set if the save fails: the housekeeping retries
560 return s.saveLocked()
561}
562
563// dropPendingKeepFileLocked releases a reservation without touching its file.
564func (s *store) dropPendingKeepFileLocked(r *reservation) {
565 s.held -= r.charged
566 delete(s.pending, r.id)
567}
568
569func syncTruncate(path string, size int64) error {
570 f, err := os.OpenFile(path, os.O_WRONLY, 0)
571 if err != nil {
572 return err
573 }
574 if err := f.Truncate(size); err != nil {
575 _ = f.Close()
576 return err
577 }
578 if err := f.Sync(); err != nil {
579 _ = f.Close()
580 return err
581 }
582 return f.Close()
583}
584
585// find is the cache lookup, as GitHub documents it ("Matching a cache key"):
586// the read refs are searched in priority order, and in each ref the keys in
587// order: the first (primary) key exactly, the restore keys as an exact match or
588// else the newest entry whose key starts with them. The match counts as a use.
589func (s *store) find(repo string, refs []string, keys []string, version string) (Entry, bool) {
590 s.mu.Lock()
591 defer s.mu.Unlock()
592 for _, ref := range refs {
593 for i, k := range keys {
594 var best *Entry
595 for _, e := range s.entries {
596 if e.Repository != repo || e.Ref != ref || e.Version != version {
597 continue
598 }
599 if e.Key == k {
600 best = e
601 break
602 }
603 if i == 0 || !strings.HasPrefix(e.Key, k) {
604 continue
605 }
606 if best == nil || e.Created.After(best.Created) || (e.Created.Equal(best.Created) && e.ID > best.ID) {
607 best = e
608 }
609 }
610 if best != nil {
611 best.LastUsed = s.now()
612 s.dirty = true
613 return *best, true
614 }
615 }
616 }
617 return Entry{}, false
618}
619
620// list returns the entries of the repository in the refs whose key starts with
621// prefix, newest first.
622func (s *store) list(repo string, refs []string, prefix string) []Entry {
623 s.mu.Lock()
624 defer s.mu.Unlock()
625 var out []Entry
626 for _, e := range s.entries {
627 if e.Repository == repo && strings.HasPrefix(e.Key, prefix) && contains(refs, e.Ref) {
628 out = append(out, *e)
629 }
630 }
631 sort.Slice(out, func(i, j int) bool { return out[i].Created.After(out[j].Created) })
632 return out
633}
634
635func contains(l []string, v string) bool {
636 for _, x := range l {
637 if x == v {
638 return true
639 }
640 }
641 return false
642}
643
644// open returns the committed entry id for reading, counts it as a use and pins
645// it until release is called: a pinned entry is not evicted.
646func (s *store) open(id int64, repo string) (f *os.File, e Entry, release func(), err error) {
647 s.mu.Lock()
648 defer s.mu.Unlock()
649 ent := s.entries[id]
650 if ent == nil || ent.Repository != repo {
651 return nil, Entry{}, nil, errNotFound
652 }
653 f, err = os.Open(s.file(id))
654 if err != nil {
655 return nil, Entry{}, nil, errNotFound
656 }
657 ent.LastUsed = s.now()
658 s.dirty = true
659 s.pins[id]++
660 var once sync.Once
661 return f, *ent, func() {
662 once.Do(func() {
663 s.mu.Lock()
664 if s.pins[id]--; s.pins[id] <= 0 {
665 delete(s.pins, id)
666 }
667 s.mu.Unlock()
668 })
669 }, nil
670}
671
672// get returns a committed entry without counting a use.
673func (s *store) get(id int64) (Entry, bool) {
674 s.mu.Lock()
675 defer s.mu.Unlock()
676 if e := s.entries[id]; e != nil {
677 return *e, true
678 }
679 return Entry{}, false
680}
681
682// stats are for tests and logs.
683func (s *store) stats() (entries int, used, held int64) {
684 s.mu.Lock()
685 defer s.mu.Unlock()
686 return len(s.entries), s.used, s.held
687}
688
689func (s *store) pendingCount() int {
690 s.mu.Lock()
691 defer s.mu.Unlock()
692 return len(s.pending)
693}