fix(app): snapshot repo visibility on push events at publish time
- commit
- 19ceb08b856095d1d751630afcb96bc990b3deae
- author
- brice · · 2026-10-10 11:41 UTC
- parent
- 3a3918d
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
5 files · +99 -28
M internal/app/events.go +4 -10
| @@ -152,8 +152,7 @@ func (e *Events) feed(ctx context.Context, sub <-chan domain.Event) { | |||
| 152 | 152 | } | |
| 153 | 153 | ||
| 154 | 154 | // snapshot records the visibility and the generation of the repository an | |
| 155 | - | // event is about. The events of the repository lifecycle carry them; for a | |
| 156 | - | // push the repository is read now. | |
| 155 | + | // event is about. Every event of the repository carries them. | |
| 157 | 156 | func (e *Events) snapshot(ctx context.Context, ev domain.Event) bufferedEvent { | |
| 158 | 157 | switch ev := ev.(type) { | |
| 159 | 158 | case domain.RepoCreated: | |
| @@ -163,15 +162,10 @@ func (e *Events) snapshot(ctx context.Context, ev domain.Event) bufferedEvent { | |||
| 163 | 162 | case domain.RepoDeleted: | |
| 164 | 163 | return bufferedEvent{known: true, vis: ev.Visibility, created: ev.CreatedAt} | |
| 165 | 164 | case domain.PushEvent: | |
| 166 | - | repo, err := lookupRepo(ctx, e.repos.meta, ev.Repo) | |
| 167 | - | if err != nil { | |
| 168 | - | if !errors.Is(err, domain.ErrNotFound) { | |
| 169 | - | log := zerowrap.FromCtx(ctx) | |
| 170 | - | log.Error().Err(err).Str("repo", ev.Repo.String()).Msg("read repository of a buffered event") | |
| 171 | - | } | |
| 172 | - | return bufferedEvent{} // gone, or unknown: nobody is told | |
| 165 | + | if ev.CreatedAt.IsZero() { | |
| 166 | + | return bufferedEvent{} // repository unknown at publish time: nobody is told | |
| 173 | 167 | } | |
| 174 | - | return bufferedEvent{known: true, vis: repo.Visibility, created: repo.CreatedAt} | |
| 168 | + | return bufferedEvent{known: true, vis: ev.Visibility, created: ev.CreatedAt} | |
| 175 | 169 | } | |
| 176 | 170 | return bufferedEvent{known: true} | |
| 177 | 171 | } | |
| @@ -152,8 +152,7 @@ func (e *Events) feed(ctx context.Context, sub <-chan domain.Event) { | |||||
| 152 | } | 152 | } | ||
| 153 | 153 | ||||
| 154 | // snapshot records the visibility and the generation of the repository an | 154 | // snapshot records the visibility and the generation of the repository an | ||
| 155 | - | // event is about. The events of the repository lifecycle carry them; for a | 155 | + | // event is about. Every event of the repository carries them. |
| 156 | - | // push the repository is read now. | |||
| 157 | func (e *Events) snapshot(ctx context.Context, ev domain.Event) bufferedEvent { | 156 | func (e *Events) snapshot(ctx context.Context, ev domain.Event) bufferedEvent { | ||
| 158 | switch ev := ev.(type) { | 157 | switch ev := ev.(type) { | ||
| 159 | case domain.RepoCreated: | 158 | case domain.RepoCreated: | ||
| @@ -163,15 +162,10 @@ func (e *Events) snapshot(ctx context.Context, ev domain.Event) bufferedEvent { | |||||
| 163 | case domain.RepoDeleted: | 162 | case domain.RepoDeleted: | ||
| 164 | return bufferedEvent{known: true, vis: ev.Visibility, created: ev.CreatedAt} | 163 | return bufferedEvent{known: true, vis: ev.Visibility, created: ev.CreatedAt} | ||
| 165 | case domain.PushEvent: | 164 | case domain.PushEvent: | ||
| 166 | - | repo, err := lookupRepo(ctx, e.repos.meta, ev.Repo) | 165 | + | if ev.CreatedAt.IsZero() { |
| 167 | - | if err != nil { | 166 | + | return bufferedEvent{} // repository unknown at publish time: nobody is told |
| 168 | - | if !errors.Is(err, domain.ErrNotFound) { | |||
| 169 | - | log := zerowrap.FromCtx(ctx) | |||
| 170 | - | log.Error().Err(err).Str("repo", ev.Repo.String()).Msg("read repository of a buffered event") | |||
| 171 | - | } | |||
| 172 | - | return bufferedEvent{} // gone, or unknown: nobody is told | |||
| 173 | } | 167 | } | ||
| 174 | - | return bufferedEvent{known: true, vis: repo.Visibility, created: repo.CreatedAt} | 168 | + | return bufferedEvent{known: true, vis: ev.Visibility, created: ev.CreatedAt} |
| 175 | } | 169 | } | ||
| 176 | return bufferedEvent{known: true} | 170 | return bufferedEvent{known: true} | ||
| 177 | } | 171 | } | ||
M internal/app/events_test.go +69 -6
| @@ -93,13 +93,23 @@ func newEventRigConfig(t *testing.T, cfg app.EventsConfig) *eventRig { | |||
| 93 | 93 | // repo declares a repository the rig knows. | |
| 94 | 94 | func (r *eventRig) repo(s string, vis domain.Visibility) domain.RepoName { | |
| 95 | 95 | n := name(r.t, s) | |
| 96 | - | r.meta.EXPECT().Get(mock.Anything, n).Return(domain.Repo{Name: n, Visibility: vis}, nil).Maybe() | |
| 96 | + | r.meta.EXPECT().Get(mock.Anything, n).Return(domain.Repo{Name: n, Visibility: vis, CreatedAt: now}, nil).Maybe() | |
| 97 | 97 | return n | |
| 98 | 98 | } | |
| 99 | 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. | |
| 102 | + | func (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 | + | ||
| 100 | 111 | func (r *eventRig) push(n domain.RepoName, i int) { | |
| 101 | - | oid := fmt.Sprintf("%040x", i+1) | |
| 102 | - | r.feed <- domain.PushEvent{Repo: n, Updates: []domain.RefUpdate{{Old: domain.ZeroOID, New: oid, Ref: "refs/heads/main"}}, At: now} | |
| 112 | + | r.feed <- r.pushEvent(n, fmt.Sprintf("%040x", i+1), "refs/heads/main") | |
| 103 | 113 | } | |
| 104 | 114 | ||
| 105 | 115 | // watcher is a running Watch. | |
| @@ -141,7 +151,7 @@ var sentinelOID = fmt.Sprintf("%040x", 0xfeed) | |||
| 141 | 151 | // Events reach a watcher in order, so anything sent before would show up first. | |
| 142 | 152 | func (w *watcher) quiet(t *testing.T, r *eventRig, n domain.RepoName) { | |
| 143 | 153 | t.Helper() | |
| 144 | - | r.feed <- domain.PushEvent{Repo: n, Updates: []domain.RefUpdate{{Old: domain.ZeroOID, New: sentinelOID, Ref: "refs/heads/sentinel"}}, At: now} | |
| 154 | + | r.feed <- r.pushEvent(n, sentinelOID, "refs/heads/sentinel") | |
| 145 | 155 | e := w.next(t) | |
| 146 | 156 | push, ok := e.Event.(domain.PushEvent) | |
| 147 | 157 | if !ok || len(push.Updates) != 1 || push.Updates[0].New != sentinelOID { | |
| @@ -189,7 +199,7 @@ func TestWatchRechecksAccessPerEvent(t *testing.T) { | |||
| 189 | 199 | r.meta.EXPECT().Get(mock.Anything, n).RunAndReturn(func(context.Context, domain.RepoName) (domain.Repo, error) { | |
| 190 | 200 | mu.Lock() | |
| 191 | 201 | defer mu.Unlock() | |
| 192 | - | return domain.Repo{Name: n, Visibility: vis}, nil | |
| 202 | + | return domain.Repo{Name: n, Visibility: vis, CreatedAt: now}, nil | |
| 193 | 203 | }) | |
| 194 | 204 | ctx, cancel := context.WithCancel(context.Background()) | |
| 195 | 205 | defer cancel() | |
| @@ -408,7 +418,7 @@ func TestEventsRepoPayloads(t *testing.T) { | |||
| 408 | 418 | require.Eventually(t, func() bool { return r.events.Watchers() == 2 }, 5*time.Second, time.Millisecond) | |
| 409 | 419 | ||
| 410 | 420 | r.feed <- domain.RepoCreated{Repo: pub, Visibility: domain.VisibilityPublic, At: now} | |
| 411 | - | r.feed <- domain.RepoUpdated{Repo: pub, At: now, Visibility: domain.VisibilityPublic, Previous: domain.VisibilityPublic} | |
| 421 | + | r.feed <- domain.RepoUpdated{Repo: pub, At: now, Visibility: domain.VisibilityPublic, Previous: domain.VisibilityPublic, CreatedAt: now} | |
| 412 | 422 | // A deleted repository cannot be read any more: its visibility decides. | |
| 413 | 423 | r.feed <- domain.RepoDeleted{Repo: priv, Visibility: domain.VisibilityPrivate, At: now} | |
| 414 | 424 | r.feed <- domain.RepoDeleted{Repo: pub, Visibility: domain.VisibilityPublic, At: now} | |
| @@ -702,3 +712,56 @@ func TestEventsKeepTheVisibilityOfTheirTime(t *testing.T) { | |||
| 702 | 712 | anon.quiet(t, r, r.repo("open", domain.VisibilityPublic)) | |
| 703 | 713 | }) | |
| 704 | 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. | |
| 718 | + | func 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 | + | } | |
| @@ -93,13 +93,23 @@ func newEventRigConfig(t *testing.T, cfg app.EventsConfig) *eventRig { | |||||
| 93 | // repo declares a repository the rig knows. | 93 | // repo declares a repository the rig knows. | ||
| 94 | func (r *eventRig) repo(s string, vis domain.Visibility) domain.RepoName { | 94 | func (r *eventRig) repo(s string, vis domain.Visibility) domain.RepoName { | ||
| 95 | n := name(r.t, s) | 95 | n := name(r.t, s) | ||
| 96 | - | r.meta.EXPECT().Get(mock.Anything, n).Return(domain.Repo{Name: n, Visibility: vis}, nil).Maybe() | 96 | + | r.meta.EXPECT().Get(mock.Anything, n).Return(domain.Repo{Name: n, Visibility: vis, CreatedAt: now}, nil).Maybe() |
| 97 | return n | 97 | return n | ||
| 98 | } | 98 | } | ||
| 99 | 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. | |||
| 102 | + | func (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 | + | ||||
| 100 | func (r *eventRig) push(n domain.RepoName, i int) { | 111 | func (r *eventRig) push(n domain.RepoName, i int) { | ||
| 101 | - | oid := fmt.Sprintf("%040x", i+1) | 112 | + | r.feed <- r.pushEvent(n, fmt.Sprintf("%040x", i+1), "refs/heads/main") |
| 102 | - | r.feed <- domain.PushEvent{Repo: n, Updates: []domain.RefUpdate{{Old: domain.ZeroOID, New: oid, Ref: "refs/heads/main"}}, At: now} | |||
| 103 | } | 113 | } | ||
| 104 | 114 | ||||
| 105 | // watcher is a running Watch. | 115 | // watcher is a running Watch. | ||
| @@ -141,7 +151,7 @@ var sentinelOID = fmt.Sprintf("%040x", 0xfeed) | |||||
| 141 | // Events reach a watcher in order, so anything sent before would show up first. | 151 | // Events reach a watcher in order, so anything sent before would show up first. | ||
| 142 | func (w *watcher) quiet(t *testing.T, r *eventRig, n domain.RepoName) { | 152 | func (w *watcher) quiet(t *testing.T, r *eventRig, n domain.RepoName) { | ||
| 143 | t.Helper() | 153 | t.Helper() | ||
| 144 | - | r.feed <- domain.PushEvent{Repo: n, Updates: []domain.RefUpdate{{Old: domain.ZeroOID, New: sentinelOID, Ref: "refs/heads/sentinel"}}, At: now} | 154 | + | r.feed <- r.pushEvent(n, sentinelOID, "refs/heads/sentinel") |
| 145 | e := w.next(t) | 155 | e := w.next(t) | ||
| 146 | push, ok := e.Event.(domain.PushEvent) | 156 | push, ok := e.Event.(domain.PushEvent) | ||
| 147 | if !ok || len(push.Updates) != 1 || push.Updates[0].New != sentinelOID { | 157 | if !ok || len(push.Updates) != 1 || push.Updates[0].New != sentinelOID { | ||
| @@ -189,7 +199,7 @@ func TestWatchRechecksAccessPerEvent(t *testing.T) { | |||||
| 189 | r.meta.EXPECT().Get(mock.Anything, n).RunAndReturn(func(context.Context, domain.RepoName) (domain.Repo, error) { | 199 | r.meta.EXPECT().Get(mock.Anything, n).RunAndReturn(func(context.Context, domain.RepoName) (domain.Repo, error) { | ||
| 190 | mu.Lock() | 200 | mu.Lock() | ||
| 191 | defer mu.Unlock() | 201 | defer mu.Unlock() | ||
| 192 | - | return domain.Repo{Name: n, Visibility: vis}, nil | 202 | + | return domain.Repo{Name: n, Visibility: vis, CreatedAt: now}, nil |
| 193 | }) | 203 | }) | ||
| 194 | ctx, cancel := context.WithCancel(context.Background()) | 204 | ctx, cancel := context.WithCancel(context.Background()) | ||
| 195 | defer cancel() | 205 | defer cancel() | ||
| @@ -408,7 +418,7 @@ func TestEventsRepoPayloads(t *testing.T) { | |||||
| 408 | require.Eventually(t, func() bool { return r.events.Watchers() == 2 }, 5*time.Second, time.Millisecond) | 418 | require.Eventually(t, func() bool { return r.events.Watchers() == 2 }, 5*time.Second, time.Millisecond) | ||
| 409 | 419 | ||||
| 410 | r.feed <- domain.RepoCreated{Repo: pub, Visibility: domain.VisibilityPublic, At: now} | 420 | r.feed <- domain.RepoCreated{Repo: pub, Visibility: domain.VisibilityPublic, At: now} | ||
| 411 | - | r.feed <- domain.RepoUpdated{Repo: pub, At: now, Visibility: domain.VisibilityPublic, Previous: domain.VisibilityPublic} | 421 | + | r.feed <- domain.RepoUpdated{Repo: pub, At: now, Visibility: domain.VisibilityPublic, Previous: domain.VisibilityPublic, CreatedAt: now} |
| 412 | // A deleted repository cannot be read any more: its visibility decides. | 422 | // A deleted repository cannot be read any more: its visibility decides. | ||
| 413 | r.feed <- domain.RepoDeleted{Repo: priv, Visibility: domain.VisibilityPrivate, At: now} | 423 | r.feed <- domain.RepoDeleted{Repo: priv, Visibility: domain.VisibilityPrivate, At: now} | ||
| 414 | r.feed <- domain.RepoDeleted{Repo: pub, Visibility: domain.VisibilityPublic, At: now} | 424 | r.feed <- domain.RepoDeleted{Repo: pub, Visibility: domain.VisibilityPublic, At: now} | ||
| @@ -702,3 +712,56 @@ func TestEventsKeepTheVisibilityOfTheirTime(t *testing.T) { | |||||
| 702 | anon.quiet(t, r, r.repo("open", domain.VisibilityPublic)) | 712 | anon.quiet(t, r, r.repo("open", domain.VisibilityPublic)) | ||
| 703 | }) | 713 | }) | ||
| 704 | } | 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. | |||
| 718 | + | func 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 | + | } | |||
M internal/app/pushes.go +16 -11
| @@ -126,13 +126,17 @@ func (u *Pushes) Record(ctx context.Context, p domain.Principal, name domain.Rep | |||
| 126 | 126 | now := u.clock.Now().UTC() | |
| 127 | 127 | var errs []error | |
| 128 | 128 | ||
| 129 | - | if err := u.touchPushedAt(ctx, name, now); err != nil { | |
| 129 | + | repo, err := u.touchPushedAt(ctx, name, now) | |
| 130 | + | if err != nil { | |
| 130 | 131 | errs = append(errs, err) | |
| 131 | 132 | } | |
| 132 | 133 | if err := u.fixDefaultBranch(ctx, name, updates); err != nil { | |
| 133 | 134 | errs = append(errs, err) | |
| 134 | 135 | } | |
| 135 | - | u.bus.Publish(ctx, domain.PushEvent{Repo: name, Updates: append([]domain.RefUpdate(nil), updates...), At: now}) | |
| 136 | + | u.bus.Publish(ctx, domain.PushEvent{ | |
| 137 | + | Repo: name, Updates: append([]domain.RefUpdate(nil), updates...), At: now, | |
| 138 | + | Visibility: repo.Visibility, CreatedAt: repo.CreatedAt, | |
| 139 | + | }) | |
| 136 | 140 | u.maint.Schedule(ctx, name) | |
| 137 | 141 | log.Info(). | |
| 138 | 142 | Str("repo", name.String()).Int("refs", len(updates)). | |
| @@ -142,18 +146,19 @@ func (u *Pushes) Record(ctx context.Context, p domain.Principal, name domain.Rep | |||
| 142 | 146 | } | |
| 143 | 147 | ||
| 144 | 148 | // touchPushedAt stores pushed_at, retrying when another write got in between. | |
| 145 | - | // A repository deleted meanwhile is not an error. | |
| 146 | - | func (u *Pushes) touchPushedAt(ctx context.Context, name domain.RepoName, now time.Time) error { | |
| 149 | + | // A repository deleted meanwhile is not an error. It returns the repository | |
| 150 | + | // as read, or the zero value when it is unknown. | |
| 151 | + | func (u *Pushes) touchPushedAt(ctx context.Context, name domain.RepoName, now time.Time) (domain.Repo, error) { | |
| 147 | 152 | for range pushRetries { | |
| 148 | 153 | repo, err := u.meta.Get(ctx, name) | |
| 149 | 154 | if errors.Is(err, domain.ErrNotFound) { | |
| 150 | - | return nil | |
| 155 | + | return domain.Repo{}, nil | |
| 151 | 156 | } | |
| 152 | 157 | if err != nil { | |
| 153 | - | return fmt.Errorf("get repository %q: %w", name, err) | |
| 158 | + | return domain.Repo{}, fmt.Errorf("get repository %q: %w", name, err) | |
| 154 | 159 | } | |
| 155 | 160 | if !now.After(repo.PushedAt) { | |
| 156 | - | return nil // a later push already recorded | |
| 161 | + | return repo, nil // a later push already recorded | |
| 157 | 162 | } | |
| 158 | 163 | repo.PushedAt = now | |
| 159 | 164 | err = u.meta.Update(ctx, repo) | |
| @@ -161,14 +166,14 @@ func (u *Pushes) touchPushedAt(ctx context.Context, name domain.RepoName, now ti | |||
| 161 | 166 | continue | |
| 162 | 167 | } | |
| 163 | 168 | if errors.Is(err, domain.ErrNotFound) { | |
| 164 | - | return nil | |
| 169 | + | return domain.Repo{}, nil | |
| 165 | 170 | } | |
| 166 | 171 | if err != nil { | |
| 167 | - | return fmt.Errorf("update repository %q: %w", name, err) | |
| 172 | + | return repo, fmt.Errorf("update repository %q: %w", name, err) | |
| 168 | 173 | } | |
| 169 | - | return nil | |
| 174 | + | return repo, nil | |
| 170 | 175 | } | |
| 171 | - | return fmt.Errorf("update pushed_at of %q: %w", name, domain.ErrConflict) | |
| 176 | + | return domain.Repo{}, fmt.Errorf("update pushed_at of %q: %w", name, domain.ErrConflict) | |
| 172 | 177 | } | |
| 173 | 178 | ||
| 174 | 179 | // fixDefaultBranch points HEAD to a branch that exists when it points to a | |
| @@ -126,13 +126,17 @@ func (u *Pushes) Record(ctx context.Context, p domain.Principal, name domain.Rep | |||||
| 126 | now := u.clock.Now().UTC() | 126 | now := u.clock.Now().UTC() | ||
| 127 | var errs []error | 127 | var errs []error | ||
| 128 | 128 | ||||
| 129 | - | if err := u.touchPushedAt(ctx, name, now); err != nil { | 129 | + | repo, err := u.touchPushedAt(ctx, name, now) |
| 130 | + | if err != nil { | |||
| 130 | errs = append(errs, err) | 131 | errs = append(errs, err) | ||
| 131 | } | 132 | } | ||
| 132 | if err := u.fixDefaultBranch(ctx, name, updates); err != nil { | 133 | if err := u.fixDefaultBranch(ctx, name, updates); err != nil { | ||
| 133 | errs = append(errs, err) | 134 | errs = append(errs, err) | ||
| 134 | } | 135 | } | ||
| 135 | - | u.bus.Publish(ctx, domain.PushEvent{Repo: name, Updates: append([]domain.RefUpdate(nil), updates...), At: now}) | 136 | + | u.bus.Publish(ctx, domain.PushEvent{ |
| 137 | + | Repo: name, Updates: append([]domain.RefUpdate(nil), updates...), At: now, | |||
| 138 | + | Visibility: repo.Visibility, CreatedAt: repo.CreatedAt, | |||
| 139 | + | }) | |||
| 136 | u.maint.Schedule(ctx, name) | 140 | u.maint.Schedule(ctx, name) | ||
| 137 | log.Info(). | 141 | log.Info(). | ||
| 138 | Str("repo", name.String()).Int("refs", len(updates)). | 142 | Str("repo", name.String()).Int("refs", len(updates)). | ||
| @@ -142,18 +146,19 @@ func (u *Pushes) Record(ctx context.Context, p domain.Principal, name domain.Rep | |||||
| 142 | } | 146 | } | ||
| 143 | 147 | ||||
| 144 | // touchPushedAt stores pushed_at, retrying when another write got in between. | 148 | // touchPushedAt stores pushed_at, retrying when another write got in between. | ||
| 145 | - | // A repository deleted meanwhile is not an error. | 149 | + | // A repository deleted meanwhile is not an error. It returns the repository |
| 146 | - | func (u *Pushes) touchPushedAt(ctx context.Context, name domain.RepoName, now time.Time) error { | 150 | + | // as read, or the zero value when it is unknown. |
| 151 | + | func (u *Pushes) touchPushedAt(ctx context.Context, name domain.RepoName, now time.Time) (domain.Repo, error) { | |||
| 147 | for range pushRetries { | 152 | for range pushRetries { | ||
| 148 | repo, err := u.meta.Get(ctx, name) | 153 | repo, err := u.meta.Get(ctx, name) | ||
| 149 | if errors.Is(err, domain.ErrNotFound) { | 154 | if errors.Is(err, domain.ErrNotFound) { | ||
| 150 | - | return nil | 155 | + | return domain.Repo{}, nil |
| 151 | } | 156 | } | ||
| 152 | if err != nil { | 157 | if err != nil { | ||
| 153 | - | return fmt.Errorf("get repository %q: %w", name, err) | 158 | + | return domain.Repo{}, fmt.Errorf("get repository %q: %w", name, err) |
| 154 | } | 159 | } | ||
| 155 | if !now.After(repo.PushedAt) { | 160 | if !now.After(repo.PushedAt) { | ||
| 156 | - | return nil // a later push already recorded | 161 | + | return repo, nil // a later push already recorded |
| 157 | } | 162 | } | ||
| 158 | repo.PushedAt = now | 163 | repo.PushedAt = now | ||
| 159 | err = u.meta.Update(ctx, repo) | 164 | err = u.meta.Update(ctx, repo) | ||
| @@ -161,14 +166,14 @@ func (u *Pushes) touchPushedAt(ctx context.Context, name domain.RepoName, now ti | |||||
| 161 | continue | 166 | continue | ||
| 162 | } | 167 | } | ||
| 163 | if errors.Is(err, domain.ErrNotFound) { | 168 | if errors.Is(err, domain.ErrNotFound) { | ||
| 164 | - | return nil | 169 | + | return domain.Repo{}, nil |
| 165 | } | 170 | } | ||
| 166 | if err != nil { | 171 | if err != nil { | ||
| 167 | - | return fmt.Errorf("update repository %q: %w", name, err) | 172 | + | return repo, fmt.Errorf("update repository %q: %w", name, err) |
| 168 | } | 173 | } | ||
| 169 | - | return nil | 174 | + | return repo, nil |
| 170 | } | 175 | } | ||
| 171 | - | return fmt.Errorf("update pushed_at of %q: %w", name, domain.ErrConflict) | 176 | + | return domain.Repo{}, fmt.Errorf("update pushed_at of %q: %w", name, domain.ErrConflict) |
| 172 | } | 177 | } | ||
| 173 | 178 | ||||
| 174 | // fixDefaultBranch points HEAD to a branch that exists when it points to a | 179 | // fixDefaultBranch points HEAD to a branch that exists when it points to a | ||
M internal/app/pushes_test.go +1 -1
| @@ -217,7 +217,7 @@ func TestPushesRecord(t *testing.T) { | |||
| 217 | 217 | f.meta.EXPECT().Get(ctx, n).Return(repo, nil) | |
| 218 | 218 | f.meta.EXPECT().Update(ctx, domain.Repo{Name: n, Visibility: domain.VisibilityPrivate, CreatedAt: repo.CreatedAt, PushedAt: now}).Return(nil) | |
| 219 | 219 | f.git.EXPECT().DefaultBranch(ctx, n).Return("main", true, nil) | |
| 220 | - | f.bus.EXPECT().Publish(ctx, domain.PushEvent{Repo: n, Updates: ups, At: now}).Return() | |
| 220 | + | f.bus.EXPECT().Publish(ctx, domain.PushEvent{Repo: n, Updates: ups, At: now, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt}).Return() | |
| 221 | 221 | ||
| 222 | 222 | require.NoError(t, f.pushes.Record(ctx, p, n, ups)) | |
| 223 | 223 | assert.Equal(t, []domain.RepoName{n}, f.sched.names) | |
| @@ -217,7 +217,7 @@ func TestPushesRecord(t *testing.T) { | |||||
| 217 | f.meta.EXPECT().Get(ctx, n).Return(repo, nil) | 217 | f.meta.EXPECT().Get(ctx, n).Return(repo, nil) | ||
| 218 | f.meta.EXPECT().Update(ctx, domain.Repo{Name: n, Visibility: domain.VisibilityPrivate, CreatedAt: repo.CreatedAt, PushedAt: now}).Return(nil) | 218 | f.meta.EXPECT().Update(ctx, domain.Repo{Name: n, Visibility: domain.VisibilityPrivate, CreatedAt: repo.CreatedAt, PushedAt: now}).Return(nil) | ||
| 219 | f.git.EXPECT().DefaultBranch(ctx, n).Return("main", true, nil) | 219 | f.git.EXPECT().DefaultBranch(ctx, n).Return("main", true, nil) | ||
| 220 | - | f.bus.EXPECT().Publish(ctx, domain.PushEvent{Repo: n, Updates: ups, At: now}).Return() | 220 | + | f.bus.EXPECT().Publish(ctx, domain.PushEvent{Repo: n, Updates: ups, At: now, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt}).Return() |
| 221 | 221 | ||||
| 222 | require.NoError(t, f.pushes.Record(ctx, p, n, ups)) | 222 | require.NoError(t, f.pushes.Record(ctx, p, n, ups)) | ||
| 223 | assert.Equal(t, []domain.RepoName{n}, f.sched.names) | 223 | assert.Equal(t, []domain.RepoName{n}, f.sched.names) | ||
M internal/domain/event.go +9 -0
| @@ -93,10 +93,19 @@ type RepoDeleted struct { | |||
| 93 | 93 | func (RepoDeleted) EventKind() string { return "repo.deleted" } | |
| 94 | 94 | ||
| 95 | 95 | // PushEvent is published after a push has been accepted. | |
| 96 | + | // | |
| 97 | + | // It carries the visibility and the creation time of the repository as read | |
| 98 | + | // when the push was recorded, so the event stream does not read them later, | |
| 99 | + | // when the repository may have changed or been recreated. A zero CreatedAt | |
| 100 | + | // means the repository was unknown: the event is not delivered. | |
| 96 | 101 | type PushEvent struct { | |
| 97 | 102 | Repo RepoName | |
| 98 | 103 | Updates []RefUpdate | |
| 99 | 104 | At time.Time | |
| 105 | + | // Visibility is the visibility of the repository at publish time. | |
| 106 | + | Visibility Visibility | |
| 107 | + | // CreatedAt is the creation time of the repository that was pushed to. | |
| 108 | + | CreatedAt time.Time | |
| 100 | 109 | } | |
| 101 | 110 | ||
| 102 | 111 | // EventKind implements Event. | |
| @@ -93,10 +93,19 @@ type RepoDeleted struct { | |||||
| 93 | func (RepoDeleted) EventKind() string { return "repo.deleted" } | 93 | func (RepoDeleted) EventKind() string { return "repo.deleted" } | ||
| 94 | 94 | ||||
| 95 | // PushEvent is published after a push has been accepted. | 95 | // PushEvent is published after a push has been accepted. | ||
| 96 | + | // | |||
| 97 | + | // It carries the visibility and the creation time of the repository as read | |||
| 98 | + | // when the push was recorded, so the event stream does not read them later, | |||
| 99 | + | // when the repository may have changed or been recreated. A zero CreatedAt | |||
| 100 | + | // means the repository was unknown: the event is not delivered. | |||
| 96 | type PushEvent struct { | 101 | type PushEvent struct { | ||
| 97 | Repo RepoName | 102 | Repo RepoName | ||
| 98 | Updates []RefUpdate | 103 | Updates []RefUpdate | ||
| 99 | At time.Time | 104 | At time.Time | ||
| 105 | + | // Visibility is the visibility of the repository at publish time. | |||
| 106 | + | Visibility Visibility | |||
| 107 | + | // CreatedAt is the creation time of the repository that was pushed to. | |||
| 108 | + | CreatedAt time.Time | |||
| 100 | } | 109 | } | ||
| 101 | 110 | ||||
| 102 | // EventKind implements Event. | 111 | // EventKind implements Event. | ||