1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package app
4
5import (
6 "bytes"
7 "context"
8 "crypto/sha256"
9 "encoding/hex"
10 "errors"
11 "fmt"
12 "net/http"
13 "os"
14 "path/filepath"
15 "strings"
16 "sync"
17 "sync/atomic"
18 "testing"
19 "time"
20
21 "connectrpc.com/connect"
22 "github.com/stretchr/testify/assert"
23 "github.com/stretchr/testify/require"
24 "google.golang.org/protobuf/types/known/timestamppb"
25
26 pb "bnema.dev/ermite-api/gen/ermite"
27 "bnema.dev/ermite-api/gen/ermite/ermiteconnect"
28 "bnema.dev/ermite/internal/adapters/ciartifacts"
29 "bnema.dev/ermite/internal/domain"
30 "bnema.dev/ermite/internal/ports"
31 "bnema.dev/ermite/internal/usecase"
32)
33
34// CI-4 D3: the artifacts of the runs, as the runner and the owners call them.
35
36const artifactWorkflow = `name: ci
37on:
38 push:
39 branches: [main]
40jobs:
41 up:
42 runs-on: ubuntu-latest
43 steps:
44 - run: echo up
45 down:
46 runs-on: ubuntu-latest
47 steps:
48 - run: echo down
49`
50
51// artifactAPI are the artifact clients of one caller.
52type artifactAPI struct {
53 job ermiteconnect.JobArtifactServiceClient
54 owner ermiteconnect.ArtifactServiceClient
55}
56
57func (c *ciHarness) artifacts(secret string) artifactAPI {
58 hc := &http.Client{Transport: bearerTransport{secret: secret, next: http.DefaultTransport}}
59 return artifactAPI{ermiteconnect.NewJobArtifactServiceClient(hc, c.url("/-/api")), ermiteconnect.NewArtifactServiceClient(hc, c.url("/-/api"))}
60}
61
62// strangerArtifacts are clients whose refused calls do not count against the
63// test's client address.
64func (c *ciHarness) strangerArtifacts(secret string) artifactAPI {
65 hc := &http.Client{Transport: bearerTransport{secret: secret, freshClient: new(atomic.Uint32), next: http.DefaultTransport}}
66 return artifactAPI{ermiteconnect.NewJobArtifactServiceClient(hc, c.url("/-/api")), ermiteconnect.NewArtifactServiceClient(hc, c.url("/-/api"))}
67}
68
69// jobArtifacts is the job side of one job.
70type jobArtifacts struct {
71 c *ciHarness
72 job *pb.Job
73 api artifactAPI
74}
75
76func (c *ciHarness) jobArtifacts(j *pb.Job) *jobArtifacts {
77 return &jobArtifacts{c: c, job: j, api: c.artifacts(j.GetJobToken())}
78}
79
80func (j *jobArtifacts) create(name string, expires *timestamppb.Timestamp) (*pb.CreateJobArtifactResponse, error) {
81 res, err := j.api.job.CreateJobArtifact(j.c.ctx, connect.NewRequest(&pb.CreateJobArtifactRequest{JobId: j.job.GetJobId(), Name: name, ExpiresAt: expires}))
82 if err != nil {
83 return nil, err
84 }
85 return res.Msg, nil
86}
87
88func (j *jobArtifacts) mustCreate(name string) int64 {
89 j.c.t.Helper()
90 res, err := j.create(name, nil)
91 require.NoError(j.c.t, err)
92 return res.GetArtifactId()
93}
94
95// block uploads data as one block, in messages of at most 1 MiB.
96func (j *jobArtifacts) block(id int64, blockID string, data []byte) (int64, error) {
97 return putBlock(j.c.ctx, j.api, j.job.GetJobId(), id, []byte(blockID), data)
98}
99
100func putBlock(ctx context.Context, api artifactAPI, jobID string, id int64, blockID, data []byte) (int64, error) {
101 stream := api.job.UploadJobArtifactBlock(ctx)
102 first := &pb.UploadJobArtifactBlockRequest{JobId: jobID, ArtifactId: id, BlockId: blockID}
103 n := min(len(data), domain.CIArtifactChunk)
104 first.Data, data = data[:n], data[n:]
105 if err := stream.Send(first); err != nil {
106 data = nil // the server answered early: CloseAndReceive has the error
107 }
108 for len(data) > 0 {
109 n := min(len(data), domain.CIArtifactChunk)
110 if err := stream.Send(&pb.UploadJobArtifactBlockRequest{Data: data[:n]}); err != nil {
111 break // the server answered early: CloseAndReceive has the error
112 }
113 data = data[n:]
114 }
115 res, err := stream.CloseAndReceive()
116 if err != nil {
117 return 0, err
118 }
119 return res.Msg.GetSize(), nil
120}
121
122func (j *jobArtifacts) commit(id int64, blocks ...string) (*pb.CommitJobArtifactResponse, error) {
123 ids := make([][]byte, 0, len(blocks))
124 for _, b := range blocks {
125 ids = append(ids, []byte(b))
126 }
127 res, err := j.api.job.CommitJobArtifact(j.c.ctx, connect.NewRequest(&pb.CommitJobArtifactRequest{JobId: j.job.GetJobId(), ArtifactId: id, BlockIds: ids}))
128 if err != nil {
129 return nil, err
130 }
131 return res.Msg, nil
132}
133
134func (j *jobArtifacts) finalize(name, digest string) (*pb.Artifact, error) {
135 res, err := j.api.job.FinalizeJobArtifact(j.c.ctx, connect.NewRequest(&pb.FinalizeJobArtifactRequest{JobId: j.job.GetJobId(), Name: name, Digest: digest}))
136 if err != nil {
137 return nil, err
138 }
139 return res.Msg.GetArtifact(), nil
140}
141
142func (j *jobArtifacts) list(name string) ([]*pb.Artifact, error) {
143 res, err := j.api.job.ListJobArtifacts(j.c.ctx, connect.NewRequest(&pb.ListJobArtifactsRequest{JobId: j.job.GetJobId(), Name: name}))
144 if err != nil {
145 return nil, err
146 }
147 return res.Msg.GetArtifacts(), nil
148}
149
150func (j *jobArtifacts) download(id, offset int64) (*pb.Artifact, []byte, error) {
151 stream, err := j.api.job.DownloadJobArtifact(j.c.ctx, connect.NewRequest(&pb.DownloadJobArtifactRequest{JobId: j.job.GetJobId(), ArtifactId: id, Offset: offset}))
152 if err != nil {
153 return nil, nil, err
154 }
155 defer func() { _ = stream.Close() }()
156 var a *pb.Artifact
157 var buf bytes.Buffer
158 for stream.Receive() {
159 if m := stream.Msg(); m.GetArtifact() != nil {
160 a = m.GetArtifact()
161 }
162 buf.Write(stream.Msg().GetChunk())
163 }
164 return a, buf.Bytes(), stream.Err()
165}
166
167func (j *jobArtifacts) remove(name string) error {
168 _, err := j.api.job.DeleteJobArtifact(j.c.ctx, connect.NewRequest(&pb.DeleteJobArtifactRequest{JobId: j.job.GetJobId(), Name: name}))
169 return err
170}
171
172// upload runs a whole upload of one block.
173func (j *jobArtifacts) upload(name string, data []byte) *pb.Artifact {
174 j.c.t.Helper()
175 id := j.mustCreate(name)
176 _, err := j.block(id, "only", data)
177 require.NoError(j.c.t, err)
178 _, err = j.commit(id, "only")
179 require.NoError(j.c.t, err)
180 a, err := j.finalize(name, "")
181 require.NoError(j.c.t, err)
182 return a
183}
184
185func digestOf(b []byte) string {
186 sum := sha256.Sum256(b)
187 return "sha256:" + hex.EncodeToString(sum[:])
188}
189
190func pattern(n int, seed byte) []byte {
191 b := make([]byte, n)
192 for i := range b {
193 b[i] = seed + byte(i%251)
194 }
195 return b
196}
197
198func (c *ciHarness) usage() int64 {
199 c.t.Helper()
200 used, _, err := c.forge.store.Artifacts().ArtifactUsage(c.ctx, c.name(c.repo))
201 require.NoError(c.t, err)
202 return used
203}
204
205// artifactFiles lists every file under ci/artifacts.
206func (c *ciHarness) artifactFiles() []string {
207 c.t.Helper()
208 var out []string
209 root := ciartifacts.Dir(c.data)
210 _ = filepath.WalkDir(root, func(path string, d os.DirEntry, err error) error {
211 if err == nil && !d.IsDir() {
212 rel, _ := filepath.Rel(root, path)
213 out = append(out, rel)
214 }
215 return nil
216 })
217 return out
218}
219
220func endedDetail(t *testing.T, err error) {
221 t.Helper()
222 require.Equal(t, connect.CodeFailedPrecondition, connect.CodeOf(err), "%v", err)
223 var ce *connect.Error
224 require.True(t, errors.As(err, &ce))
225 found := false
226 for _, d := range ce.Details() {
227 if _, derr := d.Value(); derr == nil && strings.HasSuffix(d.Type(), "JobEnded") {
228 found = true
229 }
230 }
231 assert.True(t, found, "a JobEnded detail")
232}
233
234// newArtifactRun pushes the workflow and returns the jobs of the run, given and
235// acknowledged.
236func (c *ciHarness) newArtifactRun(r *fakeRunner, files map[string]string) (runID int64, up, down *pb.Job) {
237 c.t.Helper()
238 before := len(c.runs())
239 c.push("main", files, "")
240 c.waitFor("a run", func() bool { return len(c.runs()) == before+1 })
241 runID = c.runs()[0].GetId()
242 a, b := r.await(), r.await()
243 r.heartbeat(a)
244 r.heartbeat(b)
245 if strings.HasSuffix(a.GetJobKey(), "down") {
246 a, b = b, a
247 }
248 return runID, a, b
249}
250
251func TestCIArtifactsPassBetweenJobs(t *testing.T) {
252 c := newCIHarness(t, nil)
253 runner := c.registerRunner("alpha", "ubuntu-latest")
254 _, upJob, downJob := c.newArtifactRun(runner, map[string]string{".github/workflows/ci.yml": artifactWorkflow})
255 up, down := c.jobArtifacts(upJob), c.jobArtifacts(downJob)
256
257 ver, err := c.root.server.GetVersion(c.ctx, connect.NewRequest(&pb.GetVersionRequest{}))
258 require.NoError(t, err)
259 assert.Contains(t, ver.Msg.GetFeatures(), "ci.artifacts")
260
261 // Blocks of 1.5 MiB (two messages each), uploaded at once in any order.
262 blocks := [][]byte{pattern(3<<19, 1), pattern(3<<19, 2), pattern(3<<19, 3), pattern(10, 4)}
263 var content []byte
264 for _, b := range blocks {
265 content = append(content, b...)
266 }
267 created, err := up.create("dist", nil)
268 require.NoError(t, err)
269 id := created.GetArtifactId()
270 retention := domain.CIRetentionDays * 24 * time.Hour
271 assert.WithinDuration(t, time.Now().Add(retention), created.GetExpiresAt().AsTime(), time.Minute, "the default retention")
272
273 t.Run("an upload in progress is invisible", func(t *testing.T) {
274 got, err := down.list("")
275 require.NoError(t, err)
276 assert.Empty(t, got)
277 })
278
279 var wg sync.WaitGroup
280 for i := len(blocks) - 1; i >= 0; i-- {
281 wg.Go(func() {
282 n, err := up.block(id, fmt.Sprint("b", i), blocks[i])
283 assert.NoError(t, err)
284 assert.EqualValues(t, len(blocks[i]), n)
285 })
286 }
287 wg.Wait()
288 // A retried block replaces the first one, and is charged once.
289 _, err = up.block(id, "b3", blocks[3])
290 require.NoError(t, err)
291 assert.EqualValues(t, len(content), c.usage(), "pending blocks are charged")
292
293 commit, err := up.commit(id, "b0", "b1", "b2", "b3")
294 require.NoError(t, err)
295 assert.EqualValues(t, len(content), commit.GetSize())
296 assert.Equal(t, digestOf(content), commit.GetDigest())
297 again, err := up.commit(id, "b0", "b1", "b2", "b3")
298 require.NoError(t, err, "a repeat returns the same result")
299 assert.Equal(t, commit.GetDigest(), again.GetDigest())
300 _, err = up.commit(id, "b0")
301 assert.Equal(t, connect.CodeFailedPrecondition, connect.CodeOf(err), "another list on a committed artifact")
302 _, err = up.block(id, "b9", []byte("late"))
303 assert.Equal(t, connect.CodeFailedPrecondition, connect.CodeOf(err), "no block after the commit")
304 assert.EqualValues(t, len(content), c.usage(), "the content replaces the charge of the blocks")
305 for _, f := range c.artifactFiles() {
306 assert.NotContains(t, f, ".blocks", "the blocks are dropped at the commit")
307 }
308
309 _, err = up.finalize("dist", digestOf([]byte("other")))
310 assert.Equal(t, connect.CodeInvalidArgument, connect.CodeOf(err), "a digest that differs")
311 a, err := up.finalize("dist", strings.ToUpper(digestOf(content)[7:]))
312 assert.Equal(t, connect.CodeInvalidArgument, connect.CodeOf(err), "a digest without its prefix")
313 require.Nil(t, a)
314 a, err = up.finalize("dist", digestOf(content))
315 require.NoError(t, err)
316 assert.Equal(t, id, a.GetId())
317 assert.Equal(t, upJob.GetJobId(), a.GetJobId())
318 assert.Equal(t, domain.CIArtifactZip, a.GetMediaType())
319 repeat, err := up.finalize("dist", "")
320 require.NoError(t, err, "a repeat returns the same artifact")
321 assert.Equal(t, id, repeat.GetId())
322
323 t.Run("another job of the run lists and downloads it", func(t *testing.T) {
324 got, err := down.list("")
325 require.NoError(t, err)
326 require.Len(t, got, 1)
327 assert.Equal(t, "dist", got[0].GetName())
328 assert.EqualValues(t, len(content), got[0].GetSize())
329 meta, body, err := down.download(id, 0)
330 require.NoError(t, err)
331 assert.Equal(t, "dist", meta.GetName())
332 assert.True(t, bytes.Equal(content, body))
333 _, body, err = down.download(id, 3<<19+5)
334 require.NoError(t, err, "a resumed download")
335 assert.True(t, bytes.Equal(content[3<<19+5:], body))
336 _, body, err = down.download(id, int64(len(content)))
337 require.NoError(t, err)
338 assert.Empty(t, body)
339 _, _, err = down.download(id, int64(len(content))+1)
340 assert.Equal(t, connect.CodeOutOfRange, connect.CodeOf(err))
341 })
342
343 t.Run("a name is unique in the attempt; overwrite deletes first", func(t *testing.T) {
344 _, err := down.create("dist", nil)
345 assert.Equal(t, connect.CodeAlreadyExists, connect.CodeOf(err))
346 require.NoError(t, down.remove("dist"))
347 assert.Equal(t, connect.CodeNotFound, connect.CodeOf(down.remove("dist")))
348 assert.Zero(t, c.usage(), "deleting refunds")
349 b := down.upload("dist", []byte("v2"))
350 assert.Greater(t, b.GetId(), id, "ids increase, never reused")
351 _, body, err := up.download(b.GetId(), 0)
352 require.NoError(t, err)
353 assert.Equal(t, "v2", string(body))
354 })
355
356 t.Run("names are checked, never used as paths", func(t *testing.T) {
357 for _, name := range []string{"", "../x", "a/b", `a\b`, "a:b", "nul\x00", strings.Repeat("n", 257)} {
358 _, err := up.create(name, nil)
359 assert.Equal(t, connect.CodeInvalidArgument, connect.CodeOf(err), "%q", name)
360 }
361 })
362
363 t.Run("the owners: repo:read to list and download, ci:write to delete", func(t *testing.T) {
364 res, err := c.artifacts(adminToken).owner.ListArtifacts(c.ctx, connect.NewRequest(&pb.ListArtifactsRequest{Repo: c.repo, RunId: a.GetRunId()}))
365 require.NoError(t, err)
366 require.Len(t, res.Msg.GetArtifacts(), 1)
367 assert.EqualValues(t, 2, res.Msg.GetUsedBytes())
368 assert.EqualValues(t, domain.DefaultCIArtifactQuota, res.Msg.GetQuotaBytes())
369
370 // Anonymous: a private repository does not exist, a public one needs a
371 // signed-in caller.
372 anon := c.strangerArtifacts("")
373 _, err = anon.owner.ListArtifacts(c.ctx, connect.NewRequest(&pb.ListArtifactsRequest{Repo: c.repo, RunId: a.GetRunId()}))
374 assert.Equal(t, connect.CodeNotFound, connect.CodeOf(err))
375 public := domain.VisibilityPublic
376 _, err = c.forge.Repos.Update(c.ctx, domain.Root(), c.name(c.repo), ports.UpdateRepoParams{Visibility: &public})
377 require.NoError(t, err)
378 _, err = anon.owner.ListArtifacts(c.ctx, connect.NewRequest(&pb.ListArtifactsRequest{Repo: c.repo, RunId: a.GetRunId()}))
379 assert.Equal(t, connect.CodeUnauthenticated, connect.CodeOf(err))
380 stream, err := anon.owner.DownloadArtifact(c.ctx, connect.NewRequest(&pb.DownloadArtifactRequest{Repo: c.repo, ArtifactId: res.Msg.GetArtifacts()[0].GetId()}))
381 if err == nil {
382 for stream.Receive() {
383 t.Error("anonymous got content")
384 }
385 err = stream.Err()
386 }
387 assert.Equal(t, connect.CodeUnauthenticated, connect.CodeOf(err))
388
389 // A token with ci:write only reads the repository (public) but not its
390 // artifacts; it may delete them.
391 tok, err := c.root.token.CreateToken(c.ctx, connect.NewRequest(&pb.CreateTokenRequest{Label: "ci", Scopes: []pb.Scope{pb.Scope_SCOPE_CI_WRITE}}))
392 require.NoError(t, err)
393 ciw := c.strangerArtifacts(tok.Msg.GetSecret())
394 _, err = ciw.owner.ListArtifacts(c.ctx, connect.NewRequest(&pb.ListArtifactsRequest{Repo: c.repo, RunId: a.GetRunId()}))
395 assert.Equal(t, connect.CodePermissionDenied, connect.CodeOf(err))
396 _, err = anon.owner.DeleteArtifact(c.ctx, connect.NewRequest(&pb.DeleteArtifactRequest{Repo: c.repo, ArtifactId: res.Msg.GetArtifacts()[0].GetId()}))
397 assert.Equal(t, connect.CodeUnauthenticated, connect.CodeOf(err))
398
399 root := c.artifacts(adminToken)
400 dl, err := root.owner.DownloadArtifact(c.ctx, connect.NewRequest(&pb.DownloadArtifactRequest{Repo: c.repo, ArtifactId: res.Msg.GetArtifacts()[0].GetId()}))
401 require.NoError(t, err)
402 var body []byte
403 for dl.Receive() {
404 body = append(body, dl.Msg().GetChunk()...)
405 }
406 require.NoError(t, dl.Err())
407 assert.Equal(t, "v2", string(body))
408 _, err = ciw.owner.DeleteArtifact(c.ctx, connect.NewRequest(&pb.DeleteArtifactRequest{Repo: c.repo, ArtifactId: res.Msg.GetArtifacts()[0].GetId()}))
409 require.NoError(t, err)
410 got, err := up.list("")
411 require.NoError(t, err)
412 assert.Empty(t, got)
413 assert.Zero(t, c.usage())
414 })
415}
416
417// AC3: a job token of another run reaches nothing of this run, and the calls
418// are refused once the job ended.
419func TestCIArtifactsAreIsolatedPerRunAndJob(t *testing.T) {
420 c := newCIHarness(t, nil)
421 runner := c.registerRunner("alpha", "ubuntu-latest")
422 runA, upA, downA := c.newArtifactRun(runner, map[string]string{".github/workflows/ci.yml": artifactWorkflow})
423 _, upB, _ := c.newArtifactRun(runner, map[string]string{"README": "two"})
424 a, b := c.jobArtifacts(upA), c.jobArtifacts(upB)
425
426 secret := a.upload("secret", []byte("of run A"))
427 pendingID := a.mustCreate("wip")
428 _, err := a.block(pendingID, "x", []byte("half"))
429 require.NoError(t, err)
430
431 t.Run("a job of run B", func(t *testing.T) {
432 got, err := b.list("")
433 require.NoError(t, err)
434 assert.Empty(t, got, "lists its own run only")
435 _, _, err = b.download(secret.GetId(), 0)
436 assert.Equal(t, connect.CodeNotFound, connect.CodeOf(err))
437 assert.Equal(t, connect.CodeNotFound, connect.CodeOf(b.remove("secret")))
438 _, err = b.block(pendingID, "y", []byte("intrusion"))
439 assert.Equal(t, connect.CodeNotFound, connect.CodeOf(err))
440 _, err = b.commit(pendingID, "x")
441 assert.Equal(t, connect.CodeNotFound, connect.CodeOf(err))
442 _, err = b.finalize("wip", "")
443 assert.Equal(t, connect.CodeNotFound, connect.CodeOf(err))
444 })
445
446 t.Run("another job of the same run uploads nothing into the job's upload", func(t *testing.T) {
447 d := c.jobArtifacts(downA)
448 _, err := d.block(pendingID, "y", []byte("intrusion"))
449 assert.Equal(t, connect.CodeNotFound, connect.CodeOf(err))
450 _, err = d.commit(pendingID, "x")
451 assert.Equal(t, connect.CodeNotFound, connect.CodeOf(err))
452 })
453
454 t.Run("job ids in requests must be the token's", func(t *testing.T) {
455 wrong := &jobArtifacts{c: c, job: &pb.Job{JobId: upA.GetJobId(), JobToken: upB.GetJobToken()}, api: c.strangerArtifacts(upB.GetJobToken())}
456 _, err := wrong.list("")
457 assert.Equal(t, connect.CodePermissionDenied, connect.CodeOf(err))
458 _, _, err = wrong.download(secret.GetId(), 0)
459 assert.Equal(t, connect.CodePermissionDenied, connect.CodeOf(err))
460 assert.Equal(t, connect.CodePermissionDenied, connect.CodeOf(wrong.remove("secret")))
461 _, err = wrong.block(pendingID, "z", []byte("x"))
462 assert.Equal(t, connect.CodePermissionDenied, connect.CodeOf(err))
463 })
464
465 t.Run("no token, a runner token, a repository token", func(t *testing.T) {
466 for name, secret := range map[string]string{"none": "", "runner": runner.secret, "garbage": "ermitejob_nope", "admin": adminToken} {
467 s := &jobArtifacts{c: c, job: upA, api: c.strangerArtifacts(secret)}
468 _, err := s.list("")
469 assert.Equal(t, connect.CodeUnauthenticated, connect.CodeOf(err), name)
470 _, err = s.block(pendingID, "z", []byte("x"))
471 assert.Equal(t, connect.CodeUnauthenticated, connect.CodeOf(err), name)
472 }
473 })
474
475 t.Run("the end of the job discards its uploads and refuses its token", func(t *testing.T) {
476 before := c.usage()
477 runner.runJob(upA, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil)
478 assert.EqualValues(t, before-int64(len("half")), c.usage(), "the unfinished upload is refunded")
479 _, err := c.forge.store.Artifacts().GetArtifact(c.ctx, pendingID)
480 assert.ErrorIs(t, err, domain.ErrNotFound)
481 for _, f := range c.artifactFiles() {
482 assert.NotContains(t, f, fmt.Sprintf("%020d", pendingID), "its files are gone")
483 }
484 _, err = a.list("")
485 endedDetail(t, err)
486 _, err = a.create("late", nil)
487 endedDetail(t, err)
488 _, _, err = a.download(secret.GetId(), 0)
489 endedDetail(t, err)
490
491 // The finalized artifact stays for the other jobs.
492 got, err := c.jobArtifacts(downA).list("secret")
493 require.NoError(t, err)
494 require.Len(t, got, 1)
495 assert.Equal(t, runA, got[0].GetRunId())
496 })
497}
498
499// shiftClock is the real time moved by an offset the test sets.
500type shiftClock struct{ off atomic.Int64 }
501
502func (s *shiftClock) Now() time.Time { return time.Now().Add(time.Duration(s.off.Load())) }
503func (s *shiftClock) advance(d time.Duration) { s.off.Add(int64(d)) }
504func (s *shiftClock) since(t time.Time) time.Duration { return s.Now().Sub(t) }
505
506func TestCIArtifactLimits(t *testing.T) {
507 const mib = 1 << 20
508 c := newCIHarness(t, func(cfg *usecase.CIConfig) {
509 cfg.ArtifactLimits = ports.ArtifactLimits{Retention: 10 * 24 * time.Hour, Quota: 3 * mib, MaxSize: 2 * mib}
510 })
511 runner := c.registerRunner("alpha", "ubuntu-latest")
512 _, upJob, downJob := c.newArtifactRun(runner, map[string]string{".github/workflows/ci.yml": artifactWorkflow})
513 up, down := c.jobArtifacts(upJob), c.jobArtifacts(downJob)
514
515 t.Run("the retention is kept between one day and the longest", func(t *testing.T) {
516 res, err := up.create("soon", timestamppb.New(time.Now().Add(time.Hour)))
517 require.NoError(t, err)
518 assert.WithinDuration(t, time.Now().Add(24*time.Hour), res.GetExpiresAt().AsTime(), time.Minute)
519 res, err = up.create("late", timestamppb.New(time.Now().Add(1000*24*time.Hour)))
520 require.NoError(t, err)
521 assert.WithinDuration(t, time.Now().Add(10*24*time.Hour), res.GetExpiresAt().AsTime(), time.Minute)
522 res, err = up.create("mid", timestamppb.New(time.Now().Add(5*24*time.Hour)))
523 require.NoError(t, err)
524 assert.WithinDuration(t, time.Now().Add(5*24*time.Hour), res.GetExpiresAt().AsTime(), time.Second)
525 })
526
527 id := up.mustCreate("big")
528 t.Run("a block over the size of one artifact is refused and nothing is kept", func(t *testing.T) {
529 _, err := up.block(id, "a", pattern(2*mib+1, 0))
530 require.Equal(t, connect.CodeResourceExhausted, connect.CodeOf(err))
531 assert.Contains(t, err.Error(), "2 MiB")
532 assert.Zero(t, c.usage())
533 assert.Empty(t, c.artifactFiles())
534 })
535
536 t.Run("blocks add up against the size of one artifact", func(t *testing.T) {
537 _, err := up.block(id, "a", pattern(3*mib/2, 0))
538 require.NoError(t, err)
539 _, err = up.block(id, "b", pattern(mib, 0))
540 require.Equal(t, connect.CodeResourceExhausted, connect.CodeOf(err))
541 assert.EqualValues(t, 3*mib/2, c.usage())
542 _, err = up.commit(id, "a", "a")
543 require.Equal(t, connect.CodeResourceExhausted, connect.CodeOf(err), "a repeated block counts twice")
544 _, err = up.commit(id, "a", "nope")
545 assert.Equal(t, connect.CodeInvalidArgument, connect.CodeOf(err), "a block never uploaded")
546 _, err = up.finalize("big", "")
547 assert.Equal(t, connect.CodeFailedPrecondition, connect.CodeOf(err), "not committed")
548 })
549
550 t.Run("the quota of the repository counts the uploads in progress", func(t *testing.T) {
551 other := down.mustCreate("other")
552 _, err := down.block(other, "a", pattern(mib, 1))
553 require.NoError(t, err)
554 _, err = down.block(other, "b", pattern(mib, 1))
555 require.Equal(t, connect.CodeResourceExhausted, connect.CodeOf(err))
556 assert.Contains(t, err.Error(), "quota (3 MiB)")
557 // A retry of a stored block within the quota replaces it.
558 _, err = down.block(other, "a", pattern(mib/2, 2))
559 require.NoError(t, err)
560 assert.EqualValues(t, 2*mib, c.usage())
561 _, err = down.block(other, "b", pattern(mib, 1))
562 require.NoError(t, err)
563 _, err = down.create("third", nil)
564 assert.Equal(t, connect.CodeResourceExhausted, connect.CodeOf(err), "the space is used up")
565 })
566}
567
568func TestCIArtifactsSweep(t *testing.T) {
569 clk := &shiftClock{}
570 c := newCIHarness(t, func(cfg *usecase.CIConfig) {
571 cfg.Clock = clk
572 cfg.Tick = 20 * time.Millisecond
573 cfg.ArtifactLimits.Retention = 2 * 24 * time.Hour
574 })
575 runner := c.registerRunner("alpha", "ubuntu-latest")
576 runID, upJob, _ := c.newArtifactRun(runner, map[string]string{".github/workflows/ci.yml": artifactWorkflow})
577 up := c.jobArtifacts(upJob)
578 kept := up.upload("kept", []byte("content"))
579
580 // A crash of the server between the end of the job and its cleanup: an
581 // unfinished upload stays recorded with its block; a crash between a file
582 // and its record: files that no record names.
583 crashed := up.mustCreate("crashed")
584 _, err := up.block(crashed, "x", []byte("stale"))
585 require.NoError(t, err)
586 stray := filepath.Join(ciartifacts.Dir(c.data), fmt.Sprintf("%020d", runID), fmt.Sprintf("%020d", int64(999999)))
587 require.NoError(t, os.WriteFile(stray, []byte("orphan"), 0o600))
588 strayRun := filepath.Join(ciartifacts.Dir(c.data), fmt.Sprintf("%020d", int64(888888)))
589 require.NoError(t, os.MkdirAll(strayRun, 0o700))
590 start := clk.Now()
591
592 // The day after: the upload is older than a day and swept, the stray files
593 // go, the finalized artifact stays.
594 clk.advance(25 * time.Hour)
595 c.waitFor("the sweep", func() bool {
596 _, err := c.forge.store.Artifacts().GetArtifact(c.ctx, crashed)
597 return errors.Is(err, domain.ErrNotFound)
598 })
599 c.waitFor("the stray files", func() bool {
600 _, err1 := os.Stat(stray)
601 _, err2 := os.Stat(strayRun)
602 return os.IsNotExist(err1) && os.IsNotExist(err2)
603 })
604 assert.EqualValues(t, len("content"), c.usage(), "the unfinished upload is refunded")
605 _, err = c.forge.store.Artifacts().GetArtifact(c.ctx, kept.GetId())
606 require.NoError(t, err)
607 for _, f := range c.artifactFiles() {
608 assert.Equal(t, filepath.Join(fmt.Sprintf("%020d", runID), fmt.Sprintf("%020d", kept.GetId())), f)
609 }
610
611 // Past its expiry it is invisible at once, then swept.
612 clk.advance(24 * time.Hour)
613 res, err := c.artifacts(adminToken).owner.ListArtifacts(c.ctx, connect.NewRequest(&pb.ListArtifactsRequest{Repo: c.repo, RunId: runID}))
614 require.NoError(t, err)
615 assert.Empty(t, res.Msg.GetArtifacts(), "expired")
616 c.waitFor("the expired artifact swept", func() bool { return c.usage() == 0 && len(c.artifactFiles()) == 0 })
617 assert.Greater(t, clk.since(start), 48*time.Hour)
618}
619
620// Deleting the repository takes the records and the files of its artifacts.
621func TestCIArtifactsGoWithTheRepository(t *testing.T) {
622 c := newCIHarness(t, nil)
623 runner := c.registerRunner("alpha", "ubuntu-latest")
624 _, upJob, _ := c.newArtifactRun(runner, map[string]string{".github/workflows/ci.yml": artifactWorkflow})
625 a := c.jobArtifacts(upJob).upload("dist", []byte("x"))
626 require.NotEmpty(t, c.artifactFiles())
627
628 require.NoError(t, c.forge.Repos.Delete(c.ctx, domain.Root(), c.name(c.repo)))
629 _, err := c.forge.store.Artifacts().GetArtifact(c.ctx, a.GetId())
630 assert.ErrorIs(t, err, domain.ErrNotFound)
631 assert.Empty(t, c.artifactFiles())
632}
633
634// ERMITE_CI_ARTIFACTS=0 serves neither service.
635func TestCIArtifactsOff(t *testing.T) {
636 h := newHarnessWith(t, harnessOpts{env: map[string]string{"ERMITE_CI_ARTIFACTS": "0"}}, nil)
637 ver, err := h.api("").server.GetVersion(h.ctx, connect.NewRequest(&pb.GetVersionRequest{}))
638 require.NoError(t, err)
639 assert.NotContains(t, ver.Msg.GetFeatures(), "ci.artifacts")
640 c := &ciHarness{harness: h}
641 _, err = c.artifacts(adminToken).owner.ListArtifacts(h.ctx, connect.NewRequest(&pb.ListArtifactsRequest{Repo: "x", RunId: 1}))
642 assert.Equal(t, connect.CodeUnimplemented, connect.CodeOf(err))
643 _, err = c.artifacts("ermitejob_x").job.ListJobArtifacts(h.ctx, connect.NewRequest(&pb.ListJobArtifactsRequest{JobId: "1-0-0"}))
644 assert.Equal(t, connect.CodeUnimplemented, connect.CodeOf(err))
645}
646
647// Concurrent writes never pass the limits: the bytes of a block are reserved
648// before they are written, against the charges and the other writes.
649func TestCIArtifactLimitsHoldUnderConcurrency(t *testing.T) {
650 const kib, mib = 1 << 10, 1 << 20
651 c := newCIHarness(t, func(cfg *usecase.CIConfig) {
652 cfg.ArtifactLimits = ports.ArtifactLimits{Quota: 3 * mib, MaxSize: 2 * mib}
653 })
654 runner := c.registerRunner("alpha", "ubuntu-latest")
655 _, upJob, _ := c.newArtifactRun(runner, map[string]string{".github/workflows/ci.yml": artifactWorkflow})
656 up := c.jobArtifacts(upJob)
657
658 blockFiles := func() (total int64) {
659 for _, f := range c.artifactFiles() {
660 if info, err := os.Stat(filepath.Join(ciartifacts.Dir(c.data), f)); err == nil {
661 total += info.Size()
662 }
663 }
664 return total
665 }
666
667 // peak samples the bytes on disk while fn runs.
668 peak := func(fn func()) int64 {
669 var top atomic.Int64
670 done := make(chan struct{})
671 go func() {
672 for {
673 if n := blockFiles(); n > top.Load() {
674 top.Store(n)
675 }
676 select {
677 case <-done:
678 return
679 default:
680 }
681 }
682 }()
683 fn()
684 close(done)
685 return top.Load()
686 }
687
688 t.Run("one artifact: parallel blocks stop at the size of one artifact", func(t *testing.T) {
689 id := up.mustCreate("one")
690 var stored atomic.Int64
691 var wg sync.WaitGroup
692 top := peak(func() {
693 for i := range 12 {
694 wg.Go(func() {
695 n, err := up.block(id, fmt.Sprint("b", i), pattern(512*kib, byte(i)))
696 if err == nil {
697 stored.Add(n)
698 return
699 }
700 assert.Equal(t, connect.CodeResourceExhausted, connect.CodeOf(err))
701 })
702 }
703 wg.Wait()
704 })
705 assert.LessOrEqual(t, top, int64(2*mib), "nothing beyond the limit ever reached the disk")
706 assert.LessOrEqual(t, stored.Load(), int64(2*mib), "never past the size of one artifact")
707 assert.Positive(t, stored.Load())
708 assert.Equal(t, stored.Load(), c.usage(), "what is stored is what is charged")
709 assert.Equal(t, stored.Load(), blockFiles(), "no refused block left on disk")
710 })
711
712 t.Run("several artifacts: parallel blocks stop at the quota", func(t *testing.T) {
713 ids := []int64{up.mustCreate("two"), up.mustCreate("three")}
714 before := c.usage()
715 var stored atomic.Int64
716 var wg sync.WaitGroup
717 for i := range 12 {
718 wg.Go(func() {
719 n, err := up.block(ids[i%2], fmt.Sprint("b", i), pattern(256*kib, byte(i)))
720 if err == nil {
721 stored.Add(n)
722 return
723 }
724 assert.Equal(t, connect.CodeResourceExhausted, connect.CodeOf(err))
725 })
726 }
727 wg.Wait()
728 assert.LessOrEqual(t, c.usage(), int64(3*mib), "never past the quota")
729 assert.Positive(t, stored.Load())
730 assert.Equal(t, before+stored.Load(), c.usage())
731 assert.Equal(t, c.usage(), blockFiles(), "no refused block left on disk")
732 })
733}
734
735func TestCIArtifactCaps(t *testing.T) {
736 c := newCIHarness(t, nil)
737 runner := c.registerRunner("alpha", "ubuntu-latest")
738 _, upJob, downJob := c.newArtifactRun(runner, map[string]string{".github/workflows/ci.yml": artifactWorkflow})
739 up, down := c.jobArtifacts(upJob), c.jobArtifacts(downJob)
740
741 t.Run("a job has a bounded number of unfinished uploads", func(t *testing.T) {
742 for i := range domain.CIArtifactMaxPendingPerJob {
743 up.mustCreate(fmt.Sprint("p", i))
744 }
745 _, err := up.create("one-more", nil)
746 assert.Equal(t, connect.CodeResourceExhausted, connect.CodeOf(err))
747 _, err = up.create("p0", nil)
748 require.NoError(t, err, "a new create of a pending name replaces it")
749 down.mustCreate("other job")
750 })
751
752 t.Run("a commit repeated while it runs, or after its caller left, gives one result", func(t *testing.T) {
753 content := pattern(6<<20, 9)
754 id := down.mustCreate("big")
755 _, err := down.block(id, "a", content[:3<<20])
756 require.NoError(t, err)
757 _, err = down.block(id, "b", content[3<<20:])
758 require.NoError(t, err)
759
760 // A caller that gives up at once: the commit still completes.
761 gone, cancel := context.WithTimeout(c.ctx, time.Millisecond)
762 _, _ = down.api.job.CommitJobArtifact(gone, connect.NewRequest(&pb.CommitJobArtifactRequest{JobId: downJob.GetJobId(), ArtifactId: id, BlockIds: [][]byte{[]byte("a"), []byte("b")}}))
763 cancel()
764
765 var wg sync.WaitGroup
766 results := make([]*pb.CommitJobArtifactResponse, 4)
767 for i := range results {
768 wg.Go(func() {
769 res, err := down.commit(id, "a", "b")
770 assert.NoError(t, err)
771 results[i] = res
772 })
773 }
774 wg.Wait()
775 for _, r := range results {
776 require.NotNil(t, r)
777 assert.Equal(t, digestOf(content), r.GetDigest())
778 assert.EqualValues(t, len(content), r.GetSize())
779 }
780 _, body, err := func() (*pb.Artifact, []byte, error) {
781 if _, err := down.finalize("big", ""); err != nil {
782 return nil, nil, err
783 }
784 return up.download(id, 0)
785 }()
786 require.NoError(t, err)
787 assert.True(t, bytes.Equal(content, body))
788 })
789}