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 switch a.State {
33 case domain.ReleaseAssetUploading:
34 remove = now.Sub(a.UpdatedAt) > domain.ReleaseAssetPendingMax
35 case domain.ReleaseAssetUploaded:
36 _, err := r.blobs.Stat(ctx, a.BlobKey())
37 switch {
38 case errors.Is(err, domain.ErrNotFound):
39 r.misses[a.ID]++
40 remove = r.misses[a.ID] >= 2
41 if !remove {
42 log.Error().Int64("asset", a.ID).Msg("asset blob missing")
43 }
44 case err != nil:
45 log.Error().Err(err).Int64("asset", a.ID).Msg("stat release asset")
46 default:
47 delete(r.misses, a.ID)
48 }
49 }
50 if !remove {
51 continue
52 }
53 err := r.removeAssetRecord(ctx, a, func(current domain.ReleaseAsset) bool { return current.Version == a.Version })
54 if err != nil {
55 if !errors.Is(err, domain.ErrNotFound) && !errors.Is(err, domain.ErrConflict) {
56 log.Error().Err(err).Int64("asset", a.ID).Msg("remove stale release asset")
57 }
58 continue
59 }
60 delete(r.misses, a.ID)
61 r.removeAssetBlobs(ctx, []domain.ReleaseAsset{a})
62 }
63 for id := range r.misses {
64 if !seen[id] {
65 delete(r.misses, id)
66 }
67 }
68 // Re-check the store for each old key: concurrent reservations must not be
69 // classified as orphans from a stale snapshot.
70 err = r.blobs.List(ctx, "release/", func(info ports.BlobInfo) error {
71 if now.Sub(info.ModTime) <= domain.ReleaseAssetPendingMax {
72 return nil
73 }
74 // Only a canonical key can name an asset. Check its record directly
75 // rather than repeating the full metadata scan for every object.
76 id, parseErr := strconv.ParseInt(strings.TrimPrefix(info.Key, "release/"), 10, 64)
77 if parseErr == nil && (domain.ReleaseAsset{ID: id}).BlobKey() == info.Key {
78 _, e := r.store.GetAsset(ctx, id)
79 if e == nil {
80 return nil
81 }
82 if !errors.Is(e, domain.ErrNotFound) {
83 log.Error().Err(e).Msg("check orphan release blob")
84 return nil
85 }
86 }
87 if e := r.blobs.Delete(ctx, info.Key); e != nil {
88 log.Error().Err(e).Str("key", info.Key).Msg("remove orphan release blob")
89 }
90 return nil
91 })
92 if err != nil {
93 log.Error().Err(err).Msg("list release blobs")
94 }
95 err = r.blobs.ListMultipart(ctx, "release/", func(info ports.MultipartInfo) error {
96 if now.Sub(info.Started) > domain.ReleaseAssetPendingMax {
97 if e := r.blobs.AbortMultipart(ctx, info.Key, info.ID); e != nil {
98 log.Error().Err(e).Msg("abort abandoned release multipart")
99 }
100 }
101 return nil
102 })
103 if err != nil {
104 log.Error().Err(err).Msg("list release multipart uploads")
105 }
106}