1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package usecase
4
5import (
6 "context"
7 "errors"
8 "fmt"
9 "time"
10
11 "bnema.dev/ermite/internal/domain"
12 "bnema.dev/ermite/internal/ports"
13)
14
15// mirrorReq is what a sync request asks for. Requests that coalesce merge by OR.
16type mirrorReq struct{ force, manual bool }
17
18func (r mirrorReq) merge(o mirrorReq) mirrorReq {
19 return mirrorReq{force: r.force || o.force, manual: r.manual || o.manual}
20}
21
22// mirrorFlight is the single in-flight sync of a repository: cur is running, next is the
23// merged request that arrived meanwhile (one rerun, whatever the number of requests).
24type mirrorFlight struct {
25 cur mirrorReq
26 next *mirrorReq
27 done chan struct{} // closed with the result of cur's first run
28 completed bool
29 result ports.MirrorSyncResult
30 err error
31}
32
33var (
34 _ ports.MirrorService = (*Mirrors)(nil)
35 _ ports.MirrorTrigger = (*Mirrors)(nil)
36 _ ports.MirrorWarner = (*Mirrors)(nil)
37)
38
39// Sync waits for its own run; overlapping requests coalesce into one sticky rerun.
40func (m *Mirrors) Sync(ctx context.Context, p domain.Principal, repo domain.RepoName, force bool) (ports.MirrorSyncResult, error) {
41 if err := m.admin(ctx, p, repo); err != nil {
42 return ports.MirrorSyncResult{}, err
43 }
44 m.Start()
45 f, queued, err := m.enqueue(repo, mirrorReq{force: force, manual: true})
46 if err != nil {
47 return ports.MirrorSyncResult{}, err
48 }
49 if queued {
50 return ports.MirrorSyncResult{Repo: repo, Status: domain.MirrorStatus{Repo: repo}, Queued: true}, nil
51 }
52 select {
53 case <-m.ctx.Done():
54 return ports.MirrorSyncResult{}, m.ctx.Err()
55 case <-ctx.Done():
56 return ports.MirrorSyncResult{}, ctx.Err()
57 case <-f.done:
58 f.result.Repo = repo
59 return f.result, f.err
60 }
61}
62
63// SyncAll syncs every repository whose mirror is enabled, as an owner. A failing repository
64// does not stop the others: its outcome carries the error. The returned error is only for
65// what prevents the whole request (authorization, listing, cancellation).
66func (m *Mirrors) SyncAll(ctx context.Context, p domain.Principal, force bool) ([]ports.MirrorSyncResult, error) {
67 if err := mirrorOwner(p); err != nil {
68 return nil, err
69 }
70 repos, err := m.cfg.Meta.List(ctx)
71 if err != nil {
72 return nil, err
73 }
74 out := make([]ports.MirrorSyncResult, 0, len(repos))
75 for _, r := range repos {
76 if err := ctx.Err(); err != nil {
77 return out, err
78 }
79 t, err := m.target(ctx, r.Name)
80 if err == nil && !t.Enabled {
81 continue
82 }
83 var res ports.MirrorSyncResult
84 if err == nil {
85 res, err = m.Sync(ctx, p, r.Name, force)
86 }
87 if ctxErr := ctx.Err(); ctxErr != nil {
88 return out, ctxErr
89 }
90 res.Repo = r.Name
91 if res.Status.Repo == (domain.RepoName{}) {
92 res.Status.Repo = r.Name
93 }
94 res.Err = err
95 out = append(out, res)
96 }
97 return out, nil
98}
99
100// Wait blocks until no sync is queued or running, or ctx is done. It is a test barrier: call it
101// only while nothing enqueues new syncs and before Close (queued jobs are not drained by Close).
102func (m *Mirrors) Wait(ctx context.Context) {
103 done := make(chan struct{})
104 go func() { m.flightWG.Wait(); close(done) }()
105 select {
106 case <-done:
107 case <-ctx.Done():
108 }
109}
110
111func (m *Mirrors) enqueue(repo domain.RepoName, req mirrorReq) (*mirrorFlight, bool, error) {
112 m.mu.Lock()
113 defer m.mu.Unlock()
114 if m.closed {
115 return nil, false, context.Canceled
116 }
117 if f := m.flights[repo]; f != nil {
118 next := req
119 if f.next != nil {
120 next = f.next.merge(req)
121 }
122 f.next = &next
123 return f, true, nil
124 }
125 f := &mirrorFlight{cur: req, done: make(chan struct{})}
126 select {
127 case m.jobs <- repo:
128 m.flights[repo] = f
129 m.flightWG.Add(1)
130 return f, false, nil
131 default:
132 return nil, false, domain.ErrResourceExhausted
133 }
134}
135
136func (m *Mirrors) run(repo domain.RepoName) {
137 m.mu.Lock()
138 f := m.flights[repo]
139 m.mu.Unlock()
140 for {
141 result, err := m.attempt(m.ctx, repo, f.cur)
142 m.mu.Lock()
143 if !f.completed {
144 f.result, f.err = result, err
145 f.completed = true
146 close(f.done)
147 }
148 if f.next == nil || m.closed {
149 delete(m.flights, repo)
150 m.flightWG.Done()
151 m.mu.Unlock()
152 return
153 }
154 f.cur, f.next = *f.next, nil
155 m.mu.Unlock()
156 }
157}
158
159// mirrorPlan is what attempt decided under the configuration lock, before pushing.
160type mirrorPlan struct {
161 meta domain.Repo
162 target domain.MirrorTarget
163 prev domain.MirrorStatus
164 start time.Time
165 generation uint64
166}
167
168// plan reads the target and status and applies the skip and throttle rules. When done is set
169// the attempt ends with result and err.
170func (m *Mirrors) plan(ctx context.Context, repo domain.RepoName, req mirrorReq) (pl mirrorPlan, result ports.MirrorSyncResult, done bool, err error) {
171 m.configMu.RLock()
172 defer m.configMu.RUnlock()
173 meta, err := m.cfg.Meta.Get(ctx, repo)
174 if err != nil {
175 return pl, result, true, err
176 }
177 t, err := m.target(ctx, repo)
178 if err != nil {
179 return pl, result, true, err
180 }
181 prev, err := m.status(ctx, repo)
182 if err != nil {
183 return pl, result, true, err
184 }
185 now := m.cfg.Clock.Now().UTC()
186 if !t.Enabled {
187 s := domain.MirrorDisable(prev, now)
188 if prev.State != domain.MirrorDisabled {
189 err = m.cfg.Store.PutStatus(ctx, s)
190 }
191 return pl, ports.MirrorSyncResult{Status: s}, true, err
192 }
193 if !req.force && (prev.State == domain.MirrorPaused || (!req.manual && prev.State == domain.MirrorFailing && now.Before(prev.NextRetryAt))) {
194 return pl, ports.MirrorSyncResult{Status: prev}, true, nil
195 }
196 if req.manual && !req.force && !prev.LastAttemptAt.IsZero() && now.Sub(prev.LastAttemptAt) < domain.MirrorManualInterval {
197 return pl, ports.MirrorSyncResult{Status: prev}, true, fmt.Errorf("%w: last sync %s ago; wait %s or use --force", domain.ErrResourceExhausted, now.Sub(prev.LastAttemptAt).Round(time.Second), domain.MirrorManualInterval)
198 }
199 m.mu.Lock()
200 generation := m.gens[repo].dirty
201 m.mu.Unlock()
202 return mirrorPlan{meta: meta, target: t, prev: prev, start: now, generation: generation}, result, false, nil
203}
204
205// attempt runs one sync. The configuration lock is held to read and to record, never while
206// pushing: a slow remote must not block configuration changes.
207func (m *Mirrors) attempt(ctx context.Context, repo domain.RepoName, req mirrorReq) (ports.MirrorSyncResult, error) {
208 pl, result, done, err := m.plan(ctx, repo, req)
209 if done || err != nil {
210 return result, err
211 }
212 t := pl.target
213 // A missing sealer or an unreadable key fails the attempt before any repository I/O.
214 key, runErr := m.privateKey(ctx)
215 var refs []domain.Ref
216 var selected []domain.MirrorRef
217 if runErr == nil {
218 refs, runErr = m.cfg.Reader.ListRefs(ctx, repo)
219 }
220 if runErr == nil {
221 var branch string
222 branch, _, runErr = m.cfg.Git.DefaultBranch(ctx, repo)
223 if runErr == nil {
224 selected = domain.MirrorRefspecs(refs, branch, t.AllBranches)
225 }
226 }
227 noOp := false
228 if runErr == nil && !req.force {
229 snapshot, e := m.cfg.Store.GetSnapshot(ctx, repo)
230 if e != nil && !errors.Is(e, domain.ErrNotFound) {
231 return ports.MirrorSyncResult{}, e
232 }
233 noOp = e == nil && snapshot.Equal(t.URL, selected, t.AllBranches)
234 }
235 if runErr == nil && !noOp {
236 specs := make([]string, len(selected))
237 for i, ref := range selected {
238 specs[i] = ref.Name + ":" + ref.Name
239 }
240 // An empty ref list must not invoke git's implicit default push behavior.
241 if len(specs) > 0 {
242 _, runErr = m.cfg.Pusher.Push(ctx, ports.MirrorPush{Repo: repo, URL: t.URL, Refspecs: specs, PrivateKey: key})
243 }
244 }
245 if ctx.Err() != nil {
246 return ports.MirrorSyncResult{}, ctx.Err()
247 }
248 var failure *domain.MirrorError
249 if runErr != nil && !errors.As(runErr, &failure) {
250 failure = &domain.MirrorError{Class: domain.MirrorClassInternal, Detail: "local mirror operation failed"}
251 runErr = failure
252 }
253 return m.record(ctx, repo, pl, selected, noOp, failure, runErr)
254}
255
256// record stores the outcome, unless the effective target changed during the push: the result
257// then describes a target that no longer applies and is discarded. The comparison includes the
258// cron on purpose, like resetIfChanged: a schedule change during a push costs one extra push.
259func (m *Mirrors) record(ctx context.Context, repo domain.RepoName, pl mirrorPlan, selected []domain.MirrorRef, noOp bool, failure *domain.MirrorError, runErr error) (ports.MirrorSyncResult, error) {
260 m.configMu.RLock()
261 defer m.configMu.RUnlock()
262 t, err := m.target(ctx, repo)
263 if err != nil {
264 return ports.MirrorSyncResult{}, err
265 }
266 if !sameMirrorTarget(pl.target, t) {
267 s, err := m.status(ctx, repo)
268 return ports.MirrorSyncResult{Status: s}, err
269 }
270 // The push may be slow: completion, retry and success times come from the
271 // clock after it returns; only LastAttemptAt marks when the attempt began.
272 now := m.cfg.Clock.Now().UTC()
273 transition := domain.ApplyMirrorOutcome(pl.prev, domain.MirrorOutcome{Err: failure, NoOp: noOp}, now)
274 transition.Status.LastAttemptAt = pl.start
275 if failure == nil {
276 if err := m.cfg.Store.PutSnapshot(ctx, repo, domain.MirrorSnapshot{URL: t.URL, Refs: selected, AllBranches: t.AllBranches, At: now}); err != nil {
277 return ports.MirrorSyncResult{}, err
278 }
279 }
280 if err := m.cfg.Store.PutStatus(ctx, transition.Status); err != nil {
281 return ports.MirrorSyncResult{}, err
282 }
283 if failure == nil {
284 m.mu.Lock()
285 g := m.gens[repo]
286 g.clean, g.cleanKnown = pl.generation, true
287 m.gens[repo] = g
288 m.mu.Unlock()
289 }
290 if m.cfg.Bus != nil {
291 if transition.FailedEvent {
292 m.cfg.Bus.Publish(ctx, domain.MirrorFailed{Repo: repo, State: transition.Status.State, Class: transition.Status.Class, At: now, Visibility: pl.meta.Visibility, CreatedAt: pl.meta.CreatedAt})
293 }
294 if transition.Recovered {
295 m.cfg.Bus.Publish(ctx, domain.MirrorRecovered{Repo: repo, At: now, Visibility: pl.meta.Visibility, CreatedAt: pl.meta.CreatedAt})
296 }
297 }
298 return ports.MirrorSyncResult{Status: transition.Status, NoOp: noOp}, runErr
299}