1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package usecase_test
4
5import (
6 "context"
7 "errors"
8 "runtime"
9 "sync"
10 "sync/atomic"
11 "testing"
12 "testing/synctest"
13 "time"
14
15 "github.com/stretchr/testify/assert"
16 "github.com/stretchr/testify/require"
17
18 "bnema.dev/ermite/internal/domain"
19 "bnema.dev/ermite/internal/ports/mocks"
20 "bnema.dev/ermite/internal/usecase"
21)
22
23// Every test runs in a synctest bubble with a fake clock: the delays are the
24// real ones, a run happens exactly when the clock reaches it, and a Close that
25// hangs is reported as a bubble deadlock instead of a timeout.
26
27func TestMaintainerDebounces(t *testing.T) {
28 t.Parallel()
29 synctest.Test(t, func(t *testing.T) {
30 ctx := context.Background()
31 n := name(t, "tools")
32 git := mocks.NewGitRepos(t)
33 var runs atomic.Int32
34 git.EXPECT().Maintain(mockAny, n).RunAndReturn(func(context.Context, domain.RepoName) error {
35 runs.Add(1)
36 return nil
37 })
38 m := usecase.NewMaintainer(ctx, git, usecase.MaintenanceDelay)
39 defer m.Close(ctx)
40
41 for range 10 {
42 m.Schedule(ctx, n)
43 }
44 synctest.Sleep(usecase.MaintenanceDelay - time.Nanosecond)
45 assert.Equal(t, int32(0), runs.Load(), "waits for the full delay")
46 synctest.Sleep(time.Nanosecond)
47 assert.Equal(t, int32(1), runs.Load(), "a burst makes one run")
48 synctest.Sleep(time.Hour)
49 assert.Equal(t, int32(1), runs.Load(), "no second run is coming")
50
51 m.Schedule(ctx, n)
52 synctest.Sleep(usecase.MaintenanceDelay - time.Nanosecond)
53 assert.Equal(t, int32(1), runs.Load(), "a second burst waits for its own delay")
54 synctest.Sleep(time.Nanosecond)
55 assert.Equal(t, int32(2), runs.Load(), "a second burst, a second run")
56 })
57}
58
59func TestMaintainerOneRunAtATimePerRepo(t *testing.T) {
60 t.Parallel()
61 synctest.Test(t, func(t *testing.T) {
62 ctx := context.Background()
63 n, other := name(t, "tools"), name(t, "other")
64 git := mocks.NewGitRepos(t)
65 var concurrent, maxConcurrent, runs atomic.Int32
66 release := make(chan struct{})
67 var otherRan atomic.Bool
68 git.EXPECT().Maintain(mockAny, n).RunAndReturn(func(context.Context, domain.RepoName) error {
69 c := concurrent.Add(1)
70 for {
71 m := maxConcurrent.Load()
72 if c <= m || maxConcurrent.CompareAndSwap(m, c) {
73 break
74 }
75 }
76 runs.Add(1)
77 <-release
78 concurrent.Add(-1)
79 return nil
80 })
81 git.EXPECT().Maintain(mockAny, other).RunAndReturn(func(context.Context, domain.RepoName) error {
82 otherRan.Store(true)
83 return nil
84 })
85 const d = usecase.MaintenanceDelay
86 m := usecase.NewMaintainer(ctx, git, d)
87 defer m.Close(ctx)
88
89 m.Schedule(ctx, n)
90 synctest.Sleep(d)
91 require.Equal(t, int32(1), runs.Load(), "first run started")
92 m.Schedule(ctx, n) // during the run: queues exactly one more
93 m.Schedule(ctx, n)
94 m.Schedule(ctx, other)
95 synctest.Sleep(d)
96 assert.True(t, otherRan.Load(), "another repository is not blocked")
97 assert.Equal(t, int32(1), runs.Load(), "the second run waits for the first")
98
99 release <- struct{}{} // finish the first run: the queued one is armed
100 synctest.Sleep(d - time.Nanosecond)
101 assert.Equal(t, int32(1), runs.Load(), "the queued run waits for its delay")
102 synctest.Sleep(time.Nanosecond)
103 assert.Equal(t, int32(2), runs.Load(), "queued run")
104 release <- struct{}{}
105 synctest.Sleep(time.Hour)
106 assert.Equal(t, int32(2), runs.Load(), "exactly one queued run")
107 assert.Equal(t, int32(1), maxConcurrent.Load())
108 })
109}
110
111func TestMaintainerTellsTheReaderAfterARun(t *testing.T) {
112 t.Parallel()
113 for label, failure := range map[string]error{"success": nil, "failure": errors.New("boom")} {
114 t.Run(label, func(t *testing.T) {
115 t.Parallel()
116 synctest.Test(t, func(t *testing.T) {
117 ctx := context.Background()
118 n := name(t, "tools")
119 git := mocks.NewGitRepos(t)
120 git.EXPECT().Maintain(mockAny, n).Return(failure)
121 told := make(chan struct{}, 2)
122 fresh := mocks.NewRepoFreshener(t)
123 fresh.EXPECT().Refresh(mockAny, n).Run(func(context.Context, domain.RepoName) { told <- struct{}{} }).Return().Once()
124 m := usecase.NewMaintainer(ctx, git, usecase.MaintenanceDelay).WithFreshener(fresh)
125 m.Schedule(ctx, n)
126 synctest.Sleep(usecase.MaintenanceDelay - time.Nanosecond)
127 assert.Empty(t, told, "not before the run")
128 synctest.Sleep(time.Nanosecond)
129 assert.Len(t, told, 1, "the reader is told when the run ends")
130 m.Close(ctx)
131 synctest.Sleep(time.Hour)
132 assert.Len(t, told, 1, "told exactly once")
133 })
134 })
135 }
136}
137
138func TestMaintainerFailureIsLoggedNotFatal(t *testing.T) {
139 t.Parallel()
140 synctest.Test(t, func(t *testing.T) {
141 ctx := context.Background()
142 n := name(t, "tools")
143 git := mocks.NewGitRepos(t)
144 var runs atomic.Int32
145 git.EXPECT().Maintain(mockAny, n).RunAndReturn(func(context.Context, domain.RepoName) error {
146 runs.Add(1)
147 return errors.New("boom")
148 })
149 m := usecase.NewMaintainer(ctx, git, usecase.MaintenanceDelay)
150 defer m.Close(ctx)
151 m.Schedule(ctx, n)
152 synctest.Sleep(usecase.MaintenanceDelay)
153 assert.Equal(t, int32(1), runs.Load())
154 m.Schedule(ctx, n)
155 synctest.Sleep(usecase.MaintenanceDelay)
156 assert.Equal(t, int32(2), runs.Load(), "later runs still happen")
157 })
158}
159
160func TestMaintainerClose(t *testing.T) {
161 t.Parallel()
162 ctx := context.Background()
163 n := name(t, "tools")
164
165 t.Run("pending runs are dropped", func(t *testing.T) {
166 t.Parallel()
167 synctest.Test(t, func(t *testing.T) {
168 git := mocks.NewGitRepos(t) // no Maintain expected
169 m := usecase.NewMaintainer(ctx, git, usecase.MaintenanceDelay)
170 m.Schedule(ctx, n)
171 m.Close(ctx)
172 m.Schedule(ctx, n) // after Close: ignored
173 m.Close(ctx) // twice is fine
174 synctest.Sleep(time.Hour)
175 })
176 })
177
178 t.Run("a run in progress is cancelled and awaited", func(t *testing.T) {
179 t.Parallel()
180 synctest.Test(t, func(t *testing.T) {
181 git := mocks.NewGitRepos(t)
182 var cancelled atomic.Bool
183 git.EXPECT().Maintain(mockAny, n).RunAndReturn(func(ctx context.Context, _ domain.RepoName) error {
184 <-ctx.Done()
185 cancelled.Store(true)
186 return ctx.Err()
187 })
188 m := usecase.NewMaintainer(ctx, git, usecase.MaintenanceDelay)
189 m.Schedule(ctx, n)
190 synctest.Sleep(usecase.MaintenanceDelay) // the run is blocked in Maintain
191 assert.False(t, cancelled.Load())
192 m.Close(ctx) // a hang is a bubble deadlock: reported, no timeout needed
193 assert.True(t, cancelled.Load())
194 })
195 })
196}
197
198func TestMaintainerConcurrentSchedule(t *testing.T) {
199 t.Parallel()
200 synctest.Test(t, func(t *testing.T) {
201 ctx := context.Background()
202 git := mocks.NewGitRepos(t)
203 git.EXPECT().Maintain(mockAny, mockAny).Return(nil).Maybe()
204 m := usecase.NewMaintainer(ctx, git, usecase.MaintenanceDelay)
205 var wg sync.WaitGroup
206 for i := range 8 {
207 wg.Add(1)
208 go func() {
209 defer wg.Done()
210 for range 50 {
211 m.Schedule(ctx, name(t, string(rune('a'+i%3))))
212 runtime.Gosched()
213 }
214 }()
215 }
216 wg.Wait()
217 m.Close(ctx)
218 })
219}