1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package usecase
4
5import (
6 "bytes"
7 "context"
8 "crypto/rand"
9 "crypto/sha256"
10 "encoding/binary"
11 "encoding/hex"
12 "errors"
13 "fmt"
14 "io"
15 "strconv"
16 "sync"
17 "time"
18
19 "github.com/bnema/zerowrap"
20
21 "bnema.dev/ermite/internal/domain"
22 "bnema.dev/ermite/internal/ports"
23)
24
25// The artifacts of the runs (artifacts v4, CI-4 D3). The runner translates the
26// toolkit's protocol and calls JobArtifactService with the job token; owners
27// use ArtifactService. The records are in CIConfig.Artifacts, the contents in
28// CIConfig.ArtifactFiles, named by ids only.
29//
30// Every write of the artifacts of a repository goes through its usage record
31// (ports.CIArtifactWrite): a decision is made on the records read after it and
32// written under its version, or made again. So the quota, the unique names of
33// a run and the states of an upload hold with concurrent calls, and the bytes
34// of a write are counted before it is accepted. The files are written before
35// the record that refers to them and removed after the record that dropped
36// them: a crash leaves at most files that no record names, which the sweep
37// removes, never a record without its file.
38
39var (
40 _ ports.JobArtifactService = (*CI)(nil)
41 _ ports.ArtifactService = (*CI)(nil)
42)
43
44const (
45 // artifactTries bounds the retries of a write of the artifacts of a
46 // repository that other writes keep changing.
47 artifactTries = 256
48 // commitTries bounds the assemblies of one commit whose blocks change
49 // meanwhile.
50 commitTries = 5
51)
52
53var (
54 // errNoArtifactWrite ends updateArtifacts without a write.
55 errNoArtifactWrite = errors.New("no write")
56 // errArtifactChanged makes a commit start again.
57 errArtifactChanged = errors.New("artifact changed")
58)
59
60func (c *CI) artifactsOn() bool { return c.cfg.Artifacts != nil && c.cfg.ArtifactFiles != nil }
61
62func artifactRef(a domain.CIArtifact) ports.CIArtifactRef {
63 return ports.CIArtifactRef{RunID: a.RunID, ID: a.ID}
64}
65
66// artifactJob authenticates the job token of an artifact call: the job of
67// jobID, which has not ended.
68func (c *CI) artifactJob(ctx context.Context, secret, jobID string) (domain.CIJob, error) {
69 if !c.artifactsOn() {
70 return domain.CIJob{}, fmt.Errorf("%w: artifacts are off on this server", domain.ErrPreconditionFailed)
71 }
72 j, err := c.jobFor(ctx, secret, jobID)
73 if err != nil {
74 return domain.CIJob{}, err
75 }
76 if j.Ended() {
77 return domain.CIJob{}, endedError(j)
78 }
79 return j, nil
80}
81
82// updateArtifacts reads the usage of the repository, lets fn fill the write
83// from the records it reads then, and applies it; on a conflict it starts
84// again. fn returns errNoArtifactWrite to write nothing.
85func (c *CI) updateArtifacts(ctx context.Context, repo domain.RepoName, fn func(used int64, w *ports.CIArtifactWrite) error) error {
86 for range artifactTries {
87 used, version, err := c.cfg.Artifacts.ArtifactUsage(ctx, repo)
88 if err != nil {
89 return err
90 }
91 w := ports.CIArtifactWrite{Repo: repo, Used: used, UsageVersion: version}
92 if err := fn(used, &w); err != nil {
93 return err
94 }
95 err = c.cfg.Artifacts.ApplyArtifacts(ctx, w)
96 if errors.Is(err, domain.ErrConflict) {
97 continue
98 }
99 return err
100 }
101 return fmt.Errorf("%w: the artifacts of %s keep changing, try again", domain.ErrUnavailable, repo)
102}
103
104// --- limits ---
105
106// sizeText writes a limit for a message: in GiB or MiB when it is a whole
107// number of them.
108func sizeText(n int64) string {
109 switch {
110 case n >= 1<<30 && n%(1<<30) == 0:
111 return strconv.FormatInt(n>>30, 10) + " GiB"
112 case n >= 1<<20 && n%(1<<20) == 0:
113 return strconv.FormatInt(n>>20, 10) + " MiB"
114 }
115 return strconv.FormatInt(n, 10) + " bytes"
116}
117
118func (c *CI) errArtifactTooLarge() error {
119 return fmt.Errorf("%w: the artifact would exceed the size limit of one artifact (%s)", domain.ErrResourceExhausted, sizeText(c.cfg.ArtifactLimits.MaxSize))
120}
121
122func (c *CI) errQuota(repo domain.RepoName) error {
123 return fmt.Errorf("%w: the artifacts of %s would exceed their quota (%s)", domain.ErrResourceExhausted, repo, sizeText(c.cfg.ArtifactLimits.Quota))
124}
125
126// --- reservations ---
127
128// artifactReservations are the bytes being written and not charged yet:
129// blocks streaming in, the repeats of a commit being assembled. A write
130// reserves its bytes before they reach the disk, under the lock of its
131// repository and against the charges stored and the other reservations, and
132// releases them once its record is written or refused. So concurrent writes
133// never put more on disk than the limits leave, and the write of the record
134// checks them again. The size of one artifact and the quota are both limits of
135// one repository: repositories never wait for one another.
136type artifactReservations struct {
137 mu sync.Mutex
138 repos map[domain.RepoName]*repoReservations
139}
140
141// repoReservations are the reservations of one repository. An entry stays
142// once made (a few words per repository that ever uploaded); its maps empty.
143type repoReservations struct {
144 mu sync.Mutex
145 total int64
146 art map[int64]int64
147}
148
149func (r *artifactReservations) of(repo domain.RepoName) *repoReservations {
150 r.mu.Lock()
151 defer r.mu.Unlock()
152 if r.repos == nil {
153 r.repos = map[domain.RepoName]*repoReservations{}
154 }
155 rr := r.repos[repo]
156 if rr == nil {
157 rr = &repoReservations{art: map[int64]int64{}}
158 r.repos[repo] = rr
159 }
160 return rr
161}
162
163// reserve takes n bytes for artifact a, or returns the limit it would cross.
164func (c *CI) reserve(ctx context.Context, a domain.CIArtifact, n int64) error {
165 r := c.reservations.of(a.Repo)
166 r.mu.Lock()
167 defer r.mu.Unlock()
168 used, _, err := c.cfg.Artifacts.ArtifactUsage(ctx, a.Repo)
169 if err != nil {
170 return err
171 }
172 cur, err := c.cfg.Artifacts.GetArtifact(ctx, a.ID)
173 if err != nil {
174 return err
175 }
176 limits := c.cfg.ArtifactLimits
177 switch {
178 case cur.Charged+r.art[a.ID]+n > limits.MaxSize:
179 return c.errArtifactTooLarge()
180 case used+r.total+n > limits.Quota:
181 return c.errQuota(a.Repo)
182 }
183 r.total += n
184 r.art[a.ID] += n
185 return nil
186}
187
188// release gives back n bytes reserved for a.
189func (c *CI) release(a domain.CIArtifact, n int64) {
190 if n == 0 {
191 return
192 }
193 r := c.reservations.of(a.Repo)
194 r.mu.Lock()
195 defer r.mu.Unlock()
196 r.total = max(r.total-n, 0)
197 if r.art[a.ID] -= n; r.art[a.ID] <= 0 {
198 delete(r.art, a.ID)
199 }
200}
201
202// reserveAhead is how much a reservingReader reserves at once, so that a
203// large block takes one reservation per MiB, not one per read.
204const reserveAhead = domain.CIArtifactChunk
205
206// reservingReader passes on the data of r, each read covered by a
207// reservation first: a read that the limits refuse fails without being passed
208// on, so a block is never written beyond them. It reserves reserveAhead at a
209// time, or just what a read needs when the limits leave less; reserved is all
210// it holds (written or not), released whole after the record is written.
211type reservingReader struct {
212 ctx context.Context
213 c *CI
214 a domain.CIArtifact
215 r io.Reader
216 reserved int64
217 left int64 // reserved and not yet passed on
218}
219
220func (rr *reservingReader) Read(p []byte) (int, error) {
221 n, err := rr.r.Read(p)
222 if need := int64(n) - rr.left; need > 0 {
223 got := max(need, reserveAhead)
224 if rerr := rr.c.reserve(rr.ctx, rr.a, got); rerr != nil {
225 if got == need {
226 return 0, rerr
227 }
228 got = need
229 if rerr := rr.c.reserve(rr.ctx, rr.a, got); rerr != nil {
230 return 0, rerr
231 }
232 }
233 rr.reserved += got
234 rr.left += got
235 }
236 rr.left -= int64(n)
237 return n, err
238}
239
240// --- locks of the commits ---
241
242// lockArtifact serializes the commits of an artifact: two commits must not
243// assemble its content at once. It returns the unlock function.
244func (c *CI) lockArtifact(id int64) func() {
245 c.mu.Lock()
246 if c.artifactLocks == nil {
247 c.artifactLocks = map[int64]*artifactLock{}
248 }
249 l := c.artifactLocks[id]
250 if l == nil {
251 l = &artifactLock{}
252 c.artifactLocks[id] = l
253 }
254 l.users++
255 c.mu.Unlock()
256 l.Lock()
257 return func() {
258 l.Unlock()
259 c.mu.Lock()
260 if l.users--; l.users == 0 {
261 delete(c.artifactLocks, id)
262 }
263 c.mu.Unlock()
264 }
265}
266
267type artifactLock struct {
268 sync.Mutex
269 users int
270}
271
272// --- removal ---
273
274// removeArtifacts deletes the artifacts of a run that pick chooses, judged on
275// the records as they are when the write is made, refunds them, then removes
276// their files. It returns the deleted ones.
277func (c *CI) removeArtifacts(ctx context.Context, repo domain.RepoName, runID int64, pick func(domain.CIArtifact) bool) ([]domain.CIArtifact, error) {
278 var removed []domain.CIArtifact
279 err := c.updateArtifacts(ctx, repo, func(used int64, w *ports.CIArtifactWrite) error {
280 arts, err := c.cfg.Artifacts.RunArtifacts(ctx, repo, runID)
281 if err != nil {
282 return err
283 }
284 removed = removed[:0]
285 for _, a := range arts {
286 if pick(a) {
287 removed = append(removed, a)
288 w.Used -= a.Charged
289 }
290 }
291 if len(removed) == 0 {
292 return errNoArtifactWrite
293 }
294 w.Delete = removed
295 w.Used = max(w.Used, 0)
296 return nil
297 })
298 if errors.Is(err, errNoArtifactWrite) {
299 return nil, nil
300 }
301 if err != nil {
302 return nil, err
303 }
304 c.removeArtifactFiles(ctx, removed)
305 return removed, nil
306}
307
308// removeArtifactFiles removes the files of deleted artifacts. A failure is
309// logged: the sweep removes the files that no record names.
310func (c *CI) removeArtifactFiles(ctx context.Context, arts []domain.CIArtifact) {
311 for _, a := range arts {
312 if err := c.cfg.ArtifactFiles.Remove(ctx, artifactRef(a)); err != nil {
313 ctxLog(ctx).Warn().Err(err).Int64("artifact_id", a.ID).Msg("ci: remove the files of an artifact; the sweep will")
314 }
315 }
316}
317
318// dropIfEnded discards the uploads of the job's assignment when the job ended
319// or was given again while the call ran: the cleanup of the end may have run
320// before the call wrote its record.
321func (c *CI) dropIfEnded(ctx context.Context, j domain.CIJob) {
322 cur, err := c.cfg.Store.GetJob(ctx, j.RunID, j.RunAttempt, j.ID)
323 if err == nil && !cur.Ended() && cur.TokenHash == j.TokenHash {
324 return
325 }
326 c.dropPending(ctx, j, func(a domain.CIArtifact) bool { return a.Assignment == j.TokenHash })
327}
328
329// dropPending discards the unfinished uploads of a job, of any of its
330// assignments, that pick chooses.
331func (c *CI) dropPending(ctx context.Context, j domain.CIJob, pick func(domain.CIArtifact) bool) {
332 removed, err := c.removeArtifacts(ctx, j.Repo, j.RunID, func(a domain.CIArtifact) bool {
333 return !a.Final() && a.UploadedBy(j) && pick(a)
334 })
335 if err != nil && !errors.Is(err, domain.ErrNotFound) {
336 ctxLog(ctx).Error().Err(err).Str("job_id", j.ID).Msg("ci: discard the unfinished artifacts of a job; the sweep will")
337 return
338 }
339 if len(removed) > 0 {
340 ctxLog(ctx).Info().Str("job_id", j.ID).Int64("run_id", j.RunID).Int("artifacts", len(removed)).Msg("ci: unfinished artifacts discarded")
341 }
342}
343
344// artifactsJobEnded discards the unfinished uploads of a job that ended. It is
345// called after the end is written.
346func (c *CI) artifactsJobEnded(ctx context.Context, j domain.CIJob) {
347 if !c.artifactsOn() {
348 return
349 }
350 c.dropPending(ctx, j, func(domain.CIArtifact) bool { return true })
351}
352
353// --- JobArtifactService ---
354
355// CreateJobArtifact begins an upload: a pending artifact with a new id. A
356// pending artifact of the same job and name is discarded.
357func (c *CI) CreateJobArtifact(ctx context.Context, secret string, params ports.CreateJobArtifactParams) (ports.CreatedArtifact, error) {
358 j, err := c.artifactJob(ctx, secret, params.JobID)
359 if err != nil {
360 return ports.CreatedArtifact{}, err
361 }
362 if err := domain.ParseArtifactName(params.Name); err != nil {
363 return ports.CreatedArtifact{}, err
364 }
365 mediaType, err := domain.ParseArtifactMediaType(params.MediaType)
366 if err != nil {
367 return ports.CreatedArtifact{}, err
368 }
369 now := c.now()
370 id, err := c.cfg.Artifacts.NextArtifactID(ctx)
371 if err != nil {
372 return ports.CreatedArtifact{}, err
373 }
374 a := domain.CIArtifact{
375 ID: id, Repo: j.Repo, RunID: j.RunID, Attempt: j.RunAttempt, JobID: j.ID, Assignment: j.TokenHash,
376 Name: params.Name, MediaType: mediaType, State: domain.CIArtifactPending,
377 CreatedAt: now, ExpiresAt: domain.ArtifactExpiry(params.ExpiresAt, now, c.cfg.ArtifactLimits.Retention),
378 }
379 var discarded []domain.CIArtifact
380 err = c.updateArtifacts(ctx, j.Repo, func(used int64, w *ports.CIArtifactWrite) error {
381 if used >= c.cfg.ArtifactLimits.Quota {
382 return c.errQuota(j.Repo)
383 }
384 arts, err := c.cfg.Artifacts.RunArtifacts(ctx, j.Repo, j.RunID)
385 if err != nil {
386 return err
387 }
388 discarded = discarded[:0]
389 pending := 0
390 for _, o := range arts {
391 switch {
392 case o.Name != a.Name:
393 if !o.Final() && o.UploadedBy(j) {
394 pending++
395 }
396 case o.Visible(now) && o.Attempt == j.RunAttempt:
397 return fmt.Errorf("%w: an artifact with this name already exists on the workflow run", domain.ErrAlreadyExists)
398 case !o.Final() && o.UploadedBy(j):
399 discarded = append(discarded, o) // a new create is a new upload
400 w.Used -= o.Charged
401 }
402 }
403 switch {
404 case pending >= domain.CIArtifactMaxPendingPerJob:
405 return fmt.Errorf("%w: a job has at most %d unfinished artifact uploads", domain.ErrResourceExhausted, domain.CIArtifactMaxPendingPerJob)
406 case len(arts)-len(discarded) >= domain.CIArtifactMaxPerRun:
407 return fmt.Errorf("%w: a run has at most %d artifacts", domain.ErrResourceExhausted, domain.CIArtifactMaxPerRun)
408 }
409 w.Put, w.Delete = []domain.CIArtifact{a}, discarded
410 return nil
411 })
412 if err != nil {
413 return ports.CreatedArtifact{}, err
414 }
415 c.removeArtifactFiles(ctx, discarded)
416 log := zerowrap.FromCtx(ctx)
417 log.Info().Int64("artifact_id", id).Int64("run_id", j.RunID).Str("job_id", j.ID).Msg("artifact upload begun")
418 c.dropIfEnded(ctx, j)
419 return ports.CreatedArtifact{ID: id, ExpiresAt: a.ExpiresAt}, nil
420}
421
422// pendingArtifact returns an unfinished upload of the job's assignment, or
423// domain.ErrNotFound.
424func (c *CI) pendingArtifact(ctx context.Context, j domain.CIJob, id int64) (domain.CIArtifact, error) {
425 a, err := c.cfg.Artifacts.GetArtifact(ctx, id)
426 if err != nil {
427 return domain.CIArtifact{}, err
428 }
429 if a.Repo != j.Repo || !a.PendingOf(j) {
430 return domain.CIArtifact{}, fmt.Errorf("%w: artifact %d is not an upload of this job", domain.ErrNotFound, id)
431 }
432 return a, nil
433}
434
435func errCommitted(id int64) error {
436 return fmt.Errorf("%w: artifact %d is already committed", domain.ErrPreconditionFailed, id)
437}
438
439func newBlockFile(blockID []byte) string {
440 var raw [8]byte
441 _, _ = rand.Read(raw[:])
442 return hex.EncodeToString(blockID) + "." + hex.EncodeToString(raw[:])
443}
444
445// UploadJobArtifactBlock stores a block: the data goes to a new file, counted
446// against the limits as it comes, then the record names it in place of the
447// block of the same id, if any.
448func (c *CI) UploadJobArtifactBlock(ctx context.Context, secret, jobID string, artifactID int64, blockID []byte, data io.Reader) (int64, error) {
449 j, err := c.artifactJob(ctx, secret, jobID)
450 if err != nil {
451 return 0, err
452 }
453 if err := domain.CheckArtifactBlockID(blockID); err != nil {
454 return 0, err
455 }
456 a, err := c.pendingArtifact(ctx, j, artifactID)
457 if err != nil {
458 return 0, err
459 }
460 if a.State != domain.CIArtifactPending {
461 return 0, errCommitted(a.ID)
462 }
463 // Every byte is reserved before it is written, against the charges and
464 // the other writes in flight (a block it replaces counts until the record
465 // drops it): nothing beyond the limits ever reaches the disk.
466 src := &reservingReader{ctx: ctx, c: c, a: a, r: data}
467 defer func() { c.release(a, src.reserved) }()
468 limits := c.cfg.ArtifactLimits
469 ref, file := artifactRef(a), newBlockFile(blockID)
470 size, err := c.cfg.ArtifactFiles.PutBlock(ctx, ref, file, src)
471 if err != nil {
472 return 0, err
473 }
474 var old *ports.CIArtifactBlock
475 err = c.updateArtifacts(ctx, a.Repo, func(used int64, w *ports.CIArtifactWrite) error {
476 cur, err := c.pendingArtifact(ctx, j, a.ID)
477 if err != nil {
478 return err
479 }
480 if cur.State != domain.CIArtifactPending {
481 return errCommitted(cur.ID)
482 }
483 blocks, err := c.cfg.Artifacts.ArtifactBlocks(ctx, cur.ID)
484 if err != nil {
485 return err
486 }
487 old = nil
488 for i := range blocks {
489 if bytes.Equal(blocks[i].ID, blockID) {
490 old = &blocks[i]
491 }
492 }
493 var replaced int64
494 if old != nil {
495 replaced = old.Size
496 } else if len(blocks) >= domain.CIArtifactMaxBlocks {
497 return fmt.Errorf("%w: an artifact has at most %d blocks", domain.ErrInvalid, domain.CIArtifactMaxBlocks)
498 }
499 cur.Charged += size - replaced
500 w.Used += size - replaced
501 switch {
502 case cur.Charged > limits.MaxSize:
503 return c.errArtifactTooLarge()
504 case w.Used > limits.Quota:
505 return c.errQuota(cur.Repo)
506 }
507 w.Put = []domain.CIArtifact{cur}
508 w.PutBlocks = []ports.CIArtifactBlock{{ArtifactID: cur.ID, ID: blockID, Size: size, File: file}}
509 return nil
510 })
511 if err != nil {
512 if rerr := c.cfg.ArtifactFiles.RemoveBlock(ctx, ref, file); rerr != nil {
513 ctxLog(ctx).Warn().Err(rerr).Int64("artifact_id", a.ID).Msg("ci: remove a refused block; the sweep will")
514 }
515 return 0, err
516 }
517 if old != nil {
518 if err := c.cfg.ArtifactFiles.RemoveBlock(ctx, ref, old.File); err != nil {
519 ctxLog(ctx).Warn().Err(err).Int64("artifact_id", a.ID).Msg("ci: remove a replaced block")
520 }
521 }
522 c.dropIfEnded(ctx, j)
523 return size, nil
524}
525
526// blockListDigest identifies a block list: each id with its length.
527func blockListDigest(ids [][]byte) []byte {
528 h := sha256.New()
529 var n [2]byte
530 for _, id := range ids {
531 binary.BigEndian.PutUint16(n[:], uint16(len(id))) //nolint:gosec // at most 64
532 h.Write(n[:])
533 h.Write(id)
534 }
535 return h.Sum(nil)
536}
537
538// CommitJobArtifact assembles the content of a pending artifact from its
539// blocks, then records it and drops the blocks. The size counts against the
540// limits with the repeats of the list.
541func (c *CI) CommitJobArtifact(ctx context.Context, secret, jobID string, artifactID int64, blockIDs [][]byte) (int64, string, error) {
542 j, err := c.artifactJob(ctx, secret, jobID)
543 if err != nil {
544 return 0, "", err
545 }
546 if len(blockIDs) > domain.CIArtifactMaxBlocks {
547 return 0, "", fmt.Errorf("%w: a commit names at most %d blocks", domain.ErrInvalid, domain.CIArtifactMaxBlocks)
548 }
549 for _, id := range blockIDs {
550 if err := domain.CheckArtifactBlockID(id); err != nil {
551 return 0, "", err
552 }
553 }
554 list := blockListDigest(blockIDs)
555 // A retry of the runner waits here for the commit in progress and gets
556 // its result. The commit goes on when its caller leaves: the runner gives
557 // up on a long assembly and asks again.
558 unlock := c.lockArtifact(artifactID)
559 defer unlock()
560 ctx = context.WithoutCancel(ctx)
561 limits := c.cfg.ArtifactLimits
562 for range commitTries {
563 a, err := c.cfg.Artifacts.GetArtifact(ctx, artifactID)
564 if err != nil {
565 return 0, "", err
566 }
567 if a.Repo != j.Repo || !a.UploadedBy(j) || a.Assignment != j.TokenHash {
568 return 0, "", fmt.Errorf("%w: artifact %d is not an upload of this job", domain.ErrNotFound, artifactID)
569 }
570 if a.State != domain.CIArtifactPending {
571 if bytes.Equal(a.ListDigest, list) {
572 return a.Size, a.Digest, nil // a repeat
573 }
574 return 0, "", fmt.Errorf("%w: artifact %d is committed with another block list", domain.ErrPreconditionFailed, a.ID)
575 }
576 blocks, err := c.cfg.Artifacts.ArtifactBlocks(ctx, a.ID)
577 if err != nil {
578 return 0, "", err
579 }
580 byID := make(map[string]ports.CIArtifactBlock, len(blocks))
581 for _, b := range blocks {
582 byID[string(b.ID)] = b
583 }
584 files := make([]string, 0, len(blockIDs))
585 var total int64
586 for _, id := range blockIDs {
587 b, ok := byID[string(id)]
588 if !ok {
589 return 0, "", fmt.Errorf("%w: block %x was never uploaded", domain.ErrInvalid, id)
590 }
591 files = append(files, b.File)
592 total += b.Size
593 }
594 if total > limits.MaxSize {
595 return 0, "", c.errArtifactTooLarge()
596 }
597 // The repeats of the list make the content larger than the blocks
598 // charged: that excess is reserved before the content is written.
599 excess := max(total-a.Charged, 0)
600 if excess > 0 {
601 if err := c.reserve(ctx, a, excess); err != nil {
602 return 0, "", err
603 }
604 }
605 size, digest, err := c.cfg.ArtifactFiles.Assemble(ctx, artifactRef(a), files)
606 if err != nil {
607 c.release(a, excess)
608 }
609 if errors.Is(err, domain.ErrNotFound) {
610 continue // a block was replaced meanwhile
611 }
612 if err != nil {
613 return 0, "", err
614 }
615 err = c.updateArtifacts(ctx, a.Repo, func(used int64, w *ports.CIArtifactWrite) error {
616 cur, err := c.cfg.Artifacts.GetArtifact(ctx, a.ID)
617 if err != nil {
618 return err
619 }
620 if cur.Version != a.Version {
621 return errArtifactChanged
622 }
623 w.Used += size - cur.Charged
624 if w.Used > limits.Quota {
625 return c.errQuota(cur.Repo)
626 }
627 cur.State, cur.Size, cur.Digest, cur.ListDigest, cur.Charged = domain.CIArtifactCommitted, size, digest, list, size
628 w.Put, w.DropBlocks = []domain.CIArtifact{cur}, []int64{cur.ID}
629 return nil
630 })
631 c.release(a, excess)
632 if errors.Is(err, errArtifactChanged) {
633 continue
634 }
635 if errors.Is(err, domain.ErrNotFound) {
636 c.removeArtifactFiles(ctx, []domain.CIArtifact{a}) // discarded meanwhile
637 }
638 if err != nil {
639 return 0, "", err
640 }
641 if err := c.cfg.ArtifactFiles.RemoveBlocks(ctx, artifactRef(a)); err != nil {
642 ctxLog(ctx).Warn().Err(err).Int64("artifact_id", a.ID).Msg("ci: remove the blocks of a committed artifact")
643 }
644 c.dropIfEnded(ctx, j)
645 return size, digest, nil
646 }
647 return 0, "", fmt.Errorf("%w: the blocks of artifact %d keep changing, try again", domain.ErrUnavailable, artifactID)
648}
649
650// FinalizeJobArtifact makes the committed upload of the job with that name
651// visible. An artifact of that name from an earlier attempt is deleted in the
652// same write; one finalized in this attempt is already_exists.
653func (c *CI) FinalizeJobArtifact(ctx context.Context, secret, jobID, name string, size int64, digest string) (domain.CIArtifact, error) {
654 j, err := c.artifactJob(ctx, secret, jobID)
655 if err != nil {
656 return domain.CIArtifact{}, err
657 }
658 if err := domain.ParseArtifactName(name); err != nil {
659 return domain.CIArtifact{}, err
660 }
661 digest, err = domain.ParseArtifactDigest(digest)
662 if err != nil {
663 return domain.CIArtifact{}, err
664 }
665 now := c.now()
666 var out domain.CIArtifact
667 var replaced []domain.CIArtifact
668 err = c.updateArtifacts(ctx, j.Repo, func(used int64, w *ports.CIArtifactWrite) error {
669 arts, err := c.cfg.Artifacts.RunArtifacts(ctx, j.Repo, j.RunID)
670 if err != nil {
671 return err
672 }
673 var mine, done *domain.CIArtifact
674 var finals []domain.CIArtifact
675 for i, a := range arts {
676 switch {
677 case a.Name != name:
678 case a.PendingOf(j):
679 if mine == nil || a.ID > mine.ID {
680 mine = &arts[i]
681 }
682 case a.Final():
683 finals = append(finals, a)
684 if a.UploadedBy(j) && a.Visible(now) {
685 done = &arts[i]
686 }
687 }
688 }
689 switch {
690 case mine == nil && done != nil:
691 out = *done // a repeat
692 return errNoArtifactWrite
693 case mine == nil:
694 return fmt.Errorf("%w: this job has no upload of an artifact named %q", domain.ErrNotFound, name)
695 case mine.State != domain.CIArtifactCommitted:
696 return fmt.Errorf("%w: artifact %q is not committed", domain.ErrPreconditionFailed, name)
697 case digest != "" && digest != mine.Digest:
698 return fmt.Errorf("%w: the digest %s is not the digest of the content (%s)", domain.ErrInvalid, digest, mine.Digest)
699 }
700 replaced = replaced[:0]
701 for _, f := range finals {
702 if f.Visible(now) && f.Attempt == j.RunAttempt {
703 return fmt.Errorf("%w: an artifact with this name already exists on the workflow run", domain.ErrAlreadyExists)
704 }
705 replaced = append(replaced, f) // an earlier attempt's, or expired
706 w.Used -= f.Charged
707 }
708 out = *mine
709 out.State = domain.CIArtifactFinal
710 w.Put, w.Delete = []domain.CIArtifact{out}, replaced
711 w.Used = max(w.Used, 0)
712 return nil
713 })
714 if errors.Is(err, errNoArtifactWrite) {
715 return out, nil
716 }
717 if err != nil {
718 return domain.CIArtifact{}, err
719 }
720 c.removeArtifactFiles(ctx, replaced)
721 log := zerowrap.FromCtx(ctx)
722 ev := log.Info()
723 if size != out.Size {
724 ev = ev.Int64("uploader_size", size)
725 }
726 ev.Int64("artifact_id", out.ID).Int64("run_id", out.RunID).Str("job_id", j.ID).Int64("size", out.Size).Int("replaced", len(replaced)).Msg("artifact finalized")
727 return out, nil
728}
729
730// ListJobArtifacts lists the visible artifacts of the job's run.
731func (c *CI) ListJobArtifacts(ctx context.Context, secret, jobID, name string, artifactID int64) ([]domain.CIArtifact, error) {
732 j, err := c.artifactJob(ctx, secret, jobID)
733 if err != nil {
734 return nil, err
735 }
736 arts, err := c.visibleArtifacts(ctx, j.Repo, j.RunID)
737 if err != nil {
738 return nil, err
739 }
740 out := arts[:0]
741 for _, a := range arts {
742 if (name == "" || a.Name == name) && (artifactID == 0 || a.ID == artifactID) {
743 out = append(out, a)
744 }
745 }
746 return out, nil
747}
748
749func (c *CI) visibleArtifacts(ctx context.Context, repo domain.RepoName, runID int64) ([]domain.CIArtifact, error) {
750 arts, err := c.cfg.Artifacts.RunArtifacts(ctx, repo, runID)
751 if err != nil {
752 return nil, err
753 }
754 now := c.now()
755 out := make([]domain.CIArtifact, 0, len(arts))
756 for _, a := range arts {
757 if a.Visible(now) {
758 out = append(out, a)
759 }
760 }
761 return out, nil
762}
763
764// DownloadJobArtifact streams a visible artifact of the job's run.
765func (c *CI) DownloadJobArtifact(ctx context.Context, secret, jobID string, artifactID, offset int64, send func(ports.ArtifactChunk) error) error {
766 j, err := c.artifactJob(ctx, secret, jobID)
767 if err != nil {
768 return err
769 }
770 a, err := c.cfg.Artifacts.GetArtifact(ctx, artifactID)
771 if err != nil {
772 return err
773 }
774 if a.Repo != j.Repo || a.RunID != j.RunID || !a.Visible(c.now()) {
775 return fmt.Errorf("%w: artifact %d is not an artifact of this run", domain.ErrNotFound, artifactID)
776 }
777 return c.streamArtifact(ctx, a, offset, send)
778}
779
780// streamArtifact sends the artifact, then its content from offset.
781func (c *CI) streamArtifact(ctx context.Context, a domain.CIArtifact, offset int64, send func(ports.ArtifactChunk) error) error {
782 if offset < 0 || offset > a.Size {
783 return fmt.Errorf("%w: offset %d is beyond the size %d", domain.ErrOutOfRange, offset, a.Size)
784 }
785 f, err := c.cfg.ArtifactFiles.Open(ctx, artifactRef(a), offset)
786 if err != nil {
787 return err
788 }
789 defer func() { _ = f.Close() }()
790 if err := send(ports.ArtifactChunk{Artifact: &a}); err != nil {
791 return err
792 }
793 left := a.Size - offset
794 buf := make([]byte, min(left, domain.CIArtifactChunk))
795 for left > 0 {
796 n, err := io.ReadFull(f, buf[:min(left, int64(len(buf)))])
797 if err != nil {
798 return fmt.Errorf("read artifact %d: %w", a.ID, err)
799 }
800 left -= int64(n)
801 if err := send(ports.ArtifactChunk{Data: buf[:n]}); err != nil {
802 return err
803 }
804 }
805 return nil
806}
807
808// DeleteJobArtifact deletes the visible artifact of that name of the job's
809// run.
810func (c *CI) DeleteJobArtifact(ctx context.Context, secret, jobID, name string) (int64, error) {
811 j, err := c.artifactJob(ctx, secret, jobID)
812 if err != nil {
813 return 0, err
814 }
815 if err := domain.ParseArtifactName(name); err != nil {
816 return 0, err
817 }
818 now := c.now()
819 removed, err := c.removeArtifacts(ctx, j.Repo, j.RunID, func(a domain.CIArtifact) bool {
820 return a.Name == name && a.Visible(now)
821 })
822 if err != nil {
823 return 0, err
824 }
825 if len(removed) == 0 {
826 return 0, fmt.Errorf("%w: no artifact named %q on this run", domain.ErrNotFound, name)
827 }
828 log := zerowrap.FromCtx(ctx)
829 log.Info().Int64("artifact_id", removed[0].ID).Int64("run_id", j.RunID).Str("job_id", j.ID).Msg("artifact deleted by a job")
830 return removed[0].ID, nil
831}
832
833// --- ArtifactService ---
834
835// artifactRepo is the access rule of reading artifacts: a signed-in caller
836// with repo:read, even on a public repository (GitHub serves no artifact
837// anonymously).
838func (c *CI) artifactRepo(ctx context.Context, p domain.Principal, name domain.RepoName) (domain.Repo, error) {
839 if !c.artifactsOn() {
840 return domain.Repo{}, fmt.Errorf("%w: artifacts are off on this server", domain.ErrPreconditionFailed)
841 }
842 repo, err := c.readableRepo(ctx, p, name)
843 if err != nil {
844 return domain.Repo{}, err
845 }
846 if err := scopeAccess(p, repo, domain.ScopeRepoRead); err != nil {
847 return domain.Repo{}, err
848 }
849 return repo, nil
850}
851
852// ListArtifacts lists the visible artifacts of a run with the space of the
853// repository.
854func (c *CI) ListArtifacts(ctx context.Context, p domain.Principal, name domain.RepoName, runID int64) (ports.ArtifactList, error) {
855 if _, err := c.artifactRepo(ctx, p, name); err != nil {
856 return ports.ArtifactList{}, err
857 }
858 if _, err := c.runOf(ctx, name, runID); err != nil {
859 return ports.ArtifactList{}, err
860 }
861 arts, err := c.visibleArtifacts(ctx, name, runID)
862 if err != nil {
863 return ports.ArtifactList{}, err
864 }
865 used, _, err := c.cfg.Artifacts.ArtifactUsage(ctx, name)
866 if err != nil {
867 return ports.ArtifactList{}, err
868 }
869 return ports.ArtifactList{Artifacts: arts, Used: used, Quota: c.cfg.ArtifactLimits.Quota}, nil
870}
871
872// ownerArtifact returns a visible artifact of the repository.
873func (c *CI) ownerArtifact(ctx context.Context, name domain.RepoName, id int64) (domain.CIArtifact, error) {
874 a, err := c.cfg.Artifacts.GetArtifact(ctx, id)
875 if err != nil {
876 return domain.CIArtifact{}, err
877 }
878 if a.Repo != name || !a.Visible(c.now()) {
879 return domain.CIArtifact{}, fmt.Errorf("%w: artifact %d is not an artifact of %s", domain.ErrNotFound, id, name)
880 }
881 return a, nil
882}
883
884// DownloadArtifact streams a visible artifact of the repository.
885func (c *CI) DownloadArtifact(ctx context.Context, p domain.Principal, name domain.RepoName, artifactID, offset int64, send func(ports.ArtifactChunk) error) error {
886 if _, err := c.artifactRepo(ctx, p, name); err != nil {
887 return err
888 }
889 a, err := c.ownerArtifact(ctx, name, artifactID)
890 if err != nil {
891 return err
892 }
893 return c.streamArtifact(ctx, a, offset, send)
894}
895
896// DeleteArtifact deletes an artifact. It needs ci:write on the repository and,
897// like the other writes on a run, the branch of the run within the prefixes of
898// a limited token.
899func (c *CI) DeleteArtifact(ctx context.Context, p domain.Principal, name domain.RepoName, artifactID int64) error {
900 if !c.artifactsOn() {
901 return fmt.Errorf("%w: artifacts are off on this server", domain.ErrPreconditionFailed)
902 }
903 if _, err := c.writableRepo(ctx, p, name); err != nil {
904 return err
905 }
906 a, err := c.ownerArtifact(ctx, name, artifactID)
907 if err != nil {
908 return err
909 }
910 run, err := c.runOf(ctx, name, a.RunID)
911 if err != nil {
912 return err
913 }
914 if err := runWriteAccess(p, run); err != nil {
915 return err
916 }
917 removed, err := c.removeArtifacts(ctx, name, a.RunID, func(o domain.CIArtifact) bool { return o.ID == a.ID && o.Final() })
918 if err != nil {
919 return err
920 }
921 if len(removed) == 0 {
922 return fmt.Errorf("%w: artifact %d is not an artifact of %s", domain.ErrNotFound, artifactID, name)
923 }
924 log := zerowrap.FromCtx(ctx)
925 log.Info().Int64("artifact_id", a.ID).Int64("run_id", a.RunID).Str("repo", name.String()).
926 Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("artifact deleted")
927 return nil
928}
929
930// --- sweep ---
931
932// sweepArtifacts deletes the expired artifacts, those of runs that no longer
933// exist, and the unfinished uploads of jobs that ended, were given again or
934// began more than domain.CIArtifactPendingMax ago; then the blocks and files
935// that no record names (a crash between a file and its record, a repository
936// deleted while the server was down). Every failure is logged and the sweep
937// goes on.
938func (c *CI) sweepArtifacts(ctx context.Context) {
939 if !c.artifactsOn() {
940 return
941 }
942 log := ctxLog(ctx)
943 all, err := c.cfg.Artifacts.AllArtifacts(ctx)
944 if err != nil {
945 log.Error().Err(err).Msg("ci: list the artifacts")
946 return
947 }
948 type runKey struct {
949 repo domain.RepoName
950 run int64
951 }
952 runs := map[runKey]bool{}
953 for _, a := range all {
954 runs[runKey{a.Repo, a.RunID}] = true
955 }
956 removed := 0
957 for k := range runs {
958 gone := false
959 if _, err := c.cfg.Store.GetRunSource(ctx, k.run); errors.Is(err, domain.ErrNotFound) {
960 gone = true
961 }
962 now := c.now()
963 arts, err := c.removeArtifacts(ctx, k.repo, k.run, func(a domain.CIArtifact) bool {
964 return gone || c.artifactDead(ctx, a, now)
965 })
966 if err != nil && !errors.Is(err, domain.ErrNotFound) {
967 log.Error().Err(err).Int64("run_id", k.run).Msg("ci: remove the expired artifacts of a run")
968 }
969 removed += len(arts)
970 }
971 if removed > 0 {
972 log.Info().Int("artifacts", removed).Msg("ci: expired artifacts removed")
973 }
974 c.sweepArtifactOrphans(ctx)
975}
976
977// artifactDead reports whether the sweep deletes an artifact of a run that
978// exists.
979func (c *CI) artifactDead(ctx context.Context, a domain.CIArtifact, now time.Time) bool {
980 if a.Final() {
981 return !now.Before(a.ExpiresAt)
982 }
983 if now.Sub(a.CreatedAt) >= domain.CIArtifactPendingMax {
984 return true
985 }
986 j, err := c.cfg.Store.GetJob(ctx, a.RunID, a.Attempt, a.JobID)
987 switch {
988 case errors.Is(err, domain.ErrNotFound):
989 return true
990 case err != nil:
991 return false
992 }
993 return j.Ended() || j.TokenHash != a.Assignment
994}
995
996// sweepArtifactOrphans removes the blocks and files that no record names.
997func (c *CI) sweepArtifactOrphans(ctx context.Context) {
998 log := ctxLog(ctx)
999 if ids, err := c.cfg.Artifacts.BlockArtifactIDs(ctx); err != nil {
1000 log.Error().Err(err).Msg("ci: list the staged artifact blocks")
1001 } else {
1002 for _, id := range ids {
1003 if _, err := c.cfg.Artifacts.GetArtifact(ctx, id); !errors.Is(err, domain.ErrNotFound) {
1004 continue
1005 }
1006 if err := c.cfg.Artifacts.DeleteBlocks(ctx, id); err != nil {
1007 log.Error().Err(err).Int64("artifact_id", id).Msg("ci: remove the blocks of an artifact that no longer exists")
1008 }
1009 }
1010 }
1011 c.removeOrphans(ctx, "artifacts", "artifact directories", c.cfg.ArtifactFiles.RunIDs, c.cfg.ArtifactFiles.RemoveRuns)
1012 runIDs, err := c.cfg.ArtifactFiles.RunIDs(ctx)
1013 if err != nil {
1014 log.Error().Err(err).Msg("ci: list the artifact directories")
1015 return
1016 }
1017 for _, run := range runIDs {
1018 ids, err := c.cfg.ArtifactFiles.ArtifactIDs(ctx, run)
1019 if err != nil {
1020 log.Error().Err(err).Int64("run_id", run).Msg("ci: list the artifact files of a run")
1021 continue
1022 }
1023 for _, id := range ids {
1024 a, err := c.cfg.Artifacts.GetArtifact(ctx, id)
1025 if err == nil && a.RunID == run {
1026 continue
1027 }
1028 if err != nil && !errors.Is(err, domain.ErrNotFound) {
1029 continue
1030 }
1031 ref := ports.CIArtifactRef{RunID: run, ID: id}
1032 if err := c.cfg.ArtifactFiles.Remove(ctx, ref); err != nil {
1033 log.Error().Err(err).Int64("artifact_id", id).Msg("ci: remove the files of an artifact that no longer exists")
1034 }
1035 }
1036 }
1037}