1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package web
4
5import (
6 "context"
7 "errors"
8 "fmt"
9 "sync"
10 "sync/atomic"
11 "testing"
12 "time"
13
14 "github.com/stretchr/testify/assert"
15 "github.com/stretchr/testify/require"
16)
17
18// store puts a list in the cache the way a finished build does.
19func (c *pathCache) store(id string, body []byte) {
20 c.mu.Lock()
21 defer c.mu.Unlock()
22 c.putLocked(id, body)
23}
24
25// has says whether a list is cached, marking it as used.
26func (c *pathCache) has(id string) bool {
27 c.mu.Lock()
28 defer c.mu.Unlock()
29 _, ok := c.getLocked(id)
30 return ok
31}
32
33// entry is the budget an entry of a one-letter id and an n-byte body takes.
34func entry(n int) int { return 1 + n + entryOverhead }
35
36func TestPathCacheEvictsLeastRecentlyUsed(t *testing.T) {
37 t.Parallel()
38 budget := 4 * entry(100)
39 c := newPathCache(budget, 1)
40 blob := func(n int) []byte { return make([]byte, n) }
41 c.store("a", blob(100))
42 c.store("b", blob(100))
43 c.store("e", blob(100))
44 require.True(t, c.has("a")) // a is now the most recent
45 c.store("c", blob(100)) // four fit
46 c.store("d", blob(10)) // one more: b, the least recent, goes
47 assert.False(t, c.has("b"), "b was evicted")
48 assert.True(t, c.has("a") && c.has("c") && c.has("d") && c.has("e"))
49 assert.Equal(t, 3*entry(100)+entry(10), c.size)
50
51 // Replacing an entry accounts for the old size.
52 c.store("a", blob(10))
53 assert.Equal(t, 2*entry(100)+2*entry(10), c.size)
54
55 // A big one pushes out as many as needed.
56 c.store("f", blob(100))
57 c.store("g", blob(100))
58 assert.LessOrEqual(t, c.size, budget)
59 assert.True(t, c.has("g"))
60}
61
62// Tiny bodies (a cached "no license") count their id and overhead too, so
63// their number stays bounded by the budget.
64func TestPathCacheBoundsTinyEntries(t *testing.T) {
65 t.Parallel()
66 c := newPathCache(1<<12, 1)
67 for i := range 10_000 {
68 c.store(fmt.Sprintf("repo\x00%040d", i), []byte("null"))
69 }
70 assert.LessOrEqual(t, c.size, 1<<12)
71 assert.Less(t, len(c.byID), (1<<12)/entryOverhead+1)
72 assert.Equal(t, len(c.byID), c.lru.Len())
73}
74
75// A list over a quarter of the budget is not kept, since it would push out the
76// others; one at the limit is.
77func TestPathCacheSkipsOversizedEntries(t *testing.T) {
78 t.Parallel()
79 c := newPathCache(1000, 1)
80 var builds atomic.Int32
81 big := func(n int) func() ([]byte, error) {
82 return func() ([]byte, error) { builds.Add(1); return make([]byte, n), nil }
83 }
84 for range 2 {
85 b, err := c.load(context.Background(), "r", "big", big(251))
86 require.NoError(t, err)
87 assert.Len(t, b, 251, "the caller still gets the list")
88 }
89 assert.EqualValues(t, 2, builds.Load(), "not cached, so built again")
90 assert.False(t, c.has("big"))
91 assert.Zero(t, c.size)
92
93 for range 2 {
94 _, err := c.load(context.Background(), "r", "fits", big(250))
95 require.NoError(t, err)
96 }
97 assert.EqualValues(t, 3, builds.Load(), "at a quarter: kept")
98 assert.Equal(t, len("fits")+250+entryOverhead, c.size)
99}
100
101func TestPathCacheConcurrent(t *testing.T) {
102 t.Parallel()
103 c := newPathCache(1000, 1)
104 var wg sync.WaitGroup
105 for g := range 8 {
106 wg.Add(1)
107 go func() {
108 defer wg.Done()
109 for i := range 500 {
110 c.store(fmt.Sprint((g+i)%17), make([]byte, 10+i%90))
111 c.has(fmt.Sprint(i % 17))
112 }
113 }()
114 }
115 wg.Wait()
116 assert.LessOrEqual(t, c.size, 1000)
117 sum := 0
118 for _, el := range c.byID {
119 sum += el.Value.(*pathEntry).cost()
120 }
121 assert.Equal(t, sum, c.size, "the size is the sum of the entries")
122 assert.Equal(t, len(c.byID), c.lru.Len())
123}
124
125// waiters is the number of callers waiting for the build of a key, -1 when
126// none is in flight.
127func waiters(c *pathCache, repo, id string) int {
128 c.mu.Lock()
129 defer c.mu.Unlock()
130 if f := c.building[repo+"\x00"+id]; f != nil {
131 return f.waiters
132 }
133 return -1
134}
135
136func building(c *pathCache, n int) bool {
137 c.mu.Lock()
138 defer c.mu.Unlock()
139 return len(c.building) == n
140}
141
142// Concurrent misses for one commit build once; a failure is shared and not
143// stored; a waiter that gives up leaves the build running.
144func TestPathCacheLoadBuildsOnce(t *testing.T) {
145 t.Parallel()
146 c := newPathCache(1000, 4)
147 var builds atomic.Int32
148 release := make(chan struct{})
149 build := func() ([]byte, error) {
150 builds.Add(1)
151 <-release
152 return []byte("list"), nil
153 }
154 const callers = 10
155 var wg sync.WaitGroup
156 got := make([][]byte, callers)
157 for i := range got {
158 wg.Add(1)
159 go func() {
160 defer wg.Done()
161 b, err := c.load(context.Background(), "r", "x", build)
162 assert.NoError(t, err)
163 got[i] = b
164 }()
165 }
166 // Every caller waits for the one build before it is released: none can
167 // find the list cached, so one build serving all ten shows they joined it.
168 require.Eventually(t, func() bool { return waiters(c, "r", "x") == callers }, 5*time.Second, time.Millisecond)
169 // A waiter whose context ends does not stop the others.
170 ctx, cancel := context.WithCancel(context.Background())
171 cancel()
172 _, err := c.load(ctx, "r", "x", build)
173 assert.ErrorIs(t, err, context.Canceled)
174 close(release)
175 wg.Wait()
176 assert.EqualValues(t, 1, builds.Load())
177 for _, b := range got {
178 assert.Equal(t, "list", string(b))
179 }
180 assert.True(t, c.has("x"))
181
182 boom := errors.New("boom")
183 _, err = c.load(context.Background(), "r", "y", func() ([]byte, error) { return nil, boom })
184 assert.ErrorIs(t, err, boom)
185 assert.False(t, c.has("y"), "a failure is not cached")
186 b, err := c.load(context.Background(), "r", "y", func() ([]byte, error) { return []byte("ok"), nil })
187 require.NoError(t, err)
188 assert.Equal(t, "ok", string(b), "the next request builds again")
189 assert.Empty(t, c.building)
190}
191
192// The builds of all commits share the slots: with one slot, a load of another
193// commit waits for the running build, and leaves with its ctx if it is
194// cancelled first. The slot is the build's, not the caller's.
195func TestPathCacheLoadBoundsBuilds(t *testing.T) {
196 t.Parallel()
197 c := newPathCache(1000, 1)
198 started, release := make(chan struct{}), make(chan struct{})
199 first := make(chan error, 1)
200 go func() {
201 _, err := c.load(context.Background(), "r", "a", func() ([]byte, error) { close(started); <-release; return []byte("a"), nil })
202 first <- err
203 }()
204 <-started
205
206 // A second commit waits for the slot...
207 var secondBuilt atomic.Bool
208 second := make(chan error, 1)
209 go func() {
210 _, err := c.load(context.Background(), "r", "b", func() ([]byte, error) { secondBuilt.Store(true); return []byte("b"), nil })
211 second <- err
212 }()
213 time.Sleep(30 * time.Millisecond)
214 assert.False(t, secondBuilt.Load(), "no slot, no build")
215 select {
216 case <-second:
217 t.Fatal("the second load did not wait for a slot")
218 default:
219 }
220
221 // ...or leaves with its ctx.
222 ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
223 defer cancel()
224 _, err := c.load(ctx, "r", "c", func() ([]byte, error) { t.Error("built without a slot"); return nil, nil })
225 assert.ErrorIs(t, err, context.DeadlineExceeded)
226
227 // A caller of the running build is not held by the slots.
228 _, err = c.load(ctx, "r", "a", nil)
229 assert.ErrorIs(t, err, context.DeadlineExceeded, "it joins the build and leaves with its ctx")
230
231 close(release)
232 require.NoError(t, <-first)
233 require.NoError(t, <-second)
234 assert.True(t, secondBuilt.Load())
235 assert.True(t, c.has("a") && c.has("b"))
236 assert.Empty(t, c.building)
237 assert.Empty(t, c.slots, "every slot is back")
238}
239
240// The slot of a build is kept after its callers leave, and given back when it
241// ends, whether it fails or panics.
242func TestPathCacheSlotOutlivesCallers(t *testing.T) {
243 t.Parallel()
244 c := newPathCache(1000, 1)
245 started, release := make(chan struct{}), make(chan struct{})
246 ctx, cancel := context.WithCancel(context.Background())
247 done := make(chan error, 1)
248 go func() {
249 _, err := c.load(ctx, "r", "a", func() ([]byte, error) { close(started); <-release; panic("boom") })
250 done <- err
251 }()
252 <-started
253 cancel()
254 assert.ErrorIs(t, <-done, context.Canceled)
255 assert.Len(t, c.slots, 1, "the build still holds its slot")
256 close(release)
257 require.Eventually(t, func() bool { return len(c.slots) == 0 && building(c, 0) }, time.Second, time.Millisecond)
258 b, err := c.load(context.Background(), "r", "b", func() ([]byte, error) { return []byte("ok"), nil })
259 require.NoError(t, err)
260 assert.Equal(t, "ok", string(b))
261}
262
263// A build that panics is handed to its waiters as an error, leaves nothing
264// in flight or cached, and does not take the process down.
265func TestPathCacheLoadPanic(t *testing.T) {
266 t.Parallel()
267 c := newPathCache(1000, 4)
268 started, release := make(chan struct{}), make(chan struct{})
269 var once sync.Once
270 build := func() ([]byte, error) {
271 once.Do(func() { close(started) })
272 <-release
273 panic("boom")
274 }
275 errs := make(chan error, 2)
276 for range 2 {
277 go func() {
278 _, err := c.load(context.Background(), "r", "x", build)
279 errs <- err
280 }()
281 }
282 <-started
283 require.Eventually(t, func() bool { c.mu.Lock(); defer c.mu.Unlock(); return len(c.building) == 1 }, time.Second, time.Millisecond)
284 close(release)
285 for range 2 {
286 assert.ErrorIs(t, <-errs, errBuildPanicked)
287 }
288 assert.Empty(t, c.building)
289 assert.False(t, c.has("x"), "a panic is not cached")
290
291 b, err := c.load(context.Background(), "r", "x", func() ([]byte, error) { return []byte("ok"), nil })
292 require.NoError(t, err)
293 assert.Equal(t, "ok", string(b), "the next request builds again")
294}
295
296// The caller that starts a build is released with its context too, and the
297// build goes on for the others.
298func TestPathCacheLoadStarterLeaves(t *testing.T) {
299 t.Parallel()
300 c := newPathCache(1000, 4)
301 release := make(chan struct{})
302 build := func() ([]byte, error) { <-release; return []byte("list"), nil }
303 ctx, cancel := context.WithCancel(context.Background())
304 done := make(chan error, 1)
305 go func() { _, err := c.load(ctx, "r", "x", build); done <- err }()
306 require.Eventually(t, func() bool { c.mu.Lock(); defer c.mu.Unlock(); return len(c.building) == 1 }, time.Second, time.Millisecond)
307 cancel()
308 select {
309 case err := <-done:
310 assert.ErrorIs(t, err, context.Canceled)
311 case <-time.After(time.Second):
312 t.Fatal("the starter is held by its build")
313 }
314 close(release)
315 require.Eventually(t, func() bool { return c.has("x") }, time.Second, time.Millisecond)
316 assert.Empty(t, c.building)
317}
318
319// A failure belongs to the repository that built it: another repository
320// asking for the same commit builds its own.
321func TestPathCacheLoadFailureStaysInRepo(t *testing.T) {
322 t.Parallel()
323 c := newPathCache(1000, 4)
324 started, release := make(chan struct{}), make(chan struct{})
325 boom := errors.New("boom")
326 failing := make(chan error, 1)
327 go func() {
328 _, err := c.load(context.Background(), "a", "x", func() ([]byte, error) { close(started); <-release; return nil, boom })
329 failing <- err
330 }()
331 <-started
332 b, err := c.load(context.Background(), "b", "x", func() ([]byte, error) { return []byte("list"), nil })
333 require.NoError(t, err)
334 assert.Equal(t, "list", string(b))
335 close(release)
336 assert.ErrorIs(t, <-failing, boom)
337}