1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package usecase
4
5import (
6 "context"
7 "errors"
8 "strconv"
9 "strings"
10
11 "bnema.dev/ermite/internal/domain"
12 "bnema.dev/ermite/internal/ports"
13)
14
15// Sweep collects abandoned uploads, missing contents, and old orphan blobs.
16// A failed metadata scan never authorizes blob deletion. Sweeps are serialized
17// so the two observations of a missing blob are always distinct sweeps.
18func (r *Releases) Sweep(ctx context.Context) {
19 r.sweepMu.Lock()
20 defer r.sweepMu.Unlock()
21 log := ctxLog(ctx)
22 now := r.clock.Now()
23 all, err := r.store.AllAssets(ctx)
24 if err != nil {
25 log.Error().Err(err).Msg("scan release assets")
26 return
27 }
28 seen := make(map[int64]bool, len(all))
29 for _, a := range all {
30 seen[a.ID] = true
31 remove := false
32 if a.State == domain.ReleaseAssetUploading {
33 remove = now.Sub(a.UpdatedAt) > domain.ReleaseAssetPendingMax
34 } else if a.State == domain.ReleaseAssetUploaded {
35 _, err := r.blobs.Stat(ctx, a.BlobKey())
36 switch {
37 case errors.Is(err, domain.ErrNotFound):
38 r.misses[a.ID]++
39 remove = r.misses[a.ID] >= 2
40 if !remove {
41 log.Error().Int64("asset", a.ID).Msg("asset blob missing")
42 }
43 case err != nil:
44 log.Error().Err(err).Int64("asset", a.ID).Msg("stat release asset")
45 default:
46 delete(r.misses, a.ID)
47 }
48 }
49 if !remove {
50 continue
51 }
52 err := r.removeAssetRecord(ctx, a, func(current domain.ReleaseAsset) bool { return current.Version == a.Version })
53 if err != nil {
54 if !errors.Is(err, domain.ErrNotFound) && !errors.Is(err, domain.ErrConflict) {
55 log.Error().Err(err).Int64("asset", a.ID).Msg("remove stale release asset")
56 }
57 continue
58 }
59 delete(r.misses, a.ID)
60 r.removeAssetBlobs(ctx, []domain.ReleaseAsset{a})
61 }
62 for id := range r.misses {
63 if !seen[id] {
64 delete(r.misses, id)
65 }
66 }
67 // Re-check the store for each old key: concurrent reservations must not be
68 // classified as orphans from a stale snapshot.
69 err = r.blobs.List(ctx, "release/", func(info ports.BlobInfo) error {
70 if now.Sub(info.ModTime) <= domain.ReleaseAssetPendingMax {
71 return nil
72 }
73 // Only a canonical key can name an asset. Check its record directly
74 // rather than repeating the full metadata scan for every object.
75 id, parseErr := strconv.ParseInt(strings.TrimPrefix(info.Key, "release/"), 10, 64)
76 if parseErr == nil && (domain.ReleaseAsset{ID: id}).BlobKey() == info.Key {
77 _, e := r.store.GetAsset(ctx, id)
78 if e == nil {
79 return nil
80 }
81 if !errors.Is(e, domain.ErrNotFound) {
82 log.Error().Err(e).Msg("check orphan release blob")
83 return nil
84 }
85 }
86 if e := r.blobs.Delete(ctx, info.Key); e != nil {
87 log.Error().Err(e).Str("key", info.Key).Msg("remove orphan release blob")
88 }
89 return nil
90 })
91 if err != nil {
92 log.Error().Err(err).Msg("list release blobs")
93 }
94 err = r.blobs.ListMultipart(ctx, "release/", func(info ports.MultipartInfo) error {
95 if now.Sub(info.Started) > domain.ReleaseAssetPendingMax {
96 if e := r.blobs.AbortMultipart(ctx, info.Key, info.ID); e != nil {
97 log.Error().Err(e).Msg("abort abandoned release multipart")
98 }
99 }
100 return nil
101 })
102 if err != nil {
103 log.Error().Err(err).Msg("list release multipart uploads")
104 }
105}