detached cbcd612

host-side action archive cache (#33)

commit
cbcd612ae47730a85038a6e01393a66083d6edfe
author
brice · · 2026-10-16 08:14 UTC
committer
bnema · · 2026-10-16 08:14 UTC
parent
e244502
signature
SSH UnverifiedSHA256:VOCA5lYARHcbKDZ7GhTV45VmcjWnJ6ZQIBazj17LivA The key is unknown
14 files · +1375 -49
M README.md +2 -0
@@ -88,6 +88,7 @@ The runner serves the cache of `actions/cache`, `setup-go`/`setup-node` with `ca
8888 - **Scopes.** Entries belong to (repository, ref, key, version). A job writes to its write ref only and reads its read refs, in priority order; keys are tried in the order the toolkit sends them in each ref (the primary key exactly, restore keys as prefixes, newest first). Another repository's entries are never visible. A second reservation of the same (repository, ref, key, version) while one exists or is in flight is a `409`.
8989 - **Tokens.** `ACTIONS_RUNTIME_TOKEN` is `base64url(claims).base64url(HMAC-SHA256)` over the job id, repository, write ref, read refs and expiry (job timeout plus 15 min), with a random key made at start and kept in memory only (a restart ends every job anyway). The comparison is constant time; everything wrong (no token, bad MAC, expired, ids of another job or ref) is a bare `403`. The token is not an ermite credential and ermite refuses it. The archive URL carries its own HMAC (entry, repository, issuing job, expiry) because the toolkit downloads without the bearer token; a sandbox's channel serves only URLs issued to its own job, so a URL taken out of one sandbox is refused on every other's.
9090 - **Off.** `ERMITE_RUNNER_CACHE=0`.
91+- **Action archives.** The same channel serves the tarballs of actions. When the server resolves an action to a commit, the host downloads the archive once into `<state dir>/actions-cache/` (keyed by forge, repository and commit sha, never by ref) and points the sandbox at a signed URL on the cache channel (bound to the job, the key and the expiry); the sandbox extracts it as it would a download, and "Download action repository" is printed on a hit as on a miss. Concurrent jobs share one download; each entry stores its sha256, which is checked on every read (a corrupt entry is dropped and fetched again); least recently used entries are evicted above `ERMITE_RUNNER_ACTIONS_CACHE_MAX` (default `2G`, `0` turns it off), and an entry just handed out is kept for 30 minutes, and downloads in progress are charged against the cap as they are written (an archive that does not fit is not cached and the sandbox downloads it). The sandbox never writes to this cache. Whatever cannot be cached (no cache channel for the job, no sha, a failed host download) is downloaded by the sandbox as before.
9192
9293 **Scopes from the server.** The server sends `Job.cache {repository, write_ref, read_refs}` (feature `ci.cache`). The runner checks it (`internal/host/cache.go`): the repository must be the job's own, and the scope always carries the job's canonical `owner/name`; refs are full `refs/…` names, empty ones are dropped, at most 8 read refs; an empty `write_ref` makes the job read-only (reserve answers `403 cache write denied:`). A scope that fails the checks disables the cache for that job with a log line; the job runs. **Old server fallback:** with no `Job.cache` the scope is `github.repository` and `Job.ref`, for reading and writing, with no base or default branch fallback; a job without either gets no cache.
9394
@@ -348,6 +349,7 @@ the moment it starts.
348349 | `ERMITE_RUNNER_GUEST_FORGE` | `0` | `1` lets sandboxes reach the host and port of `ERMITE_RUNNER_URL` on the host (staging only, see "Guest forge access") |
349350 | `ERMITE_RUNNER_CACHE` | `1` | `0` turns the cache service off (see "Cache") |
350351 | `ERMITE_RUNNER_CACHE_MAX` | `10G` | size cap of the cache, LRU eviction |
352+| `ERMITE_RUNNER_ACTIONS_CACHE_MAX` | `2G` | size cap of the action archive cache; `0` turns it off |
351353 | `ERMITE_RUNNER_NODE` | `1` | `0` turns the Node.js externals off (see "Externals") |
352354 | `ERMITE_RUNNER_LOCK_FILE` | `/run/ermite-runner.lock` | host-wide single-instance lock (tests only) |
353355 | `ERMITE_RUNNER_JOB_DISK_BANDWIDTH`, `_JOB_DISK_IOPS`, `_JOB_NET_BANDWIDTH` | `400`, `20000`, `1000` | VM rate limits in MiB/s, IOPS and Mbit/s; 0 disables |
@@ -88,6 +88,7 @@ The runner serves the cache of `actions/cache`, `setup-go`/`setup-node` with `ca
88 - **Scopes.** Entries belong to (repository, ref, key, version). A job writes to its write ref only and reads its read refs, in priority order; keys are tried in the order the toolkit sends them in each ref (the primary key exactly, restore keys as prefixes, newest first). Another repository's entries are never visible. A second reservation of the same (repository, ref, key, version) while one exists or is in flight is a `409`.88 - **Scopes.** Entries belong to (repository, ref, key, version). A job writes to its write ref only and reads its read refs, in priority order; keys are tried in the order the toolkit sends them in each ref (the primary key exactly, restore keys as prefixes, newest first). Another repository's entries are never visible. A second reservation of the same (repository, ref, key, version) while one exists or is in flight is a `409`.
89 - **Tokens.** `ACTIONS_RUNTIME_TOKEN` is `base64url(claims).base64url(HMAC-SHA256)` over the job id, repository, write ref, read refs and expiry (job timeout plus 15 min), with a random key made at start and kept in memory only (a restart ends every job anyway). The comparison is constant time; everything wrong (no token, bad MAC, expired, ids of another job or ref) is a bare `403`. The token is not an ermite credential and ermite refuses it. The archive URL carries its own HMAC (entry, repository, issuing job, expiry) because the toolkit downloads without the bearer token; a sandbox's channel serves only URLs issued to its own job, so a URL taken out of one sandbox is refused on every other's.89 - **Tokens.** `ACTIONS_RUNTIME_TOKEN` is `base64url(claims).base64url(HMAC-SHA256)` over the job id, repository, write ref, read refs and expiry (job timeout plus 15 min), with a random key made at start and kept in memory only (a restart ends every job anyway). The comparison is constant time; everything wrong (no token, bad MAC, expired, ids of another job or ref) is a bare `403`. The token is not an ermite credential and ermite refuses it. The archive URL carries its own HMAC (entry, repository, issuing job, expiry) because the toolkit downloads without the bearer token; a sandbox's channel serves only URLs issued to its own job, so a URL taken out of one sandbox is refused on every other's.
90 - **Off.** `ERMITE_RUNNER_CACHE=0`.90 - **Off.** `ERMITE_RUNNER_CACHE=0`.
91+- **Action archives.** The same channel serves the tarballs of actions. When the server resolves an action to a commit, the host downloads the archive once into `<state dir>/actions-cache/` (keyed by forge, repository and commit sha, never by ref) and points the sandbox at a signed URL on the cache channel (bound to the job, the key and the expiry); the sandbox extracts it as it would a download, and "Download action repository" is printed on a hit as on a miss. Concurrent jobs share one download; each entry stores its sha256, which is checked on every read (a corrupt entry is dropped and fetched again); least recently used entries are evicted above `ERMITE_RUNNER_ACTIONS_CACHE_MAX` (default `2G`, `0` turns it off), and an entry just handed out is kept for 30 minutes, and downloads in progress are charged against the cap as they are written (an archive that does not fit is not cached and the sandbox downloads it). The sandbox never writes to this cache. Whatever cannot be cached (no cache channel for the job, no sha, a failed host download) is downloaded by the sandbox as before.
91 92
92 **Scopes from the server.** The server sends `Job.cache {repository, write_ref, read_refs}` (feature `ci.cache`). The runner checks it (`internal/host/cache.go`): the repository must be the job's own, and the scope always carries the job's canonical `owner/name`; refs are full `refs/…` names, empty ones are dropped, at most 8 read refs; an empty `write_ref` makes the job read-only (reserve answers `403 cache write denied:`). A scope that fails the checks disables the cache for that job with a log line; the job runs. **Old server fallback:** with no `Job.cache` the scope is `github.repository` and `Job.ref`, for reading and writing, with no base or default branch fallback; a job without either gets no cache.93 **Scopes from the server.** The server sends `Job.cache {repository, write_ref, read_refs}` (feature `ci.cache`). The runner checks it (`internal/host/cache.go`): the repository must be the job's own, and the scope always carries the job's canonical `owner/name`; refs are full `refs/…` names, empty ones are dropped, at most 8 read refs; an empty `write_ref` makes the job read-only (reserve answers `403 cache write denied:`). A scope that fails the checks disables the cache for that job with a log line; the job runs. **Old server fallback:** with no `Job.cache` the scope is `github.repository` and `Job.ref`, for reading and writing, with no base or default branch fallback; a job without either gets no cache.
93 94
@@ -348,6 +349,7 @@ the moment it starts.
348 | `ERMITE_RUNNER_GUEST_FORGE` | `0` | `1` lets sandboxes reach the host and port of `ERMITE_RUNNER_URL` on the host (staging only, see "Guest forge access") |349 | `ERMITE_RUNNER_GUEST_FORGE` | `0` | `1` lets sandboxes reach the host and port of `ERMITE_RUNNER_URL` on the host (staging only, see "Guest forge access") |
349 | `ERMITE_RUNNER_CACHE` | `1` | `0` turns the cache service off (see "Cache") |350 | `ERMITE_RUNNER_CACHE` | `1` | `0` turns the cache service off (see "Cache") |
350 | `ERMITE_RUNNER_CACHE_MAX` | `10G` | size cap of the cache, LRU eviction |351 | `ERMITE_RUNNER_CACHE_MAX` | `10G` | size cap of the cache, LRU eviction |
352+| `ERMITE_RUNNER_ACTIONS_CACHE_MAX` | `2G` | size cap of the action archive cache; `0` turns it off |
351 | `ERMITE_RUNNER_NODE` | `1` | `0` turns the Node.js externals off (see "Externals") |353 | `ERMITE_RUNNER_NODE` | `1` | `0` turns the Node.js externals off (see "Externals") |
352 | `ERMITE_RUNNER_LOCK_FILE` | `/run/ermite-runner.lock` | host-wide single-instance lock (tests only) |354 | `ERMITE_RUNNER_LOCK_FILE` | `/run/ermite-runner.lock` | host-wide single-instance lock (tests only) |
353 | `ERMITE_RUNNER_JOB_DISK_BANDWIDTH`, `_JOB_DISK_IOPS`, `_JOB_NET_BANDWIDTH` | `400`, `20000`, `1000` | VM rate limits in MiB/s, IOPS and Mbit/s; 0 disables |355 | `ERMITE_RUNNER_JOB_DISK_BANDWIDTH`, `_JOB_DISK_IOPS`, `_JOB_NET_BANDWIDTH` | `400`, `20000`, `1000` | VM rate limits in MiB/s, IOPS and Mbit/s; 0 disables |
M host_cmd.go +35 -8
@@ -17,6 +17,7 @@ import (
1717
1818 "bnema.dev/ermite-api/gen/ermite/ermiteconnect"
1919
20+ "bnema.dev/ermite-runner/internal/actionscache"
2021 "bnema.dev/ermite-runner/internal/cachesvc"
2122 "bnema.dev/ermite-runner/internal/externals"
2223 "bnema.dev/ermite-runner/internal/host"
@@ -39,27 +40,52 @@ func loadHostConfig() (sandbox.Config, host.Config, error) {
3940 // setupCache starts the cache service of the runner (README, "Cache") and hands
4041 // it to the sandbox drivers through scfg. ERMITE_RUNNER_CACHE=0 turns it off;
4142 // ERMITE_RUNNER_CACHE_MAX is the size cap (default 10 GiB).
42-func setupCache(ctx context.Context, scfg *sandbox.Config) (*cachesvc.Service, error) {
43+func setupCache(ctx context.Context, scfg *sandbox.Config) (*cachesvc.Service, *actionscache.Cache, error) {
4344 log := zerowrap.FromCtx(ctx)
4445 switch strings.ToLower(strings.TrimSpace(os.Getenv(envCache))) {
4546 case "", "1", "true", "yes":
4647 case "0", "false", "no":
4748 log.Info().Msg("the cache service is off (" + envCache + "=0)")
48- return nil, nil
49+ return nil, nil, nil
4950 default:
50- return nil, fmt.Errorf("%s: %q is not 0 or 1", envCache, os.Getenv(envCache))
51+ return nil, nil, fmt.Errorf("%s: %q is not 0 or 1", envCache, os.Getenv(envCache))
5152 }
5253 max, err := cachesvc.ParseMax(os.Getenv(cachesvc.EnvMax))
5354 if err != nil {
54- return nil, err
55+ return nil, nil, err
5556 }
56- svc, err := cachesvc.New(cachesvc.Config{Dir: filepath.Join(scfg.StateDir, "cache"), Max: max, Log: log})
57+ archives, err := setupArchives(ctx, scfg.StateDir)
5758 if err != nil {
58- return nil, err
59+ return nil, nil, err
60+ }
61+ svc, err := cachesvc.New(cachesvc.Config{Dir: filepath.Join(scfg.StateDir, "cache"), Max: max, Log: log, Archives: archives})
62+ if err != nil {
63+ return nil, nil, err
5964 }
6065 scfg.Cache = svc
6166 log.Info().Int64("max_bytes", max).Msg("cache service on (guest loopback only, no host port)")
62- return svc, nil
67+ return svc, archives, nil
68+}
69+
70+// setupArchives opens the cache of action archives (ERMITE_RUNNER_ACTIONS_CACHE_MAX,
71+// default 2 GiB, 0 turns it off). It is served on the channel of the cache
72+// service, so it is only used when that is on.
73+func setupArchives(ctx context.Context, stateDir string) (*actionscache.Cache, error) {
74+ log := zerowrap.FromCtx(ctx)
75+ max, err := actionscache.ParseMax(os.Getenv(actionscache.EnvMax))
76+ if err != nil {
77+ return nil, err
78+ }
79+ if max == 0 {
80+ log.Info().Msg("the action archive cache is off (" + actionscache.EnvMax + "=0)")
81+ return nil, nil
82+ }
83+ c, err := actionscache.New(actionscache.Config{Dir: filepath.Join(stateDir, "actions-cache"), Max: max, Log: log})
84+ if err != nil {
85+ return nil, err
86+ }
87+ log.Info().Int64("max_bytes", max).Msg("action archive cache on")
88+ return c, nil
6389 }
6490
6591 // envCache=0 turns the cache service off.
@@ -176,7 +202,7 @@ func runCmd(ctx context.Context) int {
176202 hcfg.GuestForgeOrigin = gf.Origin
177203 }
178204 setupExternals(ctx, &scfg)
179- cache, err := setupCache(ctx, &scfg)
205+ cache, archives, err := setupCache(ctx, &scfg)
180206 if err != nil {
181207 fmt.Fprintln(os.Stderr, "ermite-runner:", err)
182208 return 2
@@ -201,6 +227,7 @@ func runCmd(ctx context.Context) int {
201227 },
202228 Features: hcfg.ServerFeaturesFunc(url),
203229 Cache: cache,
230+ Archives: archives,
204231 }
205232 r.Cfg.URL = url
206233 if err := r.Run(ctx); err != nil {
@@ -17,6 +17,7 @@ import (
17 17
18 "bnema.dev/ermite-api/gen/ermite/ermiteconnect"18 "bnema.dev/ermite-api/gen/ermite/ermiteconnect"
19 19
20+ "bnema.dev/ermite-runner/internal/actionscache"
20 "bnema.dev/ermite-runner/internal/cachesvc"21 "bnema.dev/ermite-runner/internal/cachesvc"
21 "bnema.dev/ermite-runner/internal/externals"22 "bnema.dev/ermite-runner/internal/externals"
22 "bnema.dev/ermite-runner/internal/host"23 "bnema.dev/ermite-runner/internal/host"
@@ -39,27 +40,52 @@ func loadHostConfig() (sandbox.Config, host.Config, error) {
39 // setupCache starts the cache service of the runner (README, "Cache") and hands40 // setupCache starts the cache service of the runner (README, "Cache") and hands
40 // it to the sandbox drivers through scfg. ERMITE_RUNNER_CACHE=0 turns it off;41 // it to the sandbox drivers through scfg. ERMITE_RUNNER_CACHE=0 turns it off;
41 // ERMITE_RUNNER_CACHE_MAX is the size cap (default 10 GiB).42 // ERMITE_RUNNER_CACHE_MAX is the size cap (default 10 GiB).
42-func setupCache(ctx context.Context, scfg *sandbox.Config) (*cachesvc.Service, error) {43+func setupCache(ctx context.Context, scfg *sandbox.Config) (*cachesvc.Service, *actionscache.Cache, error) {
43 log := zerowrap.FromCtx(ctx)44 log := zerowrap.FromCtx(ctx)
44 switch strings.ToLower(strings.TrimSpace(os.Getenv(envCache))) {45 switch strings.ToLower(strings.TrimSpace(os.Getenv(envCache))) {
45 case "", "1", "true", "yes":46 case "", "1", "true", "yes":
46 case "0", "false", "no":47 case "0", "false", "no":
47 log.Info().Msg("the cache service is off (" + envCache + "=0)")48 log.Info().Msg("the cache service is off (" + envCache + "=0)")
48- return nil, nil49+ return nil, nil, nil
49 default:50 default:
50- return nil, fmt.Errorf("%s: %q is not 0 or 1", envCache, os.Getenv(envCache))51+ return nil, nil, fmt.Errorf("%s: %q is not 0 or 1", envCache, os.Getenv(envCache))
51 }52 }
52 max, err := cachesvc.ParseMax(os.Getenv(cachesvc.EnvMax))53 max, err := cachesvc.ParseMax(os.Getenv(cachesvc.EnvMax))
53 if err != nil {54 if err != nil {
54- return nil, err55+ return nil, nil, err
55 }56 }
56- svc, err := cachesvc.New(cachesvc.Config{Dir: filepath.Join(scfg.StateDir, "cache"), Max: max, Log: log})57+ archives, err := setupArchives(ctx, scfg.StateDir)
57 if err != nil {58 if err != nil {
58- return nil, err59+ return nil, nil, err
60+ }
61+ svc, err := cachesvc.New(cachesvc.Config{Dir: filepath.Join(scfg.StateDir, "cache"), Max: max, Log: log, Archives: archives})
62+ if err != nil {
63+ return nil, nil, err
59 }64 }
60 scfg.Cache = svc65 scfg.Cache = svc
61 log.Info().Int64("max_bytes", max).Msg("cache service on (guest loopback only, no host port)")66 log.Info().Int64("max_bytes", max).Msg("cache service on (guest loopback only, no host port)")
62- return svc, nil67+ return svc, archives, nil
68+}
69+
70+// setupArchives opens the cache of action archives (ERMITE_RUNNER_ACTIONS_CACHE_MAX,
71+// default 2 GiB, 0 turns it off). It is served on the channel of the cache
72+// service, so it is only used when that is on.
73+func setupArchives(ctx context.Context, stateDir string) (*actionscache.Cache, error) {
74+ log := zerowrap.FromCtx(ctx)
75+ max, err := actionscache.ParseMax(os.Getenv(actionscache.EnvMax))
76+ if err != nil {
77+ return nil, err
78+ }
79+ if max == 0 {
80+ log.Info().Msg("the action archive cache is off (" + actionscache.EnvMax + "=0)")
81+ return nil, nil
82+ }
83+ c, err := actionscache.New(actionscache.Config{Dir: filepath.Join(stateDir, "actions-cache"), Max: max, Log: log})
84+ if err != nil {
85+ return nil, err
86+ }
87+ log.Info().Int64("max_bytes", max).Msg("action archive cache on")
88+ return c, nil
63 }89 }
64 90
65 // envCache=0 turns the cache service off.91 // envCache=0 turns the cache service off.
@@ -176,7 +202,7 @@ func runCmd(ctx context.Context) int {
176 hcfg.GuestForgeOrigin = gf.Origin202 hcfg.GuestForgeOrigin = gf.Origin
177 }203 }
178 setupExternals(ctx, &scfg)204 setupExternals(ctx, &scfg)
179- cache, err := setupCache(ctx, &scfg)205+ cache, archives, err := setupCache(ctx, &scfg)
180 if err != nil {206 if err != nil {
181 fmt.Fprintln(os.Stderr, "ermite-runner:", err)207 fmt.Fprintln(os.Stderr, "ermite-runner:", err)
182 return 2208 return 2
@@ -201,6 +227,7 @@ func runCmd(ctx context.Context) int {
201 },227 },
202 Features: hcfg.ServerFeaturesFunc(url),228 Features: hcfg.ServerFeaturesFunc(url),
203 Cache: cache,229 Cache: cache,
230+ Archives: archives,
204 }231 }
205 r.Cfg.URL = url232 r.Cfg.URL = url
206 if err := r.Run(ctx); err != nil {233 if err := r.Run(ctx); err != nil {
A internal/actionscache/cache.go +448 -0
@@ -0,0 +1,448 @@
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.
24+package actionscache
25+
26+import (
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.
45+const (
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.
60+var ErrNotFound = errors.New("actionscache: no such entry")
61+
62+// ErrTooLarge is an archive that does not fit under the cap on its own.
63+var ErrTooLarge = errors.New("actionscache: archive larger than the cache")
64+
65+// Config configures a Cache.
66+type 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+
74+type entry struct {
75+ size int64
76+ used time.Time
77+ pinUntil time.Time
78+}
79+
80+type flight struct {
81+ done chan struct{}
82+ err error
83+}
84+
85+// Cache is the archive cache.
86+type 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.
100+func 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.
146+func 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+
161+func 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.
170+func ValidKey(k string) bool { return validKey(k) }
171+
172+func (c *Cache) archivePath(key string) string { return filepath.Join(c.dir, key+archiveExt) }
173+func (c *Cache) sumPath(key string) string { return filepath.Join(c.dir, key+sumExt) }
174+
175+func (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.
181+type 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.
187+func (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.
227+func (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.
285+type countWriter struct {
286+ w io.Writer
287+ c *Cache
288+ n int64
289+}
290+
291+func (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.
317+func (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+
344+func (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.
357+func (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.
372+func (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.
395+func (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).
412+func (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).
421+func 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).
448+func (c *Cache) Dir() string { return c.dir }
@@ -0,0 +1,448 @@
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.
24+package actionscache
25+
26+import (
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.
45+const (
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.
60+var ErrNotFound = errors.New("actionscache: no such entry")
61+
62+// ErrTooLarge is an archive that does not fit under the cap on its own.
63+var ErrTooLarge = errors.New("actionscache: archive larger than the cache")
64+
65+// Config configures a Cache.
66+type 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+
74+type entry struct {
75+ size int64
76+ used time.Time
77+ pinUntil time.Time
78+}
79+
80+type flight struct {
81+ done chan struct{}
82+ err error
83+}
84+
85+// Cache is the archive cache.
86+type 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.
100+func 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.
146+func 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+
161+func 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.
170+func ValidKey(k string) bool { return validKey(k) }
171+
172+func (c *Cache) archivePath(key string) string { return filepath.Join(c.dir, key+archiveExt) }
173+func (c *Cache) sumPath(key string) string { return filepath.Join(c.dir, key+sumExt) }
174+
175+func (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.
181+type 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.
187+func (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.
227+func (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.
285+type countWriter struct {
286+ w io.Writer
287+ c *Cache
288+ n int64
289+}
290+
291+func (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.
317+func (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+
344+func (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.
357+func (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.
372+func (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.
395+func (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).
412+func (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).
421+func 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).
448+func (c *Cache) Dir() string { return c.dir }
A internal/actionscache/cache_test.go +398 -0
@@ -0,0 +1,398 @@
1+// SPDX-License-Identifier: MIT
2+
3+package actionscache
4+
5+import (
6+ "context"
7+ "errors"
8+ "io"
9+ "os"
10+ "strings"
11+ "sync"
12+ "sync/atomic"
13+ "testing"
14+ "time"
15+
16+ "github.com/bnema/zerowrap"
17+)
18+
19+func newCache(t *testing.T, max int64) (*Cache, *time.Time) {
20+ t.Helper()
21+ now := time.Date(2026, 10, 16, 0, 0, 0, 0, time.UTC)
22+ c, err := New(Config{Dir: t.TempDir(), Max: max, Now: func() time.Time { return now }, Log: zerowrap.Logger{}})
23+ if err != nil {
24+ t.Fatal(err)
25+ }
26+ return c, &now
27+}
28+
29+func keyOf(t *testing.T, n int) string {
30+ t.Helper()
31+ k, ok := Key("https://github.com", "o/r"+strings.Repeat("x", n), strings.Repeat("a", 40))
32+ if !ok {
33+ t.Fatal("key")
34+ }
35+ return k
36+}
37+
38+func body(s string) Fetch {
39+ return func(_ context.Context, w io.Writer) error { _, err := io.WriteString(w, s); return err }
40+}
41+
42+func read(t *testing.T, c *Cache, key string) string {
43+ t.Helper()
44+ f, err := c.Open(key)
45+ if err != nil {
46+ t.Fatal(err)
47+ }
48+ defer func() { _ = f.Close() }()
49+ b, _ := io.ReadAll(f)
50+ return string(b)
51+}
52+
53+func TestHitMiss(t *testing.T) {
54+ c, _ := newCache(t, 1<<20)
55+ k := keyOf(t, 0)
56+ var fetches int
57+ f := func(ctx context.Context, w io.Writer) error { fetches++; return body("archive")(ctx, w) }
58+ if hit, err := c.Ensure(context.Background(), k, f); err != nil || hit {
59+ t.Fatalf("first: hit=%v err=%v", hit, err)
60+ }
61+ if hit, err := c.Ensure(context.Background(), k, f); err != nil || !hit {
62+ t.Fatalf("second: hit=%v err=%v", hit, err)
63+ }
64+ if fetches != 1 || read(t, c, k) != "archive" {
65+ t.Fatalf("fetches=%d", fetches)
66+ }
67+ // A new process finds the entry.
68+ c2, err := New(Config{Dir: c.dir, Max: 1 << 20})
69+ if err != nil {
70+ t.Fatal(err)
71+ }
72+ if hit, _ := c2.Ensure(context.Background(), k, f); !hit || fetches != 1 {
73+ t.Fatalf("restart: hit=%v fetches=%d", hit, fetches)
74+ }
75+}
76+
77+func TestFailedFetchLeavesNothing(t *testing.T) {
78+ c, _ := newCache(t, 1<<20)
79+ k := keyOf(t, 0)
80+ boom := errors.New("429")
81+ _, err := c.Ensure(context.Background(), k, func(_ context.Context, w io.Writer) error { _, _ = io.WriteString(w, "part"); return boom })
82+ if !errors.Is(err, boom) {
83+ t.Fatal(err)
84+ }
85+ if _, err := c.Open(k); !errors.Is(err, ErrNotFound) {
86+ t.Fatal("a failed download must not be an entry")
87+ }
88+ des, _ := os.ReadDir(c.dir)
89+ if len(des) != 0 {
90+ t.Fatalf("leftovers: %v", des)
91+ }
92+}
93+
94+func TestConcurrentSingleFlight(t *testing.T) {
95+ c, _ := newCache(t, 1<<20)
96+ k := keyOf(t, 0)
97+ var fetches atomic.Int32
98+ release := make(chan struct{})
99+ f := func(ctx context.Context, w io.Writer) error {
100+ fetches.Add(1)
101+ <-release
102+ return body("once")(ctx, w)
103+ }
104+ var wg sync.WaitGroup
105+ var hits atomic.Int32
106+ for range 8 {
107+ wg.Add(1)
108+ go func() {
109+ defer wg.Done()
110+ hit, err := c.Ensure(context.Background(), k, f)
111+ if err != nil {
112+ t.Error(err)
113+ }
114+ if hit {
115+ hits.Add(1)
116+ }
117+ }()
118+ }
119+ time.Sleep(50 * time.Millisecond)
120+ close(release)
121+ wg.Wait()
122+ if fetches.Load() != 1 {
123+ t.Fatalf("%d downloads for one key", fetches.Load())
124+ }
125+ if hits.Load() != 7 {
126+ t.Errorf("%d hits, want 7 (all but the downloader)", hits.Load())
127+ }
128+}
129+
130+func TestCorruptEntryIsDroppedAndRefetched(t *testing.T) {
131+ c, _ := newCache(t, 1<<20)
132+ k := keyOf(t, 0)
133+ if _, err := c.Ensure(context.Background(), k, body("good bytes")); err != nil {
134+ t.Fatal(err)
135+ }
136+ // Same size, other content: only the hash can tell.
137+ if err := os.WriteFile(c.archivePath(k), []byte("evil bytes"), 0o600); err != nil {
138+ t.Fatal(err)
139+ }
140+ if _, err := c.Open(k); !errors.Is(err, ErrNotFound) {
141+ t.Fatalf("a corrupt entry was served: %v", err)
142+ }
143+ if c.Size() != 0 {
144+ t.Errorf("size %d after the drop", c.Size())
145+ }
146+ hit, err := c.Ensure(context.Background(), k, body("good bytes"))
147+ if err != nil || hit {
148+ t.Fatalf("hit=%v err=%v", hit, err)
149+ }
150+ if read(t, c, k) != "good bytes" {
151+ t.Fatal("not refetched")
152+ }
153+ // Truncation is caught too.
154+ if err := os.WriteFile(c.archivePath(k), []byte("good"), 0o600); err != nil {
155+ t.Fatal(err)
156+ }
157+ if _, err := c.Open(k); !errors.Is(err, ErrNotFound) {
158+ t.Fatal("a truncated entry was served")
159+ }
160+}
161+
162+func TestLRUEviction(t *testing.T) {
163+ c, now := newCache(t, 25)
164+ ctx := context.Background()
165+ step := func() { *now = now.Add(time.Hour + PinFor) } // past every pin
166+ a, b, d := keyOf(t, 0), keyOf(t, 1), keyOf(t, 2)
167+ for _, k := range []string{a, b} {
168+ if _, err := c.Ensure(ctx, k, body(strings.Repeat("x", 10))); err != nil {
169+ t.Fatal(err)
170+ }
171+ step()
172+ }
173+ // Use a again: b is now the least recently used.
174+ if hit, _ := c.Ensure(ctx, a, body("unused")); !hit {
175+ t.Fatal("a should hit")
176+ }
177+ step()
178+ if _, err := c.Ensure(ctx, d, body(strings.Repeat("y", 10))); err != nil {
179+ t.Fatal(err)
180+ }
181+ if _, err := c.Open(b); !errors.Is(err, ErrNotFound) {
182+ t.Error("the least recently used entry survived")
183+ }
184+ for _, k := range []string{a, d} {
185+ if _, err := c.Open(k); err != nil {
186+ t.Errorf("%s evicted: %v", k[:6], err)
187+ }
188+ }
189+ if c.Size() > 25 {
190+ t.Errorf("size %d over the cap", c.Size())
191+ }
192+}
193+
194+// A pinned entry is not evicted, even to make room: the newcomer is the one
195+// that is not cached, so the cap holds.
196+func TestPinnedEntryIsNotEvicted(t *testing.T) {
197+ c, _ := newCache(t, 15)
198+ ctx := context.Background()
199+ a, b := keyOf(t, 0), keyOf(t, 1)
200+ if _, err := c.Ensure(ctx, a, body(strings.Repeat("x", 10))); err != nil {
201+ t.Fatal(err)
202+ }
203+ if _, err := c.Ensure(ctx, b, body(strings.Repeat("y", 10))); !errors.Is(err, ErrTooLarge) {
204+ t.Fatalf("got %v", err)
205+ }
206+ if _, err := c.Open(a); err != nil {
207+ t.Fatalf("a pinned entry was evicted: %v", err)
208+ }
209+ if c.Size() > 15 {
210+ t.Fatalf("size %d", c.Size())
211+ }
212+}
213+
214+// diskUse is the bytes of every file in the cache directory, temporary files
215+// included.
216+func diskUse(t *testing.T, c *Cache) int64 {
217+ t.Helper()
218+ var n int64
219+ des, _ := os.ReadDir(c.dir)
220+ for _, de := range des {
221+ if fi, err := de.Info(); err == nil {
222+ n += fi.Size()
223+ }
224+ }
225+ return n
226+}
227+
228+// Many downloads in flight at once, each below the cap, are charged as they
229+// are written: the directory never holds more than the cap, temporary files
230+// included, and the accounting returns to the bytes of the entries.
231+func TestConcurrentDownloadsStayUnderTheCap(t *testing.T) {
232+ const max = 100
233+ c, _ := newCache(t, max)
234+ var over atomic.Int32
235+ var wg sync.WaitGroup
236+ stop := make(chan struct{})
237+ go func() {
238+ for {
239+ select {
240+ case <-stop:
241+ return
242+ default:
243+ // The charge covers the entries and the downloads in progress. (The
244+ // directory cannot be sampled atomically: a listing taken while an
245+ // entry is evicted and a download grows would count both.)
246+ c.mu.Lock()
247+ n := c.total + c.pending
248+ c.mu.Unlock()
249+ if n > max {
250+ over.Add(1)
251+ }
252+ }
253+ }
254+ }()
255+ for i := range 16 {
256+ wg.Add(1)
257+ go func() {
258+ defer wg.Done()
259+ k := keyOf(t, i)
260+ _, _ = c.Ensure(context.Background(), k, func(_ context.Context, w io.Writer) error {
261+ for range 6 { // 6 x 10 bytes, in pieces
262+ if _, err := w.Write([]byte(strings.Repeat("z", 10))); err != nil {
263+ return err
264+ }
265+ time.Sleep(time.Millisecond)
266+ }
267+ return nil
268+ })
269+ }()
270+ }
271+ wg.Wait()
272+ close(stop)
273+ if over.Load() != 0 {
274+ t.Fatal("entries and downloads in flight went over the cap")
275+ }
276+ var onDisk int64
277+ des, _ := os.ReadDir(c.dir)
278+ for _, de := range des {
279+ if strings.HasPrefix(de.Name(), tmpPrefix) {
280+ t.Errorf("temporary file left: %s", de.Name())
281+ }
282+ if strings.HasSuffix(de.Name(), archiveExt) {
283+ fi, _ := de.Info()
284+ onDisk += fi.Size()
285+ }
286+ }
287+ if onDisk > max {
288+ t.Fatalf("%d bytes on disk", onDisk)
289+ }
290+ c.mu.Lock()
291+ pending, total := c.pending, c.total
292+ c.mu.Unlock()
293+ if pending != 0 || total > max {
294+ t.Fatalf("pending=%d total=%d", pending, total)
295+ }
296+ var sum int64
297+ for _, e := range c.entries {
298+ sum += e.size
299+ }
300+ if sum != total {
301+ t.Fatalf("entries hold %d, accounted %d", sum, total)
302+ }
303+}
304+
305+// The bytes of a download in progress are counted, and a failed or refused one
306+// leaves neither a file nor a charge.
307+func TestTemporaryFilesAreCountedAndCleaned(t *testing.T) {
308+ c, _ := newCache(t, 50)
309+ seen := int64(-1)
310+ _, err := c.Ensure(context.Background(), keyOf(t, 0), func(_ context.Context, w io.Writer) error {
311+ _, _ = io.WriteString(w, strings.Repeat("a", 30))
312+ c.mu.Lock()
313+ seen = c.pending
314+ c.mu.Unlock()
315+ return errors.New("connection reset")
316+ })
317+ if err == nil || seen != 30 {
318+ t.Fatalf("err=%v pending during=%d", err, seen)
319+ }
320+ // Refused midway for size: the part already written is released too.
321+ _, err = c.Ensure(context.Background(), keyOf(t, 1), body(strings.Repeat("b", 60)))
322+ if !errors.Is(err, ErrTooLarge) {
323+ t.Fatalf("got %v", err)
324+ }
325+ if c.pending != 0 || c.Size() != 0 || diskUse(t, c) != 0 {
326+ t.Fatalf("pending=%d size=%d disk=%d", c.pending, c.Size(), diskUse(t, c))
327+ }
328+ // The size is what was written, whatever the server announced.
329+ if _, err := c.Ensure(context.Background(), keyOf(t, 2), body(strings.Repeat("c", 50))); err != nil {
330+ t.Fatal(err)
331+ }
332+ if c.Size() != 50 {
333+ t.Fatalf("size %d", c.Size())
334+ }
335+}
336+
337+func TestTooLarge(t *testing.T) {
338+ c, _ := newCache(t, 10)
339+ _, err := c.Ensure(context.Background(), keyOf(t, 0), body(strings.Repeat("x", 11)))
340+ if !errors.Is(err, ErrTooLarge) {
341+ t.Fatalf("got %v", err)
342+ }
343+ if c.Size() != 0 || diskUse(t, c) != 0 {
344+ t.Fatalf("an oversize archive left %d bytes on disk", diskUse(t, c))
345+ }
346+}
347+
348+func TestStartDropsHalfWrittenFiles(t *testing.T) {
349+ c, _ := newCache(t, 100)
350+ k := keyOf(t, 0)
351+ _ = os.WriteFile(c.archivePath(k), []byte("no sum"), 0o600)
352+ _ = os.WriteFile(c.dir+"/tmp-123", []byte("x"), 0o600)
353+ _ = os.WriteFile(c.sumPath(keyOf(t, 1)), []byte("sum without archive"), 0o600)
354+ if _, err := New(Config{Dir: c.dir, Max: 100}); err != nil {
355+ t.Fatal(err)
356+ }
357+ if des, _ := os.ReadDir(c.dir); len(des) != 0 {
358+ t.Fatalf("leftovers: %v", des)
359+ }
360+}
361+
362+// The key is made of the forge, the repository and the commit, never a ref: a
363+// tag or a branch name is not a sha and is refused, and two commits or two
364+// forges never share an entry.
365+func TestKeyNeverUsesARef(t *testing.T) {
366+ sha := strings.Repeat("a", 40)
367+ for _, bad := range []string{"", "v4", "main", "refs/tags/v4", strings.Repeat("g", 40), strings.Repeat("a", 39), strings.Repeat("a", 41)} {
368+ if _, ok := Key("https://github.com", "o/r", bad); ok {
369+ t.Errorf("%q made a key", bad)
370+ }
371+ }
372+ k1, _ := Key("https://github.com", "o/r", sha)
373+ k2, _ := Key("https://github.com", "o/r", strings.Repeat("b", 40))
374+ k3, _ := Key("https://ermite.example", "o/r", sha)
375+ k4, _ := Key("https://github.com", "O/R", strings.ToUpper(sha))
376+ if k1 == k2 || k1 == k3 {
377+ t.Error("collision")
378+ }
379+ if k1 != k4 {
380+ t.Error("the case of the repository or the sha must not split an entry")
381+ }
382+ if !ValidKey(k1) {
383+ t.Error("shape")
384+ }
385+}
386+
387+func TestParseMax(t *testing.T) {
388+ for in, want := range map[string]int64{"": DefaultMax, "0": 0, "500M": 500 << 20, "2G": 2 << 30, "2GiB": 2 << 30, "1024": 1024} {
389+ if got, err := ParseMax(in); err != nil || got != want {
390+ t.Errorf("%q: %d %v", in, got, err)
391+ }
392+ }
393+ for _, in := range []string{"-1", "x", "G"} {
394+ if _, err := ParseMax(in); err == nil {
395+ t.Errorf("%q accepted", in)
396+ }
397+ }
398+}
@@ -0,0 +1,398 @@
1+// SPDX-License-Identifier: MIT
2+
3+package actionscache
4+
5+import (
6+ "context"
7+ "errors"
8+ "io"
9+ "os"
10+ "strings"
11+ "sync"
12+ "sync/atomic"
13+ "testing"
14+ "time"
15+
16+ "github.com/bnema/zerowrap"
17+)
18+
19+func newCache(t *testing.T, max int64) (*Cache, *time.Time) {
20+ t.Helper()
21+ now := time.Date(2026, 10, 16, 0, 0, 0, 0, time.UTC)
22+ c, err := New(Config{Dir: t.TempDir(), Max: max, Now: func() time.Time { return now }, Log: zerowrap.Logger{}})
23+ if err != nil {
24+ t.Fatal(err)
25+ }
26+ return c, &now
27+}
28+
29+func keyOf(t *testing.T, n int) string {
30+ t.Helper()
31+ k, ok := Key("https://github.com", "o/r"+strings.Repeat("x", n), strings.Repeat("a", 40))
32+ if !ok {
33+ t.Fatal("key")
34+ }
35+ return k
36+}
37+
38+func body(s string) Fetch {
39+ return func(_ context.Context, w io.Writer) error { _, err := io.WriteString(w, s); return err }
40+}
41+
42+func read(t *testing.T, c *Cache, key string) string {
43+ t.Helper()
44+ f, err := c.Open(key)
45+ if err != nil {
46+ t.Fatal(err)
47+ }
48+ defer func() { _ = f.Close() }()
49+ b, _ := io.ReadAll(f)
50+ return string(b)
51+}
52+
53+func TestHitMiss(t *testing.T) {
54+ c, _ := newCache(t, 1<<20)
55+ k := keyOf(t, 0)
56+ var fetches int
57+ f := func(ctx context.Context, w io.Writer) error { fetches++; return body("archive")(ctx, w) }
58+ if hit, err := c.Ensure(context.Background(), k, f); err != nil || hit {
59+ t.Fatalf("first: hit=%v err=%v", hit, err)
60+ }
61+ if hit, err := c.Ensure(context.Background(), k, f); err != nil || !hit {
62+ t.Fatalf("second: hit=%v err=%v", hit, err)
63+ }
64+ if fetches != 1 || read(t, c, k) != "archive" {
65+ t.Fatalf("fetches=%d", fetches)
66+ }
67+ // A new process finds the entry.
68+ c2, err := New(Config{Dir: c.dir, Max: 1 << 20})
69+ if err != nil {
70+ t.Fatal(err)
71+ }
72+ if hit, _ := c2.Ensure(context.Background(), k, f); !hit || fetches != 1 {
73+ t.Fatalf("restart: hit=%v fetches=%d", hit, fetches)
74+ }
75+}
76+
77+func TestFailedFetchLeavesNothing(t *testing.T) {
78+ c, _ := newCache(t, 1<<20)
79+ k := keyOf(t, 0)
80+ boom := errors.New("429")
81+ _, err := c.Ensure(context.Background(), k, func(_ context.Context, w io.Writer) error { _, _ = io.WriteString(w, "part"); return boom })
82+ if !errors.Is(err, boom) {
83+ t.Fatal(err)
84+ }
85+ if _, err := c.Open(k); !errors.Is(err, ErrNotFound) {
86+ t.Fatal("a failed download must not be an entry")
87+ }
88+ des, _ := os.ReadDir(c.dir)
89+ if len(des) != 0 {
90+ t.Fatalf("leftovers: %v", des)
91+ }
92+}
93+
94+func TestConcurrentSingleFlight(t *testing.T) {
95+ c, _ := newCache(t, 1<<20)
96+ k := keyOf(t, 0)
97+ var fetches atomic.Int32
98+ release := make(chan struct{})
99+ f := func(ctx context.Context, w io.Writer) error {
100+ fetches.Add(1)
101+ <-release
102+ return body("once")(ctx, w)
103+ }
104+ var wg sync.WaitGroup
105+ var hits atomic.Int32
106+ for range 8 {
107+ wg.Add(1)
108+ go func() {
109+ defer wg.Done()
110+ hit, err := c.Ensure(context.Background(), k, f)
111+ if err != nil {
112+ t.Error(err)
113+ }
114+ if hit {
115+ hits.Add(1)
116+ }
117+ }()
118+ }
119+ time.Sleep(50 * time.Millisecond)
120+ close(release)
121+ wg.Wait()
122+ if fetches.Load() != 1 {
123+ t.Fatalf("%d downloads for one key", fetches.Load())
124+ }
125+ if hits.Load() != 7 {
126+ t.Errorf("%d hits, want 7 (all but the downloader)", hits.Load())
127+ }
128+}
129+
130+func TestCorruptEntryIsDroppedAndRefetched(t *testing.T) {
131+ c, _ := newCache(t, 1<<20)
132+ k := keyOf(t, 0)
133+ if _, err := c.Ensure(context.Background(), k, body("good bytes")); err != nil {
134+ t.Fatal(err)
135+ }
136+ // Same size, other content: only the hash can tell.
137+ if err := os.WriteFile(c.archivePath(k), []byte("evil bytes"), 0o600); err != nil {
138+ t.Fatal(err)
139+ }
140+ if _, err := c.Open(k); !errors.Is(err, ErrNotFound) {
141+ t.Fatalf("a corrupt entry was served: %v", err)
142+ }
143+ if c.Size() != 0 {
144+ t.Errorf("size %d after the drop", c.Size())
145+ }
146+ hit, err := c.Ensure(context.Background(), k, body("good bytes"))
147+ if err != nil || hit {
148+ t.Fatalf("hit=%v err=%v", hit, err)
149+ }
150+ if read(t, c, k) != "good bytes" {
151+ t.Fatal("not refetched")
152+ }
153+ // Truncation is caught too.
154+ if err := os.WriteFile(c.archivePath(k), []byte("good"), 0o600); err != nil {
155+ t.Fatal(err)
156+ }
157+ if _, err := c.Open(k); !errors.Is(err, ErrNotFound) {
158+ t.Fatal("a truncated entry was served")
159+ }
160+}
161+
162+func TestLRUEviction(t *testing.T) {
163+ c, now := newCache(t, 25)
164+ ctx := context.Background()
165+ step := func() { *now = now.Add(time.Hour + PinFor) } // past every pin
166+ a, b, d := keyOf(t, 0), keyOf(t, 1), keyOf(t, 2)
167+ for _, k := range []string{a, b} {
168+ if _, err := c.Ensure(ctx, k, body(strings.Repeat("x", 10))); err != nil {
169+ t.Fatal(err)
170+ }
171+ step()
172+ }
173+ // Use a again: b is now the least recently used.
174+ if hit, _ := c.Ensure(ctx, a, body("unused")); !hit {
175+ t.Fatal("a should hit")
176+ }
177+ step()
178+ if _, err := c.Ensure(ctx, d, body(strings.Repeat("y", 10))); err != nil {
179+ t.Fatal(err)
180+ }
181+ if _, err := c.Open(b); !errors.Is(err, ErrNotFound) {
182+ t.Error("the least recently used entry survived")
183+ }
184+ for _, k := range []string{a, d} {
185+ if _, err := c.Open(k); err != nil {
186+ t.Errorf("%s evicted: %v", k[:6], err)
187+ }
188+ }
189+ if c.Size() > 25 {
190+ t.Errorf("size %d over the cap", c.Size())
191+ }
192+}
193+
194+// A pinned entry is not evicted, even to make room: the newcomer is the one
195+// that is not cached, so the cap holds.
196+func TestPinnedEntryIsNotEvicted(t *testing.T) {
197+ c, _ := newCache(t, 15)
198+ ctx := context.Background()
199+ a, b := keyOf(t, 0), keyOf(t, 1)
200+ if _, err := c.Ensure(ctx, a, body(strings.Repeat("x", 10))); err != nil {
201+ t.Fatal(err)
202+ }
203+ if _, err := c.Ensure(ctx, b, body(strings.Repeat("y", 10))); !errors.Is(err, ErrTooLarge) {
204+ t.Fatalf("got %v", err)
205+ }
206+ if _, err := c.Open(a); err != nil {
207+ t.Fatalf("a pinned entry was evicted: %v", err)
208+ }
209+ if c.Size() > 15 {
210+ t.Fatalf("size %d", c.Size())
211+ }
212+}
213+
214+// diskUse is the bytes of every file in the cache directory, temporary files
215+// included.
216+func diskUse(t *testing.T, c *Cache) int64 {
217+ t.Helper()
218+ var n int64
219+ des, _ := os.ReadDir(c.dir)
220+ for _, de := range des {
221+ if fi, err := de.Info(); err == nil {
222+ n += fi.Size()
223+ }
224+ }
225+ return n
226+}
227+
228+// Many downloads in flight at once, each below the cap, are charged as they
229+// are written: the directory never holds more than the cap, temporary files
230+// included, and the accounting returns to the bytes of the entries.
231+func TestConcurrentDownloadsStayUnderTheCap(t *testing.T) {
232+ const max = 100
233+ c, _ := newCache(t, max)
234+ var over atomic.Int32
235+ var wg sync.WaitGroup
236+ stop := make(chan struct{})
237+ go func() {
238+ for {
239+ select {
240+ case <-stop:
241+ return
242+ default:
243+ // The charge covers the entries and the downloads in progress. (The
244+ // directory cannot be sampled atomically: a listing taken while an
245+ // entry is evicted and a download grows would count both.)
246+ c.mu.Lock()
247+ n := c.total + c.pending
248+ c.mu.Unlock()
249+ if n > max {
250+ over.Add(1)
251+ }
252+ }
253+ }
254+ }()
255+ for i := range 16 {
256+ wg.Add(1)
257+ go func() {
258+ defer wg.Done()
259+ k := keyOf(t, i)
260+ _, _ = c.Ensure(context.Background(), k, func(_ context.Context, w io.Writer) error {
261+ for range 6 { // 6 x 10 bytes, in pieces
262+ if _, err := w.Write([]byte(strings.Repeat("z", 10))); err != nil {
263+ return err
264+ }
265+ time.Sleep(time.Millisecond)
266+ }
267+ return nil
268+ })
269+ }()
270+ }
271+ wg.Wait()
272+ close(stop)
273+ if over.Load() != 0 {
274+ t.Fatal("entries and downloads in flight went over the cap")
275+ }
276+ var onDisk int64
277+ des, _ := os.ReadDir(c.dir)
278+ for _, de := range des {
279+ if strings.HasPrefix(de.Name(), tmpPrefix) {
280+ t.Errorf("temporary file left: %s", de.Name())
281+ }
282+ if strings.HasSuffix(de.Name(), archiveExt) {
283+ fi, _ := de.Info()
284+ onDisk += fi.Size()
285+ }
286+ }
287+ if onDisk > max {
288+ t.Fatalf("%d bytes on disk", onDisk)
289+ }
290+ c.mu.Lock()
291+ pending, total := c.pending, c.total
292+ c.mu.Unlock()
293+ if pending != 0 || total > max {
294+ t.Fatalf("pending=%d total=%d", pending, total)
295+ }
296+ var sum int64
297+ for _, e := range c.entries {
298+ sum += e.size
299+ }
300+ if sum != total {
301+ t.Fatalf("entries hold %d, accounted %d", sum, total)
302+ }
303+}
304+
305+// The bytes of a download in progress are counted, and a failed or refused one
306+// leaves neither a file nor a charge.
307+func TestTemporaryFilesAreCountedAndCleaned(t *testing.T) {
308+ c, _ := newCache(t, 50)
309+ seen := int64(-1)
310+ _, err := c.Ensure(context.Background(), keyOf(t, 0), func(_ context.Context, w io.Writer) error {
311+ _, _ = io.WriteString(w, strings.Repeat("a", 30))
312+ c.mu.Lock()
313+ seen = c.pending
314+ c.mu.Unlock()
315+ return errors.New("connection reset")
316+ })
317+ if err == nil || seen != 30 {
318+ t.Fatalf("err=%v pending during=%d", err, seen)
319+ }
320+ // Refused midway for size: the part already written is released too.
321+ _, err = c.Ensure(context.Background(), keyOf(t, 1), body(strings.Repeat("b", 60)))
322+ if !errors.Is(err, ErrTooLarge) {
323+ t.Fatalf("got %v", err)
324+ }
325+ if c.pending != 0 || c.Size() != 0 || diskUse(t, c) != 0 {
326+ t.Fatalf("pending=%d size=%d disk=%d", c.pending, c.Size(), diskUse(t, c))
327+ }
328+ // The size is what was written, whatever the server announced.
329+ if _, err := c.Ensure(context.Background(), keyOf(t, 2), body(strings.Repeat("c", 50))); err != nil {
330+ t.Fatal(err)
331+ }
332+ if c.Size() != 50 {
333+ t.Fatalf("size %d", c.Size())
334+ }
335+}
336+
337+func TestTooLarge(t *testing.T) {
338+ c, _ := newCache(t, 10)
339+ _, err := c.Ensure(context.Background(), keyOf(t, 0), body(strings.Repeat("x", 11)))
340+ if !errors.Is(err, ErrTooLarge) {
341+ t.Fatalf("got %v", err)
342+ }
343+ if c.Size() != 0 || diskUse(t, c) != 0 {
344+ t.Fatalf("an oversize archive left %d bytes on disk", diskUse(t, c))
345+ }
346+}
347+
348+func TestStartDropsHalfWrittenFiles(t *testing.T) {
349+ c, _ := newCache(t, 100)
350+ k := keyOf(t, 0)
351+ _ = os.WriteFile(c.archivePath(k), []byte("no sum"), 0o600)
352+ _ = os.WriteFile(c.dir+"/tmp-123", []byte("x"), 0o600)
353+ _ = os.WriteFile(c.sumPath(keyOf(t, 1)), []byte("sum without archive"), 0o600)
354+ if _, err := New(Config{Dir: c.dir, Max: 100}); err != nil {
355+ t.Fatal(err)
356+ }
357+ if des, _ := os.ReadDir(c.dir); len(des) != 0 {
358+ t.Fatalf("leftovers: %v", des)
359+ }
360+}
361+
362+// The key is made of the forge, the repository and the commit, never a ref: a
363+// tag or a branch name is not a sha and is refused, and two commits or two
364+// forges never share an entry.
365+func TestKeyNeverUsesARef(t *testing.T) {
366+ sha := strings.Repeat("a", 40)
367+ for _, bad := range []string{"", "v4", "main", "refs/tags/v4", strings.Repeat("g", 40), strings.Repeat("a", 39), strings.Repeat("a", 41)} {
368+ if _, ok := Key("https://github.com", "o/r", bad); ok {
369+ t.Errorf("%q made a key", bad)
370+ }
371+ }
372+ k1, _ := Key("https://github.com", "o/r", sha)
373+ k2, _ := Key("https://github.com", "o/r", strings.Repeat("b", 40))
374+ k3, _ := Key("https://ermite.example", "o/r", sha)
375+ k4, _ := Key("https://github.com", "O/R", strings.ToUpper(sha))
376+ if k1 == k2 || k1 == k3 {
377+ t.Error("collision")
378+ }
379+ if k1 != k4 {
380+ t.Error("the case of the repository or the sha must not split an entry")
381+ }
382+ if !ValidKey(k1) {
383+ t.Error("shape")
384+ }
385+}
386+
387+func TestParseMax(t *testing.T) {
388+ for in, want := range map[string]int64{"": DefaultMax, "0": 0, "500M": 500 << 20, "2G": 2 << 30, "2GiB": 2 << 30, "1024": 1024} {
389+ if got, err := ParseMax(in); err != nil || got != want {
390+ t.Errorf("%q: %d %v", in, got, err)
391+ }
392+ }
393+ for _, in := range []string{"-1", "x", "G"} {
394+ if _, err := ParseMax(in); err == nil {
395+ t.Errorf("%q accepted", in)
396+ }
397+ }
398+}
A internal/actionscache/client.go +56 -0
@@ -0,0 +1,56 @@
1+// SPDX-License-Identifier: MIT
2+
3+package actionscache
4+
5+import (
6+ "errors"
7+ "fmt"
8+ "net/http"
9+ "slices"
10+
11+ "bnema.dev/ermite-runner/internal/urlorigin"
12+)
13+
14+func originOf(raw string) string { return urlorigin.Canonical(raw) }
15+
16+// NewClient returns the client that downloads the archives. It follows
17+// redirects (GitHub's codeload does redirect) but never carries credentials
18+// across an origin: the Authorization header is dropped as soon as the scheme,
19+// the host or the port changes, and a redirect from https to http is refused.
20+// (net/http alone would keep the header for subdomains and ignore the scheme and
21+// the port.) Tests may supply their own transport through Actions.HTTPClient;
22+// its redirect policy is replaced.
23+func NewClient(base *http.Client) *http.Client {
24+ c := &http.Client{}
25+ if base != nil {
26+ cp := *base
27+ c = &cp
28+ }
29+ c.CheckRedirect = func(req *http.Request, via []*http.Request) error {
30+ if len(via) >= 10 {
31+ return errors.New("stopped after 10 redirects")
32+ }
33+ // Go builds every redirected request from the headers of the first one
34+ // (copying Authorization when the host matches), so the header must be
35+ // removed again on every hop that is not on the origin the token was meant
36+ // for: the first request's. Once the chain left that origin, it never gets
37+ // the header back, even if a later hop returns to it (A -> B -> A would
38+ // otherwise be re-authorised by the copy).
39+ origin := originOf(via[0].URL.String())
40+ chain := slices.Concat(via[1:], []*http.Request{req}) // not append: via belongs to the client
41+ for i, r := range chain {
42+ prev := via[i]
43+ if prev.URL.Scheme == "https" && r.URL.Scheme != "https" {
44+ return fmt.Errorf("refusing to follow a redirect from https to %s", r.URL.Scheme)
45+ }
46+ }
47+ for _, r := range chain {
48+ if originOf(r.URL.String()) != origin {
49+ req.Header.Del("Authorization")
50+ break
51+ }
52+ }
53+ return nil
54+ }
55+ return c
56+}
@@ -0,0 +1,56 @@
1+// SPDX-License-Identifier: MIT
2+
3+package actionscache
4+
5+import (
6+ "errors"
7+ "fmt"
8+ "net/http"
9+ "slices"
10+
11+ "bnema.dev/ermite-runner/internal/urlorigin"
12+)
13+
14+func originOf(raw string) string { return urlorigin.Canonical(raw) }
15+
16+// NewClient returns the client that downloads the archives. It follows
17+// redirects (GitHub's codeload does redirect) but never carries credentials
18+// across an origin: the Authorization header is dropped as soon as the scheme,
19+// the host or the port changes, and a redirect from https to http is refused.
20+// (net/http alone would keep the header for subdomains and ignore the scheme and
21+// the port.) Tests may supply their own transport through Actions.HTTPClient;
22+// its redirect policy is replaced.
23+func NewClient(base *http.Client) *http.Client {
24+ c := &http.Client{}
25+ if base != nil {
26+ cp := *base
27+ c = &cp
28+ }
29+ c.CheckRedirect = func(req *http.Request, via []*http.Request) error {
30+ if len(via) >= 10 {
31+ return errors.New("stopped after 10 redirects")
32+ }
33+ // Go builds every redirected request from the headers of the first one
34+ // (copying Authorization when the host matches), so the header must be
35+ // removed again on every hop that is not on the origin the token was meant
36+ // for: the first request's. Once the chain left that origin, it never gets
37+ // the header back, even if a later hop returns to it (A -> B -> A would
38+ // otherwise be re-authorised by the copy).
39+ origin := originOf(via[0].URL.String())
40+ chain := slices.Concat(via[1:], []*http.Request{req}) // not append: via belongs to the client
41+ for i, r := range chain {
42+ prev := via[i]
43+ if prev.URL.Scheme == "https" && r.URL.Scheme != "https" {
44+ return fmt.Errorf("refusing to follow a redirect from https to %s", r.URL.Scheme)
45+ }
46+ }
47+ for _, r := range chain {
48+ if originOf(r.URL.String()) != origin {
49+ req.Header.Del("Authorization")
50+ break
51+ }
52+ }
53+ return nil
54+ }
55+ return c
56+}
A internal/cachesvc/archives.go +70 -0
@@ -0,0 +1,70 @@
1+// SPDX-License-Identifier: MIT
2+
3+package cachesvc
4+
5+import (
6+ "crypto/hmac"
7+ "errors"
8+ "net/http"
9+ "net/url"
10+ "strconv"
11+ "time"
12+
13+ "bnema.dev/ermite-runner/internal/actionscache"
14+)
15+
16+// ArchivePath is the prefix of the action archive route: the host downloads the
17+// archives of actions into its own cache (package actionscache) and the sandbox
18+// reads them here instead of from the forge. It is not part of the toolkit's
19+// protocol.
20+const ArchivePath = "/_ermite/action-archives/"
21+
22+// ArchiveURL is the URL, as the guest sees it (its loopback, DefaultPort), from
23+// which the job can read the archive of key for ttl. Like the toolkit's archive
24+// URLs the signature is the credential: it is bound to the key, the job and the
25+// expiry, so a URL that left its sandbox is useless on any other channel, and a
26+// job can read only the archives the host resolved for it. It is "" when the
27+// service holds no archive cache.
28+func (s *Service) ArchiveURL(job, key string, ttl time.Duration) string {
29+ if s.archives == nil || !actionscache.ValidKey(key) {
30+ return ""
31+ }
32+ exp := s.now().Add(ttl).Unix()
33+ q := url.Values{"job": {job}, "exp": {strconv.FormatInt(exp, 10)}, "sig": {b64.EncodeToString(s.sign.mac("action-archive", []byte(key), []byte(job), []byte(strconv.FormatInt(exp, 10))))}}
34+ return "http://127.0.0.1:" + strconv.Itoa(DefaultPort) + ArchivePath + key + "?" + q.Encode()
35+}
36+
37+func (s *Service) archiveOK(key, job string, exp int64, sig string) bool {
38+ got, err := b64.DecodeString(sig)
39+ if err != nil || !s.now().Before(time.Unix(exp, 0)) {
40+ return false
41+ }
42+ return hmac.Equal(got, s.sign.mac("action-archive", []byte(key), []byte(job), []byte(strconv.FormatInt(exp, 10))))
43+}
44+
45+func (s *Service) serveActionArchive(channelJob string) http.HandlerFunc {
46+ return func(w http.ResponseWriter, r *http.Request) {
47+ key := r.PathValue("key")
48+ q := r.URL.Query()
49+ exp, err := strconv.ParseInt(q.Get("exp"), 10, 64)
50+ job := q.Get("job")
51+ if s.archives == nil || err != nil || job == "" || (channelJob != "" && job != channelJob) || !actionscache.ValidKey(key) || !s.archiveOK(key, job, exp, q.Get("sig")) {
52+ reply(w, http.StatusForbidden, "forbidden")
53+ return
54+ }
55+ f, err := s.archives.Open(key)
56+ if err != nil {
57+ if !errors.Is(err, actionscache.ErrNotFound) {
58+ s.log.Warn().Err(err).Msg("action archive")
59+ }
60+ reply(w, http.StatusNotFound, "no such archive")
61+ return
62+ }
63+ defer func() { _ = f.Close() }()
64+ rc := http.NewResponseController(w)
65+ remaining := s.remaining(exp)
66+ defer func() { _ = rc.SetWriteDeadline(deadline(s.progress, remaining)) }()
67+ w.Header().Set("Content-Type", "application/gzip")
68+ http.ServeContent(&progressWriter{ResponseWriter: w, rc: rc, idle: s.progress, remaining: remaining}, r, "", time.Time{}, f)
69+ }
70+}
@@ -0,0 +1,70 @@
1+// SPDX-License-Identifier: MIT
2+
3+package cachesvc
4+
5+import (
6+ "crypto/hmac"
7+ "errors"
8+ "net/http"
9+ "net/url"
10+ "strconv"
11+ "time"
12+
13+ "bnema.dev/ermite-runner/internal/actionscache"
14+)
15+
16+// ArchivePath is the prefix of the action archive route: the host downloads the
17+// archives of actions into its own cache (package actionscache) and the sandbox
18+// reads them here instead of from the forge. It is not part of the toolkit's
19+// protocol.
20+const ArchivePath = "/_ermite/action-archives/"
21+
22+// ArchiveURL is the URL, as the guest sees it (its loopback, DefaultPort), from
23+// which the job can read the archive of key for ttl. Like the toolkit's archive
24+// URLs the signature is the credential: it is bound to the key, the job and the
25+// expiry, so a URL that left its sandbox is useless on any other channel, and a
26+// job can read only the archives the host resolved for it. It is "" when the
27+// service holds no archive cache.
28+func (s *Service) ArchiveURL(job, key string, ttl time.Duration) string {
29+ if s.archives == nil || !actionscache.ValidKey(key) {
30+ return ""
31+ }
32+ exp := s.now().Add(ttl).Unix()
33+ q := url.Values{"job": {job}, "exp": {strconv.FormatInt(exp, 10)}, "sig": {b64.EncodeToString(s.sign.mac("action-archive", []byte(key), []byte(job), []byte(strconv.FormatInt(exp, 10))))}}
34+ return "http://127.0.0.1:" + strconv.Itoa(DefaultPort) + ArchivePath + key + "?" + q.Encode()
35+}
36+
37+func (s *Service) archiveOK(key, job string, exp int64, sig string) bool {
38+ got, err := b64.DecodeString(sig)
39+ if err != nil || !s.now().Before(time.Unix(exp, 0)) {
40+ return false
41+ }
42+ return hmac.Equal(got, s.sign.mac("action-archive", []byte(key), []byte(job), []byte(strconv.FormatInt(exp, 10))))
43+}
44+
45+func (s *Service) serveActionArchive(channelJob string) http.HandlerFunc {
46+ return func(w http.ResponseWriter, r *http.Request) {
47+ key := r.PathValue("key")
48+ q := r.URL.Query()
49+ exp, err := strconv.ParseInt(q.Get("exp"), 10, 64)
50+ job := q.Get("job")
51+ if s.archives == nil || err != nil || job == "" || (channelJob != "" && job != channelJob) || !actionscache.ValidKey(key) || !s.archiveOK(key, job, exp, q.Get("sig")) {
52+ reply(w, http.StatusForbidden, "forbidden")
53+ return
54+ }
55+ f, err := s.archives.Open(key)
56+ if err != nil {
57+ if !errors.Is(err, actionscache.ErrNotFound) {
58+ s.log.Warn().Err(err).Msg("action archive")
59+ }
60+ reply(w, http.StatusNotFound, "no such archive")
61+ return
62+ }
63+ defer func() { _ = f.Close() }()
64+ rc := http.NewResponseController(w)
65+ remaining := s.remaining(exp)
66+ defer func() { _ = rc.SetWriteDeadline(deadline(s.progress, remaining)) }()
67+ w.Header().Set("Content-Type", "application/gzip")
68+ http.ServeContent(&progressWriter{ResponseWriter: w, rc: rc, idle: s.progress, remaining: remaining}, r, "", time.Time{}, f)
69+ }
70+}
A internal/cachesvc/archives_test.go +90 -0
@@ -0,0 +1,90 @@
1+// SPDX-License-Identifier: MIT
2+
3+package cachesvc
4+
5+import (
6+ "context"
7+ "io"
8+ "net/http"
9+ "net/url"
10+ "os"
11+ "path/filepath"
12+ "strings"
13+ "testing"
14+ "time"
15+
16+ "bnema.dev/ermite-runner/internal/actionscache"
17+)
18+
19+func archiveRig(t *testing.T) (*rig, *actionscache.Cache, string) {
20+ t.Helper()
21+ ac, err := actionscache.New(actionscache.Config{Dir: filepath.Join(t.TempDir(), "a"), Max: 1 << 20})
22+ if err != nil {
23+ t.Fatal(err)
24+ }
25+ r := newRigWith(t, Config{Max: 1 << 20, MinCharge: 1, Archives: ac})
26+ key, _ := actionscache.Key("https://github.com", "o/r", strings.Repeat("a", 40))
27+ if _, err := ac.Ensure(context.Background(), key, func(_ context.Context, w io.Writer) error {
28+ _, err := io.WriteString(w, "tarball")
29+ return err
30+ }); err != nil {
31+ t.Fatal(err)
32+ }
33+ return r, ac, key
34+}
35+
36+// The guest reads the archive the host fetched on the channel of its job, with
37+// the signed URL and nothing else; a URL for another job, another key, a stale
38+// expiry or a forged signature gets nothing.
39+func TestActionArchiveRoute(t *testing.T) {
40+ r, ac, key := archiveRig(t)
41+ a := channel(t, r.svc, "er0000000a", "job-a")
42+ b := channel(t, r.svc, "er0000000b", "job-b")
43+
44+ loc := r.svc.ArchiveURL("job-a", key, time.Hour)
45+ if !strings.HasPrefix(loc, "http://127.0.0.1:9089"+ArchivePath+key+"?") {
46+ t.Fatalf("url %q", loc)
47+ }
48+ if code, body := archiveStatus(t, a, "GET", loc); code != http.StatusOK || body != "tarball" {
49+ t.Fatalf("own channel: %d %q", code, body)
50+ }
51+ if code, _ := archiveStatus(t, b, "GET", loc); code != http.StatusForbidden {
52+ t.Errorf("another job's channel: %d", code)
53+ }
54+ u, _ := url.Parse(loc)
55+ otherKey, _ := actionscache.Key("https://github.com", "o/private", strings.Repeat("a", 40))
56+ if code, _ := archiveStatus(t, a, "GET", strings.Replace(loc, key, otherKey, 1)); code != http.StatusForbidden {
57+ t.Errorf("a signature moved to another key: %d", code)
58+ }
59+ q := u.Query()
60+ q.Set("sig", "AAAA")
61+ u.RawQuery = q.Encode()
62+ if code, _ := archiveStatus(t, a, "GET", u.String()); code != http.StatusForbidden {
63+ t.Errorf("forged signature: %d", code)
64+ }
65+ r.clk.t = r.clk.t.Add(2 * time.Hour)
66+ if code, _ := archiveStatus(t, a, "GET", loc); code != http.StatusForbidden {
67+ t.Errorf("expired: %d", code)
68+ }
69+ r.clk.t = r.clk.t.Add(-2 * time.Hour)
70+
71+ // A corrupt archive is never served.
72+ files, _ := filepath.Glob(filepath.Join(ac.Dir(), "*.tgz"))
73+ if len(files) != 1 {
74+ t.Fatalf("files %v", files)
75+ }
76+ if err := os.WriteFile(files[0], []byte("tampered"), 0o600); err != nil {
77+ t.Fatal(err)
78+ }
79+ if code, body := archiveStatus(t, a, "GET", loc); code != http.StatusNotFound || strings.Contains(body, "tampered") {
80+ t.Errorf("corrupt: %d %q", code, body)
81+ }
82+}
83+
84+func TestActionArchiveURLNeedsACache(t *testing.T) {
85+ r := newRig(t, 1<<20)
86+ key, _ := actionscache.Key("https://github.com", "o/r", strings.Repeat("a", 40))
87+ if u := r.svc.ArchiveURL("j", key, time.Hour); u != "" {
88+ t.Errorf("url without a cache: %q", u)
89+ }
90+}
@@ -0,0 +1,90 @@
1+// SPDX-License-Identifier: MIT
2+
3+package cachesvc
4+
5+import (
6+ "context"
7+ "io"
8+ "net/http"
9+ "net/url"
10+ "os"
11+ "path/filepath"
12+ "strings"
13+ "testing"
14+ "time"
15+
16+ "bnema.dev/ermite-runner/internal/actionscache"
17+)
18+
19+func archiveRig(t *testing.T) (*rig, *actionscache.Cache, string) {
20+ t.Helper()
21+ ac, err := actionscache.New(actionscache.Config{Dir: filepath.Join(t.TempDir(), "a"), Max: 1 << 20})
22+ if err != nil {
23+ t.Fatal(err)
24+ }
25+ r := newRigWith(t, Config{Max: 1 << 20, MinCharge: 1, Archives: ac})
26+ key, _ := actionscache.Key("https://github.com", "o/r", strings.Repeat("a", 40))
27+ if _, err := ac.Ensure(context.Background(), key, func(_ context.Context, w io.Writer) error {
28+ _, err := io.WriteString(w, "tarball")
29+ return err
30+ }); err != nil {
31+ t.Fatal(err)
32+ }
33+ return r, ac, key
34+}
35+
36+// The guest reads the archive the host fetched on the channel of its job, with
37+// the signed URL and nothing else; a URL for another job, another key, a stale
38+// expiry or a forged signature gets nothing.
39+func TestActionArchiveRoute(t *testing.T) {
40+ r, ac, key := archiveRig(t)
41+ a := channel(t, r.svc, "er0000000a", "job-a")
42+ b := channel(t, r.svc, "er0000000b", "job-b")
43+
44+ loc := r.svc.ArchiveURL("job-a", key, time.Hour)
45+ if !strings.HasPrefix(loc, "http://127.0.0.1:9089"+ArchivePath+key+"?") {
46+ t.Fatalf("url %q", loc)
47+ }
48+ if code, body := archiveStatus(t, a, "GET", loc); code != http.StatusOK || body != "tarball" {
49+ t.Fatalf("own channel: %d %q", code, body)
50+ }
51+ if code, _ := archiveStatus(t, b, "GET", loc); code != http.StatusForbidden {
52+ t.Errorf("another job's channel: %d", code)
53+ }
54+ u, _ := url.Parse(loc)
55+ otherKey, _ := actionscache.Key("https://github.com", "o/private", strings.Repeat("a", 40))
56+ if code, _ := archiveStatus(t, a, "GET", strings.Replace(loc, key, otherKey, 1)); code != http.StatusForbidden {
57+ t.Errorf("a signature moved to another key: %d", code)
58+ }
59+ q := u.Query()
60+ q.Set("sig", "AAAA")
61+ u.RawQuery = q.Encode()
62+ if code, _ := archiveStatus(t, a, "GET", u.String()); code != http.StatusForbidden {
63+ t.Errorf("forged signature: %d", code)
64+ }
65+ r.clk.t = r.clk.t.Add(2 * time.Hour)
66+ if code, _ := archiveStatus(t, a, "GET", loc); code != http.StatusForbidden {
67+ t.Errorf("expired: %d", code)
68+ }
69+ r.clk.t = r.clk.t.Add(-2 * time.Hour)
70+
71+ // A corrupt archive is never served.
72+ files, _ := filepath.Glob(filepath.Join(ac.Dir(), "*.tgz"))
73+ if len(files) != 1 {
74+ t.Fatalf("files %v", files)
75+ }
76+ if err := os.WriteFile(files[0], []byte("tampered"), 0o600); err != nil {
77+ t.Fatal(err)
78+ }
79+ if code, body := archiveStatus(t, a, "GET", loc); code != http.StatusNotFound || strings.Contains(body, "tampered") {
80+ t.Errorf("corrupt: %d %q", code, body)
81+ }
82+}
83+
84+func TestActionArchiveURLNeedsACache(t *testing.T) {
85+ r := newRig(t, 1<<20)
86+ key, _ := actionscache.Key("https://github.com", "o/r", strings.Repeat("a", 40))
87+ if u := r.svc.ArchiveURL("j", key, time.Hour); u != "" {
88+ t.Errorf("url without a cache: %q", u)
89+ }
90+}
M internal/cachesvc/handler.go +1 -0
@@ -59,6 +59,7 @@ func (s *Service) handler(job string) http.Handler {
5959 mux.HandleFunc("POST "+APIPath+"caches/{id}", s.auth(job, s.commit))
6060 mux.HandleFunc("GET "+APIPath+"artifacts/{id}", s.download(job))
6161 mux.HandleFunc("HEAD "+APIPath+"artifacts/{id}", s.download(job))
62+ mux.HandleFunc("GET "+ArchivePath+"{key}", s.serveActionArchive(job))
6263 // net/http resets the write deadline per request only when the server has a
6364 // WriteTimeout, which would cut long uploads and downloads: set a fresh
6465 // progress deadline for each request here, so that a deadline left by the
@@ -59,6 +59,7 @@ func (s *Service) handler(job string) http.Handler {
59 mux.HandleFunc("POST "+APIPath+"caches/{id}", s.auth(job, s.commit))59 mux.HandleFunc("POST "+APIPath+"caches/{id}", s.auth(job, s.commit))
60 mux.HandleFunc("GET "+APIPath+"artifacts/{id}", s.download(job))60 mux.HandleFunc("GET "+APIPath+"artifacts/{id}", s.download(job))
61 mux.HandleFunc("HEAD "+APIPath+"artifacts/{id}", s.download(job))61 mux.HandleFunc("HEAD "+APIPath+"artifacts/{id}", s.download(job))
62+ mux.HandleFunc("GET "+ArchivePath+"{key}", s.serveActionArchive(job))
62 // net/http resets the write deadline per request only when the server has a63 // net/http resets the write deadline per request only when the server has a
63 // WriteTimeout, which would cut long uploads and downloads: set a fresh64 // WriteTimeout, which would cut long uploads and downloads: set a fresh
64 // progress deadline for each request here, so that a deadline left by the65 // progress deadline for each request here, so that a deadline left by the
M internal/cachesvc/service.go +7 -1
@@ -30,6 +30,8 @@ import (
3030 "time"
3131
3232 "github.com/bnema/zerowrap"
33+
34+ "bnema.dev/ermite-runner/internal/actionscache"
3335 )
3436
3537 // Defaults.
@@ -69,6 +71,9 @@ type Config struct {
6971 // Now is the clock; nil means time.Now.
7072 Now func() time.Time
7173 Log zerowrap.Logger
74+ // Archives, when not nil, is the host's cache of action archives, served on
75+ // the channels (see ArchiveURL).
76+ Archives *actionscache.Cache
7277 // The limits below take the Default… values (or those of the store) when
7378 // zero; tests set them.
7479 ProgressTimeout time.Duration
@@ -93,6 +98,7 @@ type Service struct {
9398 wg sync.WaitGroup
9499 closed bool
95100
101+ archives *actionscache.Cache
96102 hookChunk func(id int64)
97103 progress time.Duration
98104 perListener int
@@ -135,7 +141,7 @@ func New(cfg Config) (*Service, error) {
135141 cfg.Conns = DefaultConns
136142 }
137143 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)}
144+ archives: cfg.Archives, hookChunk: cfg.hookChunk, progress: cfg.ProgressTimeout, perListener: cfg.ConnsPerListener, connsGlobal: make(chan struct{}, cfg.Conns)}
139145 s.wg.Add(1)
140146 go s.housekeeping()
141147 return s, nil
@@ -30,6 +30,8 @@ import (
30 "time"30 "time"
31 31
32 "github.com/bnema/zerowrap"32 "github.com/bnema/zerowrap"
33+
34+ "bnema.dev/ermite-runner/internal/actionscache"
33 )35 )
34 36
35 // Defaults.37 // Defaults.
@@ -69,6 +71,9 @@ type Config struct {
69 // Now is the clock; nil means time.Now.71 // Now is the clock; nil means time.Now.
70 Now func() time.Time72 Now func() time.Time
71 Log zerowrap.Logger73 Log zerowrap.Logger
74+ // Archives, when not nil, is the host's cache of action archives, served on
75+ // the channels (see ArchiveURL).
76+ Archives *actionscache.Cache
72 // The limits below take the Default… values (or those of the store) when77 // The limits below take the Default… values (or those of the store) when
73 // zero; tests set them.78 // zero; tests set them.
74 ProgressTimeout time.Duration79 ProgressTimeout time.Duration
@@ -93,6 +98,7 @@ type Service struct {
93 wg sync.WaitGroup98 wg sync.WaitGroup
94 closed bool99 closed bool
95 100
101+ archives *actionscache.Cache
96 hookChunk func(id int64)102 hookChunk func(id int64)
97 progress time.Duration103 progress time.Duration
98 perListener int104 perListener int
@@ -135,7 +141,7 @@ func New(cfg Config) (*Service, error) {
135 cfg.Conns = DefaultConns141 cfg.Conns = DefaultConns
136 }142 }
137 s := &Service{store: st, sign: sg, now: cfg.Now, log: cfg.Log, channels: map[string]*listener{}, stop: make(chan struct{}),143 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)}144+ archives: cfg.Archives, hookChunk: cfg.hookChunk, progress: cfg.ProgressTimeout, perListener: cfg.ConnsPerListener, connsGlobal: make(chan struct{}, cfg.Conns)}
139 s.wg.Add(1)145 s.wg.Add(1)
140 go s.housekeeping()146 go s.housekeeping()
141 return s, nil147 return s, nil
M internal/executor/actions_download.go +6 -40
@@ -12,11 +12,11 @@ import (
1212 "net/http"
1313 "os"
1414 "path/filepath"
15- "slices"
1615 "time"
1716
1817 "github.com/google/uuid"
1918
19+ "bnema.dev/ermite-runner/internal/actionscache"
2020 "bnema.dev/ermite-runner/internal/urlorigin"
2121 )
2222
@@ -57,46 +57,12 @@ func tokenAllowed(ra ResolvedAction) bool {
5757 return ra.Token != "" && o != "" && o == originOf(ra.TarballURL)
5858 }
5959
60-// archiveClient returns the client that downloads the archives. It follows
61-// redirects (GitHub's codeload does redirect) but never carries credentials
62-// across an origin: the Authorization header is dropped as soon as the scheme,
63-// the host or the port changes, and a redirect from https to http is refused.
64-// (net/http alone would keep the header for subdomains and ignore the scheme and
65-// the port.) Tests may supply their own transport through Actions.HTTPClient;
66-// its redirect policy is replaced.
60+// archiveClient returns the client that downloads the archives
61+// (actionscache.NewClient): it follows redirects but never carries credentials
62+// across an origin. Tests may supply their own transport through
63+// Actions.HTTPClient; its redirect policy is replaced.
6764 func (j *job) archiveClient() *http.Client {
68- c := &http.Client{}
69- if base := j.cfg.Actions.HTTPClient; base != nil {
70- cp := *base
71- c = &cp
72- }
73- c.CheckRedirect = func(req *http.Request, via []*http.Request) error {
74- if len(via) >= 10 {
75- return errors.New("stopped after 10 redirects")
76- }
77- // Go builds every redirected request from the headers of the first one
78- // (copying Authorization when the host matches), so the header must be
79- // removed again on every hop that is not on the origin the token was meant
80- // for: the first request's. Once the chain left that origin, it never gets
81- // the header back, even if a later hop returns to it (A -> B -> A would
82- // otherwise be re-authorised by the copy).
83- origin := originOf(via[0].URL.String())
84- chain := slices.Concat(via[1:], []*http.Request{req}) // not append: via belongs to the client
85- for i, r := range chain {
86- prev := via[i]
87- if prev.URL.Scheme == "https" && r.URL.Scheme != "https" {
88- return fmt.Errorf("refusing to follow a redirect from https to %s", r.URL.Scheme)
89- }
90- }
91- for _, r := range chain {
92- if originOf(r.URL.String()) != origin {
93- req.Header.Del("Authorization")
94- break
95- }
96- }
97- return nil
98- }
99- return c
65+ return actionscache.NewClient(j.cfg.Actions.HTTPClient)
10066 }
10167
10268 // sleepBackoff waits a random 10 to 30 seconds (BackoffTimerHelper.
@@ -12,11 +12,11 @@ import (
12 "net/http"12 "net/http"
13 "os"13 "os"
14 "path/filepath"14 "path/filepath"
15- "slices"
16 "time"15 "time"
17 16
18 "github.com/google/uuid"17 "github.com/google/uuid"
19 18
19+ "bnema.dev/ermite-runner/internal/actionscache"
20 "bnema.dev/ermite-runner/internal/urlorigin"20 "bnema.dev/ermite-runner/internal/urlorigin"
21 )21 )
22 22
@@ -57,46 +57,12 @@ func tokenAllowed(ra ResolvedAction) bool {
57 return ra.Token != "" && o != "" && o == originOf(ra.TarballURL)57 return ra.Token != "" && o != "" && o == originOf(ra.TarballURL)
58 }58 }
59 59
60-// archiveClient returns the client that downloads the archives. It follows60+// archiveClient returns the client that downloads the archives
61-// redirects (GitHub's codeload does redirect) but never carries credentials61+// (actionscache.NewClient): it follows redirects but never carries credentials
62-// across an origin: the Authorization header is dropped as soon as the scheme,62+// across an origin. Tests may supply their own transport through
63-// the host or the port changes, and a redirect from https to http is refused.63+// Actions.HTTPClient; its redirect policy is replaced.
64-// (net/http alone would keep the header for subdomains and ignore the scheme and
65-// the port.) Tests may supply their own transport through Actions.HTTPClient;
66-// its redirect policy is replaced.
67 func (j *job) archiveClient() *http.Client {64 func (j *job) archiveClient() *http.Client {
68- c := &http.Client{}65+ return actionscache.NewClient(j.cfg.Actions.HTTPClient)
69- if base := j.cfg.Actions.HTTPClient; base != nil {
70- cp := *base
71- c = &cp
72- }
73- c.CheckRedirect = func(req *http.Request, via []*http.Request) error {
74- if len(via) >= 10 {
75- return errors.New("stopped after 10 redirects")
76- }
77- // Go builds every redirected request from the headers of the first one
78- // (copying Authorization when the host matches), so the header must be
79- // removed again on every hop that is not on the origin the token was meant
80- // for: the first request's. Once the chain left that origin, it never gets
81- // the header back, even if a later hop returns to it (A -> B -> A would
82- // otherwise be re-authorised by the copy).
83- origin := originOf(via[0].URL.String())
84- chain := slices.Concat(via[1:], []*http.Request{req}) // not append: via belongs to the client
85- for i, r := range chain {
86- prev := via[i]
87- if prev.URL.Scheme == "https" && r.URL.Scheme != "https" {
88- return fmt.Errorf("refusing to follow a redirect from https to %s", r.URL.Scheme)
89- }
90- }
91- for _, r := range chain {
92- if originOf(r.URL.String()) != origin {
93- req.Header.Del("Authorization")
94- break
95- }
96- }
97- return nil
98- }
99- return c
100 }66 }
101 67
102 // sleepBackoff waits a random 10 to 30 seconds (BackoffTimerHelper.68 // sleepBackoff waits a random 10 to 30 seconds (BackoffTimerHelper.
A internal/host/archives.go +104 -0
@@ -0,0 +1,104 @@
1+// SPDX-License-Identifier: MIT
2+
3+package host
4+
5+import (
6+ "context"
7+ "encoding/base64"
8+ "errors"
9+ "fmt"
10+ "io"
11+ "net/http"
12+ "time"
13+
14+ ermite "bnema.dev/ermite-api/gen/ermite"
15+
16+ "bnema.dev/ermite-runner/internal/actionscache"
17+ "bnema.dev/ermite-runner/internal/execstream/execpb"
18+)
19+
20+const (
21+ // archiveFetchTimeout and maxArchiveBytes are the executor's limits for an
22+ // archive (actions_download.go).
23+ archiveFetchTimeout = 20 * time.Minute
24+ maxArchiveBytes = 1 << 30
25+)
26+
27+// cacheArchives points the actions of a resolution at the host's archive cache.
28+// The host downloads each archive once per (forge, repository, resolved sha) and
29+// the sandbox reads it from the cache channel instead of the forge, so that a
30+// busy runner does not trip the forge's rate limit (codeload 429).
31+//
32+// It only ever improves a resolution: an action that cannot be cached (no cache,
33+// no channel for this job, a ref that did not resolve to a sha, a download that
34+// failed, a full disk) keeps its original URL and token, and the sandbox
35+// downloads it as before, with its retries and its messages. The sandbox prints
36+// "Download action repository" in either case, so a hit reads like a miss in the
37+// job's log; the hit or miss is logged here.
38+func (j *jobRun) cacheArchives(ctx context.Context, orig []*ermite.ResolvedAction, conv []*execpb.ResolvedAction) {
39+ rt := j.r
40+ if rt == nil || rt.Archives == nil || rt.Cache == nil || len(orig) != len(conv) {
41+ return
42+ }
43+ if _, ok, err := cacheScope(j.job, j.contexts); err != nil || !ok {
44+ return // no channel for this job: the guest has no relay either
45+ }
46+ ttl := time.Duration(j.timeoutMinutes())*time.Minute + cacheTokenMargin
47+ for i, a := range orig {
48+ c := conv[i]
49+ if c.Error != "" || a.Error != "" || a.TarballUrl == "" {
50+ continue
51+ }
52+ origin := originOf(a.TarballUrl)
53+ key, ok := actionscache.Key(origin, a.NameWithOwner, a.ResolvedSha)
54+ if !ok {
55+ continue
56+ }
57+ hit, err := rt.Archives.Ensure(ctx, key, func(ctx context.Context, w io.Writer) error {
58+ return j.fetchArchive(ctx, a, origin, w)
59+ })
60+ log := j.log.With().Str("action", a.NameWithOwner).Str("sha", a.ResolvedSha).Logger()
61+ if err != nil {
62+ log.Warn().Err(err).Msg("action archive not cached; the sandbox downloads it")
63+ continue
64+ }
65+ u := rt.Cache.ArchiveURL(j.job.JobId, key, ttl)
66+ if u == "" {
67+ continue
68+ }
69+ log.Info().Bool("hit", hit).Msg("action archive cache")
70+ c.TarballUrl, c.Token, c.AuthOrigin = u, "", ""
71+ }
72+}
73+
74+// fetchArchive downloads the archive of a from the forge, with the token only
75+// when the server gave one for an origin that is ermite (like resolveResponse).
76+func (j *jobRun) fetchArchive(ctx context.Context, a *ermite.ResolvedAction, origin string, w io.Writer) error {
77+ token := ""
78+ if a.Source == ermite.ActionSource_ACTION_SOURCE_ERMITE && a.Token != "" && (origin == j.publicOrigin || origin == j.serverOrigin) {
79+ token = a.Token
80+ }
81+ ctx, cancel := context.WithTimeout(ctx, archiveFetchTimeout)
82+ defer cancel()
83+ req, err := http.NewRequestWithContext(ctx, http.MethodGet, a.TarballUrl, nil)
84+ if err != nil {
85+ return err
86+ }
87+ if token != "" {
88+ req.Header.Set("Authorization", "Basic "+base64.StdEncoding.EncodeToString([]byte("x-access-token:"+token)))
89+ }
90+ req.Header.Set("User-Agent", "ermite-runner")
91+ resp, err := actionscache.NewClient(j.r.ArchiveHTTP).Do(req)
92+ if err != nil {
93+ return err
94+ }
95+ defer func() { _ = resp.Body.Close() }()
96+ if resp.StatusCode < 200 || resp.StatusCode > 299 {
97+ return fmt.Errorf("status %d", resp.StatusCode)
98+ }
99+ n, err := io.Copy(w, io.LimitReader(resp.Body, maxArchiveBytes+1))
100+ if err == nil && n > maxArchiveBytes {
101+ err = errors.New("the archive is larger than 1 GiB")
102+ }
103+ return err
104+}
@@ -0,0 +1,104 @@
1+// SPDX-License-Identifier: MIT
2+
3+package host
4+
5+import (
6+ "context"
7+ "encoding/base64"
8+ "errors"
9+ "fmt"
10+ "io"
11+ "net/http"
12+ "time"
13+
14+ ermite "bnema.dev/ermite-api/gen/ermite"
15+
16+ "bnema.dev/ermite-runner/internal/actionscache"
17+ "bnema.dev/ermite-runner/internal/execstream/execpb"
18+)
19+
20+const (
21+ // archiveFetchTimeout and maxArchiveBytes are the executor's limits for an
22+ // archive (actions_download.go).
23+ archiveFetchTimeout = 20 * time.Minute
24+ maxArchiveBytes = 1 << 30
25+)
26+
27+// cacheArchives points the actions of a resolution at the host's archive cache.
28+// The host downloads each archive once per (forge, repository, resolved sha) and
29+// the sandbox reads it from the cache channel instead of the forge, so that a
30+// busy runner does not trip the forge's rate limit (codeload 429).
31+//
32+// It only ever improves a resolution: an action that cannot be cached (no cache,
33+// no channel for this job, a ref that did not resolve to a sha, a download that
34+// failed, a full disk) keeps its original URL and token, and the sandbox
35+// downloads it as before, with its retries and its messages. The sandbox prints
36+// "Download action repository" in either case, so a hit reads like a miss in the
37+// job's log; the hit or miss is logged here.
38+func (j *jobRun) cacheArchives(ctx context.Context, orig []*ermite.ResolvedAction, conv []*execpb.ResolvedAction) {
39+ rt := j.r
40+ if rt == nil || rt.Archives == nil || rt.Cache == nil || len(orig) != len(conv) {
41+ return
42+ }
43+ if _, ok, err := cacheScope(j.job, j.contexts); err != nil || !ok {
44+ return // no channel for this job: the guest has no relay either
45+ }
46+ ttl := time.Duration(j.timeoutMinutes())*time.Minute + cacheTokenMargin
47+ for i, a := range orig {
48+ c := conv[i]
49+ if c.Error != "" || a.Error != "" || a.TarballUrl == "" {
50+ continue
51+ }
52+ origin := originOf(a.TarballUrl)
53+ key, ok := actionscache.Key(origin, a.NameWithOwner, a.ResolvedSha)
54+ if !ok {
55+ continue
56+ }
57+ hit, err := rt.Archives.Ensure(ctx, key, func(ctx context.Context, w io.Writer) error {
58+ return j.fetchArchive(ctx, a, origin, w)
59+ })
60+ log := j.log.With().Str("action", a.NameWithOwner).Str("sha", a.ResolvedSha).Logger()
61+ if err != nil {
62+ log.Warn().Err(err).Msg("action archive not cached; the sandbox downloads it")
63+ continue
64+ }
65+ u := rt.Cache.ArchiveURL(j.job.JobId, key, ttl)
66+ if u == "" {
67+ continue
68+ }
69+ log.Info().Bool("hit", hit).Msg("action archive cache")
70+ c.TarballUrl, c.Token, c.AuthOrigin = u, "", ""
71+ }
72+}
73+
74+// fetchArchive downloads the archive of a from the forge, with the token only
75+// when the server gave one for an origin that is ermite (like resolveResponse).
76+func (j *jobRun) fetchArchive(ctx context.Context, a *ermite.ResolvedAction, origin string, w io.Writer) error {
77+ token := ""
78+ if a.Source == ermite.ActionSource_ACTION_SOURCE_ERMITE && a.Token != "" && (origin == j.publicOrigin || origin == j.serverOrigin) {
79+ token = a.Token
80+ }
81+ ctx, cancel := context.WithTimeout(ctx, archiveFetchTimeout)
82+ defer cancel()
83+ req, err := http.NewRequestWithContext(ctx, http.MethodGet, a.TarballUrl, nil)
84+ if err != nil {
85+ return err
86+ }
87+ if token != "" {
88+ req.Header.Set("Authorization", "Basic "+base64.StdEncoding.EncodeToString([]byte("x-access-token:"+token)))
89+ }
90+ req.Header.Set("User-Agent", "ermite-runner")
91+ resp, err := actionscache.NewClient(j.r.ArchiveHTTP).Do(req)
92+ if err != nil {
93+ return err
94+ }
95+ defer func() { _ = resp.Body.Close() }()
96+ if resp.StatusCode < 200 || resp.StatusCode > 299 {
97+ return fmt.Errorf("status %d", resp.StatusCode)
98+ }
99+ n, err := io.Copy(w, io.LimitReader(resp.Body, maxArchiveBytes+1))
100+ if err == nil && n > maxArchiveBytes {
101+ err = errors.New("the archive is larger than 1 GiB")
102+ }
103+ return err
104+}
A internal/host/archives_test.go +147 -0
@@ -0,0 +1,147 @@
1+// SPDX-License-Identifier: MIT
2+
3+package host
4+
5+import (
6+ "context"
7+ "net/http"
8+ "net/http/httptest"
9+ "path/filepath"
10+ "strings"
11+ "sync/atomic"
12+ "testing"
13+
14+ "github.com/bnema/zerowrap"
15+
16+ ermite "bnema.dev/ermite-api/gen/ermite"
17+
18+ "bnema.dev/ermite-runner/internal/actionscache"
19+ "bnema.dev/ermite-runner/internal/cachesvc"
20+ "bnema.dev/ermite-runner/internal/execstream/execpb"
21+)
22+
23+type archiveRig struct {
24+ r *Runner
25+ srv *httptest.Server
26+ hits atomic.Int32
27+ status atomic.Int32
28+ archive *actionscache.Cache
29+}
30+
31+func newArchiveRig(t *testing.T) *archiveRig {
32+ t.Helper()
33+ a := &archiveRig{}
34+ a.status.Store(http.StatusOK)
35+ a.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
36+ a.hits.Add(1)
37+ if s := int(a.status.Load()); s != http.StatusOK {
38+ http.Error(w, "slow down", s)
39+ return
40+ }
41+ _, _ = w.Write([]byte("tarball of " + r.URL.Path))
42+ }))
43+ t.Cleanup(a.srv.Close)
44+ ac, err := actionscache.New(actionscache.Config{Dir: filepath.Join(t.TempDir(), "a"), Max: 1 << 20})
45+ if err != nil {
46+ t.Fatal(err)
47+ }
48+ svc, err := cachesvc.New(cachesvc.Config{Dir: filepath.Join(t.TempDir(), "c"), Archives: ac})
49+ if err != nil {
50+ t.Fatal(err)
51+ }
52+ t.Cleanup(func() { _ = svc.CloseAll() })
53+ a.archive = ac
54+ a.r = &Runner{Cache: svc, Archives: ac}
55+ return a
56+}
57+
58+func (a *archiveRig) job(id string) *jobRun {
59+ j := &jobRun{r: a.r, log: zerowrap.Logger{}, ctx: context.Background(),
60+ job: &ermite.Job{JobId: id, Ref: "refs/heads/main"},
61+ contexts: `{"github":{"repository":"o/r"}}`}
62+ return j
63+}
64+
65+func (a *archiveRig) action(sha string) (*ermite.ResolvedAction, *execpb.ResolvedAction) {
66+ o := &ermite.ResolvedAction{Ref: "x/y@v1", NameWithOwner: "x/y", ResolvedSha: sha, TarballUrl: a.srv.URL + "/x/y/tarball/" + sha, Token: "ignored-for-github"}
67+ c := &execpb.ResolvedAction{Ref: o.Ref, NameWithOwner: "x/y", ResolvedSha: sha, TarballUrl: o.TarballUrl, Token: "t", AuthOrigin: "o"}
68+ return o, c
69+}
70+
71+// Two jobs resolving the same commit cost one download, and the sandbox is
72+// pointed at the cache with no credentials.
73+func TestCacheArchivesSharedAcrossJobs(t *testing.T) {
74+ a := newArchiveRig(t)
75+ sha := strings.Repeat("a", 40)
76+ for _, id := range []string{"j1", "j2"} {
77+ o, c := a.action(sha)
78+ a.job(id).cacheArchives(context.Background(), []*ermite.ResolvedAction{o}, []*execpb.ResolvedAction{c})
79+ if !strings.HasPrefix(c.TarballUrl, "http://127.0.0.1:9089"+cachesvc.ArchivePath) || c.Token != "" || c.AuthOrigin != "" {
80+ t.Fatalf("%s: not redirected to the cache: %+v", id, c)
81+ }
82+ if !strings.Contains(c.TarballUrl, "job="+id) {
83+ t.Errorf("%s: the URL is not bound to the job: %s", id, c.TarballUrl)
84+ }
85+ }
86+ if a.hits.Load() != 1 {
87+ t.Fatalf("%d downloads for one commit", a.hits.Load())
88+ }
89+}
90+
91+// A moved tag resolves to a new sha, hence a new entry: the ref is never a key.
92+func TestCacheArchivesKeyedBySha(t *testing.T) {
93+ a := newArchiveRig(t)
94+ for _, sha := range []string{strings.Repeat("a", 40), strings.Repeat("b", 40)} {
95+ o, c := a.action(sha)
96+ a.job("j").cacheArchives(context.Background(), []*ermite.ResolvedAction{o}, []*execpb.ResolvedAction{c})
97+ }
98+ if a.hits.Load() != 2 {
99+ t.Fatalf("%d downloads for two commits", a.hits.Load())
100+ }
101+}
102+
103+// Whatever cannot be cached keeps its original URL and token, so the sandbox
104+// downloads it with its own retries and messages.
105+func TestCacheArchivesFallsBack(t *testing.T) {
106+ a := newArchiveRig(t)
107+ a.status.Store(http.StatusTooManyRequests)
108+ o, c := a.action(strings.Repeat("a", 40))
109+ origURL := c.TarballUrl
110+ a.job("j").cacheArchives(context.Background(), []*ermite.ResolvedAction{o}, []*execpb.ResolvedAction{c})
111+ if c.TarballUrl != origURL || c.Token != "t" || c.AuthOrigin != "o" {
112+ t.Errorf("429: %+v", c)
113+ }
114+
115+ a.status.Store(http.StatusOK)
116+ for name, mutate := range map[string]func(*ermite.ResolvedAction){
117+ "a ref instead of a sha": func(o *ermite.ResolvedAction) { o.ResolvedSha = "v1" },
118+ "no sha": func(o *ermite.ResolvedAction) { o.ResolvedSha = "" },
119+ "an error": func(o *ermite.ResolvedAction) { o.Error = "boom" },
120+ "no url": func(o *ermite.ResolvedAction) { o.TarballUrl = "" },
121+ } {
122+ o, c := a.action(strings.Repeat("c", 40))
123+ mutate(o)
124+ before := a.hits.Load()
125+ a.job("j").cacheArchives(context.Background(), []*ermite.ResolvedAction{o}, []*execpb.ResolvedAction{c})
126+ if c.Token != "t" || strings.Contains(c.TarballUrl, cachesvc.ArchivePath) || a.hits.Load() != before {
127+ t.Errorf("%s: cached anyway: %+v", name, c)
128+ }
129+ }
130+
131+ // No cache channel for the job (no repository): the guest has no relay.
132+ o, c = a.action(strings.Repeat("d", 40))
133+ j := a.job("j")
134+ j.contexts = `{}`
135+ j.cacheArchives(context.Background(), []*ermite.ResolvedAction{o}, []*execpb.ResolvedAction{c})
136+ if c.Token != "t" || strings.Contains(c.TarballUrl, cachesvc.ArchivePath) {
137+ t.Errorf("job without a cache scope: %+v", c)
138+ }
139+
140+ // Cache off.
141+ a.r.Archives = nil
142+ o, c = a.action(strings.Repeat("e", 40))
143+ a.job("j").cacheArchives(context.Background(), []*ermite.ResolvedAction{o}, []*execpb.ResolvedAction{c})
144+ if c.Token != "t" {
145+ t.Errorf("cache off: %+v", c)
146+ }
147+}
@@ -0,0 +1,147 @@
1+// SPDX-License-Identifier: MIT
2+
3+package host
4+
5+import (
6+ "context"
7+ "net/http"
8+ "net/http/httptest"
9+ "path/filepath"
10+ "strings"
11+ "sync/atomic"
12+ "testing"
13+
14+ "github.com/bnema/zerowrap"
15+
16+ ermite "bnema.dev/ermite-api/gen/ermite"
17+
18+ "bnema.dev/ermite-runner/internal/actionscache"
19+ "bnema.dev/ermite-runner/internal/cachesvc"
20+ "bnema.dev/ermite-runner/internal/execstream/execpb"
21+)
22+
23+type archiveRig struct {
24+ r *Runner
25+ srv *httptest.Server
26+ hits atomic.Int32
27+ status atomic.Int32
28+ archive *actionscache.Cache
29+}
30+
31+func newArchiveRig(t *testing.T) *archiveRig {
32+ t.Helper()
33+ a := &archiveRig{}
34+ a.status.Store(http.StatusOK)
35+ a.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
36+ a.hits.Add(1)
37+ if s := int(a.status.Load()); s != http.StatusOK {
38+ http.Error(w, "slow down", s)
39+ return
40+ }
41+ _, _ = w.Write([]byte("tarball of " + r.URL.Path))
42+ }))
43+ t.Cleanup(a.srv.Close)
44+ ac, err := actionscache.New(actionscache.Config{Dir: filepath.Join(t.TempDir(), "a"), Max: 1 << 20})
45+ if err != nil {
46+ t.Fatal(err)
47+ }
48+ svc, err := cachesvc.New(cachesvc.Config{Dir: filepath.Join(t.TempDir(), "c"), Archives: ac})
49+ if err != nil {
50+ t.Fatal(err)
51+ }
52+ t.Cleanup(func() { _ = svc.CloseAll() })
53+ a.archive = ac
54+ a.r = &Runner{Cache: svc, Archives: ac}
55+ return a
56+}
57+
58+func (a *archiveRig) job(id string) *jobRun {
59+ j := &jobRun{r: a.r, log: zerowrap.Logger{}, ctx: context.Background(),
60+ job: &ermite.Job{JobId: id, Ref: "refs/heads/main"},
61+ contexts: `{"github":{"repository":"o/r"}}`}
62+ return j
63+}
64+
65+func (a *archiveRig) action(sha string) (*ermite.ResolvedAction, *execpb.ResolvedAction) {
66+ o := &ermite.ResolvedAction{Ref: "x/y@v1", NameWithOwner: "x/y", ResolvedSha: sha, TarballUrl: a.srv.URL + "/x/y/tarball/" + sha, Token: "ignored-for-github"}
67+ c := &execpb.ResolvedAction{Ref: o.Ref, NameWithOwner: "x/y", ResolvedSha: sha, TarballUrl: o.TarballUrl, Token: "t", AuthOrigin: "o"}
68+ return o, c
69+}
70+
71+// Two jobs resolving the same commit cost one download, and the sandbox is
72+// pointed at the cache with no credentials.
73+func TestCacheArchivesSharedAcrossJobs(t *testing.T) {
74+ a := newArchiveRig(t)
75+ sha := strings.Repeat("a", 40)
76+ for _, id := range []string{"j1", "j2"} {
77+ o, c := a.action(sha)
78+ a.job(id).cacheArchives(context.Background(), []*ermite.ResolvedAction{o}, []*execpb.ResolvedAction{c})
79+ if !strings.HasPrefix(c.TarballUrl, "http://127.0.0.1:9089"+cachesvc.ArchivePath) || c.Token != "" || c.AuthOrigin != "" {
80+ t.Fatalf("%s: not redirected to the cache: %+v", id, c)
81+ }
82+ if !strings.Contains(c.TarballUrl, "job="+id) {
83+ t.Errorf("%s: the URL is not bound to the job: %s", id, c.TarballUrl)
84+ }
85+ }
86+ if a.hits.Load() != 1 {
87+ t.Fatalf("%d downloads for one commit", a.hits.Load())
88+ }
89+}
90+
91+// A moved tag resolves to a new sha, hence a new entry: the ref is never a key.
92+func TestCacheArchivesKeyedBySha(t *testing.T) {
93+ a := newArchiveRig(t)
94+ for _, sha := range []string{strings.Repeat("a", 40), strings.Repeat("b", 40)} {
95+ o, c := a.action(sha)
96+ a.job("j").cacheArchives(context.Background(), []*ermite.ResolvedAction{o}, []*execpb.ResolvedAction{c})
97+ }
98+ if a.hits.Load() != 2 {
99+ t.Fatalf("%d downloads for two commits", a.hits.Load())
100+ }
101+}
102+
103+// Whatever cannot be cached keeps its original URL and token, so the sandbox
104+// downloads it with its own retries and messages.
105+func TestCacheArchivesFallsBack(t *testing.T) {
106+ a := newArchiveRig(t)
107+ a.status.Store(http.StatusTooManyRequests)
108+ o, c := a.action(strings.Repeat("a", 40))
109+ origURL := c.TarballUrl
110+ a.job("j").cacheArchives(context.Background(), []*ermite.ResolvedAction{o}, []*execpb.ResolvedAction{c})
111+ if c.TarballUrl != origURL || c.Token != "t" || c.AuthOrigin != "o" {
112+ t.Errorf("429: %+v", c)
113+ }
114+
115+ a.status.Store(http.StatusOK)
116+ for name, mutate := range map[string]func(*ermite.ResolvedAction){
117+ "a ref instead of a sha": func(o *ermite.ResolvedAction) { o.ResolvedSha = "v1" },
118+ "no sha": func(o *ermite.ResolvedAction) { o.ResolvedSha = "" },
119+ "an error": func(o *ermite.ResolvedAction) { o.Error = "boom" },
120+ "no url": func(o *ermite.ResolvedAction) { o.TarballUrl = "" },
121+ } {
122+ o, c := a.action(strings.Repeat("c", 40))
123+ mutate(o)
124+ before := a.hits.Load()
125+ a.job("j").cacheArchives(context.Background(), []*ermite.ResolvedAction{o}, []*execpb.ResolvedAction{c})
126+ if c.Token != "t" || strings.Contains(c.TarballUrl, cachesvc.ArchivePath) || a.hits.Load() != before {
127+ t.Errorf("%s: cached anyway: %+v", name, c)
128+ }
129+ }
130+
131+ // No cache channel for the job (no repository): the guest has no relay.
132+ o, c = a.action(strings.Repeat("d", 40))
133+ j := a.job("j")
134+ j.contexts = `{}`
135+ j.cacheArchives(context.Background(), []*ermite.ResolvedAction{o}, []*execpb.ResolvedAction{c})
136+ if c.Token != "t" || strings.Contains(c.TarballUrl, cachesvc.ArchivePath) {
137+ t.Errorf("job without a cache scope: %+v", c)
138+ }
139+
140+ // Cache off.
141+ a.r.Archives = nil
142+ o, c = a.action(strings.Repeat("e", 40))
143+ a.job("j").cacheArchives(context.Background(), []*ermite.ResolvedAction{o}, []*execpb.ResolvedAction{c})
144+ if c.Token != "t" {
145+ t.Errorf("cache off: %+v", c)
146+ }
147+}
M internal/host/resolve.go +3 -0
@@ -197,6 +197,9 @@ func (j *jobRun) callResolve(ctx context.Context, uses []string) ([]*execpb.Reso
197197 ccancel()
198198 if err == nil {
199199 conv := j.resolveResponse(resp.Msg, len(uses))
200+ if conv.Error == "" {
201+ j.cacheArchives(ctx, resp.Msg.Actions, conv.Actions)
202+ }
200203 if conv.Error != "" {
201204 return fail("%s", conv.Error)
202205 }
@@ -197,6 +197,9 @@ func (j *jobRun) callResolve(ctx context.Context, uses []string) ([]*execpb.Reso
197 ccancel()197 ccancel()
198 if err == nil {198 if err == nil {
199 conv := j.resolveResponse(resp.Msg, len(uses))199 conv := j.resolveResponse(resp.Msg, len(uses))
200+ if conv.Error == "" {
201+ j.cacheArchives(ctx, resp.Msg.Actions, conv.Actions)
202+ }
200 if conv.Error != "" {203 if conv.Error != "" {
201 return fail("%s", conv.Error)204 return fail("%s", conv.Error)
202 }205 }
M internal/host/runner.go +8 -0
@@ -8,6 +8,7 @@ import (
88 "encoding/base64"
99 "errors"
1010 "fmt"
11+ "net/http"
1112 "runtime"
1213 "sync"
1314 "sync/atomic"
@@ -19,6 +20,7 @@ import (
1920 ermite "bnema.dev/ermite-api/gen/ermite"
2021 "bnema.dev/ermite-api/gen/ermite/ermiteconnect"
2122
23+ "bnema.dev/ermite-runner/internal/actionscache"
2224 "bnema.dev/ermite-runner/internal/cachesvc"
2325 "bnema.dev/ermite-runner/internal/execstream/execpb"
2426 "bnema.dev/ermite-runner/internal/sandbox"
@@ -42,6 +44,12 @@ type Runner struct {
4244 // which the guest reaches on its loopback (cachesvc.DefaultPort), and each
4345 // job gets a token scoped to its repository and refs.
4446 Cache *cachesvc.Service
47+ // Archives, when not nil (and Cache is), is the host's cache of action
48+ // archives: the host downloads them and the sandbox reads them from the cache
49+ // channel (see cacheArchives). ArchiveHTTP is its HTTP client (tests); nil
50+ // means the default transport.
51+ Archives *actionscache.Cache
52+ ArchiveHTTP *http.Client
4553
4654 // now returns the current time; tests replace it.
4755 now func() time.Time
@@ -8,6 +8,7 @@ import (
8 "encoding/base64"8 "encoding/base64"
9 "errors"9 "errors"
10 "fmt"10 "fmt"
11+ "net/http"
11 "runtime"12 "runtime"
12 "sync"13 "sync"
13 "sync/atomic"14 "sync/atomic"
@@ -19,6 +20,7 @@ import (
19 ermite "bnema.dev/ermite-api/gen/ermite"20 ermite "bnema.dev/ermite-api/gen/ermite"
20 "bnema.dev/ermite-api/gen/ermite/ermiteconnect"21 "bnema.dev/ermite-api/gen/ermite/ermiteconnect"
21 22
23+ "bnema.dev/ermite-runner/internal/actionscache"
22 "bnema.dev/ermite-runner/internal/cachesvc"24 "bnema.dev/ermite-runner/internal/cachesvc"
23 "bnema.dev/ermite-runner/internal/execstream/execpb"25 "bnema.dev/ermite-runner/internal/execstream/execpb"
24 "bnema.dev/ermite-runner/internal/sandbox"26 "bnema.dev/ermite-runner/internal/sandbox"
@@ -42,6 +44,12 @@ type Runner struct {
42 // which the guest reaches on its loopback (cachesvc.DefaultPort), and each44 // which the guest reaches on its loopback (cachesvc.DefaultPort), and each
43 // job gets a token scoped to its repository and refs.45 // job gets a token scoped to its repository and refs.
44 Cache *cachesvc.Service46 Cache *cachesvc.Service
47+ // Archives, when not nil (and Cache is), is the host's cache of action
48+ // archives: the host downloads them and the sandbox reads them from the cache
49+ // channel (see cacheArchives). ArchiveHTTP is its HTTP client (tests); nil
50+ // means the default transport.
51+ Archives *actionscache.Cache
52+ ArchiveHTTP *http.Client
45 53
46 // now returns the current time; tests replace it.54 // now returns the current time; tests replace it.
47 now func() time.Time55 now func() time.Time

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 4

Tags 11

↑↓ moveenter openesc close