1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package app_test
4
5import (
6 "context"
7 "errors"
8 "fmt"
9 "sync"
10 "testing"
11 "time"
12
13 "github.com/stretchr/testify/assert"
14 "github.com/stretchr/testify/mock"
15 "github.com/stretchr/testify/require"
16
17 "bnema.dev/ermite/internal/app"
18 "bnema.dev/ermite/internal/domain"
19 "bnema.dev/ermite/internal/ports/mocks"
20)
21
22// fakeAuth is the Revalidator of the rig: tokens are valid until revoked.
23type fakeAuth struct {
24 mu sync.Mutex
25 revoked map[string]bool
26 calls int
27}
28
29func (a *fakeAuth) Revalidate(_ context.Context, p domain.Principal) error {
30 a.mu.Lock()
31 defer a.mu.Unlock()
32 a.calls++
33 if a.revoked[p.TokenID()] {
34 return domain.ErrUnauthenticated
35 }
36 return nil
37}
38
39func (a *fakeAuth) revoke(id string) {
40 a.mu.Lock()
41 defer a.mu.Unlock()
42 if a.revoked == nil {
43 a.revoked = map[string]bool{}
44 }
45 a.revoked[id] = true
46}
47
48func (a *fakeAuth) count() int {
49 a.mu.Lock()
50 defer a.mu.Unlock()
51 return a.calls
52}
53
54// eventRig is an Events use case fed by a channel the test writes to.
55type eventRig struct {
56 t *testing.T
57 meta *mocks.RepoMetaStore
58 git *mocks.GitRepos
59 feed chan domain.Event
60 auth *fakeAuth
61 events *app.Events
62}
63
64func newEventRig(t *testing.T, size int) *eventRig {
65 t.Helper()
66 return newEventRigConfig(t, app.EventsConfig{Size: size})
67}
68
69func newEventRigConfig(t *testing.T, cfg app.EventsConfig) *eventRig {
70 t.Helper()
71 r := &eventRig{
72 t: t,
73 meta: mocks.NewRepoMetaStore(t),
74 git: mocks.NewGitRepos(t),
75 feed: make(chan domain.Event, 4096),
76 auth: &fakeAuth{},
77 }
78 cfg.Auth = r.auth
79 bus := mocks.NewEventBus(t)
80 bus.EXPECT().Subscribe(mock.Anything).RunAndReturn(func(ctx context.Context) <-chan domain.Event {
81 go func() { <-ctx.Done(); close(r.feed) }()
82 return r.feed
83 })
84 clock := mocks.NewClock(t)
85 clock.EXPECT().Now().Return(now).Maybe()
86 r.git.EXPECT().HeadBranch(mock.Anything, mock.Anything).Return("main", nil).Maybe()
87 repos := app.NewRepos(r.meta, r.git, bus, clock)
88 r.events = app.NewEvents(context.Background(), repos, bus, clock, cfg)
89 t.Cleanup(func() { r.events.Close(context.Background()) })
90 return r
91}
92
93// repo declares a repository the rig knows.
94func (r *eventRig) repo(s string, vis domain.Visibility) domain.RepoName {
95 n := name(r.t, s)
96 r.meta.EXPECT().Get(mock.Anything, n).Return(domain.Repo{Name: n, Visibility: vis, CreatedAt: now}, nil).Maybe()
97 return n
98}
99
100// pushEvent builds the PushEvent the Pushes use case would publish now: it
101// carries the repository as the store answers at this moment.
102func (r *eventRig) pushEvent(n domain.RepoName, oid, ref string) domain.PushEvent {
103 repo, err := r.meta.Get(context.Background(), n)
104 require.NoError(r.t, err)
105 return domain.PushEvent{
106 Repo: n, Updates: []domain.RefUpdate{{Old: domain.ZeroOID, New: oid, Ref: ref}}, At: now,
107 Visibility: repo.Visibility, CreatedAt: repo.CreatedAt,
108 }
109}
110
111func (r *eventRig) push(n domain.RepoName, i int) {
112 r.feed <- r.pushEvent(n, fmt.Sprintf("%040x", i+1), "refs/heads/main")
113}
114
115// watcher is a running Watch.
116type watcher struct {
117 got chan app.WatchEvent
118 done chan error
119}
120
121func (r *eventRig) watch(ctx context.Context, p domain.Principal, params app.WatchParams) *watcher {
122 w := &watcher{got: make(chan app.WatchEvent, 4096), done: make(chan error, 1)}
123 go func() {
124 w.done <- r.events.Watch(ctx, p, params, func(e app.WatchEvent) error {
125 w.got <- e
126 return nil
127 })
128 }()
129 return w
130}
131
132func (w *watcher) next(t *testing.T) app.WatchEvent {
133 t.Helper()
134 select {
135 case e := <-w.got:
136 return e
137 case err := <-w.done:
138 t.Fatalf("watch ended: %v", err)
139 case <-time.After(5 * time.Second):
140 t.Fatal("no event")
141 }
142 return app.WatchEvent{}
143}
144
145// sentinelOID marks the push events that only prove the ones before them were
146// handled.
147var sentinelOID = fmt.Sprintf("%040x", 0xfeed)
148
149// quiet proves that the watcher got nothing since its last read: it pushes an
150// event on n, which the watcher may see, and requires it to be the next one.
151// Events reach a watcher in order, so anything sent before would show up first.
152func (w *watcher) quiet(t *testing.T, r *eventRig, n domain.RepoName) {
153 t.Helper()
154 r.feed <- r.pushEvent(n, sentinelOID, "refs/heads/sentinel")
155 e := w.next(t)
156 push, ok := e.Event.(domain.PushEvent)
157 if !ok || len(push.Updates) != 1 || push.Updates[0].New != sentinelOID {
158 t.Fatalf("expected nothing before the sentinel, got %+v", e)
159 }
160}
161
162func TestWatchLiveAndFiltered(t *testing.T) {
163 r := newEventRig(t, 0)
164 pub := r.repo("pub", domain.VisibilityPublic)
165 priv := r.repo("priv", domain.VisibilityPrivate)
166 ctx, cancel := context.WithCancel(context.Background())
167 defer cancel()
168
169 anon := r.watch(ctx, domain.Anonymous(), app.WatchParams{})
170 root := r.watch(ctx, domain.Root(), app.WatchParams{Topics: []string{"home"}})
171 scoped := r.watch(ctx, tokenPrincipal(domain.ScopeRepoRead, "other"), app.WatchParams{})
172 onlyPriv := r.watch(ctx, domain.Root(), app.WatchParams{Topics: []string{"repo:priv"}})
173 require.Eventually(t, func() bool { return r.events.Watchers() == 4 }, 5*time.Second, time.Millisecond)
174
175 r.push(priv, 1)
176 r.push(pub, 2)
177
178 // Anonymous and a token limited to another repository never see priv.
179 for name, w := range map[string]*watcher{"anonymous": anon, "scoped": scoped} {
180 e := w.next(t)
181 assert.Equal(t, pub, e.Event.(domain.PushEvent).Repo, name)
182 w.quiet(t, r, pub)
183 }
184 first, second := root.next(t), root.next(t)
185 assert.Equal(t, priv, first.Event.(domain.PushEvent).Repo)
186 assert.Equal(t, pub, second.Event.(domain.PushEvent).Repo)
187 assert.NotEmpty(t, first.ID)
188 assert.NotEqual(t, first.ID, second.ID)
189
190 assert.Equal(t, priv, onlyPriv.next(t).Event.(domain.PushEvent).Repo)
191 onlyPriv.quiet(t, r, priv)
192}
193
194func TestWatchRechecksAccessPerEvent(t *testing.T) {
195 r := newEventRig(t, 0)
196 n := name(t, "flip")
197 vis := domain.VisibilityPublic
198 var mu sync.Mutex
199 r.meta.EXPECT().Get(mock.Anything, n).RunAndReturn(func(context.Context, domain.RepoName) (domain.Repo, error) {
200 mu.Lock()
201 defer mu.Unlock()
202 return domain.Repo{Name: n, Visibility: vis, CreatedAt: now}, nil
203 })
204 ctx, cancel := context.WithCancel(context.Background())
205 defer cancel()
206 w := r.watch(ctx, domain.Anonymous(), app.WatchParams{})
207 require.Eventually(t, func() bool { return r.events.Watchers() == 1 }, 5*time.Second, time.Millisecond)
208
209 r.push(n, 1)
210 w.next(t)
211
212 mu.Lock()
213 vis = domain.VisibilityPrivate
214 mu.Unlock()
215 r.push(n, 2)
216 w.quiet(t, r, r.repo("open", domain.VisibilityPublic)) // push 2 is not delivered to anonymous
217}
218
219func TestWatchResumeAndGap(t *testing.T) {
220 const size = 8
221 r := newEventRig(t, size)
222 n := r.repo("pub", domain.VisibilityPublic)
223 ctx, cancel := context.WithCancel(context.Background())
224 defer cancel()
225
226 // Record the ids of 20 events through a first watcher.
227 first := r.watch(ctx, domain.Root(), app.WatchParams{})
228 require.Eventually(t, func() bool { return r.events.Watchers() == 1 }, 5*time.Second, time.Millisecond)
229 // One at a time: this watcher must not fall behind the small ring.
230 var ids []string
231 for i := range 20 {
232 r.push(n, i)
233 ids = append(ids, first.next(t).ID)
234 }
235 require.Len(t, ids, 20)
236
237 // The subtests that read the state of the ring come first: the ones that
238 // need a sentinel add events to it.
239 t.Run("an id that fell out of the buffer gives a gap, then the oldest buffered event", func(t *testing.T) {
240 // 20 events were taken, the buffer holds the last 8: ids[12] is the
241 // oldest still present; ids[0] is long gone.
242 w := r.watch(ctx, domain.Root(), app.WatchParams{ResumeAfter: ids[0]})
243 gap := w.next(t)
244 assert.True(t, gap.Gap)
245 assert.Empty(t, gap.ID)
246 oldest := w.next(t)
247 assert.False(t, oldest.Gap)
248 assert.Equal(t, ids[12], oldest.ID)
249 })
250
251 t.Run("an unknown id gives a gap", func(t *testing.T) {
252 for _, id := range []string{"nonsense", "zz-1", "-", "0-0"} {
253 w := r.watch(ctx, domain.Root(), app.WatchParams{ResumeAfter: id})
254 assert.True(t, w.next(t).Gap, id)
255 }
256 })
257
258 t.Run("an id of a later run is a gap, not a newer event", func(t *testing.T) {
259 w := r.watch(ctx, domain.Root(), app.WatchParams{ResumeAfter: "otherboot-999"})
260 assert.True(t, w.next(t).Gap)
261 })
262
263 t.Run("resume replays the buffered events after the id, then goes live", func(t *testing.T) {
264 w := r.watch(ctx, domain.Root(), app.WatchParams{ResumeAfter: ids[16]})
265 for _, want := range ids[17:] {
266 e := w.next(t)
267 assert.False(t, e.Gap)
268 assert.Equal(t, want, e.ID)
269 }
270 // Nothing else was replayed: the next event is the one pushed now.
271 r.push(n, 98)
272 live := w.next(t)
273 assert.False(t, live.Gap)
274 assert.NotContains(t, ids, live.ID)
275 ids = append(ids, live.ID)
276 })
277
278 t.Run("resume at the last event replays nothing, then goes live", func(t *testing.T) {
279 w := r.watch(ctx, domain.Root(), app.WatchParams{ResumeAfter: ids[len(ids)-1]})
280 r.push(n, 99)
281 e := w.next(t)
282 assert.False(t, e.Gap)
283 assert.NotContains(t, ids, e.ID, "the first event is the new one, nothing was replayed")
284 })
285}
286
287func TestWatchTopicErrors(t *testing.T) {
288 r := newEventRig(t, 0)
289 r.repo("pub", domain.VisibilityPublic)
290 r.repo("priv", domain.VisibilityPrivate)
291 r.meta.EXPECT().Get(mock.Anything, name(t, "gone")).Return(domain.Repo{}, domain.ErrNotFound).Maybe()
292 noop := func(app.WatchEvent) error { return nil }
293 ctx := context.Background()
294
295 for topic, want := range map[string]error{
296 "nope": domain.ErrInvalid,
297 "repo:Bad": domain.ErrInvalidName,
298 "repo:": domain.ErrInvalidName,
299 "repo:gone": domain.ErrNotFound,
300 "repo:priv": domain.ErrNotFound, // hidden from anonymous
301 "repo:pub:foo": domain.ErrInvalidName,
302 } {
303 err := r.events.Watch(ctx, domain.Anonymous(), app.WatchParams{Topics: []string{topic}}, noop)
304 assert.ErrorIs(t, err, want, topic)
305 }
306}
307
308func TestWatchEndsCleanly(t *testing.T) {
309 t.Run("on context cancellation", func(t *testing.T) {
310 r := newEventRig(t, 0)
311 ctx, cancel := context.WithCancel(context.Background())
312 w := r.watch(ctx, domain.Root(), app.WatchParams{})
313 require.Eventually(t, func() bool { return r.events.Watchers() == 1 }, 5*time.Second, time.Millisecond)
314 cancel()
315 select {
316 case err := <-w.done:
317 assert.NoError(t, err)
318 case <-time.After(5 * time.Second):
319 t.Fatal("watch did not end")
320 }
321 assert.Zero(t, r.events.Watchers())
322 })
323
324 t.Run("on Close", func(t *testing.T) {
325 r := newEventRig(t, 0)
326 w := r.watch(context.Background(), domain.Root(), app.WatchParams{})
327 require.Eventually(t, func() bool { return r.events.Watchers() == 1 }, 5*time.Second, time.Millisecond)
328 r.events.Close(context.Background())
329 r.events.Close(context.Background()) // twice is fine
330 select {
331 case err := <-w.done:
332 assert.NoError(t, err)
333 case <-time.After(5 * time.Second):
334 t.Fatal("watch did not end")
335 }
336 })
337
338 t.Run("a failing send ends the watch with its error", func(t *testing.T) {
339 r := newEventRig(t, 0)
340 n := r.repo("pub", domain.VisibilityPublic)
341 boom := errors.New("client went away")
342 done := make(chan error, 1)
343 go func() {
344 done <- r.events.Watch(context.Background(), domain.Root(), app.WatchParams{}, func(app.WatchEvent) error { return boom })
345 }()
346 require.Eventually(t, func() bool { return r.events.Watchers() == 1 }, 5*time.Second, time.Millisecond)
347 r.push(n, 1)
348 assert.ErrorIs(t, <-done, boom)
349 })
350}
351
352func TestWatchSlowWatcherGetsAGap(t *testing.T) {
353 const size = 4
354 r := newEventRig(t, size)
355 n := r.repo("pub", domain.VisibilityPublic)
356
357 release := make(chan struct{})
358 var mu sync.Mutex
359 var got []app.WatchEvent
360 done := make(chan error, 1)
361 blocked := make(chan struct{})
362 first := true
363 go func() {
364 done <- r.events.Watch(context.Background(), domain.Root(), app.WatchParams{}, func(e app.WatchEvent) error {
365 if first { // block on the first event while the ring turns over
366 first = false
367 close(blocked)
368 <-release
369 }
370 mu.Lock()
371 got = append(got, e)
372 mu.Unlock()
373 return nil
374 })
375 }()
376 // A second, fast watcher tells when the use case has taken every event:
377 // it receives them in order.
378 fast := r.watch(context.Background(), domain.Root(), app.WatchParams{})
379 require.Eventually(t, func() bool { return r.events.Watchers() == 2 }, 5*time.Second, time.Millisecond)
380 r.push(n, 0)
381 <-blocked
382 for i := 1; i < 30; i++ {
383 r.push(n, i)
384 }
385 // The ring holds four events: the fast watcher may get a gap, but it ends
386 // on the last push, which proves the use case took all of them.
387 last := fmt.Sprintf("%040x", 29+1)
388 for {
389 e := fast.next(t)
390 if p, ok := e.Event.(domain.PushEvent); ok && p.Updates[0].New == last {
391 break
392 }
393 }
394 close(release)
395
396 require.Eventually(t, func() bool {
397 mu.Lock()
398 defer mu.Unlock()
399 return len(got) >= 2
400 }, 5*time.Second, time.Millisecond)
401 mu.Lock()
402 defer mu.Unlock()
403 hasGap := false
404 for _, e := range got[1:] {
405 hasGap = hasGap || e.Gap
406 }
407 assert.True(t, hasGap, "events were lost while the watcher was blocked: %+v", got)
408}
409
410func TestEventsRepoPayloads(t *testing.T) {
411 r := newEventRig(t, 0)
412 pub := r.repo("pub", domain.VisibilityPublic)
413 priv := r.repo("priv", domain.VisibilityPrivate)
414 ctx, cancel := context.WithCancel(context.Background())
415 defer cancel()
416 anon := r.watch(ctx, domain.Anonymous(), app.WatchParams{})
417 root := r.watch(ctx, domain.Root(), app.WatchParams{})
418 require.Eventually(t, func() bool { return r.events.Watchers() == 2 }, 5*time.Second, time.Millisecond)
419
420 r.feed <- domain.RepoCreated{Repo: pub, Visibility: domain.VisibilityPublic, At: now}
421 r.feed <- domain.RepoUpdated{Repo: pub, At: now, Visibility: domain.VisibilityPublic, Previous: domain.VisibilityPublic, CreatedAt: now}
422 // A deleted repository cannot be read any more: its visibility decides.
423 r.feed <- domain.RepoDeleted{Repo: priv, Visibility: domain.VisibilityPrivate, At: now}
424 r.feed <- domain.RepoDeleted{Repo: pub, Visibility: domain.VisibilityPublic, At: now}
425
426 for _, w := range []*watcher{anon, root} {
427 created := w.next(t)
428 require.NotNil(t, created.Repo)
429 assert.Equal(t, "main", created.Repo.DefaultBranch)
430 assert.Equal(t, pub, created.Repo.Name)
431 require.NotNil(t, w.next(t).Repo)
432 }
433 assert.Equal(t, priv, root.next(t).Event.(domain.RepoDeleted).Repo)
434 assert.Equal(t, pub, root.next(t).Event.(domain.RepoDeleted).Repo)
435 assert.Equal(t, pub, anon.next(t).Event.(domain.RepoDeleted).Repo, "the private deletion was skipped")
436 anon.quiet(t, r, pub)
437}
438
439// tokenWith builds the principal of a stored token with an id and an expiry.
440func tokenWith(id string, expires time.Time, repos ...string) domain.Principal {
441 tok := domain.Token{ID: id, Scopes: []domain.Scope{domain.ScopeRepoRead}, ExpiresAt: expires}
442 for _, r := range repos {
443 n, err := domain.ParseRepoName(r)
444 if err != nil {
445 panic(err)
446 }
447 tok.Repos = append(tok.Repos, n)
448 }
449 return domain.TokenPrincipal(tok)
450}
451
452func TestWatchEndsWhenTheTokenIsRevoked(t *testing.T) {
453 // Asking at every turn keeps the test independent of the clock.
454 r := newEventRigConfig(t, app.EventsConfig{RevalidateEvery: -1})
455 pub := r.repo("pub", domain.VisibilityPublic)
456 priv := r.repo("priv", domain.VisibilityPrivate)
457 p := tokenWith("tok", time.Time{})
458 ctx, cancel := context.WithCancel(context.Background())
459 defer cancel()
460
461 w := r.watch(ctx, p, app.WatchParams{})
462 require.Eventually(t, func() bool { return r.events.Watchers() == 1 }, 5*time.Second, time.Millisecond)
463 r.push(pub, 1)
464 w.next(t)
465
466 r.auth.revoke("tok")
467 r.push(priv, 2) // would be delivered to a token that may read priv
468 select {
469 case err := <-w.done:
470 assert.ErrorIs(t, err, domain.ErrUnauthenticated)
471 case <-time.After(5 * time.Second):
472 t.Fatal("the watch of a revoked token did not end")
473 }
474 select {
475 case e := <-w.got:
476 t.Fatalf("an event was sent after the revocation: %+v", e)
477 default:
478 }
479}
480
481func TestWatchEndsWhenTheTokenExpires(t *testing.T) {
482 r := newEventRig(t, 0)
483 pub := r.repo("pub", domain.VisibilityPublic)
484 // The rig's clock stands at now: this token expired at now.
485 p := tokenWith("tok", now)
486 w := r.watch(context.Background(), p, app.WatchParams{})
487 r.push(pub, 1) // whether or not the watch had started, it ends unauthenticated
488 select {
489 case err := <-w.done:
490 assert.ErrorIs(t, err, domain.ErrUnauthenticated)
491 case <-time.After(5 * time.Second):
492 t.Fatal("the watch of an expired token did not end")
493 }
494 assert.Empty(t, w.got)
495}
496
497func TestWatchRevalidatesIdleWatchersAndCachesTheAnswer(t *testing.T) {
498 t.Run("a quiet forge still notices a revocation", func(t *testing.T) {
499 r := newEventRigConfig(t, app.EventsConfig{RevalidateEvery: 10 * time.Millisecond})
500 w := r.watch(context.Background(), tokenWith("tok", time.Time{}), app.WatchParams{})
501 require.Eventually(t, func() bool { return r.events.Watchers() == 1 }, 5*time.Second, time.Millisecond)
502 r.auth.revoke("tok")
503 select {
504 case err := <-w.done:
505 assert.ErrorIs(t, err, domain.ErrUnauthenticated)
506 case <-time.After(5 * time.Second):
507 t.Fatal("an idle watch of a revoked token did not end")
508 }
509 })
510
511 t.Run("the answer is kept between checks", func(t *testing.T) {
512 r := newEventRigConfig(t, app.EventsConfig{RevalidateEvery: time.Hour})
513 pub := r.repo("pub", domain.VisibilityPublic)
514 w := r.watch(context.Background(), tokenWith("tok", time.Time{}), app.WatchParams{})
515 require.Eventually(t, func() bool { return r.events.Watchers() == 1 }, 5*time.Second, time.Millisecond)
516 for i := range 20 {
517 r.push(pub, i)
518 }
519 for range 20 {
520 w.next(t)
521 }
522 assert.Zero(t, r.auth.count(), "no store lookup within the interval")
523 })
524
525 t.Run("root and anonymous never ask", func(t *testing.T) {
526 r := newEventRigConfig(t, app.EventsConfig{RevalidateEvery: -1})
527 pub := r.repo("pub", domain.VisibilityPublic)
528 a := r.watch(context.Background(), domain.Anonymous(), app.WatchParams{})
529 b := r.watch(context.Background(), domain.Root(), app.WatchParams{})
530 require.Eventually(t, func() bool { return r.events.Watchers() == 2 }, 5*time.Second, time.Millisecond)
531 r.push(pub, 1)
532 a.next(t)
533 b.next(t)
534 assert.Zero(t, r.auth.count())
535 })
536}
537
538func TestWatchLimits(t *testing.T) {
539 noop := func(app.WatchEvent) error { return nil }
540
541 t.Run("too many topics", func(t *testing.T) {
542 r := newEventRig(t, 0)
543 r.repo("pub", domain.VisibilityPublic)
544 ok := make([]string, app.MaxWatchTopics)
545 for i := range ok {
546 ok[i] = "home"
547 }
548 ctx, cancel := context.WithCancel(context.Background())
549 cancel() // returns at once once the topics are accepted
550 assert.NoError(t, r.events.Watch(ctx, domain.Root(), app.WatchParams{Topics: ok}, noop))
551 err := r.events.Watch(context.Background(), domain.Root(), app.WatchParams{Topics: append(ok, "home")}, noop)
552 assert.ErrorIs(t, err, domain.ErrInvalid)
553 })
554
555 t.Run("total watchers", func(t *testing.T) {
556 r := newEventRig(t, 0)
557 ctx, cancel := context.WithCancel(context.Background())
558 defer cancel()
559 for range app.MaxWatchers {
560 r.watch(ctx, domain.Root(), app.WatchParams{})
561 }
562 require.Eventually(t, func() bool { return r.events.Watchers() == app.MaxWatchers }, 5*time.Second, time.Millisecond)
563 err := r.events.Watch(ctx, domain.Root(), app.WatchParams{}, noop)
564 assert.ErrorIs(t, err, domain.ErrResourceExhausted)
565 assert.Equal(t, app.MaxWatchers, r.events.Watchers(), "a refused watch is not counted")
566
567 cancel() // all end; a new one is accepted again
568 require.Eventually(t, func() bool { return r.events.Watchers() == 0 }, 5*time.Second, time.Millisecond)
569 again, stop := context.WithCancel(context.Background())
570 stop()
571 assert.NoError(t, r.events.Watch(again, domain.Root(), app.WatchParams{}, noop))
572 })
573
574 t.Run("watchers per client", func(t *testing.T) {
575 r := newEventRig(t, 0)
576 ctx, cancel := context.WithCancel(context.Background())
577 defer cancel()
578 for range app.MaxWatchersPerClient {
579 r.watch(ctx, domain.Root(), app.WatchParams{Client: "198.51.100.1"})
580 }
581 require.Eventually(t, func() bool { return r.events.Watchers() == app.MaxWatchersPerClient }, 5*time.Second, time.Millisecond)
582
583 err := r.events.Watch(ctx, domain.Root(), app.WatchParams{Client: "198.51.100.1"}, noop)
584 assert.ErrorIs(t, err, domain.ErrResourceExhausted)
585
586 // Another client, and callers in process, are not affected.
587 r.watch(ctx, domain.Root(), app.WatchParams{Client: "198.51.100.2"})
588 r.watch(ctx, domain.Root(), app.WatchParams{})
589 require.Eventually(t, func() bool { return r.events.Watchers() == app.MaxWatchersPerClient+2 }, 5*time.Second, time.Millisecond)
590
591 cancel()
592 require.Eventually(t, func() bool { return r.events.Watchers() == 0 }, 5*time.Second, time.Millisecond)
593 again, stop := context.WithCancel(context.Background())
594 stop()
595 assert.NoError(t, r.events.Watch(again, domain.Root(), app.WatchParams{Client: "198.51.100.1"}, noop), "the slots were released")
596 })
597}
598
599// repoAt declares a repository whose metadata the test changes.
600type liveRepo struct {
601 mu sync.Mutex
602 repo domain.Repo
603}
604
605func (l *liveRepo) set(f func(*domain.Repo)) {
606 l.mu.Lock()
607 defer l.mu.Unlock()
608 f(&l.repo)
609}
610
611func (r *eventRig) liveRepo(s string, vis domain.Visibility, created time.Time) (*liveRepo, domain.RepoName) {
612 n := name(r.t, s)
613 l := &liveRepo{repo: domain.Repo{Name: n, Visibility: vis, CreatedAt: created}}
614 r.meta.EXPECT().Get(mock.Anything, n).RunAndReturn(func(context.Context, domain.RepoName) (domain.Repo, error) {
615 l.mu.Lock()
616 defer l.mu.Unlock()
617 return l.repo, nil
618 }).Maybe()
619 return l, n
620}
621
622func TestEventsKeepTheVisibilityOfTheirTime(t *testing.T) {
623 ctx, cancel := context.WithCancel(context.Background())
624 defer cancel()
625
626 t.Run("a repository made public does not reveal what happened while it was private", func(t *testing.T) {
627 r := newEventRig(t, 0)
628 l, n := r.liveRepo("secret", domain.VisibilityPrivate, now)
629 root := r.watch(ctx, domain.Root(), app.WatchParams{})
630 require.Eventually(t, func() bool { return r.events.Watchers() >= 1 }, 5*time.Second, time.Millisecond)
631 r.push(n, 1)
632 r.push(n, 2)
633 root.next(t)
634 root.next(t) // both are buffered now
635
636 l.set(func(repo *domain.Repo) { repo.Visibility = domain.VisibilityPublic })
637 r.feed <- domain.RepoUpdated{Repo: n, At: now, Visibility: domain.VisibilityPublic, Previous: domain.VisibilityPrivate, CreatedAt: now}
638 root.next(t)
639 r.push(n, 3) // public era
640
641 // An anonymous caller that resumes with an id nobody issued gets what
642 // is buffered, filtered: the public update and the public push, not the
643 // pushes of the private era.
644 anon := r.watch(ctx, domain.Anonymous(), app.WatchParams{ResumeAfter: "bogus"})
645 assert.True(t, anon.next(t).Gap)
646 upd := anon.next(t)
647 require.IsType(t, domain.RepoUpdated{}, upd.Event)
648 push := anon.next(t)
649 require.IsType(t, domain.PushEvent{}, push.Event)
650 assert.Equal(t, fmt.Sprintf("%040x", 3+1), push.Event.(domain.PushEvent).Updates[0].New)
651 anon.quiet(t, r, n)
652 })
653
654 t.Run("a repository deleted and created again does not inherit old events", func(t *testing.T) {
655 r := newEventRig(t, 0)
656 first := now.Add(-time.Hour)
657 l, n := r.liveRepo("again", domain.VisibilityPublic, first)
658 root := r.watch(ctx, domain.Root(), app.WatchParams{})
659 require.Eventually(t, func() bool { return r.events.Watchers() >= 1 }, 5*time.Second, time.Millisecond)
660 r.push(n, 1)
661 root.next(t)
662
663 l.set(func(repo *domain.Repo) { repo.CreatedAt = now }) // deleted, created again
664 r.push(n, 2)
665 root.next(t)
666
667 anon := r.watch(ctx, domain.Anonymous(), app.WatchParams{ResumeAfter: "bogus"})
668 assert.True(t, anon.next(t).Gap)
669 e := anon.next(t)
670 require.IsType(t, domain.PushEvent{}, e.Event)
671 assert.Equal(t, fmt.Sprintf("%040x", 2+1), e.Event.(domain.PushEvent).Updates[0].New, "only the push to the new repository")
672 anon.quiet(t, r, n)
673 })
674
675 t.Run("a repository made private tells those who could read it, once", func(t *testing.T) {
676 r := newEventRig(t, 0)
677 l, n := r.liveRepo("hide", domain.VisibilityPublic, now)
678 anon := r.watch(ctx, domain.Anonymous(), app.WatchParams{})
679 reader := r.watch(ctx, tokenPrincipal(domain.ScopeRepoRead, "hide"), app.WatchParams{})
680 other := r.watch(ctx, tokenPrincipal(domain.ScopeRepoRead, "elsewhere"), app.WatchParams{})
681 root := r.watch(ctx, domain.Root(), app.WatchParams{})
682 require.Eventually(t, func() bool { return r.events.Watchers() >= 4 }, 5*time.Second, time.Millisecond)
683
684 l.set(func(repo *domain.Repo) { repo.Visibility = domain.VisibilityPrivate })
685 r.feed <- domain.RepoUpdated{Repo: n, At: now, Visibility: domain.VisibilityPrivate, Previous: domain.VisibilityPublic, CreatedAt: now}
686
687 for name, w := range map[string]*watcher{"anonymous": anon, "reader": reader, "other": other, "root": root} {
688 e := w.next(t)
689 up, ok := e.Event.(domain.RepoUpdated)
690 require.True(t, ok, name)
691 assert.Equal(t, n, up.Repo, name)
692 require.NotNil(t, e.Repo, name)
693 assert.Equal(t, domain.VisibilityPrivate, e.Repo.Visibility, name)
694 assert.Empty(t, e.Repo.Description, "nothing but the name and the visibility", name)
695 }
696
697 // Later events of the repository reach those who may read it now, only.
698 r.push(n, 1)
699 assert.IsType(t, domain.PushEvent{}, reader.next(t).Event)
700 assert.IsType(t, domain.PushEvent{}, root.next(t).Event)
701 anon.quiet(t, r, r.repo("open", domain.VisibilityPublic))
702 other.quiet(t, r, r.repo("open2", domain.VisibilityPublic))
703 })
704
705 t.Run("a private repository that never was public stays hidden", func(t *testing.T) {
706 r := newEventRig(t, 0)
707 _, n := r.liveRepo("closed", domain.VisibilityPrivate, now)
708 anon := r.watch(ctx, domain.Anonymous(), app.WatchParams{})
709 require.Eventually(t, func() bool { return r.events.Watchers() >= 1 }, 5*time.Second, time.Millisecond)
710 r.feed <- domain.RepoUpdated{Repo: n, At: now, Visibility: domain.VisibilityPrivate, Previous: domain.VisibilityPrivate, CreatedAt: now}
711 r.push(n, 1)
712 anon.quiet(t, r, r.repo("open", domain.VisibilityPublic))
713 })
714}
715
716// TestPushEventsKeepTheirSnapshot: a push is told by the repository as it was
717// when published, not as the store says when the feeder gets to it.
718func TestPushEventsKeepTheirSnapshot(t *testing.T) {
719 ctx, cancel := context.WithCancel(context.Background())
720 defer cancel()
721
722 tests := []struct {
723 name string
724 swap func(repo *domain.Repo)
725 rootSees bool // the repository is still the one the push is about
726 }{
727 {"made public before the feeder appends", func(repo *domain.Repo) { repo.Visibility = domain.VisibilityPublic }, true},
728 {"recreated before the feeder appends", func(repo *domain.Repo) {
729 repo.Visibility = domain.VisibilityPublic
730 repo.CreatedAt = now.Add(time.Hour)
731 }, false},
732 }
733 for _, tt := range tests {
734 t.Run(tt.name, func(t *testing.T) {
735 r := newEventRig(t, 0)
736 l, n := r.liveRepo("secret", domain.VisibilityPrivate, now)
737 anon := r.watch(ctx, domain.Anonymous(), app.WatchParams{})
738 root := r.watch(ctx, domain.Root(), app.WatchParams{})
739 require.Eventually(t, func() bool { return r.events.Watchers() >= 2 }, 5*time.Second, time.Millisecond)
740
741 // Published while private; the store changes before the feeder
742 // can have looked at it: the event is built, then the state is
743 // swapped, and only then is it handed to the feeder.
744 ev := r.pushEvent(n, fmt.Sprintf("%040x", 1), "refs/heads/main")
745 l.set(tt.swap)
746 r.feed <- ev
747 if tt.rootSees {
748 root.next(t)
749 }
750
751 anon.quiet(t, r, r.repo("open", domain.VisibilityPublic))
752
753 late := r.watch(ctx, domain.Anonymous(), app.WatchParams{ResumeAfter: "bogus"})
754 assert.True(t, late.next(t).Gap)
755 late.quiet(t, r, r.repo("open2", domain.VisibilityPublic))
756 })
757 }
758
759 t.Run("a push of an unknown repository is told to nobody", func(t *testing.T) {
760 r := newEventRig(t, 0)
761 n := r.repo("ghost", domain.VisibilityPublic)
762 root := r.watch(ctx, domain.Root(), app.WatchParams{})
763 require.Eventually(t, func() bool { return r.events.Watchers() >= 1 }, 5*time.Second, time.Millisecond)
764 r.feed <- domain.PushEvent{Repo: n, Visibility: domain.VisibilityPublic, At: now}
765 root.quiet(t, r, n)
766 })
767}