1// SPDX-License-Identifier: MIT
2
3// Package cachesvc is the GitHub Actions cache service (API v1) of the runner
4// (plan CI-3, D8). Actions that use @actions/cache (setup-go, setup-node,
5// actions/cache…) talk to it through ACTIONS_CACHE_URL: the toolkit takes the v1
6// protocol for any server URL that is not github.com, *.ghe.com or *.localhost
7// (toolkit/packages/cache/src/internal/config.ts:1-20).
8//
9// One Service per runner host. Entries live under <state>/cache/ (files plus a
10// JSON index, atomic finalisation, LRU eviction by total size) and are scoped by
11// (repository, ref, key, version). A job reaches the service with a bearer token
12// (ACTIONS_RUNTIME_TOKEN) that is an HMAC, made with a key that exists only in
13// the memory of the runner, over its job id, repository, write ref, read refs
14// and expiry.
15//
16// The service never listens on a network. Each sandbox has a channel of its own
17// (Serve): a Unix socket, bind-mounted into a container or reached through
18// Firecracker's vsock proxy for a VM, on which the relay in the guest delivers
19// the connections of the actions, which reach it on the guest's loopback
20// (DefaultPort). The host opens no port and its firewall has no rule for it.
21package cachesvc
22
23import (
24 "errors"
25 "fmt"
26 "net"
27 "net/http"
28 "strconv"
29 "sync"
30 "time"
31
32 "github.com/bnema/zerowrap"
33)
34
35// Defaults.
36const (
37 // DefaultMax is the size cap, GitHub's limit per repository.
38 DefaultMax = 10 << 30
39 // DefaultPort is the TCP port of the cache on the loopback of the guest
40 // (ACTIONS_CACHE_URL=http://127.0.0.1:9089/), and the vsock port of the
41 // cache streams of a VM. It is not a port of the host.
42 DefaultPort = 9089
43 // EnvMax is ERMITE_RUNNER_CACHE_MAX, the size cap (1G, 500M…).
44 EnvMax = "ERMITE_RUNNER_CACHE_MAX"
45 // DefaultReservationTTL is how long an upload may stay uncommitted.
46 DefaultReservationTTL = time.Hour
47 flushEvery = 30 * time.Second
48 // DefaultProgressTimeout is how long an upload or a download may make no
49 // progress before its connection is cut.
50 DefaultProgressTimeout = 60 * time.Second
51 // DefaultConnsPerListener and DefaultConns bound the open connections of one
52 // sandbox's listener and of the whole service.
53 DefaultConnsPerListener = 16
54 DefaultConns = 64
55)
56
57// Config configures a Service.
58type Config struct {
59 // Dir is the cache directory (<state>/cache).
60 Dir string
61 // Max is the size cap in bytes; 0 means DefaultMax.
62 Max int64
63 // ReservationTTL is the life of an uncommitted reservation; 0 means
64 // DefaultReservationTTL.
65 ReservationTTL time.Duration
66 // ReservationMaxLife is the absolute life of a reservation from its creation
67 // (the token's expiry still ends it sooner); 0 means DefaultReservationTTL.
68 ReservationMaxLife time.Duration
69 // Now is the clock; nil means time.Now.
70 Now func() time.Time
71 Log zerowrap.Logger
72 // The limits below take the Default… values (or those of the store) when
73 // zero; tests set them.
74 ProgressTimeout time.Duration
75 ConnsPerListener, Conns int
76 MinCharge int64
77 MaxPending, MaxPendingPerJob int
78 // hookChunk, when set, is called by an upload once its range is admitted and
79 // before its body is read (tests: a barrier with a chunk in flight).
80 hookChunk func(id int64)
81}
82
83// Service is the cache service.
84type Service struct {
85 store *store
86 sign *signer
87 now func() time.Time
88 log zerowrap.Logger
89
90 mu sync.Mutex
91 channels map[string]*listener
92 stop chan struct{}
93 wg sync.WaitGroup
94 closed bool
95
96 hookChunk func(id int64)
97 progress time.Duration
98 perListener int
99 connsGlobal chan struct{}
100}
101
102type listener struct {
103 srv *http.Server
104 done chan struct{}
105 bw *bandwidth // the channel's limiter; nil when unlimited (read by tests)
106}
107
108// New opens the store and starts the housekeeping goroutine; Close stops it.
109func New(cfg Config) (*Service, error) {
110 if cfg.Max <= 0 {
111 cfg.Max = DefaultMax
112 }
113 if cfg.ReservationTTL <= 0 {
114 cfg.ReservationTTL = DefaultReservationTTL
115 }
116 if cfg.Now == nil {
117 cfg.Now = time.Now
118 }
119 sg, err := newSigner()
120 if err != nil {
121 return nil, err
122 }
123 st, err := openStore(storeConfig{dir: cfg.Dir, max: cfg.Max, ttl: cfg.ReservationTTL, now: cfg.Now,
124 minCharge: cfg.MinCharge, maxPending: cfg.MaxPending, maxPerJob: cfg.MaxPendingPerJob, maxLife: cfg.ReservationMaxLife, log: cfg.Log})
125 if err != nil {
126 return nil, err
127 }
128 if cfg.ProgressTimeout <= 0 {
129 cfg.ProgressTimeout = DefaultProgressTimeout
130 }
131 if cfg.ConnsPerListener <= 0 {
132 cfg.ConnsPerListener = DefaultConnsPerListener
133 }
134 if cfg.Conns <= 0 {
135 cfg.Conns = DefaultConns
136 }
137 s := &Service{store: st, sign: sg, now: cfg.Now, log: cfg.Log, channels: map[string]*listener{}, stop: make(chan struct{}),
138 hookChunk: cfg.hookChunk, progress: cfg.ProgressTimeout, perListener: cfg.ConnsPerListener, connsGlobal: make(chan struct{}, cfg.Conns)}
139 s.wg.Add(1)
140 go s.housekeeping()
141 return s, nil
142}
143
144// housekeeping writes the usage times of the index and drops stale reservations.
145func (s *Service) housekeeping() {
146 defer s.wg.Done()
147 t := time.NewTicker(flushEvery)
148 defer t.Stop()
149 for {
150 select {
151 case <-s.stop:
152 return
153 case <-t.C:
154 s.store.expire()
155 if err := s.store.flush(); err != nil {
156 s.log.Warn().Err(err).Msg("cache index")
157 }
158 }
159 }
160}
161
162// Token returns the ACTIONS_RUNTIME_TOKEN of a job. It is valid for ttl, which
163// should be the job timeout plus a margin.
164func (s *Service) Token(jobID string, sc Scope, ttl time.Duration) string {
165 return s.sign.token(Claims{JobID: jobID, Scope: sc, Expiry: s.now().Add(ttl).Unix()})
166}
167
168// start serves HTTP on ln, subject to the connection caps, until the listener
169// is closed. job, when not empty, restricts the tokens accepted to that job's.
170func (s *Service) start(name string, ln net.Listener, job string, bw *bandwidth) *listener {
171 if bw != nil {
172 ln = throttleListener{ln, bw}
173 }
174 ln = newLimitListener(ln, s.perListener, s.connsGlobal)
175 l := &listener{
176 srv: &http.Server{
177 Handler: s.handler(job),
178 ReadHeaderTimeout: 30 * time.Second,
179 // Bodies are chunks of up to tens of MiB and archives of GiB: only
180 // idle connections time out.
181 IdleTimeout: 2 * time.Minute,
182 },
183 done: make(chan struct{}),
184 bw: bw,
185 }
186 go func() {
187 defer close(l.done)
188 if err := l.srv.Serve(ln); err != nil && !errors.Is(err, http.ErrServerClosed) {
189 s.log.Error().Err(err).Str("listener", name).Msg("cache listener")
190 }
191 }()
192 return l
193}
194
195// Serve serves the cache on ln, the host end of the channel of one sandbox
196// (id): a listener that is not on the network, such as the Unix socket on which
197// the relay of that sandbox's guest delivers its connections. Each sandbox has
198// its own listener and so its own handler, and nothing but that sandbox's guest
199// can connect to it. job is the id of the job the sandbox runs: only tokens made
200// for it are accepted on this channel, besides the checks of every token. The
201// connection caps apply as on any listener, and so does WithBandwidth when given:
202// one limiter per call, shared by all the connections of the channel, so that
203// the sandbox's cache traffic as a whole stays under its rate. Serve takes
204// ownership of ln: Stop (or CloseAll) closes it, and so does a failed Serve.
205func (s *Service) Serve(id string, ln net.Listener, job string, opts ...ServeOption) error {
206 var sc serveConfig
207 for _, o := range opts {
208 o(&sc)
209 }
210 if job == "" {
211 _ = ln.Close()
212 return errors.New("cache: a channel needs the job it serves")
213 }
214 s.mu.Lock()
215 defer s.mu.Unlock()
216 if s.closed {
217 _ = ln.Close()
218 return errors.New("cache: service closed")
219 }
220 if _, ok := s.channels[id]; ok {
221 _ = ln.Close()
222 return fmt.Errorf("cache: already serving %s", id)
223 }
224 s.channels[id] = s.start(id, ln, job, newBandwidth(sc.bytesPerSec))
225 return nil
226}
227
228// Stop ends the service of the channel of sandbox id, closing its listener and
229// every connection. Unknown ids are ignored.
230func (s *Service) Stop(id string) {
231 s.mu.Lock()
232 l := s.channels[id]
233 delete(s.channels, id)
234 s.mu.Unlock()
235 if l != nil {
236 _ = l.srv.Close()
237 <-l.done
238 }
239}
240
241// CloseAll stops every channel and the housekeeping, and writes the index.
242func (s *Service) CloseAll() error {
243 s.mu.Lock()
244 var ids []string
245 for id := range s.channels {
246 ids = append(ids, id)
247 }
248 already := s.closed
249 s.closed = true
250 s.mu.Unlock()
251 for _, id := range ids {
252 s.Stop(id)
253 }
254 if already {
255 return nil
256 }
257 close(s.stop)
258 s.wg.Wait()
259 return s.store.flush()
260}
261
262// ParseMax reads ERMITE_RUNNER_CACHE_MAX: a number of bytes or a size with a K,
263// M, G or T suffix (optionally followed by iB or B). Empty means DefaultMax.
264func ParseMax(v string) (int64, error) {
265 if v == "" {
266 return DefaultMax, nil
267 }
268 s := []byte(v)
269 for len(s) > 0 && (s[len(s)-1] == 'B' || s[len(s)-1] == 'b' || s[len(s)-1] == 'i') {
270 s = s[:len(s)-1]
271 }
272 mult := int64(1)
273 if n := len(s); n > 0 {
274 switch s[n-1] {
275 case 'k', 'K':
276 mult, s = 1<<10, s[:n-1]
277 case 'm', 'M':
278 mult, s = 1<<20, s[:n-1]
279 case 'g', 'G':
280 mult, s = 1<<30, s[:n-1]
281 case 't', 'T':
282 mult, s = 1<<40, s[:n-1]
283 }
284 }
285 n, err := strconv.ParseInt(string(s), 10, 64)
286 if err != nil || n <= 0 || n > (1<<62)/mult {
287 return 0, fmt.Errorf("%s: %q is not a size (examples: 10G, 500M)", EnvMax, v)
288 }
289 return n * mult, nil
290}