1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package usecase_test
4
5import (
6 "context"
7 "errors"
8 "strconv"
9 "strings"
10 "sync/atomic"
11 "testing"
12 "time"
13
14 "github.com/stretchr/testify/assert"
15 "github.com/stretchr/testify/mock"
16 "github.com/stretchr/testify/require"
17
18 "bnema.dev/ermite/internal/domain"
19 "bnema.dev/ermite/internal/ports"
20 "bnema.dev/ermite/internal/ports/mocks"
21 "bnema.dev/ermite/internal/usecase"
22)
23
24func fakeOID(t *testing.T, c string) domain.ObjectID {
25 t.Helper()
26 return oidFrom(t, strings.Repeat(c, 40))
27}
28
29// mergeRig runs the mergeability service over the real adapters of mrRig.
30type mergeRig struct {
31 *mrRig
32 svc *usecase.Mergeability
33 mev <-chan domain.Event
34 history []domain.MergeRequestMergeability
35}
36
37func newMergeRig(t *testing.T) *mergeRig {
38 t.Helper()
39 r := newMRRig(t)
40 svc := usecase.NewMergeability(r.ctx, usecase.MergeabilityConfig{
41 Store: r.store.MergeRequests(), Meta: r.store, Git: r.git, Bus: r.bus, Clock: fixedClock{}, Workers: 2,
42 })
43 r.mrs = usecase.NewMergeRequests(usecase.MergeRequestsConfig{
44 Meta: r.store, Store: r.store.MergeRequests(), Repos: r.git, Git: r.git, Signer: nil,
45 Committer: domain.Identity{Name: "ermite", Email: "ermite@git.example.org"}, Bus: r.bus, Clock: fixedClock{}, Maint: r.sched,
46 Mergeability: svc,
47 })
48 t.Cleanup(func() { svc.Close(context.Background()) })
49 return &mergeRig{mrRig: r, svc: svc, mev: r.bus.Subscribe(r.ctx)}
50}
51
52func (r *mergeRig) record(n int64) (domain.Mergeability, error) {
53 return r.store.MergeRequests().GetMergeability(r.ctx, r.repo, n)
54}
55
56func (r *mergeRig) waitState(n int64, want domain.MergeabilityState, target, head domain.ObjectID) domain.Mergeability {
57 r.t.Helper()
58 var m domain.Mergeability
59 deadline := time.After(10 * time.Second)
60 for {
61 var err error
62 m, err = r.record(n)
63 if err == nil && m.State == want && m.IsFor(target, head) {
64 break
65 }
66 select {
67 case ev := <-r.mev:
68 if e, ok := ev.(domain.MergeRequestMergeability); ok {
69 r.history = append(r.history, e)
70 }
71 case <-deadline:
72 r.t.Fatalf("mergeability of !%d did not become %s", n, want)
73 }
74 }
75
76 return m
77}
78
79// events returns the mergeability events published so far.
80func (r *mergeRig) events() []domain.MergeRequestMergeability {
81 out := r.history
82 r.history = nil
83 for {
84 select {
85 case ev := <-r.mev:
86 if m, ok := ev.(domain.MergeRequestMergeability); ok {
87 out = append(out, m)
88 }
89 default:
90 return out
91 }
92 }
93}
94
95// awaitEvents receives mergeability events until n arrived, or fails after a
96// bound: the workers publish after they stored the record the tests wait for.
97func (r *mergeRig) awaitEvents(n int) []domain.MergeRequestMergeability {
98 r.t.Helper()
99 out := r.history
100 r.history = nil
101 deadline := time.After(10 * time.Second)
102 for len(out) < n {
103 select {
104 case ev := <-r.mev:
105 if m, ok := ev.(domain.MergeRequestMergeability); ok {
106 out = append(out, m)
107 }
108 case <-deadline:
109 r.t.Fatalf("got %d mergeability events, want %d", len(out), n)
110 }
111 }
112 return out
113}
114
115func TestMergeabilityStatesEndToEnd(t *testing.T) {
116 r := newMergeRig(t)
117 r.svc.Start()
118 main := r.seed()
119 head := r.feature("feat")
120 mr := r.open("feat")
121
122 // up to date: the target is an ancestor of the head.
123 m := r.waitState(mr.Number, domain.MergeabilityUpToDate, main, head)
124 assert.False(t, m.CheckedAt.IsZero())
125
126 // behind: the target moved with a change that merges cleanly.
127 r.sh("checkout", "-q", "main")
128 main2 := r.commit(author, "d.txt", "dee\n", "add d")
129 r.push("main")
130 r.svc.OnBranchesMoved(r.ctx, r.repo, []domain.RefUpdate{{Old: main.String(), New: main2.String(), Ref: "refs/heads/main"}})
131 m = r.waitState(mr.Number, domain.MergeabilityBehind, main2, head)
132 assert.Equal(t, 1, m.Behind)
133 assert.False(t, m.BehindCapped)
134
135 // conflict: the target changes a line the head changed.
136 main3 := r.commit(author, "a.txt", "one\nTWO\nthree\n", "conflict a")
137 r.push("main")
138 r.svc.OnBranchesMoved(r.ctx, r.repo, []domain.RefUpdate{{Old: main2.String(), New: main3.String(), Ref: "refs/heads/main"}})
139 m = r.waitState(mr.Number, domain.MergeabilityConflict, main3, head)
140 assert.Equal(t, []string{"a.txt"}, m.ConflictPaths)
141
142 got := r.awaitEvents(3)
143 for i, s := range []domain.MergeabilityState{domain.MergeabilityUpToDate, domain.MergeabilityBehind, domain.MergeabilityConflict} {
144 assert.Equal(t, s, got[i].Mergeability.State)
145 assert.Equal(t, mr.Number, got[i].Number)
146 }
147}
148
149func TestMergeabilityReaderIsLazy(t *testing.T) {
150 r := newMergeRig(t)
151 main := r.seed()
152 head := r.feature("feat")
153 mr := r.open("feat") // queued, but the workers are not started
154
155 d, err := r.mrs.Get(r.ctx, root, r.repo, mr.Number, ports.GetMergeRequestOptions{})
156 require.NoError(t, err)
157 assert.Equal(t, domain.MergeabilityChecking, d.Mergeability.State)
158
159 r.svc.Start()
160 r.waitState(mr.Number, domain.MergeabilityUpToDate, main, head)
161 d, err = r.mrs.Get(r.ctx, root, r.repo, mr.Number, ports.GetMergeRequestOptions{})
162 require.NoError(t, err)
163 assert.Equal(t, domain.MergeabilityUpToDate, d.Mergeability.State)
164
165 page, err := r.mrs.List(r.ctx, root, r.repo, ports.ListMergeRequestsParams{})
166 require.NoError(t, err)
167 assert.Equal(t, domain.MergeabilityUpToDate, page.Mergeability[mr.Number].State)
168}
169
170func TestMergeabilityPublishesEachSettledResultOnce(t *testing.T) {
171 r := newMergeRig(t)
172 r.seed()
173 r.feature("feat")
174 mr := r.open("feat")
175 r.svc.ReconcileForTest(r.repo)
176 require.Len(t, r.events(), 1)
177
178 // Same tips: nothing is computed, nothing is published.
179 r.svc.ReconcileForTest(r.repo)
180 assert.Empty(t, r.events())
181
182 // New tips publish even when the state and counts stay the same.
183 r.sh("checkout", "-q", "feat")
184 r.commit(author, "e.txt", "e\n", "more")
185 r.push("feat")
186 r.svc.ReconcileForTest(r.repo)
187 assert.Len(t, r.events(), 1)
188 m, err := r.record(mr.Number)
189 require.NoError(t, err)
190 assert.Equal(t, domain.MergeabilityUpToDate, m.State)
191}
192
193func TestMergeabilityMergeReevaluatesOtherRequests(t *testing.T) {
194 r := newMergeRig(t)
195 r.svc.Start()
196 main := r.seed()
197 r.sh("checkout", "-q", "-B", "one", "main")
198 oneHead := r.commit(author, "one.txt", "1\n", "one")
199 r.push("one")
200 r.sh("checkout", "-q", "-B", "two", "main")
201 twoHead := r.commit(author, "two.txt", "2\n", "two")
202 r.push("two")
203 one, two := r.open("one"), r.open("two")
204 r.waitState(one.Number, domain.MergeabilityUpToDate, main, oneHead)
205 r.waitState(two.Number, domain.MergeabilityUpToDate, main, twoHead)
206
207 _, err := r.mrs.Merge(r.ctx, mrM, r.repo, one.Number, ports.MergeParams{HeadID: oneHead})
208 require.NoError(t, err)
209 m := r.waitState(two.Number, domain.MergeabilityBehind, oneHead, twoHead)
210 assert.Equal(t, 1, m.Behind)
211 _, err = r.record(one.Number)
212 assert.ErrorIs(t, err, domain.ErrNotFound, "the merged request keeps no record")
213}
214
215func TestMergeabilityClosedRequestRecordDeleted(t *testing.T) {
216 r := newMergeRig(t)
217 r.seed()
218 r.feature("feat")
219 mr := r.open("feat")
220 r.svc.ReconcileForTest(r.repo)
221 _, err := r.record(mr.Number)
222 require.NoError(t, err)
223
224 _, err = r.mrs.Close(r.ctx, mrW, r.repo, mr.Number)
225 require.NoError(t, err)
226 _, err = r.record(mr.Number)
227 assert.ErrorIs(t, err, domain.ErrNotFound)
228 r.svc.ReconcileForTest(r.repo) // a late check writes nothing
229 _, err = r.record(mr.Number)
230 assert.ErrorIs(t, err, domain.ErrNotFound)
231}
232
233func TestMergeabilityStartupRescanComputesMissingRecords(t *testing.T) {
234 r := newMergeRig(t)
235 main := r.seed()
236 head := r.feature("feat")
237 mr := r.open("feat")
238 _, err := r.record(mr.Number)
239 require.ErrorIs(t, err, domain.ErrNotFound)
240
241 // A fresh service has received no OnOpened or Read trigger.
242 svc := usecase.NewMergeability(r.ctx, usecase.MergeabilityConfig{
243 Store: r.store.MergeRequests(), Meta: r.store, Git: r.git, Bus: r.bus, Clock: fixedClock{},
244 })
245 defer svc.Close(context.Background())
246 svc.Start()
247 r.waitState(mr.Number, domain.MergeabilityUpToDate, main, head)
248}
249
250func TestMergeabilityNeedsNoCI(t *testing.T) {
251 // The service is built without any CI: a target move still updates the
252 // record, and the repository gets no pull ref (the service has no CI to create a run with).
253 r := newMergeRig(t)
254 r.svc.Start()
255 main := r.seed()
256 head := r.feature("feat")
257 mr := r.open("feat")
258 r.waitState(mr.Number, domain.MergeabilityUpToDate, main, head)
259 r.sh("checkout", "-q", "main")
260 main2 := r.commit(author, "d.txt", "dee\n", "add d")
261 r.push("main")
262 r.svc.OnBranchesMoved(r.ctx, r.repo, []domain.RefUpdate{{Old: main.String(), New: main2.String(), Ref: "refs/heads/main"}})
263 r.waitState(mr.Number, domain.MergeabilityBehind, main2, head)
264
265 assert.Empty(t, r.run(r.bare(), r.env(author), "for-each-ref", "refs/pull"))
266}
267
268// --- unit tests over mocks ---
269
270type mockRig struct {
271 store *mocks.MergeRequestStore
272 meta *mocks.RepoMetaStore
273 git *mocks.MergeGit
274 bus *mocks.EventBus
275 svc *usecase.Mergeability
276 mr domain.MergeRequest
277}
278
279func newMockRig(t *testing.T) *mockRig {
280 t.Helper()
281 repo := name(t, "proj")
282 r := &mockRig{
283 store: mocks.NewMergeRequestStore(t), meta: mocks.NewRepoMetaStore(t), git: mocks.NewMergeGit(t), bus: mocks.NewEventBus(t),
284 mr: domain.MergeRequest{Repo: repo, Number: 7, Source: "feat", Target: "main", State: domain.MRStateOpen},
285 }
286 r.store.EXPECT().ListOpen(mock.Anything, repo).Return([]domain.MergeRequest{r.mr}, nil).Maybe()
287 r.store.EXPECT().GetMergeability(mock.Anything, repo, int64(7)).Return(domain.Mergeability{}, domain.ErrNotFound).Maybe()
288 r.meta.EXPECT().List(mock.Anything).Return(nil, nil).Maybe()
289 r.meta.EXPECT().Get(mock.Anything, repo).Return(domain.Repo{Name: repo, CreatedAt: now}, nil).Maybe()
290 r.svc = usecase.NewMergeability(context.Background(), usecase.MergeabilityConfig{
291 Store: r.store, Meta: r.meta, Git: r.git, Bus: r.bus, Clock: fixedClock{},
292 })
293 t.Cleanup(func() { r.svc.Close(context.Background()) })
294 return r
295}
296
297func TestMergeabilityDedupsTriggers(t *testing.T) {
298 r := newMockRig(t)
299 repo := r.mr.Repo
300 head, tip := fakeOID(t, "a"), fakeOID(t, "b")
301 r.git.EXPECT().BranchHead(mock.Anything, repo, "feat").Return(head, nil)
302 r.git.EXPECT().BranchHead(mock.Anything, repo, "main").Return(tip, nil)
303 computed := make(chan struct{}, 1)
304 var computes atomic.Int32
305 r.git.EXPECT().IsAncestor(mock.Anything, repo, tip, head).RunAndReturn(
306 func(context.Context, domain.RepoName, domain.ObjectID, domain.ObjectID) (bool, error) {
307 computes.Add(1)
308 computed <- struct{}{}
309 return true, nil
310 })
311 r.store.EXPECT().PutMergeability(mock.Anything, mock.Anything).Return(nil)
312 r.bus.EXPECT().Publish(mock.Anything, mock.Anything).Return()
313
314 for range 50 {
315 r.svc.OnOpened(context.Background(), r.mr) // queued before any worker runs
316 }
317 r.svc.Start()
318 <-computed
319 r.svc.Close(context.Background()) // the workers have each looked at the queue once more
320 assert.EqualValues(t, 1, computes.Load())
321}
322
323func TestMergeabilityNoWriteForClosedRequest(t *testing.T) {
324 r := newMockRig(t)
325 repo := r.mr.Repo
326 head, tip := fakeOID(t, "a"), fakeOID(t, "b")
327 r.git.EXPECT().BranchHead(mock.Anything, repo, "feat").Return(head, nil)
328 r.git.EXPECT().BranchHead(mock.Anything, repo, "main").Return(tip, nil)
329 r.git.EXPECT().IsAncestor(mock.Anything, repo, tip, head).Return(true, nil)
330 r.store.EXPECT().PutMergeability(mock.Anything, mock.Anything).Return(domain.ErrNotFound).Once() // closed meanwhile
331 r.svc.ReconcileForTest(repo) // no event, no further write
332}
333
334func TestMergeabilityBoundsConflictPathsAndBehind(t *testing.T) {
335 t.Run("conflict paths", func(t *testing.T) {
336 r := newMockRig(t)
337 repo := r.mr.Repo
338 head, tip := fakeOID(t, "a"), fakeOID(t, "b")
339 r.git.EXPECT().BranchHead(mock.Anything, repo, "feat").Return(head, nil)
340 r.git.EXPECT().BranchHead(mock.Anything, repo, "main").Return(tip, nil)
341 r.git.EXPECT().IsAncestor(mock.Anything, repo, tip, head).Return(false, nil)
342 var paths []string
343 for i := 0; i < domain.MaxConflictPaths+3; i++ {
344 paths = append(paths, strings.Repeat("p", i+1))
345 }
346 r.git.EXPECT().MergeTree(mock.Anything, repo, tip, head).Return(domain.MergeTreeResult{ConflictPaths: paths}, nil)
347 var put domain.Mergeability
348 r.store.EXPECT().PutMergeability(mock.Anything, mock.Anything).RunAndReturn(
349 func(_ context.Context, m domain.Mergeability) error { put = m; return nil })
350 r.bus.EXPECT().Publish(mock.Anything, mock.Anything).Return()
351 r.svc.ReconcileForTest(repo)
352 assert.Equal(t, domain.MergeabilityConflict, put.State)
353 assert.Len(t, put.ConflictPaths, domain.MaxConflictPaths)
354 assert.True(t, put.ConflictPathsTruncated)
355 })
356 t.Run("behind", func(t *testing.T) {
357 r := newMockRig(t)
358 repo := r.mr.Repo
359 head, tip := fakeOID(t, "a"), fakeOID(t, "b")
360 r.git.EXPECT().BranchHead(mock.Anything, repo, "feat").Return(head, nil)
361 r.git.EXPECT().BranchHead(mock.Anything, repo, "main").Return(tip, nil)
362 r.git.EXPECT().IsAncestor(mock.Anything, repo, tip, head).Return(false, nil)
363 r.git.EXPECT().MergeTree(mock.Anything, repo, tip, head).Return(domain.MergeTreeResult{Clean: true}, nil)
364 r.git.EXPECT().CountBehind(mock.Anything, repo, head, tip, domain.MaxBehindCount+1).Return(domain.MaxBehindCount+1, nil)
365 var put domain.Mergeability
366 r.store.EXPECT().PutMergeability(mock.Anything, mock.Anything).RunAndReturn(
367 func(_ context.Context, m domain.Mergeability) error { put = m; return nil })
368 r.bus.EXPECT().Publish(mock.Anything, mock.Anything).Return()
369 r.svc.ReconcileForTest(repo)
370 assert.Equal(t, domain.MaxBehindCount, put.Behind)
371 assert.True(t, put.BehindCapped)
372 })
373}
374
375// A transient Git failure is retried once after the backoff, by rescanning
376// the repository through the queue.
377func TestMergeabilityRetriesATransientFailureWithBackoff(t *testing.T) {
378 repo := name(t, "proj")
379 head, tip := fakeOID(t, "a"), fakeOID(t, "b")
380 mr := domain.MergeRequest{Repo: repo, Number: 7, Source: "feat", Target: "main", State: domain.MRStateOpen}
381 store, meta, git, bus := mocks.NewMergeRequestStore(t), mocks.NewRepoMetaStore(t), mocks.NewMergeGit(t), mocks.NewEventBus(t)
382 store.EXPECT().ListOpen(mock.Anything, repo).Return([]domain.MergeRequest{mr}, nil)
383 store.EXPECT().GetMergeability(mock.Anything, repo, int64(7)).Return(domain.Mergeability{}, domain.ErrNotFound)
384 store.EXPECT().PutMergeability(mock.Anything, mock.Anything).Return(nil)
385 meta.EXPECT().List(mock.Anything).Return(nil, nil)
386 meta.EXPECT().Get(mock.Anything, repo).Return(domain.Repo{Name: repo, CreatedAt: now}, nil)
387 var fails atomic.Int32
388 git.EXPECT().BranchHead(mock.Anything, repo, "feat").RunAndReturn(
389 func(context.Context, domain.RepoName, string) (domain.ObjectID, error) {
390 if fails.Add(1) == 1 {
391 return domain.ObjectID{}, errors.New("transient")
392 }
393 return head, nil
394 })
395 git.EXPECT().BranchHead(mock.Anything, repo, "main").Return(tip, nil)
396 git.EXPECT().IsAncestor(mock.Anything, repo, tip, head).Return(true, nil)
397 published := make(chan struct{})
398 bus.EXPECT().Publish(mock.Anything, mock.Anything).Run(func(context.Context, domain.Event) { close(published) }).Return().Once()
399
400 asked, fire := make(chan time.Duration, 4), make(chan time.Time)
401 svc := usecase.NewMergeability(context.Background(), usecase.MergeabilityConfig{
402 Store: store, Meta: meta, Git: git, Bus: bus, Clock: fixedClock{},
403 After: func(d time.Duration) <-chan time.Time { asked <- d; return fire },
404 })
405 t.Cleanup(func() { svc.Close(context.Background()) })
406 svc.Start()
407 svc.OnOpened(context.Background(), mr)
408
409 assert.Equal(t, time.Second, <-asked)
410 select {
411 case <-published:
412 t.Fatal("published before the backoff elapsed")
413 default:
414 }
415 fire <- time.Time{}
416 <-published
417 assert.Empty(t, asked, "one retry was enough and only one was scheduled")
418}
419
420// Close stops a pending retry timer's goroutine.
421func TestMergeabilityCloseStopsPendingRetries(t *testing.T) {
422 repo := name(t, "proj")
423 mr := domain.MergeRequest{Repo: repo, Number: 7, Source: "feat", Target: "main", State: domain.MRStateOpen}
424 store, meta, git, bus := mocks.NewMergeRequestStore(t), mocks.NewRepoMetaStore(t), mocks.NewMergeGit(t), mocks.NewEventBus(t)
425 store.EXPECT().ListOpen(mock.Anything, repo).Return(nil, errors.New("db down"))
426 meta.EXPECT().Get(mock.Anything, repo).Return(domain.Repo{Name: repo}, nil)
427 meta.EXPECT().List(mock.Anything).Return(nil, nil)
428 asked := make(chan time.Duration, 4)
429 svc := usecase.NewMergeability(context.Background(), usecase.MergeabilityConfig{
430 Store: store, Meta: meta, Git: git, Bus: bus, Clock: fixedClock{},
431 After: func(d time.Duration) <-chan time.Time { asked <- d; return make(chan time.Time) },
432 })
433 svc.Start()
434 svc.OnOpened(context.Background(), mr)
435 <-asked
436 done := make(chan struct{})
437 go func() { svc.Close(context.Background()); close(done) }()
438 <-done // Close waits for the timer goroutine: it returns only if it stopped
439}
440
441// Workers are not stopped by the cancellation of the context given to the
442// constructor; Close stops them.
443func TestMergeabilityWorkersOutliveTheParentContext(t *testing.T) {
444 repo := name(t, "proj")
445 head, tip := fakeOID(t, "a"), fakeOID(t, "b")
446 mr := domain.MergeRequest{Repo: repo, Number: 7, Source: "feat", Target: "main", State: domain.MRStateOpen}
447 store, meta, git, bus := mocks.NewMergeRequestStore(t), mocks.NewRepoMetaStore(t), mocks.NewMergeGit(t), mocks.NewEventBus(t)
448 store.EXPECT().ListOpen(mock.Anything, repo).Return([]domain.MergeRequest{mr}, nil)
449 store.EXPECT().GetMergeability(mock.Anything, repo, int64(7)).Return(domain.Mergeability{}, domain.ErrNotFound)
450 store.EXPECT().PutMergeability(mock.Anything, mock.Anything).Return(nil)
451 meta.EXPECT().List(mock.Anything).Return(nil, nil)
452 meta.EXPECT().Get(mock.Anything, repo).Return(domain.Repo{Name: repo, CreatedAt: now}, nil)
453 git.EXPECT().BranchHead(mock.Anything, repo, "feat").Return(head, nil)
454 git.EXPECT().BranchHead(mock.Anything, repo, "main").Return(tip, nil)
455 git.EXPECT().IsAncestor(mock.Anything, repo, tip, head).Return(true, nil)
456 published := make(chan struct{})
457 bus.EXPECT().Publish(mock.Anything, mock.Anything).Run(func(context.Context, domain.Event) { close(published) }).Return().Once()
458
459 parent, cancel := context.WithCancel(context.Background())
460 svc := usecase.NewMergeability(parent, usecase.MergeabilityConfig{Store: store, Meta: meta, Git: git, Bus: bus, Clock: fixedClock{}})
461 svc.Start()
462 cancel()
463 svc.OnOpened(context.Background(), mr)
464 <-published // computed although the parent is cancelled
465 svc.Close(context.Background())
466}
467
468// Failures of many requests of 3 repositories schedule at most one retry per
469// repository on a single scheduler goroutine.
470func TestMergeabilityRetriesAreBounded(t *testing.T) {
471 const repos = 3
472 store, meta := mocks.NewMergeRequestStore(t), mocks.NewRepoMetaStore(t)
473 barrier := name(t, "barrier")
474 done := make(chan struct{})
475 store.EXPECT().ListOpen(mock.Anything, barrier).RunAndReturn(func(context.Context, domain.RepoName) ([]domain.MergeRequest, error) { close(done); return nil, nil }).Once()
476 store.EXPECT().ListOpen(mock.Anything, mock.Anything).Return(nil, errors.New("db down"))
477 meta.EXPECT().Get(mock.Anything, mock.Anything).Return(domain.Repo{}, nil)
478 meta.EXPECT().List(mock.Anything).Return(nil, nil)
479 asked := make(chan time.Duration, repos)
480 svc := usecase.NewMergeability(context.Background(), usecase.MergeabilityConfig{
481 Store: store, Meta: meta, Clock: fixedClock{}, Workers: 1,
482 After: func(d time.Duration) <-chan time.Time { asked <- d; return make(chan time.Time) },
483 })
484 svc.Start()
485 for r := range repos {
486 repo := name(t, "proj"+strconv.Itoa(r))
487 for range 20 {
488 svc.OnOpened(context.Background(), domain.MergeRequest{Repo: repo})
489 }
490 }
491 svc.OnOpened(context.Background(), domain.MergeRequest{Repo: barrier})
492 <-done
493 // The shared scheduler only arms the earliest pending retry.
494 <-asked
495 svc.Close(context.Background())
496 assert.Equal(t, repos, svc.ScheduledRetries())
497 assert.Equal(t, repos, svc.RetryAttempts())
498}
499
500// retryRig is a service over mocks whose only repository holds request !7,
501// whose read fails for its first failFor calls. Each repository rescan signals passes.
502type retryRig struct {
503 svc *usecase.Mergeability
504 repo domain.RepoName
505 reads atomic.Int32
506 failFor atomic.Int32
507 passes chan struct{}
508 afters chan chan time.Time
509 barrier chan struct{}
510}
511
512func newRetryRig(t *testing.T) *retryRig {
513 t.Helper()
514 r := &retryRig{repo: name(t, "proj"), passes: make(chan struct{}, 32), afters: make(chan chan time.Time, 32), barrier: make(chan struct{})}
515 store, meta, git, bus := mocks.NewMergeRequestStore(t), mocks.NewRepoMetaStore(t), mocks.NewMergeGit(t), mocks.NewEventBus(t)
516 r.failFor.Store(1 << 30)
517 meta.EXPECT().Get(mock.Anything, r.repo).Return(domain.Repo{Name: r.repo}, nil)
518 store.EXPECT().ListOpen(mock.Anything, r.repo).RunAndReturn(
519 func(context.Context, domain.RepoName) ([]domain.MergeRequest, error) {
520 defer func() { r.passes <- struct{}{} }()
521 if r.reads.Add(1) <= r.failFor.Load() {
522 return nil, errors.New("db down")
523 }
524 return nil, nil
525 })
526 barrier := name(t, "barrier")
527 meta.EXPECT().Get(mock.Anything, barrier).RunAndReturn(func(context.Context, domain.RepoName) (domain.Repo, error) {
528 r.barrier <- struct{}{}
529 return domain.Repo{}, domain.ErrNotFound
530 })
531 meta.EXPECT().List(mock.Anything).Return([]domain.Repo{{Name: r.repo}}, nil)
532 r.svc = usecase.NewMergeability(context.Background(), usecase.MergeabilityConfig{
533 Store: store, Meta: meta, Git: git, Bus: bus, Clock: fixedClock{}, Workers: 1,
534 After: func(time.Duration) <-chan time.Time { ch := make(chan time.Time, 1); r.afters <- ch; return ch },
535 })
536 t.Cleanup(func() { r.svc.Close(context.Background()) })
537 return r
538}
539
540// next fires the pending retry and waits for the rescan it queues.
541func (r *retryRig) next() {
542 ch := <-r.afters
543 ch <- time.Time{}
544 <-r.passes
545}
546
547func (r *retryRig) settled(t *testing.T) {
548 t.Helper()
549 r.svc.OnOpened(context.Background(), domain.MergeRequest{Repo: name(t, "barrier")})
550 <-r.barrier
551 require.Zero(t, r.svc.RetryAttempts())
552 require.Zero(t, r.svc.ScheduledRetries())
553}
554
555// A persistent failure is retried 5 times, then given up with no state kept.
556func TestMergeabilityGivesUpAfterFiveRetries(t *testing.T) {
557 r := newRetryRig(t)
558 r.svc.Start()
559 <-r.passes // the startup rescan fails
560 for range 5 {
561 r.next()
562 }
563 r.settled(t)
564 assert.Empty(t, r.afters, "no sixth retry")
565 assert.Empty(t, r.passes)
566}
567
568// A successful full rescan resets the budget: a later failure gets 5 again.
569func TestMergeabilitySuccessfulPassForgetsTheRetryBudget(t *testing.T) {
570 r := newRetryRig(t)
571 r.failFor.Store(2) // the first two reads fail, the third succeeds
572 r.svc.Start()
573 <-r.passes
574 r.next() // second failing pass
575 r.next() // success
576 r.settled(t)
577 r.failFor.Store(1 << 30)
578 r.svc.OnOpened(context.Background(), domain.MergeRequest{Repo: r.repo})
579 <-r.passes
580 for range 5 {
581 r.next()
582 }
583 r.settled(t)
584 assert.Empty(t, r.afters)
585 assert.Empty(t, r.passes)
586}
587
588// stepClock is a clock the test moves.
589type stepClock struct{ at atomic.Int64 }
590
591func (c *stepClock) Now() time.Time { return time.Unix(0, c.at.Load()).UTC() }
592func (c *stepClock) advance(d time.Duration) { c.at.Add(int64(d)) }
593
594// An expired timer must not consume a replacement retry of its repository.
595func TestMergeabilityStaleTimerKeepsReplacementRetry(t *testing.T) {
596 repo := name(t, "proj")
597 meta := mocks.NewRepoMetaStore(t)
598 meta.EXPECT().List(mock.Anything).Return(nil, nil)
599 clock := &stepClock{}
600 clock.advance(time.Hour)
601 armed := make(chan time.Duration, 8)
602 svc := usecase.NewMergeability(context.Background(), usecase.MergeabilityConfig{
603 Meta: meta, Clock: clock, Workers: 1,
604 After: func(d time.Duration) <-chan time.Time { armed <- d; return make(chan time.Time) },
605 })
606 t.Cleanup(func() { svc.Close(context.Background()) })
607 svc.Start()
608
609 svc.RetryForTest(repo)
610 assert.Equal(t, time.Second, <-armed, "the scheduler armed the timer of retry A")
611 oldDue, ok := svc.RetryDueForTest(repo)
612 require.True(t, ok)
613
614 svc.RetryDoneForTest(repo)
615 clock.advance(10 * time.Second)
616 svc.RetryForTest(repo)
617 newDue, ok := svc.RetryDueForTest(repo)
618 require.True(t, ok)
619 require.True(t, newDue.After(oldDue))
620
621 // The old timer fires while the replacement is not yet due.
622 assert.Empty(t, svc.PopDueForTest(repo, oldDue))
623 assert.Equal(t, 1, svc.ScheduledRetries())
624 // The timer armed for the replacement itself releases it.
625 assert.Equal(t, []domain.RepoName{repo}, svc.PopDueForTest(repo, newDue))
626 assert.Zero(t, svc.ScheduledRetries())
627}
628
629func reconcileMocks(t *testing.T, repo domain.RepoName) (*mocks.MergeRequestStore, *mocks.RepoMetaStore, *mocks.MergeGit, domain.MergeRequest) {
630 t.Helper()
631 mr := domain.MergeRequest{Repo: repo, Number: 7, Source: "feat", Target: "main", State: domain.MRStateOpen}
632 store, meta, git := mocks.NewMergeRequestStore(t), mocks.NewRepoMetaStore(t), mocks.NewMergeGit(t)
633 meta.EXPECT().Get(mock.Anything, repo).Return(domain.Repo{Name: repo, CreatedAt: now}, nil)
634 store.EXPECT().ListOpen(mock.Anything, repo).Return([]domain.MergeRequest{mr}, nil)
635 return store, meta, git, mr
636}
637
638func TestMergeabilityDisappearedBranchDeletesTheRecord(t *testing.T) {
639 repo := name(t, "proj")
640 store, meta, git, _ := reconcileMocks(t, repo)
641 git.EXPECT().BranchHead(mock.Anything, repo, "feat").Return(domain.ObjectID{}, domain.ErrNotFound)
642 git.EXPECT().BranchHead(mock.Anything, repo, "main").Return(fakeOID(t, "b"), nil)
643 store.EXPECT().DeleteMergeability(mock.Anything, repo, int64(7)).Return(nil).Once()
644 svc := usecase.NewMergeability(context.Background(), usecase.MergeabilityConfig{
645 Store: store, Meta: meta, Git: git, Clock: fixedClock{},
646 })
647 t.Cleanup(func() { svc.Close(context.Background()) })
648 assert.True(t, svc.ReconcileForTest(repo))
649}
650
651func TestMergeabilityConflictMakesThePassFailAndRetries(t *testing.T) {
652 repo := name(t, "proj")
653 head, tip := fakeOID(t, "a"), fakeOID(t, "b")
654 store, meta, git, mr := reconcileMocks(t, repo)
655 meta.EXPECT().List(mock.Anything).Return(nil, nil)
656 git.EXPECT().BranchHead(mock.Anything, repo, "feat").Return(head, nil)
657 git.EXPECT().BranchHead(mock.Anything, repo, "main").Return(tip, nil)
658 git.EXPECT().IsAncestor(mock.Anything, repo, tip, head).Return(true, nil)
659 store.EXPECT().GetMergeability(mock.Anything, repo, int64(7)).Return(domain.Mergeability{}, domain.ErrNotFound)
660 store.EXPECT().PutMergeability(mock.Anything, mock.Anything).Return(domain.ErrConflict)
661
662 asked := make(chan time.Duration, 4)
663 svc := usecase.NewMergeability(context.Background(), usecase.MergeabilityConfig{
664 Store: store, Meta: meta, Git: git, Clock: fixedClock{}, Workers: 1,
665 After: func(d time.Duration) <-chan time.Time { asked <- d; return make(chan time.Time) },
666 })
667 t.Cleanup(func() { svc.Close(context.Background()) })
668 assert.False(t, svc.ReconcileForTest(repo))
669
670 svc.Start()
671 svc.OnOpened(context.Background(), mr)
672 assert.Equal(t, time.Second, <-asked, "the conflict scheduled a retry")
673}