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// DefaultMirrorSyncWait is how long a manual sync holds its caller before answering "queued":
40// under the 10 s the edge proxy allows a response, and well under the client timeout.
41const DefaultMirrorSyncWait = 8 * time.Second
42
43// Sync waits for its own run; overlapping requests coalesce into one sticky rerun. When the
44// push outlasts the wait bound the result is Queued: the sync goes on in the background and
45// its outcome is in the mirror status.
46func (m *Mirrors) Sync(ctx context.Context, p domain.Principal, repo domain.RepoName, force bool) (ports.MirrorSyncResult, error) {
47 if err := m.admin(ctx, p, repo); err != nil {
48 return ports.MirrorSyncResult{}, err
49 }
50 m.Start()
51 f, queued, err := m.enqueue(repo, mirrorReq{force: force, manual: true})
52 if err != nil {
53 return ports.MirrorSyncResult{}, err
54 }
55 if queued {
56 return pendingSync(repo), nil
57 }
58 wctx, cancel := context.WithTimeout(ctx, m.cfg.SyncWait)
59 defer cancel()
60 return m.await(ctx, wctx, repo, f)
61}
62
63func pendingSync(repo domain.RepoName) ports.MirrorSyncResult {
64 return ports.MirrorSyncResult{Repo: repo, Status: domain.MirrorStatus{Repo: repo}, Queued: true}
65}
66
67// await waits for the first run of f. wctx carries the wait bound and ctx the caller's own
68// cancellation: the bound expiring alone yields a Queued result, not an error.
69func (m *Mirrors) await(ctx, wctx context.Context, repo domain.RepoName, f *mirrorFlight) (ports.MirrorSyncResult, error) {
70 select {
71 case <-m.ctx.Done():
72 return ports.MirrorSyncResult{}, m.ctx.Err()
73 case <-wctx.Done():
74 if err := ctx.Err(); err != nil {
75 return ports.MirrorSyncResult{}, err
76 }
77 select { // a run that finished as the bound expired wins
78 case <-f.done:
79 default:
80 return pendingSync(repo), nil
81 }
82 case <-f.done:
83 }
84 // run writes result and err before close(f.done): the close orders them before this read.
85 res, err := f.result, f.err
86 res.Repo = repo
87 return res, err
88}
89
90// SyncAll syncs every repository whose mirror is enabled, as an owner. Every repository is
91// enqueued first, then all outcomes are awaited together under one wait bound, so the total
92// time is about one push, not one per repository. A failing repository does not stop the
93// others: its outcome carries the error; one still running at the bound is Queued. The
94// returned error is only for what prevents the whole request (authorization, listing,
95// cancellation). Results are in listing order.
96func (m *Mirrors) SyncAll(ctx context.Context, p domain.Principal, force bool) ([]ports.MirrorSyncResult, error) {
97 if err := mirrorOwner(p); err != nil {
98 return nil, err
99 }
100 repos, err := m.cfg.Meta.List(ctx)
101 if err != nil {
102 return nil, err
103 }
104 m.Start()
105 type entry struct {
106 res ports.MirrorSyncResult
107 f *mirrorFlight // nil once the outcome is known
108 }
109 entries := make([]entry, 0, len(repos))
110 for _, r := range repos {
111 if err := ctx.Err(); err != nil {
112 return nil, err
113 }
114 t, err := m.target(ctx, r.Name)
115 if err == nil && !t.Enabled {
116 continue
117 }
118 e := entry{res: ports.MirrorSyncResult{Repo: r.Name, Err: err}}
119 if err == nil {
120 f, queued, qerr := m.enqueue(r.Name, mirrorReq{force: force, manual: true})
121 switch {
122 case qerr != nil:
123 e.res.Err = qerr
124 case queued:
125 e.res = pendingSync(r.Name)
126 default:
127 e.f = f
128 }
129 }
130 entries = append(entries, e)
131 }
132 wctx, cancel := context.WithTimeout(ctx, m.cfg.SyncWait)
133 defer cancel()
134 out := make([]ports.MirrorSyncResult, 0, len(entries))
135 for _, e := range entries {
136 if e.f != nil {
137 res, err := m.await(ctx, wctx, e.res.Repo, e.f)
138 if cerr := ctx.Err(); cerr != nil {
139 return nil, cerr
140 }
141 e.res, e.res.Err = res, err
142 }
143 if e.res.Status.Repo == (domain.RepoName{}) {
144 e.res.Status.Repo = e.res.Repo
145 }
146 out = append(out, e.res)
147 }
148 return out, nil
149}
150
151// Wait blocks until no sync is queued or running, or ctx is done. It is a test barrier: call it
152// only while nothing enqueues new syncs and before Close (queued jobs are not drained by Close).
153func (m *Mirrors) Wait(ctx context.Context) {
154 done := make(chan struct{})
155 go func() { m.flightWG.Wait(); close(done) }()
156 select {
157 case <-done:
158 case <-ctx.Done():
159 }
160}
161
162func (m *Mirrors) enqueue(repo domain.RepoName, req mirrorReq) (*mirrorFlight, bool, error) {
163 m.mu.Lock()
164 defer m.mu.Unlock()
165 if m.closed {
166 return nil, false, context.Canceled
167 }
168 if f := m.flights[repo]; f != nil {
169 next := req
170 if f.next != nil {
171 next = f.next.merge(req)
172 }
173 f.next = &next
174 return f, true, nil
175 }
176 f := &mirrorFlight{cur: req, done: make(chan struct{})}
177 select {
178 case m.jobs <- repo:
179 m.flights[repo] = f
180 m.flightWG.Add(1)
181 return f, false, nil
182 default:
183 return nil, false, domain.ErrResourceExhausted
184 }
185}
186
187func (m *Mirrors) run(repo domain.RepoName) {
188 m.mu.Lock()
189 f := m.flights[repo]
190 m.mu.Unlock()
191 for {
192 result, err := m.attempt(m.ctx, repo, f.cur)
193 m.mu.Lock()
194 if !f.completed {
195 f.result, f.err = result, err
196 f.completed = true
197 close(f.done)
198 }
199 if f.next == nil || m.closed {
200 delete(m.flights, repo)
201 m.flightWG.Done()
202 m.mu.Unlock()
203 return
204 }
205 f.cur, f.next = *f.next, nil
206 m.mu.Unlock()
207 }
208}
209
210// mirrorPlan is what attempt decided under the configuration lock, before pushing.
211type mirrorPlan struct {
212 meta domain.Repo
213 target domain.MirrorTarget
214 prev domain.MirrorStatus
215 start time.Time
216 generation uint64
217}
218
219// plan reads the target and status and applies the skip and throttle rules. When done is set
220// the attempt ends with result and err.
221func (m *Mirrors) plan(ctx context.Context, repo domain.RepoName, req mirrorReq) (pl mirrorPlan, result ports.MirrorSyncResult, done bool, err error) {
222 m.configMu.RLock()
223 defer m.configMu.RUnlock()
224 meta, err := m.cfg.Meta.Get(ctx, repo)
225 if err != nil {
226 return pl, result, true, err
227 }
228 t, err := m.target(ctx, repo)
229 if err != nil {
230 return pl, result, true, err
231 }
232 prev, err := m.status(ctx, repo)
233 if err != nil {
234 return pl, result, true, err
235 }
236 now := m.cfg.Clock.Now().UTC()
237 if !t.Enabled {
238 s := domain.MirrorDisable(prev, now)
239 if prev.State != domain.MirrorDisabled {
240 err = m.cfg.Store.PutStatus(ctx, s)
241 }
242 return pl, ports.MirrorSyncResult{Status: s}, true, err
243 }
244 if !req.force && (prev.State == domain.MirrorPaused || (!req.manual && prev.State == domain.MirrorFailing && now.Before(prev.NextRetryAt))) {
245 return pl, ports.MirrorSyncResult{Status: prev}, true, nil
246 }
247 if req.manual && !req.force && !prev.LastAttemptAt.IsZero() && now.Sub(prev.LastAttemptAt) < domain.MirrorManualInterval {
248 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)
249 }
250 m.mu.Lock()
251 generation := m.gens[repo].dirty
252 m.mu.Unlock()
253 return mirrorPlan{meta: meta, target: t, prev: prev, start: now, generation: generation}, result, false, nil
254}
255
256// attempt runs one sync. The configuration lock is held to read and to record, never while
257// pushing: a slow remote must not block configuration changes.
258func (m *Mirrors) attempt(ctx context.Context, repo domain.RepoName, req mirrorReq) (ports.MirrorSyncResult, error) {
259 pl, result, done, err := m.plan(ctx, repo, req)
260 if done || err != nil {
261 return result, err
262 }
263 t := pl.target
264 // A missing sealer or an unreadable key fails the attempt before any repository I/O.
265 key, runErr := m.privateKey(ctx)
266 var refs []domain.Ref
267 var selected []domain.MirrorRef
268 if runErr == nil {
269 refs, runErr = m.cfg.Reader.ListRefs(ctx, repo)
270 }
271 if runErr == nil {
272 var branch string
273 branch, _, runErr = m.cfg.Git.DefaultBranch(ctx, repo)
274 if runErr == nil {
275 selected = domain.MirrorRefspecs(refs, branch, t.AllBranches)
276 }
277 }
278 noOp := false
279 if runErr == nil && !req.force {
280 snapshot, e := m.cfg.Store.GetSnapshot(ctx, repo)
281 if e != nil && !errors.Is(e, domain.ErrNotFound) {
282 return ports.MirrorSyncResult{}, e
283 }
284 noOp = e == nil && snapshot.Equal(t.URL, selected, t.AllBranches)
285 }
286 if runErr == nil && !noOp {
287 specs := make([]string, len(selected))
288 for i, ref := range selected {
289 specs[i] = ref.Name + ":" + ref.Name
290 }
291 // An empty ref list must not invoke git's implicit default push behavior.
292 if len(specs) > 0 {
293 _, runErr = m.cfg.Pusher.Push(ctx, ports.MirrorPush{Repo: repo, URL: t.URL, Refspecs: specs, PrivateKey: key})
294 }
295 }
296 if ctx.Err() != nil {
297 return ports.MirrorSyncResult{}, ctx.Err()
298 }
299 var failure *domain.MirrorError
300 if runErr != nil && !errors.As(runErr, &failure) {
301 failure = &domain.MirrorError{Class: domain.MirrorClassInternal, Detail: "local mirror operation failed"}
302 runErr = failure
303 }
304 return m.record(ctx, repo, pl, selected, noOp, failure, runErr)
305}
306
307// record stores the outcome, unless the effective target changed during the push: the result
308// then describes a target that no longer applies and is discarded. The comparison includes the
309// cron on purpose, like resetIfChanged: a schedule change during a push costs one extra push.
310func (m *Mirrors) record(ctx context.Context, repo domain.RepoName, pl mirrorPlan, selected []domain.MirrorRef, noOp bool, failure *domain.MirrorError, runErr error) (ports.MirrorSyncResult, error) {
311 m.configMu.RLock()
312 defer m.configMu.RUnlock()
313 t, err := m.target(ctx, repo)
314 if err != nil {
315 return ports.MirrorSyncResult{}, err
316 }
317 if !sameMirrorTarget(pl.target, t) {
318 s, err := m.status(ctx, repo)
319 return ports.MirrorSyncResult{Status: s}, err
320 }
321 // The push may be slow: completion, retry and success times come from the
322 // clock after it returns; only LastAttemptAt marks when the attempt began.
323 now := m.cfg.Clock.Now().UTC()
324 transition := domain.ApplyMirrorOutcome(pl.prev, domain.MirrorOutcome{Err: failure, NoOp: noOp}, now)
325 transition.Status.LastAttemptAt = pl.start
326 if failure == nil {
327 if err := m.cfg.Store.PutSnapshot(ctx, repo, domain.MirrorSnapshot{URL: t.URL, Refs: selected, AllBranches: t.AllBranches, At: now}); err != nil {
328 return ports.MirrorSyncResult{}, err
329 }
330 }
331 if err := m.cfg.Store.PutStatus(ctx, transition.Status); err != nil {
332 return ports.MirrorSyncResult{}, err
333 }
334 if failure == nil {
335 m.mu.Lock()
336 g := m.gens[repo]
337 g.clean, g.cleanKnown = pl.generation, true
338 m.gens[repo] = g
339 m.mu.Unlock()
340 }
341 if m.cfg.Bus != nil {
342 if transition.FailedEvent {
343 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})
344 }
345 if transition.Recovered {
346 m.cfg.Bus.Publish(ctx, domain.MirrorRecovered{Repo: repo, At: now, Visibility: pl.meta.Visibility, CreatedAt: pl.meta.CreatedAt})
347 }
348 }
349 return ports.MirrorSyncResult{Status: transition.Status, NoOp: noOp}, runErr
350}