1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package usecase_test
4
5import (
6 "bytes"
7 "context"
8 "errors"
9 "strings"
10 "sync"
11 "testing"
12 "time"
13
14 "github.com/bnema/zerowrap"
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/usecase"
21)
22
23var errBoom = errors.New("boom")
24
25func (f *mirrorFixture) pushes(t *testing.T, result error) *int {
26 t.Helper()
27 var mu sync.Mutex
28 n := new(int)
29 f.push.EXPECT().Push(mock.Anything, mock.Anything).RunAndReturn(func(context.Context, ports.MirrorPush) (ports.MirrorPushResult, error) {
30 mu.Lock()
31 defer mu.Unlock()
32 *n++
33 return ports.MirrorPushResult{}, result
34 }).Maybe()
35 return n
36}
37
38func TestMirrorsNoOpThrottleForce(t *testing.T) {
39 f := newMirrorFixture(t)
40 f.setSnapshot(f.repo, domain.MirrorSnapshot{URL: "git@example.com:acme/tools", Refs: []domain.MirrorRef{{Name: "refs/heads/main", ID: mirrorOID(t)}}})
41 v, err := f.m.Sync(context.Background(), domain.Root(), f.repo, false)
42 require.NoError(t, err)
43 require.True(t, v.NoOp)
44 _, err = f.m.Sync(context.Background(), domain.Root(), f.repo, false)
45 require.ErrorIs(t, err, domain.ErrResourceExhausted)
46 require.Contains(t, err.Error(), "wait 1m0s", "the interval comes from the domain constant")
47 require.Contains(t, err.Error(), "last sync 0s ago")
48 f.push.EXPECT().Push(mock.Anything, mock.MatchedBy(func(p ports.MirrorPush) bool {
49 return len(p.Refspecs) == 1 && p.Refspecs[0] == "refs/heads/main:refs/heads/main"
50 })).Return(ports.MirrorPushResult{}, nil).Once()
51 v, err = f.m.Sync(context.Background(), domain.Root(), f.repo, true)
52 require.NoError(t, err)
53 require.False(t, v.NoOp)
54}
55
56func TestMirrorsIncidentAndRecovery(t *testing.T) {
57 f := newMirrorFixture(t)
58 f.push.EXPECT().Push(mock.Anything, mock.Anything).Return(ports.MirrorPushResult{}, &domain.MirrorError{Class: domain.MirrorClassTransient}).Times(5)
59 for i := 1; i <= 5; i++ {
60 v, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true)
61 require.Error(t, err)
62 require.Equal(t, i, v.Status.ConsecutiveFailures)
63 if i < 5 {
64 require.Equal(t, domain.MirrorFailing, v.Status.State)
65 require.Equal(t, domain.MirrorBackoff(i), v.Status.NextRetryAt.Sub(v.Status.LastAttemptAt))
66 } else {
67 require.Equal(t, domain.MirrorPaused, v.Status.State)
68 }
69 }
70 events := f.eventList()
71 require.Len(t, events, 1, "mirror.failed once per incident")
72 failed, ok := events[0].(domain.MirrorFailed)
73 require.True(t, ok)
74 require.Equal(t, domain.MirrorFailing, failed.State)
75 // A paused mirror is not retried by the schedule.
76 f.advance(24 * time.Hour)
77 f.m.ScheduleNow(context.Background())
78 f.drain(t)
79 f.push.AssertNumberOfCalls(t, "Push", 5)
80 f.push.EXPECT().Push(mock.Anything, mock.Anything).Return(ports.MirrorPushResult{}, nil).Once()
81 v, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true)
82 require.NoError(t, err)
83 require.Equal(t, domain.MirrorOK, v.Status.State)
84 events = f.eventList()
85 require.Len(t, events, 2)
86 require.IsType(t, domain.MirrorRecovered{}, events[1])
87}
88
89func TestMirrorsSingleFlight(t *testing.T) {
90 f := newMirrorFixture(t)
91 entered, release := make(chan struct{}), make(chan struct{})
92 var mu sync.Mutex
93 calls := 0
94 f.push.EXPECT().Push(mock.Anything, mock.Anything).RunAndReturn(func(context.Context, ports.MirrorPush) (ports.MirrorPushResult, error) {
95 mu.Lock()
96 calls++
97 first := calls == 1
98 mu.Unlock()
99 if first {
100 close(entered)
101 <-release
102 }
103 return ports.MirrorPushResult{}, nil
104 })
105 done := make(chan error, 1)
106 go func() { _, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true); done <- err }()
107 <-entered
108 type outcome struct {
109 res ports.MirrorSyncResult
110 err error
111 }
112 results := make(chan outcome, 20)
113 var wg sync.WaitGroup
114 for range 20 {
115 wg.Add(1)
116 go func() {
117 defer wg.Done()
118 res, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true)
119 results <- outcome{res, err}
120 }()
121 }
122 wg.Wait()
123 close(results)
124 for o := range results {
125 require.NoError(t, o.err)
126 require.True(t, o.res.Queued)
127 require.Equal(t, f.repo, o.res.Repo)
128 }
129 close(release)
130 require.NoError(t, <-done)
131 f.drain(t)
132 mu.Lock()
133 defer mu.Unlock()
134 require.Equal(t, 2, calls, "twenty overlapping requests coalesce into one rerun")
135}
136
137func TestMirrorsCoalescedManualAppliesThrottle(t *testing.T) {
138 f := newMirrorFixture(t)
139 entered, release := make(chan struct{}), make(chan struct{})
140 f.push.EXPECT().Push(mock.Anything, mock.Anything).RunAndReturn(func(context.Context, ports.MirrorPush) (ports.MirrorPushResult, error) {
141 close(entered)
142 <-release
143 return ports.MirrorPushResult{}, nil
144 }).Once()
145 done := make(chan error, 1)
146 go func() { _, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true); done <- err }()
147 <-entered
148 v, err := f.m.Sync(context.Background(), domain.Root(), f.repo, false)
149 require.NoError(t, err)
150 require.True(t, v.Queued)
151 close(release)
152 require.NoError(t, <-done)
153 f.drain(t)
154 // The rerun is manual and not forced: the throttle (just attempted) stops it, so no second push
155 // (a second call would fail the Once expectation).
156 f.push.AssertNumberOfCalls(t, "Push", 1)
157}
158
159func TestMirrorsDivergedPausedAtOnce(t *testing.T) {
160 f := newMirrorFixture(t)
161 f.push.EXPECT().Push(mock.Anything, mock.Anything).Return(ports.MirrorPushResult{}, &domain.MirrorError{Class: domain.MirrorClassDiverged}).Once()
162 v, err := f.m.Sync(context.Background(), domain.Root(), f.repo, false)
163 require.Error(t, err)
164 require.Equal(t, domain.MirrorPaused, v.Status.State)
165 for range 3 {
166 f.advance(time.Hour)
167 f.m.ScheduleNow(context.Background())
168 }
169 f.drain(t)
170 f.push.AssertNumberOfCalls(t, "Push", 1)
171 require.Len(t, f.eventList(), 1)
172}
173
174func TestMirrorsCloseCancelsWorker(t *testing.T) {
175 f := newMirrorFixture(t)
176 entered, exited := make(chan struct{}), make(chan struct{})
177 f.push.EXPECT().Push(mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, _ ports.MirrorPush) (ports.MirrorPushResult, error) {
178 close(entered)
179 <-ctx.Done()
180 close(exited)
181 return ports.MirrorPushResult{}, context.Canceled
182 }).Once()
183 done := make(chan error, 1)
184 go func() { _, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true); done <- err }()
185 <-entered
186 ctx, cancel := context.WithTimeout(context.Background(), time.Second)
187 defer cancel()
188 f.m.Close(ctx)
189 require.NoError(t, ctx.Err())
190 select {
191 case <-exited:
192 default:
193 t.Fatal("Close returned before worker exited")
194 }
195 require.ErrorIs(t, <-done, context.Canceled)
196 f.m.Close(ctx) // idempotent; Start cannot resurrect workers.
197 f.m.Start()
198 _, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true)
199 require.ErrorIs(t, err, context.Canceled)
200}
201
202func TestMirrorsCronBackoff(t *testing.T) {
203 f := newMirrorFixture(t)
204 f.push.EXPECT().Push(mock.Anything, mock.Anything).Return(ports.MirrorPushResult{}, &domain.MirrorError{Class: domain.MirrorClassTransient}).Twice()
205 _, err := f.m.Sync(context.Background(), domain.Root(), f.repo, false)
206 require.Error(t, err)
207 f.advance(time.Minute)
208 f.m.ScheduleNow(context.Background())
209 f.drain(t)
210 f.push.AssertNumberOfCalls(t, "Push", 1)
211 f.advance(4 * time.Minute)
212 f.m.ScheduleNow(context.Background())
213 f.drain(t)
214 f.push.AssertNumberOfCalls(t, "Push", 2)
215}
216
217func TestMirrorsCronDueAndDirtySkip(t *testing.T) {
218 f := newMirrorFixture(t)
219 f.pushes(t, nil)
220 _, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true)
221 require.NoError(t, err)
222 require.Equal(t, 1, f.refReads())
223 f.advance(5 * time.Minute)
224 f.m.ScheduleNow(context.Background()) // due and clean: skipped without reading refs
225 f.drain(t)
226 require.Equal(t, 1, f.refReads())
227 // Merges and default branch changes both end in OnPush: an idempotent dirty mark.
228 f.m.OnPush(context.Background(), f.repo, []domain.RefUpdate{{Ref: "refs/heads/main"}})
229 f.m.OnPush(context.Background(), f.repo, []domain.RefUpdate{{Ref: "HEAD"}})
230 f.m.ScheduleNow(context.Background()) // same minute: nothing runs, although dirty
231 f.drain(t)
232 require.Equal(t, 1, f.refReads())
233 f.advance(5 * time.Minute)
234 f.m.ScheduleNow(context.Background())
235 f.drain(t)
236 require.Equal(t, 2, f.refReads(), "the dirty repository is examined on the next due tick")
237}
238
239func TestMirrorsScheduleKeepsFullQueueRepositoriesForNextTicks(t *testing.T) {
240 f := newMirrorFixtureWith(t, func(c *usecase.MirrorsConfig) { c.Workers, c.QueueSize = 1, 1 })
241 repos := []domain.RepoName{f.repo, name(t, "b"), name(t, "c"), name(t, "d")}
242 for _, r := range repos[1:] {
243 f.addRepo(r)
244 }
245 gate := make(chan struct{})
246 var mu sync.Mutex
247 pushed := map[string]int{}
248 f.push.EXPECT().Push(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, p ports.MirrorPush) (ports.MirrorPushResult, error) {
249 <-gate
250 mu.Lock()
251 pushed[p.URL]++
252 mu.Unlock()
253 return ports.MirrorPushResult{}, nil
254 })
255 f.advance(5 * time.Minute)
256 f.m.ScheduleNow(context.Background())
257 // The worker holds one repository and the queue one: the other two found it full.
258 close(gate)
259 for range 6 { // later ticks of the same minute drain the deferred set
260 f.drain(t)
261 f.m.ScheduleNow(context.Background())
262 }
263 f.drain(t)
264 mu.Lock()
265 defer mu.Unlock()
266 for _, r := range repos {
267 require.Equal(t, 1, pushed["git@example.com:acme/"+r.String()], r.String())
268 }
269}
270
271func TestMirrorsSyncRefusedWhenQueueFull(t *testing.T) {
272 f := newMirrorFixtureWith(t, func(c *usecase.MirrorsConfig) { c.Workers, c.QueueSize = 1, 1 })
273 second, third := name(t, "b"), name(t, "c")
274 f.addRepo(second)
275 f.addRepo(third)
276 gate := make(chan struct{})
277 entered := make(chan struct{}, 8)
278 f.push.EXPECT().Push(mock.Anything, mock.Anything).RunAndReturn(func(context.Context, ports.MirrorPush) (ports.MirrorPushResult, error) {
279 entered <- struct{}{}
280 <-gate
281 return ports.MirrorPushResult{}, nil
282 })
283 first := make(chan error, 1)
284 go func() { _, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true); first <- err }()
285 <-entered // the single worker is busy
286 // Enqueueing is synchronous: this request takes the only queue slot, then waits and gives up.
287 ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
288 defer cancel()
289 _, err := f.m.Sync(ctx, domain.Root(), second, true)
290 require.ErrorIs(t, err, context.DeadlineExceeded)
291 _, err = f.m.Sync(context.Background(), domain.Root(), third, true)
292 require.ErrorIs(t, err, domain.ErrResourceExhausted)
293 close(gate)
294 require.NoError(t, <-first)
295 f.drain(t)
296}
297
298func TestMirrorsSlowPushTimestamps(t *testing.T) {
299 f := newMirrorFixture(t)
300 start := f.at
301 f.push.EXPECT().Push(mock.Anything, mock.Anything).RunAndReturn(func(context.Context, ports.MirrorPush) (ports.MirrorPushResult, error) {
302 f.advance(10 * time.Minute)
303 return ports.MirrorPushResult{}, &domain.MirrorError{Class: domain.MirrorClassTransient}
304 }).Once()
305 v, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true)
306 require.Error(t, err)
307 require.True(t, v.Status.LastAttemptAt.Equal(start))
308 require.True(t, v.Status.NextRetryAt.Equal(start.Add(10*time.Minute+domain.MirrorBackoff(1))))
309}
310
311func TestMirrorsPushDoesNotHoldConfigLock(t *testing.T) {
312 f := newMirrorFixture(t)
313 entered, release := make(chan struct{}), make(chan struct{})
314 f.push.EXPECT().Push(mock.Anything, mock.Anything).RunAndReturn(func(context.Context, ports.MirrorPush) (ports.MirrorPushResult, error) {
315 close(entered)
316 <-release
317 return ports.MirrorPushResult{}, nil
318 }).Once()
319 done := make(chan error, 1)
320 go func() { _, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true); done <- err }()
321 <-entered
322 changed := make(chan error, 1)
323 go func() {
324 _, err := f.m.SetGlobalConfig(context.Background(), domain.Root(), ports.MirrorGlobalPatch{Cron: ptr("*/10 * * * *")})
325 changed <- err
326 }()
327 select {
328 case err := <-changed:
329 require.NoError(t, err)
330 case <-time.After(5 * time.Second): // only on failure: the push holds the configuration lock
331 t.Fatal("configuration change blocked by a running push")
332 }
333 close(release)
334 require.NoError(t, <-done)
335}
336
337func TestMirrorsResultDiscardedWhenTargetChangedDuringPush(t *testing.T) {
338 f := newMirrorFixture(t)
339 f.setStatus(domain.MirrorStatus{Repo: f.repo, State: domain.MirrorFailing, Class: domain.MirrorClassTransient, ConsecutiveFailures: 1})
340 entered, release := make(chan struct{}), make(chan struct{})
341 f.push.EXPECT().Push(mock.Anything, mock.Anything).RunAndReturn(func(context.Context, ports.MirrorPush) (ports.MirrorPushResult, error) {
342 close(entered)
343 <-release
344 return ports.MirrorPushResult{}, nil
345 }).Once()
346 done := make(chan error, 1)
347 go func() { _, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true); done <- err }()
348 <-entered
349 _, err := f.m.SetGlobalConfig(context.Background(), domain.Root(), ports.MirrorGlobalPatch{URLTemplate: ptr("git@other.example:acme/{repo}")})
350 require.NoError(t, err)
351 close(release)
352 require.NoError(t, <-done)
353 _, ok := f.snapshotOf(f.repo)
354 require.False(t, ok, "no snapshot for a target that no longer applies")
355 require.Empty(t, f.eventList(), "no mirror.recovered for the old target")
356 s := f.statusOf(f.repo)
357 require.Equal(t, domain.MirrorOK, s.State)
358 require.True(t, s.LastSuccessAt.IsZero(), "the push result was not recorded")
359}
360
361func TestMirrorsSyncAllContinuesPastFailures(t *testing.T) {
362 f := newMirrorFixture(t)
363 bad, off := name(t, "bad"), name(t, "off")
364 f.addRepo(bad)
365 f.addRepo(off)
366 f.setRepoConfig(domain.MirrorRepoConfig{Repo: off, Excluded: true})
367 f.push.EXPECT().Push(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, p ports.MirrorPush) (ports.MirrorPushResult, error) {
368 if strings.HasSuffix(p.URL, "/bad") {
369 return ports.MirrorPushResult{}, &domain.MirrorError{Class: domain.MirrorClassDiverged, Detail: "rejected"}
370 }
371 return ports.MirrorPushResult{}, nil
372 })
373 got, err := f.m.SyncAll(context.Background(), domain.Root(), true)
374 require.NoError(t, err)
375 require.Len(t, got, 2, "the excluded repository is skipped")
376 byRepo := map[domain.RepoName]ports.MirrorSyncResult{}
377 for _, r := range got {
378 byRepo[r.Repo] = r
379 }
380 require.NoError(t, byRepo[f.repo].Err)
381 require.Equal(t, domain.MirrorOK, byRepo[f.repo].Status.State)
382 var me *domain.MirrorError
383 require.ErrorAs(t, byRepo[bad].Err, &me)
384 require.Equal(t, domain.MirrorPaused, byRepo[bad].Status.State)
385 _, listed := byRepo[off]
386 require.False(t, listed)
387 _, err = f.m.SyncAll(context.Background(), tokenPrincipal(domain.ScopeRepoAdmin), true)
388 require.ErrorIs(t, err, domain.ErrForbidden)
389}
390
391func TestMirrorsLocalFailuresAreInternalAndNeverPush(t *testing.T) {
392 for _, tc := range []struct {
393 name string
394 set func(*mirrorFixture)
395 }{
396 {"ListRefs", func(f *mirrorFixture) { f.mu.Lock(); f.refsErr = errBoom; f.mu.Unlock() }},
397 {"DefaultBranch", func(f *mirrorFixture) { f.mu.Lock(); f.branchErr = errBoom; f.mu.Unlock() }},
398 } {
399 t.Run(tc.name, func(t *testing.T) {
400 f := newMirrorFixture(t)
401 tc.set(f)
402 v, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true) // a Push call would fail the strict mock
403 var me *domain.MirrorError
404 require.ErrorAs(t, err, &me)
405 require.Equal(t, domain.MirrorClassInternal, me.Class)
406 require.NotContains(t, me.Detail, "boom")
407 require.Equal(t, domain.MirrorFailing, v.Status.State)
408 })
409 }
410}
411
412func TestMirrorsMissingSecretsKeyFailsBeforeReadingRefs(t *testing.T) {
413 f := newMirrorFixtureWith(t, func(c *usecase.MirrorsConfig) { c.Sealer = nil })
414 v, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true)
415 var me *domain.MirrorError
416 require.ErrorAs(t, err, &me)
417 require.Equal(t, domain.MirrorClassInternal, me.Class)
418 require.Contains(t, me.Detail, "ERMITE_SECRETS_KEY")
419 require.Equal(t, domain.MirrorFailing, v.Status.State)
420 require.Zero(t, f.refReads())
421}
422
423func TestMirrorsErrorsAndLogsDoNotExposeSecrets(t *testing.T) {
424 f := newMirrorFixture(t)
425 var logs bytes.Buffer
426 ctx := zerowrap.WithCtx(context.Background(), zerowrap.New(zerowrap.Config{Level: "debug", Format: "json", Output: &logs}))
427 credentialURL := "ssh://alice:secret-password@example.com/acme/tools"
428 _, err := f.m.SetRepoMirror(ctx, domain.Root(), domain.MirrorRepoConfig{Repo: f.repo, URL: credentialURL})
429 require.Error(t, err)
430 require.NotContains(t, err.Error(), "secret-password")
431 require.NotContains(t, err.Error(), "alice")
432 f.push.EXPECT().Push(mock.Anything, mock.Anything).Return(ports.MirrorPushResult{}, errors.New("local key PRIVATE-KEY-MARKER ssh://alice:secret-password@example.com/acme/tools")).Once()
433 _, err = f.m.Sync(ctx, domain.Root(), f.repo, true)
434 require.Error(t, err)
435 require.NotContains(t, err.Error(), "PRIVATE-KEY-MARKER")
436 require.NotContains(t, err.Error(), "secret-password")
437 require.NotContains(t, logs.String(), "PRIVATE-KEY-MARKER")
438 require.NotContains(t, logs.String(), "secret-password")
439}
440
441func TestMirrorsKeyLazyGenerationAndRotation(t *testing.T) {
442 f := newKeyFixture(t)
443 f.keys.EXPECT().GetMirrorKey(mock.Anything).Return([]byte(nil), time.Time{}, domain.ErrNotFound).Once()
444 f.keys.EXPECT().PutMirrorKey(mock.Anything, mock.Anything, now, true).Return(nil).Once()
445 f.keys.EXPECT().PutMirrorKey(mock.Anything, mock.Anything, now, false).Return(nil).Once()
446 first, err := f.m.ShowKey(context.Background(), domain.Root())
447 require.NoError(t, err)
448 cached, err := f.m.ShowKey(context.Background(), domain.Root())
449 require.NoError(t, err)
450 require.Equal(t, first, cached)
451 rotated, err := f.m.RotateKey(context.Background(), domain.Root())
452 require.NoError(t, err)
453 require.Equal(t, first.Fingerprint, rotated.Previous.Fingerprint)
454 require.NotEqual(t, first.Fingerprint, rotated.Key.Fingerprint)
455 require.Contains(t, rotated.Key.AuthorizedKey, "ermite-mirror@forge.example.com")
456 require.Len(t, f.seeds(), 2)
457 require.NotEqual(t, f.seeds()[0], f.seeds()[1])
458}
459
460func TestMirrorsRotateWithoutExistingKeyGeneratesOnce(t *testing.T) {
461 f := newKeyFixture(t)
462 f.keys.EXPECT().GetMirrorKey(mock.Anything).Return([]byte(nil), time.Time{}, domain.ErrNotFound).Once()
463 f.keys.EXPECT().PutMirrorKey(mock.Anything, mock.Anything, now, false).Return(nil).Once() // never ifAbsent=true
464 rotated, err := f.m.RotateKey(context.Background(), domain.Root())
465 require.NoError(t, err)
466 require.Empty(t, rotated.Previous.Fingerprint)
467 require.NotEmpty(t, rotated.Key.Fingerprint)
468 require.Len(t, f.seeds(), 1, "no throwaway key")
469}
470
471func TestMirrorsRotateReplacesUnreadableKey(t *testing.T) {
472 f := newKeyFixture(t)
473 f.keys.EXPECT().GetMirrorKey(mock.Anything).Return([]byte("garbage"), now, nil).Once()
474 f.sealerOpenFails = true
475 f.keys.EXPECT().PutMirrorKey(mock.Anything, mock.Anything, now, false).Return(nil).Once()
476 rotated, err := f.m.RotateKey(context.Background(), domain.Root())
477 require.NoError(t, err)
478 require.Empty(t, rotated.Previous.Fingerprint)
479}
480
481func TestMirrorsAuthorizationAndMissingSecretsKey(t *testing.T) {
482 f := newMirrorFixtureWith(t, func(c *usecase.MirrorsConfig) { c.Sealer = nil })
483 _, err := f.m.ShowKey(context.Background(), domain.Anonymous())
484 require.ErrorIs(t, err, domain.ErrUnauthenticated)
485 _, err = f.m.ShowKey(context.Background(), domain.Root())
486 require.ErrorIs(t, err, domain.ErrPreconditionFailed)
487 require.Contains(t, err.Error(), "ERMITE_SECRETS_KEY")
488 _, err = f.m.RotateKey(context.Background(), domain.Root())
489 require.ErrorIs(t, err, domain.ErrPreconditionFailed)
490 _, err = f.m.RotateKey(context.Background(), tokenPrincipal(domain.ScopeRepoRead))
491 require.ErrorIs(t, err, domain.ErrForbidden)
492}
493
494// gatedPush makes every push announce itself on entered and block until release is closed.
495func (f *mirrorFixture) gatedPush(release <-chan struct{}) <-chan struct{} {
496 entered := make(chan struct{}, 16)
497 f.push.EXPECT().Push(mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, _ ports.MirrorPush) (ports.MirrorPushResult, error) {
498 entered <- struct{}{}
499 select {
500 case <-release:
501 case <-ctx.Done():
502 }
503 return ports.MirrorPushResult{}, nil
504 }).Maybe()
505 return entered
506}
507
508func TestMirrorsSyncAllRunsRepositoriesConcurrently(t *testing.T) {
509 f := newMirrorFixtureWith(t, func(c *usecase.MirrorsConfig) { c.Workers = 3; c.SyncWait = time.Minute })
510 f.addRepo(name(t, "b"))
511 f.addRepo(name(t, "c"))
512 release := make(chan struct{})
513 entered := f.gatedPush(release)
514 type outcome struct {
515 got []ports.MirrorSyncResult
516 err error
517 }
518 done := make(chan outcome, 1)
519 go func() {
520 got, err := f.m.SyncAll(context.Background(), domain.Root(), true)
521 done <- outcome{got, err}
522 }()
523 // All three pushes are in flight together: a sequential SyncAll would never get past the first.
524 for i := 0; i < 3; i++ {
525 select {
526 case <-entered:
527 case <-time.After(10 * time.Second):
528 t.Fatalf("only %d pushes started: repositories are not synced concurrently", i)
529 }
530 }
531 close(release)
532 o := <-done
533 require.NoError(t, o.err)
534 require.Len(t, o.got, 3)
535 for i, want := range []string{"tools", "b", "c"} {
536 r := o.got[i]
537 require.Equal(t, want, r.Repo.String(), "results keep the listing order")
538 require.NoError(t, r.Err)
539 require.False(t, r.Queued)
540 require.Equal(t, domain.MirrorOK, r.Status.State)
541 }
542}
543
544func TestMirrorsSyncReturnsQueuedWhenPushOutlastsTheWait(t *testing.T) {
545 f := newMirrorFixtureWith(t, func(c *usecase.MirrorsConfig) { c.SyncWait = 20 * time.Millisecond })
546 other := name(t, "other")
547 f.addRepo(other)
548 release := make(chan struct{})
549 entered := f.gatedPush(release)
550
551 res, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true)
552 require.NoError(t, err)
553 require.True(t, res.Queued, "the push is still running at the bound")
554 require.Equal(t, f.repo, res.Repo)
555 <-entered
556
557 all, err := f.m.SyncAll(context.Background(), domain.Root(), true)
558 require.NoError(t, err)
559 require.Len(t, all, 2)
560 for _, r := range all {
561 require.NoError(t, r.Err)
562 require.True(t, r.Queued, "%s is coalesced or still running", r.Repo)
563 require.Equal(t, r.Repo, r.Status.Repo)
564 }
565
566 close(release)
567 f.drain(t)
568 require.Equal(t, domain.MirrorOK, f.statusOf(f.repo).State, "the sync completed in the background")
569}
570
571func TestMirrorsSyncAllHonoursCallerCancellation(t *testing.T) {
572 f := newMirrorFixtureWith(t, func(c *usecase.MirrorsConfig) { c.SyncWait = time.Minute })
573 release := make(chan struct{})
574 entered := f.gatedPush(release)
575 ctx, cancel := context.WithCancel(context.Background())
576 errc := make(chan error, 1)
577 go func() { _, err := f.m.SyncAll(ctx, domain.Root(), true); errc <- err }()
578 <-entered
579 cancel()
580 require.ErrorIs(t, <-errc, context.Canceled)
581 close(release)
582}