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