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