1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package usecase_test
4
5import (
6 "bytes"
7 "context"
8 "errors"
9 "testing"
10 "time"
11
12 "bnema.dev/ermite/internal/domain"
13 "bnema.dev/ermite/internal/ports"
14 "bnema.dev/ermite/internal/ports/mocks"
15 "bnema.dev/ermite/internal/usecase"
16 "github.com/stretchr/testify/mock"
17 "github.com/stretchr/testify/require"
18)
19
20func (r *releaseRig) setClock(t *testing.T, at time.Time, blobs ports.BlobStore) {
21 t.Helper()
22 clock := mocks.NewClock(t)
23 clock.EXPECT().Now().Return(at).Maybe()
24 bus := mocks.NewEventBus(t)
25 bus.EXPECT().Publish(mock.Anything, mock.Anything).Return().Maybe()
26 r.clock = clock
27 r.svc = usecase.NewReleases(r.store.Releases(), blobs, r.store, r.reader, r.tags, r.pushes, bus, clock, ports.ReleaseLimits{})
28}
29func TestReleaseSweepPendingAndOrphans(t *testing.T) {
30 r := newReleaseRig(t, ports.ReleaseLimits{})
31 rel := r.create(t, "v1", false)
32 pending := r.upload(t, rel, "pending", []byte("abc"))
33 good := r.upload(t, rel, "good", []byte("xyz"))
34 fresh := r.upload(t, rel, "fresh", nil)
35 old, err := r.store.Releases().GetAsset(r.ctx, pending.ID)
36 require.NoError(t, err)
37 old.State = domain.ReleaseAssetUploading
38 mp, err := r.blobs.NewMultipart(r.ctx, old.BlobKey())
39 require.NoError(t, err)
40 old.MultipartID = mp.ID()
41 fresh.State = domain.ReleaseAssetUploading
42 fresh.UpdatedAt = time.Now().Add(25 * time.Hour)
43 used, v, err := r.store.Releases().ReleaseUsage(r.ctx, r.repo)
44 require.NoError(t, err)
45 require.NoError(t, r.store.Releases().ApplyReleases(r.ctx, ports.ReleaseWrite{Repo: r.repo, Used: used, UsageVersion: v, PutAssets: []domain.ReleaseAsset{old, fresh}}))
46 _, err = r.blobs.Put(r.ctx, "release/99999999999999999999", bytes.NewReader([]byte("orphan")), 6)
47 require.NoError(t, err)
48 // The clock is advanced beyond the disk adapter's real modification times.
49 at := time.Now().Add(26 * time.Hour)
50 r.setClock(t, at, r.blobs)
51 r.svc.Sweep(r.ctx)
52 _, err = r.store.Releases().GetAsset(r.ctx, pending.ID)
53 require.ErrorIs(t, err, domain.ErrNotFound)
54 _, err = r.blobs.Stat(r.ctx, pending.BlobKey())
55 require.ErrorIs(t, err, domain.ErrNotFound)
56 _, err = r.blobs.Stat(r.ctx, "release/99999999999999999999")
57 require.ErrorIs(t, err, domain.ErrNotFound)
58 _, err = r.store.Releases().GetAsset(r.ctx, good.ID)
59 require.NoError(t, err)
60 // The pending record inside the grace period is retained.
61 _, err = r.store.Releases().GetAsset(r.ctx, fresh.ID)
62 require.NoError(t, err)
63 require.NoError(t, r.blobs.ListMultipart(r.ctx, "release/", func(ports.MultipartInfo) error { t.Error("multipart survived"); return nil }))
64}
65func TestReleaseSweepMissingBlobTwoObservations(t *testing.T) {
66 r := newReleaseRig(t, ports.ReleaseLimits{})
67 rel := r.create(t, "v1", false)
68 a := r.upload(t, rel, "binary", []byte("abc"))
69 require.NoError(t, r.blobs.Delete(r.ctx, a.BlobKey()))
70 r.svc.Sweep(r.ctx)
71 _, err := r.store.Releases().GetAsset(r.ctx, a.ID)
72 require.NoError(t, err)
73 r.svc.Sweep(r.ctx)
74 _, err = r.store.Releases().GetAsset(r.ctx, a.ID)
75 require.ErrorIs(t, err, domain.ErrNotFound)
76 usage, err := r.svc.Usage(r.ctx, domain.Root(), r.repo)
77 require.NoError(t, err)
78 require.Zero(t, usage.Used)
79}
80
81type releaseStatOverride struct {
82 ports.BlobStore
83 err error
84}
85
86func (b *releaseStatOverride) Stat(ctx context.Context, key string) (ports.BlobInfo, error) {
87 if b.err != nil {
88 return ports.BlobInfo{}, b.err
89 }
90 return b.BlobStore.Stat(ctx, key)
91}
92func TestReleaseSweepTransientStatAndRecovery(t *testing.T) {
93 for _, recover := range []bool{false, true} {
94 t.Run(map[bool]string{false: "transient preserves miss", true: "success resets miss"}[recover], func(t *testing.T) {
95 r := newReleaseRig(t, ports.ReleaseLimits{})
96 rel := r.create(t, "v1", false)
97 a := r.upload(t, rel, "binary", []byte("abc"))
98 b := &releaseStatOverride{BlobStore: r.blobs, err: domain.ErrNotFound}
99 r.setClock(t, now, b)
100 r.svc.Sweep(r.ctx)
101 b.err = errors.New("outage")
102 r.svc.Sweep(r.ctx)
103 _, err := r.store.Releases().GetAsset(r.ctx, a.ID)
104 require.NoError(t, err)
105 if recover {
106 b.err = nil
107 r.svc.Sweep(r.ctx)
108 }
109 b.err = domain.ErrNotFound
110 r.svc.Sweep(r.ctx)
111 _, err = r.store.Releases().GetAsset(r.ctx, a.ID)
112 if recover {
113 require.NoError(t, err)
114 } else {
115 require.ErrorIs(t, err, domain.ErrNotFound)
116 }
117 })
118 }
119}
120func TestReleaseSweepGraceAndFailureIsolation(t *testing.T) {
121 r := newReleaseRig(t, ports.ReleaseLimits{})
122 rel := r.create(t, "v1", false)
123 a := r.upload(t, rel, "pending", nil)
124 a.State = domain.ReleaseAssetUploading
125 used, v, err := r.store.Releases().ReleaseUsage(r.ctx, r.repo)
126 require.NoError(t, err)
127 require.NoError(t, r.store.Releases().ApplyReleases(r.ctx, ports.ReleaseWrite{Repo: r.repo, Used: used, UsageVersion: v, PutAssets: []domain.ReleaseAsset{a}}))
128 b := mocks.NewBlobStore(t)
129 b.EXPECT().List(r.ctx, "release/", mock.Anything).RunAndReturn(func(_ context.Context, _ string, fn func(ports.BlobInfo) error) error {
130 require.NoError(t, fn(ports.BlobInfo{Key: "release/old", ModTime: now.Add(-25 * time.Hour)}))
131 require.NoError(t, fn(ports.BlobInfo{Key: "release/new", ModTime: now.Add(23 * time.Hour)}))
132 return nil
133 }).Once()
134 b.EXPECT().Delete(r.ctx, "release/old").Return(errors.New("failed deletion")).Once()
135 b.EXPECT().ListMultipart(r.ctx, "release/", mock.Anything).RunAndReturn(func(_ context.Context, _ string, fn func(ports.MultipartInfo) error) error {
136 require.NoError(t, fn(ports.MultipartInfo{Key: "release/old", ID: "old", Started: now.Add(-25 * time.Hour)}))
137 require.NoError(t, fn(ports.MultipartInfo{Key: "release/new", ID: "new", Started: now}))
138 return nil
139 }).Once()
140 b.EXPECT().AbortMultipart(r.ctx, "release/old", "old").Return(nil).Once()
141 r.setClock(t, now.Add(24*time.Hour), b)
142 // Use times relative to the sweep, keeping both new objects within grace.
143 // pending exactly 24h old survives; failures in orphan deletion do not stop multipart GC.
144 r.svc.Sweep(r.ctx)
145 _, err = r.store.Releases().GetAsset(r.ctx, a.ID)
146 require.NoError(t, err)
147}