mirrors_schedule.go RawHistoryFolder

1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package usecase
4
5import (
6	"context"
7	"errors"
8	"time"
9
10	"bnema.dev/ermite/internal/domain"
11	"github.com/bnema/zerowrap"
12)
13
14// Start starts the bounded worker pool and UTC scheduler exactly once.
15func (m *Mirrors) Start() {
16	m.start.Do(func() {
17		m.mu.Lock()
18		defer m.mu.Unlock()
19		if m.closed {
20			return
21		}
22		m.wg.Add(m.cfg.Workers + 1)
23		for i := 0; i < m.cfg.Workers; i++ {
24			go func() {
25				defer m.wg.Done()
26				for {
27					select {
28					case <-m.ctx.Done():
29						return
30					case repo := <-m.jobs:
31						m.run(repo)
32					}
33				}
34			}()
35		}
36		go func() {
37			defer m.wg.Done()
38			ticker := time.NewTicker(m.cfg.ScheduleTick)
39			defer ticker.Stop()
40			for {
41				select {
42				case <-m.ctx.Done():
43					return
44				case <-ticker.C:
45					m.ScheduleNow(m.ctx)
46				}
47			}
48		}()
49	})
50}
51
52// Close cancels workers and waits until they stop or the caller's deadline expires.
53func (m *Mirrors) Close(ctx context.Context) {
54	m.mu.Lock()
55	m.closed = true
56	m.cancel()
57	m.mu.Unlock()
58	done := make(chan struct{})
59	go func() { m.wg.Wait(); close(done) }()
60	select {
61	case <-ctx.Done():
62	case <-done:
63	}
64}
65
66// OnPush marks the repository dirty without I/O or waiting for a worker.
67func (m *Mirrors) OnPush(_ context.Context, repo domain.RepoName, _ []domain.RefUpdate) {
68	m.mu.Lock()
69	defer m.mu.Unlock()
70	if !m.closed {
71		g := m.gens[repo]
72		g.dirty++
73		m.gens[repo] = g
74	}
75}
76
77// ScheduleNow examines the current minute; missed minutes are not replayed. A repository that
78// was due but found the queue full is kept and retried on the next calls, so a full queue
79// delays a sync instead of skipping a period. Workers must be running (Start).
80func (m *Mirrors) ScheduleNow(ctx context.Context) {
81	m.scheduleMu.Lock()
82	defer m.scheduleMu.Unlock()
83	now := m.cfg.Clock.Now().UTC()
84	minute := now.Truncate(time.Minute)
85	newMinute := minute.After(m.lastMinute)
86	if !newMinute && len(m.deferred) == 0 {
87		return
88	}
89	log := zerowrap.FromCtx(ctx)
90	repos, err := m.cfg.Meta.List(ctx)
91	if err != nil {
92		log.Warn().Err(err).Msg("mirror schedule: cannot list repositories")
93		return
94	}
95	if newMinute {
96		m.lastMinute = minute
97	}
98	live := make(map[domain.RepoName]bool, len(repos))
99	for _, r := range repos {
100		if ctx.Err() != nil {
101			return
102		}
103		live[r.Name] = true
104		_, wasDeferred := m.deferred[r.Name]
105		if !newMinute && !wasDeferred {
106			continue
107		}
108		delete(m.deferred, r.Name)
109		t, err := m.target(ctx, r.Name)
110		if err != nil || !t.Enabled {
111			continue
112		}
113		s, err := m.status(ctx, r.Name)
114		if err != nil || s.State == domain.MirrorPaused {
115			continue
116		}
117		due := wasDeferred
118		if !due {
119			if s.State == domain.MirrorFailing {
120				due = !now.Before(s.NextRetryAt)
121			} else {
122				due = (domain.CronSchedule{Cron: t.Cron, Loc: time.UTC}).Due(minute)
123			}
124		}
125		if !due {
126			continue
127		}
128		m.mu.Lock()
129		g := m.gens[r.Name]
130		m.mu.Unlock()
131		if s.State == domain.MirrorOK && g.cleanKnown && g.clean == g.dirty {
132			snapshot, e := m.cfg.Store.GetSnapshot(ctx, r.Name)
133			if e == nil && snapshot.URL == t.URL && snapshot.AllBranches == t.AllBranches {
134				continue
135			}
136		}
137		if _, _, err := m.enqueue(r.Name, mirrorReq{}); errors.Is(err, domain.ErrResourceExhausted) {
138			m.deferred[r.Name] = struct{}{}
139			log.Warn().Str("repo", r.Name.String()).Msg("mirror queue full: sync deferred")
140		}
141	}
142	for repo := range m.deferred {
143		if !live[repo] {
144			delete(m.deferred, repo)
145		}
146	}
147	m.mu.Lock()
148	for repo := range m.gens {
149		if !live[repo] {
150			delete(m.gens, repo)
151		}
152	}
153	m.mu.Unlock()
154}