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 "strings"
17 "sync"
18 "time"
19
20 "github.com/bnema/zerowrap"
21
22 "bnema.dev/ermite/internal/domain"
23 "bnema.dev/ermite/internal/ports"
24)
25
26// The artifacts of the runs (artifacts v4, CI-4 D3). The runner translates the
27// toolkit's protocol and calls JobArtifactService with the job token; owners
28// use ArtifactService. The records are in CIConfig.Artifacts, the contents in
29// CIConfig.Blobs (ArtifactFiles for legacy records), named by ids only.
30//
31// Every write of the artifacts of a repository goes through its usage record
32// (ports.CIArtifactWrite): a decision is made on the records read after it and
33// written under its version, or made again. So the quota, the unique names of
34// a run and the states of an upload hold with concurrent calls, and the bytes
35// of a write are counted before it is accepted. The files are written before
36// the record that refers to them and removed after the record that dropped
37// them: a crash leaves at most files that no record names, which the sweep
38// removes, never a record without its file.
39
40var (
41 _ ports.JobArtifactService = (*CI)(nil)
42 _ ports.ArtifactService = (*CI)(nil)
43)
44
45const (
46 // artifactTries bounds the retries of a write of the artifacts of a
47 // repository that other writes keep changing.
48 artifactTries = 256
49 // commitTries bounds the assemblies of one commit whose blocks change
50 // meanwhile.
51 commitTries = 5
52)
53
54// ciArtifactUploadSlots bounds the artifact block uploads and assemblies that
55// hold an 8 MiB buffer at the same time, server-wide.
56const ciArtifactUploadSlots = 16
57
58// blockBufs pools the 8 MiB buffers of uploads and assemblies. CIArtifactMaxBlockSize
59// and ports.BlobPartSize are both 8 MiB.
60var blockBufs = sync.Pool{New: func() any {
61 b := make([]byte, max(domain.CIArtifactMaxBlockSize, ports.BlobPartSize))
62 return &b
63}}
64
65// acquireUploadSlot takes an upload slot without waiting; release frees it.
66func (c *CI) acquireUploadSlot() (release func(), err error) {
67 if c.uploadSlots == nil {
68 return func() {}, nil
69 }
70 select {
71 case c.uploadSlots <- struct{}{}:
72 return func() { <-c.uploadSlots }, nil
73 default:
74 return nil, fmt.Errorf("%w: too many artifact uploads in progress, retry", domain.ErrResourceExhausted)
75 }
76}
77
78// waitUploadSlot takes an upload slot, waiting for one until ctx ends: a
79// commit must not fail because block uploads are in flight.
80func (c *CI) waitUploadSlot(ctx context.Context) (release func(), err error) {
81 if c.uploadSlots == nil {
82 return func() {}, nil
83 }
84 select {
85 case c.uploadSlots <- struct{}{}:
86 return func() { <-c.uploadSlots }, nil
87 case <-ctx.Done():
88 return nil, ctx.Err()
89 }
90}
91
92var (
93 // errNoArtifactWrite ends updateArtifacts without a write.
94 errNoArtifactWrite = errors.New("no write")
95 // errArtifactChanged makes a commit start again.
96 errArtifactChanged = errors.New("artifact changed")
97)
98
99func (c *CI) artifactsOn() bool { return c.cfg.Artifacts != nil && c.cfg.Blobs != nil }
100
101func artifactRef(a domain.CIArtifact) ports.CIArtifactRef {
102 return ports.CIArtifactRef{RunID: a.RunID, ID: a.ID}
103}
104
105// artifactJob authenticates the job token of an artifact call: the job of
106// jobID, which has not ended.
107func (c *CI) artifactJob(ctx context.Context, secret, jobID string) (domain.CIJob, error) {
108 if !c.artifactsOn() {
109 return domain.CIJob{}, fmt.Errorf("%w: artifacts are off on this server", domain.ErrPreconditionFailed)
110 }
111 j, err := c.jobFor(ctx, secret, jobID)
112 if err != nil {
113 return domain.CIJob{}, err
114 }
115 if j.Ended() {
116 return domain.CIJob{}, endedError(j)
117 }
118 return j, nil
119}
120
121// updateArtifacts reads the usage of the repository, lets fn fill the write
122// from the records it reads then, and applies it; on a conflict it starts
123// again. fn returns errNoArtifactWrite to write nothing.
124func (c *CI) updateArtifacts(ctx context.Context, repo domain.RepoName, fn func(used int64, w *ports.CIArtifactWrite) error) error {
125 for range artifactTries {
126 used, version, err := c.cfg.Artifacts.ArtifactUsage(ctx, repo)
127 if err != nil {
128 return err
129 }
130 w := ports.CIArtifactWrite{Repo: repo, Used: used, UsageVersion: version}
131 if err := fn(used, &w); err != nil {
132 return err
133 }
134 err = c.cfg.Artifacts.ApplyArtifacts(ctx, w)
135 if errors.Is(err, domain.ErrConflict) {
136 continue
137 }
138 return err
139 }
140 return fmt.Errorf("%w: the artifacts of %s keep changing, try again", domain.ErrUnavailable, repo)
141}
142
143// --- limits ---
144
145// sizeText writes a limit for a message: in GiB or MiB when it is a whole
146// number of them.
147func sizeText(n int64) string {
148 switch {
149 case n >= 1<<30 && n%(1<<30) == 0:
150 return strconv.FormatInt(n>>30, 10) + " GiB"
151 case n >= 1<<20 && n%(1<<20) == 0:
152 return strconv.FormatInt(n>>20, 10) + " MiB"
153 }
154 return strconv.FormatInt(n, 10) + " bytes"
155}
156
157func (c *CI) errArtifactTooLarge() error {
158 return fmt.Errorf("%w: the artifact would exceed the size limit of one artifact (%s)", domain.ErrResourceExhausted, sizeText(c.cfg.ArtifactLimits.MaxSize))
159}
160
161func (c *CI) errQuota(repo domain.RepoName) error {
162 return fmt.Errorf("%w: the artifacts of %s would exceed their quota (%s)", domain.ErrResourceExhausted, repo, sizeText(c.cfg.ArtifactLimits.Quota))
163}
164
165// --- reservations ---
166
167// artifactReservations are the bytes being written and not charged yet:
168// blocks streaming in, the repeats of a commit being assembled. A write
169// reserves its bytes before they reach the disk, under the lock of its
170// repository and against the charges stored and the other reservations, and
171// releases them once its record is written or refused. So concurrent writes
172// never put more on disk than the limits leave, and the write of the record
173// checks them again. The size of one artifact and the quota are both limits of
174// one repository: repositories never wait for one another.
175type artifactReservations struct {
176 mu sync.Mutex
177 repos map[domain.RepoName]*repoReservations
178}
179
180// repoReservations are the reservations of one repository. An entry stays
181// once made (a few words per repository that ever uploaded); its maps empty.
182type repoReservations struct {
183 mu sync.Mutex
184 total int64
185 art map[int64]int64
186}
187
188func (r *artifactReservations) of(repo domain.RepoName) *repoReservations {
189 r.mu.Lock()
190 defer r.mu.Unlock()
191 if r.repos == nil {
192 r.repos = map[domain.RepoName]*repoReservations{}
193 }
194 rr := r.repos[repo]
195 if rr == nil {
196 rr = &repoReservations{art: map[int64]int64{}}
197 r.repos[repo] = rr
198 }
199 return rr
200}
201
202// reserve takes n bytes for artifact a, or returns the limit it would cross.
203func (c *CI) reserve(ctx context.Context, a domain.CIArtifact, n int64) error {
204 r := c.reservations.of(a.Repo)
205 r.mu.Lock()
206 defer r.mu.Unlock()
207 used, _, err := c.cfg.Artifacts.ArtifactUsage(ctx, a.Repo)
208 if err != nil {
209 return err
210 }
211 cur, err := c.cfg.Artifacts.GetArtifact(ctx, a.ID)
212 if err != nil {
213 return err
214 }
215 limits := c.cfg.ArtifactLimits
216 switch {
217 case cur.Charged+r.art[a.ID]+n > limits.MaxSize:
218 return c.errArtifactTooLarge()
219 case used+r.total+n > limits.Quota:
220 return c.errQuota(a.Repo)
221 }
222 r.total += n
223 r.art[a.ID] += n
224 return nil
225}
226
227// release gives back n bytes reserved for a.
228func (c *CI) release(a domain.CIArtifact, n int64) {
229 if n == 0 {
230 return
231 }
232 r := c.reservations.of(a.Repo)
233 r.mu.Lock()
234 defer r.mu.Unlock()
235 r.total = max(r.total-n, 0)
236 if r.art[a.ID] -= n; r.art[a.ID] <= 0 {
237 delete(r.art, a.ID)
238 }
239}
240
241// reserveAhead is how much a reservingReader reserves at once, so that a
242// large block takes one reservation per MiB, not one per read.
243const reserveAhead = domain.CIArtifactChunk
244
245// reservingReader passes on the data of r, each read covered by a
246// reservation first: a read that the limits refuse fails without being passed
247// on, so a block is never written beyond them. It reserves reserveAhead at a
248// time, or just what a read needs when the limits leave less; reserved is all
249// it holds (written or not), released whole after the record is written.
250type reservingReader struct {
251 ctx context.Context
252 c *CI
253 a domain.CIArtifact
254 r io.Reader
255 reserved int64
256 left int64 // reserved and not yet passed on
257}
258
259func (rr *reservingReader) Read(p []byte) (int, error) {
260 n, err := rr.r.Read(p)
261 if need := int64(n) - rr.left; need > 0 {
262 got := max(need, reserveAhead)
263 if rerr := rr.c.reserve(rr.ctx, rr.a, got); rerr != nil {
264 if got == need {
265 return 0, rerr
266 }
267 got = need
268 if rerr := rr.c.reserve(rr.ctx, rr.a, got); rerr != nil {
269 return 0, rerr
270 }
271 }
272 rr.reserved += got
273 rr.left += got
274 }
275 rr.left -= int64(n)
276 return n, err
277}
278
279// --- locks of the commits ---
280
281// lockArtifact serializes the commits of an artifact: two commits must not
282// assemble its content at once. It returns the unlock function.
283func (c *CI) lockArtifact(id int64) func() {
284 c.mu.Lock()
285 if c.artifactLocks == nil {
286 c.artifactLocks = map[int64]*artifactLock{}
287 }
288 l := c.artifactLocks[id]
289 if l == nil {
290 l = &artifactLock{}
291 c.artifactLocks[id] = l
292 }
293 l.users++
294 c.mu.Unlock()
295 l.Lock()
296 return func() {
297 l.Unlock()
298 c.mu.Lock()
299 if l.users--; l.users == 0 {
300 delete(c.artifactLocks, id)
301 }
302 c.mu.Unlock()
303 }
304}
305
306type artifactLock struct {
307 sync.Mutex
308 users int
309}
310
311// --- removal ---
312
313// removeArtifacts deletes the artifacts of a run that pick chooses, judged on
314// the records as they are when the write is made, refunds them, then removes
315// their files. It returns the deleted ones.
316func (c *CI) removeArtifacts(ctx context.Context, repo domain.RepoName, runID int64, pick func(domain.CIArtifact) bool) ([]domain.CIArtifact, error) {
317 var removed []domain.CIArtifact
318 err := c.updateArtifacts(ctx, repo, func(used int64, w *ports.CIArtifactWrite) error {
319 arts, err := c.cfg.Artifacts.RunArtifacts(ctx, repo, runID)
320 if err != nil {
321 return err
322 }
323 removed = removed[:0]
324 for _, a := range arts {
325 if pick(a) {
326 removed = append(removed, a)
327 w.Used -= a.Charged
328 }
329 }
330 if len(removed) == 0 {
331 return errNoArtifactWrite
332 }
333 w.Delete = removed
334 w.Used = max(w.Used, 0)
335 return nil
336 })
337 if errors.Is(err, errNoArtifactWrite) {
338 return nil, nil
339 }
340 if err != nil {
341 return nil, err
342 }
343 c.removeArtifactFiles(ctx, removed)
344 return removed, nil
345}
346
347// removeArtifactFiles removes the files of deleted artifacts. A failure is
348// logged: the sweep removes the files that no record names.
349func (c *CI) removeArtifactFiles(ctx context.Context, arts []domain.CIArtifact) {
350 for _, a := range arts {
351 var err error
352 if a.Store == domain.CIArtifactStoreBlob {
353 err = errors.Join(c.cfg.Blobs.Delete(ctx, a.BlobKey()), c.removeBlobBlocks(ctx, a))
354 } else if c.cfg.ArtifactFiles != nil {
355 err = c.cfg.ArtifactFiles.Remove(ctx, artifactRef(a))
356 }
357 if err != nil {
358 ctxLog(ctx).Warn().Err(err).Int64("artifact_id", a.ID).Msg("ci: remove the files of an artifact; the sweep will")
359 }
360 }
361}
362
363// dropIfEnded discards the uploads of the job's assignment when the job ended
364// or was given again while the call ran: the cleanup of the end may have run
365// before the call wrote its record.
366func (c *CI) dropIfEnded(ctx context.Context, j domain.CIJob) {
367 cur, err := c.cfg.Store.GetJob(ctx, j.RunID, j.RunAttempt, j.ID)
368 if err == nil && !cur.Ended() && cur.TokenHash == j.TokenHash {
369 return
370 }
371 c.dropPending(ctx, j, func(a domain.CIArtifact) bool { return a.Assignment == j.TokenHash })
372}
373
374// dropPending discards the unfinished uploads of a job, of any of its
375// assignments, that pick chooses.
376func (c *CI) dropPending(ctx context.Context, j domain.CIJob, pick func(domain.CIArtifact) bool) {
377 removed, err := c.removeArtifacts(ctx, j.Repo, j.RunID, func(a domain.CIArtifact) bool {
378 return !a.Final() && a.UploadedBy(j) && pick(a)
379 })
380 if err != nil && !errors.Is(err, domain.ErrNotFound) {
381 ctxLog(ctx).Error().Err(err).Str("job_id", j.ID).Msg("ci: discard the unfinished artifacts of a job; the sweep will")
382 return
383 }
384 if len(removed) > 0 {
385 ctxLog(ctx).Info().Str("job_id", j.ID).Int64("run_id", j.RunID).Int("artifacts", len(removed)).Msg("ci: unfinished artifacts discarded")
386 }
387}
388
389// artifactsJobEnded discards the unfinished uploads of a job that ended. It is
390// called after the end is written.
391func (c *CI) artifactsJobEnded(ctx context.Context, j domain.CIJob) {
392 if !c.artifactsOn() {
393 return
394 }
395 c.dropPending(ctx, j, func(domain.CIArtifact) bool { return true })
396}
397
398// --- JobArtifactService ---
399
400// CreateJobArtifact begins an upload: a pending artifact with a new id. A
401// pending artifact of the same job and name is discarded.
402func (c *CI) CreateJobArtifact(ctx context.Context, secret string, params ports.CreateJobArtifactParams) (ports.CreatedArtifact, error) {
403 j, err := c.artifactJob(ctx, secret, params.JobID)
404 if err != nil {
405 return ports.CreatedArtifact{}, err
406 }
407 if err := domain.ParseArtifactName(params.Name); err != nil {
408 return ports.CreatedArtifact{}, err
409 }
410 mediaType, err := domain.ParseArtifactMediaType(params.MediaType)
411 if err != nil {
412 return ports.CreatedArtifact{}, err
413 }
414 now := c.now()
415 id, err := c.cfg.Artifacts.NextArtifactID(ctx)
416 if err != nil {
417 return ports.CreatedArtifact{}, err
418 }
419 a := domain.CIArtifact{
420 ID: id, Repo: j.Repo, RunID: j.RunID, Attempt: j.RunAttempt, JobID: j.ID, Assignment: j.TokenHash,
421 Name: params.Name, MediaType: mediaType, State: domain.CIArtifactPending, Store: domain.CIArtifactStoreBlob,
422 CreatedAt: now, ExpiresAt: domain.ArtifactExpiry(params.ExpiresAt, now, c.cfg.ArtifactLimits.Retention),
423 }
424 var discarded []domain.CIArtifact
425 err = c.updateArtifacts(ctx, j.Repo, func(used int64, w *ports.CIArtifactWrite) error {
426 if used >= c.cfg.ArtifactLimits.Quota {
427 return c.errQuota(j.Repo)
428 }
429 arts, err := c.cfg.Artifacts.RunArtifacts(ctx, j.Repo, j.RunID)
430 if err != nil {
431 return err
432 }
433 discarded = discarded[:0]
434 pending := 0
435 for _, o := range arts {
436 switch {
437 case o.Name != a.Name:
438 if !o.Final() && o.UploadedBy(j) {
439 pending++
440 }
441 case o.Visible(now) && o.Attempt == j.RunAttempt:
442 return fmt.Errorf("%w: an artifact with this name already exists on the workflow run", domain.ErrAlreadyExists)
443 case !o.Final() && o.UploadedBy(j):
444 discarded = append(discarded, o) // a new create is a new upload
445 w.Used -= o.Charged
446 }
447 }
448 switch {
449 case pending >= domain.CIArtifactMaxPendingPerJob:
450 return fmt.Errorf("%w: a job has at most %d unfinished artifact uploads", domain.ErrResourceExhausted, domain.CIArtifactMaxPendingPerJob)
451 case len(arts)-len(discarded) >= domain.CIArtifactMaxPerRun:
452 return fmt.Errorf("%w: a run has at most %d artifacts", domain.ErrResourceExhausted, domain.CIArtifactMaxPerRun)
453 }
454 w.Put, w.Delete = []domain.CIArtifact{a}, discarded
455 return nil
456 })
457 if err != nil {
458 return ports.CreatedArtifact{}, err
459 }
460 c.removeArtifactFiles(ctx, discarded)
461 log := zerowrap.FromCtx(ctx)
462 log.Info().Int64("artifact_id", id).Int64("run_id", j.RunID).Str("job_id", j.ID).Msg("artifact upload begun")
463 c.dropIfEnded(ctx, j)
464 return ports.CreatedArtifact{ID: id, ExpiresAt: a.ExpiresAt}, nil
465}
466
467// pendingArtifact returns an unfinished upload of the job's assignment, or
468// domain.ErrNotFound.
469func (c *CI) pendingArtifact(ctx context.Context, j domain.CIJob, id int64) (domain.CIArtifact, error) {
470 a, err := c.cfg.Artifacts.GetArtifact(ctx, id)
471 if err != nil {
472 return domain.CIArtifact{}, err
473 }
474 if a.Repo != j.Repo || !a.PendingOf(j) {
475 return domain.CIArtifact{}, fmt.Errorf("%w: artifact %d is not an upload of this job", domain.ErrNotFound, id)
476 }
477 return a, nil
478}
479
480func errCommitted(id int64) error {
481 return fmt.Errorf("%w: artifact %d is already committed", domain.ErrPreconditionFailed, id)
482}
483
484func newBlockFile(blockID []byte) string {
485 var raw [8]byte
486 _, _ = rand.Read(raw[:])
487 return hex.EncodeToString(blockID) + "." + hex.EncodeToString(raw[:])
488}
489
490// UploadJobArtifactBlock stores a block: the data goes to a new file, counted
491// against the limits as it comes, then the record names it in place of the
492// block of the same id, if any.
493func (c *CI) UploadJobArtifactBlock(ctx context.Context, secret, jobID string, artifactID int64, blockID []byte, data io.Reader) (int64, error) {
494 j, err := c.artifactJob(ctx, secret, jobID)
495 if err != nil {
496 return 0, err
497 }
498 if err := domain.CheckArtifactBlockID(blockID); err != nil {
499 return 0, err
500 }
501 a, err := c.pendingArtifact(ctx, j, artifactID)
502 if err != nil {
503 return 0, err
504 }
505 if a.State != domain.CIArtifactPending {
506 return 0, errCommitted(a.ID)
507 }
508 // The slot is taken before anything is reserved, read or allocated.
509 freeSlot, err := c.acquireUploadSlot()
510 if err != nil {
511 return 0, err
512 }
513 defer freeSlot()
514 // Every byte is reserved before it is written, against the charges and
515 // the other writes in flight (a block it replaces counts until the record
516 // drops it): nothing beyond the limits ever reaches the disk.
517 src := &reservingReader{ctx: ctx, c: c, a: a, r: data}
518 defer func() { c.release(a, src.reserved) }()
519 limits := c.cfg.ArtifactLimits
520 file := newBlockFile(blockID)
521 size, err := c.putUnsized(ctx, a.BlockKey(file), src)
522 if err != nil {
523 if derr := c.cfg.Blobs.Delete(context.WithoutCancel(ctx), a.BlockKey(file)); derr != nil {
524 ctxLog(ctx).Warn().Err(derr).Msg("ci: remove a refused block; the sweep will")
525 }
526 return 0, err
527 }
528 var old *ports.CIArtifactBlock
529 err = c.updateArtifacts(ctx, a.Repo, func(used int64, w *ports.CIArtifactWrite) error {
530 cur, err := c.pendingArtifact(ctx, j, a.ID)
531 if err != nil {
532 return err
533 }
534 if cur.State != domain.CIArtifactPending {
535 return errCommitted(cur.ID)
536 }
537 blocks, err := c.cfg.Artifacts.ArtifactBlocks(ctx, cur.ID)
538 if err != nil {
539 return err
540 }
541 old = nil
542 for i := range blocks {
543 if bytes.Equal(blocks[i].ID, blockID) {
544 old = &blocks[i]
545 }
546 }
547 var replaced int64
548 if old != nil {
549 replaced = old.Size
550 } else if len(blocks) >= domain.CIArtifactMaxBlocks {
551 return fmt.Errorf("%w: an artifact has at most %d blocks", domain.ErrInvalid, domain.CIArtifactMaxBlocks)
552 }
553 cur.Charged += size - replaced
554 w.Used += size - replaced
555 switch {
556 case cur.Charged > limits.MaxSize:
557 return c.errArtifactTooLarge()
558 case w.Used > limits.Quota:
559 return c.errQuota(cur.Repo)
560 }
561 w.Put = []domain.CIArtifact{cur}
562 w.PutBlocks = []ports.CIArtifactBlock{{ArtifactID: cur.ID, ID: blockID, Size: size, File: file}}
563 return nil
564 })
565 if err != nil {
566 if rerr := c.cfg.Blobs.Delete(context.WithoutCancel(ctx), a.BlockKey(file)); rerr != nil {
567 ctxLog(ctx).Warn().Err(rerr).Int64("artifact_id", a.ID).Msg("ci: remove a refused block; the sweep will")
568 }
569 return 0, err
570 }
571 if old != nil {
572 if err := c.cfg.Blobs.Delete(ctx, a.BlockKey(old.File)); err != nil {
573 ctxLog(ctx).Warn().Err(err).Int64("artifact_id", a.ID).Msg("ci: remove a replaced block")
574 }
575 }
576 c.dropIfEnded(ctx, j)
577 return size, nil
578}
579
580// blockListDigest identifies a block list: each id with its length.
581func blockListDigest(ids [][]byte) []byte {
582 h := sha256.New()
583 var n [2]byte
584 for _, id := range ids {
585 binary.BigEndian.PutUint16(n[:], uint16(len(id))) //nolint:gosec // at most 64
586 h.Write(n[:])
587 h.Write(id)
588 }
589 return h.Sum(nil)
590}
591
592// CommitJobArtifact assembles the content of a pending artifact from its
593// blocks, then records it and drops the blocks. The size counts against the
594// limits with the repeats of the list.
595func (c *CI) CommitJobArtifact(ctx context.Context, secret, jobID string, artifactID int64, blockIDs [][]byte) (int64, string, error) {
596 j, err := c.artifactJob(ctx, secret, jobID)
597 if err != nil {
598 return 0, "", err
599 }
600 if len(blockIDs) > domain.CIArtifactMaxBlocks {
601 return 0, "", fmt.Errorf("%w: a commit names at most %d blocks", domain.ErrInvalid, domain.CIArtifactMaxBlocks)
602 }
603 for _, id := range blockIDs {
604 if err := domain.CheckArtifactBlockID(id); err != nil {
605 return 0, "", err
606 }
607 }
608 list := blockListDigest(blockIDs)
609 // A retry of the runner waits here for the commit in progress and gets
610 // its result. The commit goes on when its caller leaves: the runner gives
611 // up on a long assembly and asks again.
612 unlock := c.lockArtifact(artifactID)
613 defer unlock()
614 ctx = context.WithoutCancel(ctx)
615 limits := c.cfg.ArtifactLimits
616 for range commitTries {
617 a, err := c.cfg.Artifacts.GetArtifact(ctx, artifactID)
618 if err != nil {
619 return 0, "", err
620 }
621 if a.Repo != j.Repo || !a.UploadedBy(j) || a.Assignment != j.TokenHash {
622 return 0, "", fmt.Errorf("%w: artifact %d is not an upload of this job", domain.ErrNotFound, artifactID)
623 }
624 if a.State != domain.CIArtifactPending {
625 if bytes.Equal(a.ListDigest, list) {
626 return a.Size, a.Digest, nil // a repeat
627 }
628 return 0, "", fmt.Errorf("%w: artifact %d is committed with another block list", domain.ErrPreconditionFailed, a.ID)
629 }
630 blocks, err := c.cfg.Artifacts.ArtifactBlocks(ctx, a.ID)
631 if err != nil {
632 return 0, "", err
633 }
634 byID := make(map[string]ports.CIArtifactBlock, len(blocks))
635 for _, b := range blocks {
636 byID[string(b.ID)] = b
637 }
638 files := make([]string, 0, len(blockIDs))
639 var total int64
640 for _, id := range blockIDs {
641 b, ok := byID[string(id)]
642 if !ok {
643 return 0, "", fmt.Errorf("%w: block %x was never uploaded", domain.ErrInvalid, id)
644 }
645 files = append(files, b.File)
646 total += b.Size
647 }
648 if total > limits.MaxSize {
649 return 0, "", c.errArtifactTooLarge()
650 }
651 // The repeats of the list make the content larger than the blocks
652 // charged: that excess is reserved before the content is written.
653 excess := max(total-a.Charged, 0)
654 if excess > 0 {
655 if err := c.reserve(ctx, a, excess); err != nil {
656 return 0, "", err
657 }
658 }
659 size, digest, err := c.assembleBlobs(ctx, a, files, total)
660 if err != nil {
661 c.release(a, excess)
662 }
663 if errors.Is(err, domain.ErrNotFound) {
664 continue // a block was replaced meanwhile
665 }
666 if err != nil {
667 return 0, "", err
668 }
669 err = c.updateArtifacts(ctx, a.Repo, func(used int64, w *ports.CIArtifactWrite) error {
670 cur, err := c.cfg.Artifacts.GetArtifact(ctx, a.ID)
671 if err != nil {
672 return err
673 }
674 if cur.Version != a.Version {
675 return errArtifactChanged
676 }
677 w.Used += size - cur.Charged
678 if w.Used > limits.Quota {
679 return c.errQuota(cur.Repo)
680 }
681 cur.State, cur.Size, cur.Digest, cur.ListDigest, cur.Charged = domain.CIArtifactCommitted, size, digest, list, size
682 w.Put, w.DropBlocks = []domain.CIArtifact{cur}, []int64{cur.ID}
683 return nil
684 })
685 c.release(a, excess)
686 if errors.Is(err, errArtifactChanged) {
687 continue
688 }
689 if errors.Is(err, domain.ErrNotFound) {
690 c.removeArtifactFiles(ctx, []domain.CIArtifact{a}) // discarded meanwhile
691 }
692 if err != nil {
693 return 0, "", err
694 }
695 if err := c.removeBlobBlocks(ctx, a); err != nil {
696 ctxLog(ctx).Warn().Err(err).Int64("artifact_id", a.ID).Msg("ci: remove the blocks of a committed artifact")
697 }
698 c.dropIfEnded(ctx, j)
699 return size, digest, nil
700 }
701 return 0, "", fmt.Errorf("%w: the blocks of artifact %d keep changing, try again", domain.ErrUnavailable, artifactID)
702}
703
704// FinalizeJobArtifact makes the committed upload of the job with that name
705// visible. An artifact of that name from an earlier attempt is deleted in the
706// same write; one finalized in this attempt is already_exists.
707func (c *CI) FinalizeJobArtifact(ctx context.Context, secret, jobID, name string, size int64, digest string) (domain.CIArtifact, error) {
708 j, err := c.artifactJob(ctx, secret, jobID)
709 if err != nil {
710 return domain.CIArtifact{}, err
711 }
712 if err := domain.ParseArtifactName(name); err != nil {
713 return domain.CIArtifact{}, err
714 }
715 digest, err = domain.ParseArtifactDigest(digest)
716 if err != nil {
717 return domain.CIArtifact{}, err
718 }
719 now := c.now()
720 var out domain.CIArtifact
721 var replaced []domain.CIArtifact
722 err = c.updateArtifacts(ctx, j.Repo, func(used int64, w *ports.CIArtifactWrite) error {
723 arts, err := c.cfg.Artifacts.RunArtifacts(ctx, j.Repo, j.RunID)
724 if err != nil {
725 return err
726 }
727 var mine, done *domain.CIArtifact
728 var finals []domain.CIArtifact
729 for i, a := range arts {
730 switch {
731 case a.Name != name:
732 case a.PendingOf(j):
733 if mine == nil || a.ID > mine.ID {
734 mine = &arts[i]
735 }
736 case a.Final():
737 finals = append(finals, a)
738 if a.UploadedBy(j) && a.Visible(now) {
739 done = &arts[i]
740 }
741 }
742 }
743 switch {
744 case mine == nil && done != nil:
745 out = *done // a repeat
746 return errNoArtifactWrite
747 case mine == nil:
748 return fmt.Errorf("%w: this job has no upload of an artifact named %q", domain.ErrNotFound, name)
749 case mine.State != domain.CIArtifactCommitted:
750 return fmt.Errorf("%w: artifact %q is not committed", domain.ErrPreconditionFailed, name)
751 case digest != "" && digest != mine.Digest:
752 return fmt.Errorf("%w: the digest %s is not the digest of the content (%s)", domain.ErrInvalid, digest, mine.Digest)
753 }
754 replaced = replaced[:0]
755 for _, f := range finals {
756 if f.Visible(now) && f.Attempt == j.RunAttempt {
757 return fmt.Errorf("%w: an artifact with this name already exists on the workflow run", domain.ErrAlreadyExists)
758 }
759 replaced = append(replaced, f) // an earlier attempt's, or expired
760 w.Used -= f.Charged
761 }
762 out = *mine
763 out.State = domain.CIArtifactFinal
764 w.Put, w.Delete = []domain.CIArtifact{out}, replaced
765 w.Used = max(w.Used, 0)
766 return nil
767 })
768 if errors.Is(err, errNoArtifactWrite) {
769 return out, nil
770 }
771 if err != nil {
772 return domain.CIArtifact{}, err
773 }
774 c.removeArtifactFiles(ctx, replaced)
775 log := zerowrap.FromCtx(ctx)
776 ev := log.Info()
777 if size != out.Size {
778 ev = ev.Int64("uploader_size", size)
779 }
780 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")
781 return out, nil
782}
783
784// ListJobArtifacts lists the visible artifacts of the job's run.
785func (c *CI) ListJobArtifacts(ctx context.Context, secret, jobID, name string, artifactID int64) ([]domain.CIArtifact, error) {
786 j, err := c.artifactJob(ctx, secret, jobID)
787 if err != nil {
788 return nil, err
789 }
790 arts, err := c.visibleArtifacts(ctx, j.Repo, j.RunID)
791 if err != nil {
792 return nil, err
793 }
794 out := arts[:0]
795 for _, a := range arts {
796 if (name == "" || a.Name == name) && (artifactID == 0 || a.ID == artifactID) {
797 out = append(out, a)
798 }
799 }
800 return out, nil
801}
802
803func (c *CI) visibleArtifacts(ctx context.Context, repo domain.RepoName, runID int64) ([]domain.CIArtifact, error) {
804 arts, err := c.cfg.Artifacts.RunArtifacts(ctx, repo, runID)
805 if err != nil {
806 return nil, err
807 }
808 now := c.now()
809 out := make([]domain.CIArtifact, 0, len(arts))
810 for _, a := range arts {
811 if a.Visible(now) {
812 out = append(out, a)
813 }
814 }
815 return out, nil
816}
817
818// DownloadJobArtifact streams a visible artifact of the job's run.
819func (c *CI) DownloadJobArtifact(ctx context.Context, secret, jobID string, artifactID, offset int64, send func(ports.ArtifactChunk) error) error {
820 j, err := c.artifactJob(ctx, secret, jobID)
821 if err != nil {
822 return err
823 }
824 a, err := c.cfg.Artifacts.GetArtifact(ctx, artifactID)
825 if err != nil {
826 return err
827 }
828 if a.Repo != j.Repo || a.RunID != j.RunID || !a.Visible(c.now()) {
829 return fmt.Errorf("%w: artifact %d is not an artifact of this run", domain.ErrNotFound, artifactID)
830 }
831 return c.streamArtifact(ctx, a, offset, send)
832}
833
834// streamArtifact sends the artifact, then its content from offset.
835func (c *CI) streamArtifact(ctx context.Context, a domain.CIArtifact, offset int64, send func(ports.ArtifactChunk) error) error {
836 if offset < 0 || offset > a.Size {
837 return fmt.Errorf("%w: offset %d is beyond the size %d", domain.ErrOutOfRange, offset, a.Size)
838 }
839 var f io.ReadCloser
840 var err error
841 if a.Store == domain.CIArtifactStoreBlob {
842 // BlobStore rejects offsets at EOF, while artifact downloads permit
843 // an empty resumed stream (including zero-byte artifacts).
844 if offset == a.Size {
845 _, err = c.cfg.Blobs.Stat(ctx, a.BlobKey())
846 f = io.NopCloser(bytes.NewReader(nil))
847 } else {
848 f, _, err = c.cfg.Blobs.Open(ctx, a.BlobKey(), offset, -1)
849 }
850 } else if c.cfg.ArtifactFiles != nil {
851 f, err = c.cfg.ArtifactFiles.Open(ctx, artifactRef(a), offset)
852 } else {
853 ctxLog(ctx).Error().Int64("artifact_id", a.ID).Msg("ci: legacy artifact reader is unavailable")
854 return domain.ErrNotFound
855 }
856 if err != nil {
857 return err
858 }
859 defer func() { _ = f.Close() }()
860 if err := send(ports.ArtifactChunk{Artifact: &a}); err != nil {
861 return err
862 }
863 left := a.Size - offset
864 buf := make([]byte, min(left, domain.CIArtifactChunk))
865 for left > 0 {
866 n, err := io.ReadFull(f, buf[:min(left, int64(len(buf)))])
867 if err != nil {
868 return fmt.Errorf("read artifact %d: %w", a.ID, err)
869 }
870 left -= int64(n)
871 if err := send(ports.ArtifactChunk{Data: buf[:n]}); err != nil {
872 return err
873 }
874 }
875 return nil
876}
877
878// DeleteJobArtifact deletes the visible artifact of that name of the job's
879// run.
880func (c *CI) DeleteJobArtifact(ctx context.Context, secret, jobID, name string) (int64, error) {
881 j, err := c.artifactJob(ctx, secret, jobID)
882 if err != nil {
883 return 0, err
884 }
885 if err := domain.ParseArtifactName(name); err != nil {
886 return 0, err
887 }
888 now := c.now()
889 removed, err := c.removeArtifacts(ctx, j.Repo, j.RunID, func(a domain.CIArtifact) bool {
890 return a.Name == name && a.Visible(now)
891 })
892 if err != nil {
893 return 0, err
894 }
895 if len(removed) == 0 {
896 return 0, fmt.Errorf("%w: no artifact named %q on this run", domain.ErrNotFound, name)
897 }
898 log := zerowrap.FromCtx(ctx)
899 log.Info().Int64("artifact_id", removed[0].ID).Int64("run_id", j.RunID).Str("job_id", j.ID).Msg("artifact deleted by a job")
900 return removed[0].ID, nil
901}
902
903// --- ArtifactService ---
904
905// artifactRepo is the access rule of reading artifacts: a signed-in caller
906// with repo:read, even on a public repository (GitHub serves no artifact
907// anonymously).
908func (c *CI) artifactRepo(ctx context.Context, p domain.Principal, name domain.RepoName) (domain.Repo, error) {
909 if !c.artifactsOn() {
910 return domain.Repo{}, fmt.Errorf("%w: artifacts are off on this server", domain.ErrPreconditionFailed)
911 }
912 repo, err := c.readableRepo(ctx, p, name)
913 if err != nil {
914 return domain.Repo{}, err
915 }
916 if err := scopeAccess(p, repo, domain.ScopeRepoRead); err != nil {
917 return domain.Repo{}, err
918 }
919 return repo, nil
920}
921
922// ListArtifacts lists the visible artifacts of a run with the space of the
923// repository.
924func (c *CI) ListArtifacts(ctx context.Context, p domain.Principal, name domain.RepoName, runID int64) (ports.ArtifactList, error) {
925 if _, err := c.artifactRepo(ctx, p, name); err != nil {
926 return ports.ArtifactList{}, err
927 }
928 if _, err := c.runOf(ctx, name, runID); err != nil {
929 return ports.ArtifactList{}, err
930 }
931 arts, err := c.visibleArtifacts(ctx, name, runID)
932 if err != nil {
933 return ports.ArtifactList{}, err
934 }
935 used, _, err := c.cfg.Artifacts.ArtifactUsage(ctx, name)
936 if err != nil {
937 return ports.ArtifactList{}, err
938 }
939 return ports.ArtifactList{Artifacts: arts, Used: used, Quota: c.cfg.ArtifactLimits.Quota}, nil
940}
941
942// ownerArtifact returns a visible artifact of the repository.
943func (c *CI) ownerArtifact(ctx context.Context, name domain.RepoName, id int64) (domain.CIArtifact, error) {
944 a, err := c.cfg.Artifacts.GetArtifact(ctx, id)
945 if err != nil {
946 return domain.CIArtifact{}, err
947 }
948 if a.Repo != name || !a.Visible(c.now()) {
949 return domain.CIArtifact{}, fmt.Errorf("%w: artifact %d is not an artifact of %s", domain.ErrNotFound, id, name)
950 }
951 return a, nil
952}
953
954// DownloadArtifact streams a visible artifact of the repository.
955func (c *CI) DownloadArtifact(ctx context.Context, p domain.Principal, name domain.RepoName, artifactID, offset int64, send func(ports.ArtifactChunk) error) error {
956 if _, err := c.artifactRepo(ctx, p, name); err != nil {
957 return err
958 }
959 a, err := c.ownerArtifact(ctx, name, artifactID)
960 if err != nil {
961 return err
962 }
963 return c.streamArtifact(ctx, a, offset, send)
964}
965
966// DeleteArtifact deletes an artifact. It needs ci:write on the repository and,
967// like the other writes on a run, the branch of the run within the prefixes of
968// a limited token.
969func (c *CI) DeleteArtifact(ctx context.Context, p domain.Principal, name domain.RepoName, artifactID int64) error {
970 if !c.artifactsOn() {
971 return fmt.Errorf("%w: artifacts are off on this server", domain.ErrPreconditionFailed)
972 }
973 if _, err := c.writableRepo(ctx, p, name); err != nil {
974 return err
975 }
976 a, err := c.ownerArtifact(ctx, name, artifactID)
977 if err != nil {
978 return err
979 }
980 run, err := c.runOf(ctx, name, a.RunID)
981 if err != nil {
982 return err
983 }
984 if err := runWriteAccess(p, run); err != nil {
985 return err
986 }
987 removed, err := c.removeArtifacts(ctx, name, a.RunID, func(o domain.CIArtifact) bool { return o.ID == a.ID && o.Final() })
988 if err != nil {
989 return err
990 }
991 if len(removed) == 0 {
992 return fmt.Errorf("%w: artifact %d is not an artifact of %s", domain.ErrNotFound, artifactID, name)
993 }
994 log := zerowrap.FromCtx(ctx)
995 log.Info().Int64("artifact_id", a.ID).Int64("run_id", a.RunID).Str("repo", name.String()).
996 Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("artifact deleted")
997 return nil
998}
999
1000// --- sweep ---
1001
1002// sweepArtifacts deletes the expired artifacts, those of runs that no longer
1003// exist, and the unfinished uploads of jobs that ended, were given again or
1004// began more than domain.CIArtifactPendingMax ago; then the blocks and files
1005// that no record names (a crash between a file and its record, a repository
1006// deleted while the server was down). Every failure is logged and the sweep
1007// goes on.
1008func (c *CI) sweepArtifacts(ctx context.Context) {
1009 if !c.artifactsOn() {
1010 return
1011 }
1012 log := ctxLog(ctx)
1013 all, err := c.cfg.Artifacts.AllArtifacts(ctx)
1014 if err != nil {
1015 log.Error().Err(err).Msg("ci: list the artifacts")
1016 return
1017 }
1018 type runKey struct {
1019 repo domain.RepoName
1020 run int64
1021 }
1022 runs := map[runKey]bool{}
1023 for _, a := range all {
1024 runs[runKey{a.Repo, a.RunID}] = true
1025 }
1026 removed := 0
1027 for k := range runs {
1028 gone := false
1029 if _, err := c.cfg.Store.GetRunSource(ctx, k.run); errors.Is(err, domain.ErrNotFound) {
1030 gone = true
1031 }
1032 now := c.now()
1033 arts, err := c.removeArtifacts(ctx, k.repo, k.run, func(a domain.CIArtifact) bool {
1034 return gone || c.artifactDead(ctx, a, now)
1035 })
1036 if err != nil && !errors.Is(err, domain.ErrNotFound) {
1037 log.Error().Err(err).Int64("run_id", k.run).Msg("ci: remove the expired artifacts of a run")
1038 }
1039 removed += len(arts)
1040 }
1041 if removed > 0 {
1042 log.Info().Int("artifacts", removed).Msg("ci: expired artifacts removed")
1043 }
1044 c.sweepArtifactOrphans(ctx)
1045}
1046
1047// artifactDead reports whether the sweep deletes an artifact of a run that
1048// exists.
1049func (c *CI) artifactDead(ctx context.Context, a domain.CIArtifact, now time.Time) bool {
1050 if a.Final() {
1051 return !now.Before(a.ExpiresAt)
1052 }
1053 if now.Sub(a.CreatedAt) >= domain.CIArtifactPendingMax {
1054 return true
1055 }
1056 j, err := c.cfg.Store.GetJob(ctx, a.RunID, a.Attempt, a.JobID)
1057 switch {
1058 case errors.Is(err, domain.ErrNotFound):
1059 return true
1060 case err != nil:
1061 return false
1062 }
1063 return j.Ended() || j.TokenHash != a.Assignment
1064}
1065
1066// sweepArtifactOrphans removes the blocks and files that no record names.
1067func (c *CI) sweepArtifactOrphans(ctx context.Context) {
1068 log := ctxLog(ctx)
1069 if ids, err := c.cfg.Artifacts.BlockArtifactIDs(ctx); err != nil {
1070 log.Error().Err(err).Msg("ci: list the staged artifact blocks")
1071 } else {
1072 for _, id := range ids {
1073 if _, err := c.cfg.Artifacts.GetArtifact(ctx, id); !errors.Is(err, domain.ErrNotFound) {
1074 continue
1075 }
1076 if err := c.cfg.Artifacts.DeleteBlocks(ctx, id); err != nil {
1077 log.Error().Err(err).Int64("artifact_id", id).Msg("ci: remove the blocks of an artifact that no longer exists")
1078 }
1079 }
1080 }
1081 c.sweepArtifactBlobOrphans(ctx)
1082 if c.cfg.ArtifactFiles == nil {
1083 return
1084 }
1085 c.removeOrphans(ctx, "artifacts", "artifact directories", c.cfg.ArtifactFiles.RunIDs, c.cfg.ArtifactFiles.RemoveRuns)
1086 runIDs, err := c.cfg.ArtifactFiles.RunIDs(ctx)
1087 if err != nil {
1088 log.Error().Err(err).Msg("ci: list the artifact directories")
1089 return
1090 }
1091 for _, run := range runIDs {
1092 ids, err := c.cfg.ArtifactFiles.ArtifactIDs(ctx, run)
1093 if err != nil {
1094 log.Error().Err(err).Int64("run_id", run).Msg("ci: list the artifact files of a run")
1095 continue
1096 }
1097 for _, id := range ids {
1098 a, err := c.cfg.Artifacts.GetArtifact(ctx, id)
1099 if err == nil && a.RunID == run && a.Store != domain.CIArtifactStoreBlob {
1100 continue
1101 }
1102 if err != nil && !errors.Is(err, domain.ErrNotFound) {
1103 continue
1104 }
1105 ref := ports.CIArtifactRef{RunID: run, ID: id}
1106 if err := c.cfg.ArtifactFiles.Remove(ctx, ref); err != nil {
1107 log.Error().Err(err).Int64("artifact_id", id).Msg("ci: remove the files of an artifact that no longer exists")
1108 }
1109 }
1110 }
1111}
1112
1113// putUnsized buffers at most one toolkit block; Put requires an exact size.
1114func (c *CI) putUnsized(ctx context.Context, key string, r io.Reader) (int64, error) {
1115 // A fixed backing array prevents bytes.Buffer's geometric growth from
1116 // allocating beyond the block bound. Probe the next byte separately.
1117 bp := blockBufs.Get().(*[]byte)
1118 defer blockBufs.Put(bp)
1119 buf := (*bp)[:domain.CIArtifactMaxBlockSize]
1120 n, err := io.ReadFull(r, buf)
1121 if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF {
1122 return 0, err
1123 }
1124 if n == len(buf) {
1125 var extra [1]byte
1126 m, err := io.ReadFull(r, extra[:])
1127 if m != 0 {
1128 return 0, fmt.Errorf("%w: a block has at most 8 MiB", domain.ErrInvalid)
1129 }
1130 if err != io.EOF {
1131 return 0, err
1132 }
1133 }
1134 if err := ctx.Err(); err != nil {
1135 return 0, err
1136 }
1137 _, err = c.cfg.Blobs.Put(ctx, key, bytes.NewReader(buf[:n]), int64(n))
1138 return int64(n), err
1139}
1140
1141func (c *CI) removeBlobBlocks(ctx context.Context, a domain.CIArtifact) error {
1142 return c.cfg.Blobs.List(ctx, a.BlobKey()+".blocks/", func(b ports.BlobInfo) error {
1143 return c.cfg.Blobs.Delete(ctx, b.Key)
1144 })
1145}
1146
1147// assembleBlobs keeps one 8 MiB part in memory, even when a block crosses
1148// part boundaries or is repeated in the commit list.
1149func (c *CI) assembleBlobs(ctx context.Context, a domain.CIArtifact, files []string, total int64) (int64, string, error) {
1150 freeSlot, err := c.waitUploadSlot(ctx)
1151 if err != nil {
1152 return 0, "", err
1153 }
1154 defer freeSlot()
1155 var upload ports.MultipartUpload
1156 if total > ports.BlobPartSize {
1157 var err error
1158 upload, err = c.cfg.Blobs.NewMultipart(ctx, a.BlobKey())
1159 if err != nil {
1160 return 0, "", err
1161 }
1162 }
1163 completed := false
1164 defer func() {
1165 if upload != nil && !completed {
1166 if err := upload.Abort(context.WithoutCancel(ctx)); err != nil {
1167 ctxLog(ctx).Warn().Err(err).Msg("ci: abort artifact assembly; the sweep will")
1168 }
1169 }
1170 }()
1171 bp := blockBufs.Get().(*[]byte)
1172 defer blockBufs.Put(bp)
1173 buf := (*bp)[:ports.BlobPartSize]
1174 used, part := 0, 1
1175 var size int64
1176 h := sha256.New()
1177 flush := func() error {
1178 if err := upload.UploadPart(ctx, part, bytes.NewReader(buf[:used]), int64(used)); err != nil {
1179 return err
1180 }
1181 used = 0
1182 part++
1183 return nil
1184 }
1185 for _, file := range files {
1186 r, info, err := c.cfg.Blobs.Open(ctx, a.BlockKey(file), 0, -1)
1187 // Empty blocks cannot be opened under BlobStore's EOF contract.
1188 if errors.Is(err, domain.ErrOutOfRange) {
1189 info, err = c.cfg.Blobs.Stat(ctx, a.BlockKey(file))
1190 if err == nil && info.Size == 0 {
1191 continue
1192 }
1193 if err == nil {
1194 return 0, "", domain.ErrOutOfRange
1195 }
1196 }
1197 if err != nil {
1198 return 0, "", err
1199 }
1200 if info.Size > total-size {
1201 _ = r.Close()
1202 return 0, "", fmt.Errorf("%w: artifact block size changed", domain.ErrInvalid)
1203 }
1204 left := info.Size
1205 for left > 0 {
1206 if err := ctx.Err(); err != nil {
1207 _ = r.Close()
1208 return 0, "", err
1209 }
1210 want := min(left, int64(len(buf)-used))
1211 if want == 0 {
1212 _ = r.Close()
1213 return 0, "", fmt.Errorf("%w: artifact block size changed", domain.ErrInvalid)
1214 }
1215 n, err := io.ReadFull(r, buf[used:used+int(want)])
1216 if err != nil {
1217 _ = r.Close()
1218 return 0, "", err
1219 }
1220 h.Write(buf[used : used+n])
1221 used += n
1222 size += int64(n)
1223 left -= int64(n)
1224 if used == len(buf) && upload != nil {
1225 if err := flush(); err != nil {
1226 _ = r.Close()
1227 return 0, "", err
1228 }
1229 }
1230 }
1231 var extra [1]byte
1232 n, err := io.ReadFull(r, extra[:])
1233 cerr := r.Close()
1234 if n != 0 || err != io.EOF {
1235 return 0, "", fmt.Errorf("%w: artifact block size changed", domain.ErrInvalid)
1236 }
1237 if cerr != nil {
1238 return 0, "", cerr
1239 }
1240 }
1241 if size != total {
1242 return 0, "", fmt.Errorf("%w: artifact block size changed", domain.ErrInvalid)
1243 }
1244 if upload == nil {
1245 if _, err := c.cfg.Blobs.Put(ctx, a.BlobKey(), bytes.NewReader(buf[:used]), size); err != nil {
1246 return 0, "", err
1247 }
1248 } else {
1249 if used > 0 {
1250 if err := flush(); err != nil {
1251 return 0, "", err
1252 }
1253 }
1254 if _, err := upload.Complete(ctx); err != nil {
1255 return 0, "", err
1256 }
1257 completed = true
1258 }
1259 return size, "sha256:" + hex.EncodeToString(h.Sum(nil)), nil
1260}
1261
1262func (c *CI) sweepArtifactBlobOrphans(ctx context.Context) {
1263 log := ctxLog(ctx)
1264 cutoff := c.now().Add(-domain.CIArtifactPendingMax)
1265 err := c.cfg.Blobs.List(ctx, "ci/", func(b ports.BlobInfo) error {
1266 if !b.ModTime.Before(cutoff) {
1267 return nil
1268 }
1269 run, id, block, ok := artifactBlobIDs(b.Key)
1270 if !ok {
1271 return nil
1272 }
1273 a, err := c.cfg.Artifacts.GetArtifact(ctx, id)
1274 if err != nil && !errors.Is(err, domain.ErrNotFound) {
1275 return nil
1276 }
1277 if err == nil && a.RunID == run && a.Store == domain.CIArtifactStoreBlob && (!block || a.State == domain.CIArtifactPending) {
1278 return nil
1279 }
1280 if err := c.cfg.Blobs.Delete(ctx, b.Key); err != nil {
1281 log.Error().Err(err).Msg("ci: remove orphan artifact blob")
1282 }
1283 return nil
1284 })
1285 if err != nil {
1286 log.Error().Err(err).Msg("ci: list artifact blobs")
1287 }
1288 err = c.cfg.Blobs.ListMultipart(ctx, "ci/", func(m ports.MultipartInfo) error {
1289 if m.Started.Before(cutoff) {
1290 if err := c.cfg.Blobs.AbortMultipart(ctx, m.Key, m.ID); err != nil {
1291 log.Error().Err(err).Msg("ci: abort abandoned artifact upload")
1292 }
1293 }
1294 return nil
1295 })
1296 if err != nil {
1297 log.Error().Err(err).Msg("ci: list artifact multipart uploads")
1298 }
1299}
1300
1301func artifactBlobIDs(key string) (run, id int64, block, ok bool) {
1302 rest, found := strings.CutPrefix(key, "ci/")
1303 if !found {
1304 return
1305 }
1306 r, tail, found := strings.Cut(rest, "/")
1307 if !found || len(r) != 20 {
1308 return
1309 }
1310 name, file, block := strings.Cut(tail, ".blocks/")
1311 if len(name) != 20 || (block && file == "") {
1312 return 0, 0, false, false
1313 }
1314 run, err := strconv.ParseInt(r, 10, 64)
1315 if err != nil || run <= 0 {
1316 return 0, 0, false, false
1317 }
1318 id, err = strconv.ParseInt(name, 10, 64)
1319 return run, id, block, err == nil && id > 0
1320}
1321
1322// removeArtifactRuns follows repository deletion, when records are already
1323// gone. Run ids are global, so their blob prefixes cannot name another repo.
1324func (c *CI) removeArtifactRuns(ctx context.Context, runIDs []int64) error {
1325 var errs []error
1326 if c.cfg.ArtifactFiles != nil {
1327 errs = append(errs, c.cfg.ArtifactFiles.RemoveRuns(ctx, runIDs))
1328 }
1329 if c.cfg.Blobs != nil {
1330 for _, run := range runIDs {
1331 if run <= 0 {
1332 continue
1333 }
1334 prefix := fmt.Sprintf("ci/%020d/", run)
1335 errs = append(errs, c.cfg.Blobs.List(ctx, prefix, func(b ports.BlobInfo) error {
1336 if err := c.cfg.Blobs.Delete(ctx, b.Key); err != nil {
1337 errs = append(errs, err)
1338 }
1339 return nil
1340 }))
1341 errs = append(errs, c.cfg.Blobs.ListMultipart(ctx, prefix, func(m ports.MultipartInfo) error {
1342 if err := c.cfg.Blobs.AbortMultipart(ctx, m.Key, m.ID); err != nil {
1343 errs = append(errs, err)
1344 }
1345 return nil
1346 }))
1347 }
1348 }
1349 return errors.Join(errs...)
1350}