1// SPDX-License-Identifier: MIT
2
3package host
4
5import (
6 "bytes"
7 "context"
8 "encoding/json"
9 "io"
10 "net/http"
11 "net/http/httptest"
12 "path/filepath"
13 "strings"
14 "sync/atomic"
15 "testing"
16
17 "github.com/bnema/zerowrap"
18
19 ermite "bnema.dev/ermite-api/gen/ermite"
20
21 "bnema.dev/ermite-runner/internal/actionscache"
22 "bnema.dev/ermite-runner/internal/cachesvc"
23 "bnema.dev/ermite-runner/internal/execstream/execpb"
24)
25
26type archiveRig struct {
27 r *Runner
28 srv *httptest.Server
29 hits atomic.Int32
30 status atomic.Int32
31 archive *actionscache.Cache
32}
33
34func newArchiveRig(t *testing.T) *archiveRig {
35 t.Helper()
36 a := &archiveRig{}
37 a.status.Store(http.StatusOK)
38 a.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
39 a.hits.Add(1)
40 if s := int(a.status.Load()); s != http.StatusOK {
41 http.Error(w, "slow down", s)
42 return
43 }
44 _, _ = w.Write([]byte("tarball of " + r.URL.Path))
45 }))
46 t.Cleanup(a.srv.Close)
47 ac, err := actionscache.New(actionscache.Config{Dir: filepath.Join(t.TempDir(), "a"), Max: 1 << 20})
48 if err != nil {
49 t.Fatal(err)
50 }
51 svc, err := cachesvc.New(cachesvc.Config{Dir: filepath.Join(t.TempDir(), "c"), Archives: ac})
52 if err != nil {
53 t.Fatal(err)
54 }
55 t.Cleanup(func() { _ = svc.CloseAll() })
56 a.archive = ac
57 a.r = &Runner{Cache: svc, Archives: ac}
58 return a
59}
60
61func (a *archiveRig) job(id string) *jobRun {
62 j := &jobRun{r: a.r, log: zerowrap.Logger{}, ctx: context.Background(),
63 job: &ermite.Job{JobId: id, Ref: "refs/heads/main"},
64 contexts: `{"github":{"repository":"o/r"}}`, timeline: timeline{}}
65 return j
66}
67
68func (a *archiveRig) action(sha string) (*ermite.ResolvedAction, *execpb.ResolvedAction) {
69 o := &ermite.ResolvedAction{Ref: "x/y@v1", NameWithOwner: "x/y", ResolvedSha: sha, TarballUrl: a.srv.URL + "/x/y/tarball/" + sha, Token: "ignored-for-github"}
70 c := &execpb.ResolvedAction{Ref: o.Ref, NameWithOwner: "x/y", ResolvedSha: sha, TarballUrl: o.TarballUrl, Token: "t", AuthOrigin: "o"}
71 return o, c
72}
73
74// Two jobs resolving the same commit cost one download, and the sandbox is
75// pointed at the cache with no credentials.
76func TestCacheArchivesSharedAcrossJobs(t *testing.T) {
77 a := newArchiveRig(t)
78 sha := strings.Repeat("a", 40)
79 for _, id := range []string{"j1", "j2"} {
80 o, c := a.action(sha)
81 a.job(id).cacheArchives(context.Background(), []*ermite.ResolvedAction{o}, []*execpb.ResolvedAction{c})
82 if !strings.HasPrefix(c.TarballUrl, "http://127.0.0.1:9089"+cachesvc.ArchivePath) || c.Token != "" || c.AuthOrigin != "" {
83 t.Fatalf("%s: not redirected to the cache: %+v", id, c)
84 }
85 if !strings.Contains(c.TarballUrl, "job="+id) {
86 t.Errorf("%s: the URL is not bound to the job: %s", id, c.TarballUrl)
87 }
88 }
89 if a.hits.Load() != 1 {
90 t.Fatalf("%d downloads for one commit", a.hits.Load())
91 }
92}
93
94// A moved tag resolves to a new sha, hence a new entry: the ref is never a key.
95func TestCacheArchivesKeyedBySha(t *testing.T) {
96 a := newArchiveRig(t)
97 for _, sha := range []string{strings.Repeat("a", 40), strings.Repeat("b", 40)} {
98 o, c := a.action(sha)
99 a.job("j").cacheArchives(context.Background(), []*ermite.ResolvedAction{o}, []*execpb.ResolvedAction{c})
100 }
101 if a.hits.Load() != 2 {
102 t.Fatalf("%d downloads for two commits", a.hits.Load())
103 }
104}
105
106// Whatever cannot be cached keeps its original URL and token, so the sandbox
107// downloads it with its own retries and messages.
108func TestCacheArchivesFallsBack(t *testing.T) {
109 a := newArchiveRig(t)
110 a.status.Store(http.StatusTooManyRequests)
111 o, c := a.action(strings.Repeat("a", 40))
112 origURL := c.TarballUrl
113 a.job("j").cacheArchives(context.Background(), []*ermite.ResolvedAction{o}, []*execpb.ResolvedAction{c})
114 if c.TarballUrl != origURL || c.Token != "t" || c.AuthOrigin != "o" {
115 t.Errorf("429: %+v", c)
116 }
117
118 a.status.Store(http.StatusOK)
119 for name, mutate := range map[string]func(*ermite.ResolvedAction){
120 "a ref instead of a sha": func(o *ermite.ResolvedAction) { o.ResolvedSha = "v1" },
121 "no sha": func(o *ermite.ResolvedAction) { o.ResolvedSha = "" },
122 "an error": func(o *ermite.ResolvedAction) { o.Error = "boom" },
123 "no url": func(o *ermite.ResolvedAction) { o.TarballUrl = "" },
124 } {
125 o, c := a.action(strings.Repeat("c", 40))
126 mutate(o)
127 before := a.hits.Load()
128 a.job("j").cacheArchives(context.Background(), []*ermite.ResolvedAction{o}, []*execpb.ResolvedAction{c})
129 if c.Token != "t" || strings.Contains(c.TarballUrl, cachesvc.ArchivePath) || a.hits.Load() != before {
130 t.Errorf("%s: cached anyway: %+v", name, c)
131 }
132 }
133
134 // No cache scope for the job (no repository): archives are not served.
135 o, c = a.action(strings.Repeat("d", 40))
136 j := a.job("j")
137 j.contexts = `{}`
138 j.cacheArchives(context.Background(), []*ermite.ResolvedAction{o}, []*execpb.ResolvedAction{c})
139 if c.Token != "t" || strings.Contains(c.TarballUrl, cachesvc.ArchivePath) {
140 t.Errorf("job without a cache scope: %+v", c)
141 }
142
143 // Cache off.
144 a.r.Archives = nil
145 o, c = a.action(strings.Repeat("e", 40))
146 a.job("j").cacheArchives(context.Background(), []*ermite.ResolvedAction{o}, []*execpb.ResolvedAction{c})
147 if c.Token != "t" {
148 t.Errorf("cache off: %+v", c)
149 }
150}
151
152func TestArchiveTimingLogs(t *testing.T) {
153 a := newArchiveRig(t)
154 var logs bytes.Buffer
155 j := a.job("timing")
156 j.log = zerowrap.New(zerowrap.Config{Level: "info", Format: "json", Output: &logs})
157 for i := 0; i < 2; i++ {
158 logs.Reset()
159 o, c := a.action(strings.Repeat("a", 40))
160 j.cacheArchives(context.Background(), []*ermite.ResolvedAction{o}, []*execpb.ResolvedAction{c})
161 var fields map[string]any
162 if err := json.Unmarshal(logs.Bytes(), &fields); err != nil {
163 t.Fatal(err, logs.String())
164 }
165 if value, ok := fields["verify_ms"].(float64); !ok || value < 0 {
166 t.Fatal(fields)
167 }
168 if fields["hit"] != (i == 1) {
169 t.Fatal(fields)
170 }
171 if i == 0 {
172 if value, ok := fields["fetch_ms"].(float64); !ok || value < 0 {
173 t.Fatal(fields)
174 }
175 if value, ok := fields["bytes"].(float64); !ok || value <= 0 {
176 t.Fatal(fields)
177 }
178 } else if _, ok := fields["fetch_ms"]; ok {
179 t.Fatal("hit has fetch_ms", fields)
180 }
181 }
182 if j.timeline["archive_misses"] != 1 || j.timeline["archive_miss_bytes"] <= 0 {
183 t.Fatal(j.timeline)
184 }
185}
186
187// Only a download the job ran itself is a miss: a failed one counts, a job
188// whose wait for another job's download is cancelled does not.
189func TestArchiveMissesCountFetches(t *testing.T) {
190 a := newArchiveRig(t)
191 a.status.Store(http.StatusTooManyRequests)
192 failed := a.job("failed")
193 o, c := a.action(strings.Repeat("a", 40))
194 failed.cacheArchives(context.Background(), []*ermite.ResolvedAction{o}, []*execpb.ResolvedAction{c})
195 if failed.timeline["archive_misses"] != 1 {
196 t.Fatalf("failed fetch: %v", failed.timeline)
197 }
198
199 a.status.Store(http.StatusOK)
200 // A joiner whose wait is cancelled while another job downloads.
201 release := make(chan struct{})
202 started, done := make(chan struct{}), make(chan struct{})
203 sha := strings.Repeat("b", 40)
204 key, ok := actionscache.Key(originOf(a.srv.URL), "x/y", sha)
205 if !ok {
206 t.Fatal("key")
207 }
208 go func() {
209 defer close(done)
210 _, _, _ = a.r.Archives.Ensure(context.Background(), key, func(_ context.Context, w io.Writer) error {
211 close(started)
212 <-release
213 _, err := io.WriteString(w, "tarball")
214 return err
215 })
216 }()
217 <-started
218 ctx, cancel := context.WithCancel(context.Background())
219 cancel()
220 joiner := a.job("joiner")
221 o, c = a.action(sha)
222 joiner.cacheArchives(ctx, []*ermite.ResolvedAction{o}, []*execpb.ResolvedAction{c})
223 close(release)
224 <-done // the download writes under t.TempDir: let it finish before cleanup
225 if _, ok := joiner.timeline["archive_misses"]; ok {
226 t.Fatalf("cancelled joiner: %v", joiner.timeline)
227 }
228}