1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package usecase
4
5import (
6 "context"
7 "errors"
8 "fmt"
9 "runtime/debug"
10 "slices"
11 "strings"
12 "sync"
13 "time"
14
15 "github.com/bnema/zerowrap"
16
17 "bnema.dev/ermite-actions/workflow"
18 "bnema.dev/ermite/internal/domain"
19)
20
21// The schedule event (CI-3, D5). A scheduler goroutine looks, once per minute of
22// its clock, at the "cron" entries of the workflows of every repository's default
23// branch and starts the ones that match that minute (UTC; "timezone" is ignored).
24//
25// - The run is on the latest commit of the default branch. Its actor and its
26// trust are those of whoever pushed that commit (ci_tips.go), as on GitHub:
27// the default branch is not necessarily the owner's work.
28// - A minute is looked at once, when the clock first reaches it. Ticks missed
29// while ermite was down or busy are not made up: a pass that is still going
30// when the next minute starts is dropped, as GitHub drops late runs.
31// - The schedules of a repository are read once and cached with the commit
32// they were read at. A push (of any principal, a job token's included, which
33// starts no run but may change the file) marks the repository, and the
34// cache is dropped when its default branch moved. A repository that no
35// longer exists is forgotten at the next pass.
36// - The work is bounded: at most ScheduleBudget repositories are looked at per
37// tick, so a pass over many repositories spreads over several ticks (and
38// never starts a run late for a minute that is over). A pass dropped
39// before its end leaves a cursor: the next pass starts after the last
40// repository handled, so that no repository is starved.
41//
42// The cron entries of one workflow that match the same minute start one run of
43// it (the first in file order); the event payload is that entry's text.
44
45const (
46 // DefaultCIScheduleTick is how often the scheduler looks at its clock.
47 DefaultCIScheduleTick = time.Second
48 // DefaultCIScheduleBudget is the repositories the scheduler handles at most
49 // per tick.
50 DefaultCIScheduleBudget = 64
51)
52
53// scheduledCron is one cron entry of a workflow.
54type scheduledCron struct {
55 spec string
56 cron domain.Cron
57}
58
59// scheduledWorkflow is a workflow with at least one valid cron entry.
60type scheduledWorkflow struct {
61 path string
62 yaml []byte
63 crons []scheduledCron
64}
65
66// scheduleEntry is what was read from one repository at tip.
67type scheduleEntry struct {
68 tip domain.ObjectID
69 files []scheduledWorkflow
70}
71
72// schedulePass is the look at one minute over the repositories.
73type schedulePass struct {
74 minute time.Time
75 repos []domain.RepoName
76 next int
77}
78
79type scheduler struct {
80 mu sync.Mutex
81 dirty map[domain.RepoName]struct{}
82 cache map[domain.RepoName]*scheduleEntry
83 // passed is the minute of the last pass that went through every repository.
84 passed time.Time
85
86 // The fields below belong to the scheduler goroutine.
87 last time.Time // the latest minute a pass was started for
88 pass *schedulePass
89 // cursor is the last repository a pass handled when it was dropped before
90 // the end; the next pass starts after it, so that repositories beyond the
91 // budget of a tick are not starved by the ones that come first. Zero when
92 // the last pass went through every repository.
93 cursor domain.RepoName
94}
95
96func newScheduler(boot time.Time) *scheduler {
97 return &scheduler{
98 dirty: map[domain.RepoName]struct{}{}, cache: map[domain.RepoName]*scheduleEntry{},
99 // The minute ermite starts in belongs to the process before it.
100 last: boot.UTC().Truncate(time.Minute),
101 }
102}
103
104func (s *scheduler) get(repo domain.RepoName) *scheduleEntry {
105 s.mu.Lock()
106 defer s.mu.Unlock()
107 return s.cache[repo]
108}
109
110func (s *scheduler) put(repo domain.RepoName, e *scheduleEntry) {
111 s.mu.Lock()
112 defer s.mu.Unlock()
113 s.cache[repo] = e
114}
115
116func (s *scheduler) drop(repo domain.RepoName) {
117 s.mu.Lock()
118 defer s.mu.Unlock()
119 delete(s.cache, repo)
120}
121
122// markDirty remembers that the default branch of a repository may have moved.
123func (s *scheduler) markDirty(repo domain.RepoName) {
124 s.mu.Lock()
125 defer s.mu.Unlock()
126 if _, cached := s.cache[repo]; cached {
127 s.dirty[repo] = struct{}{}
128 }
129}
130
131func (s *scheduler) forgetDirty(repo domain.RepoName) {
132 s.mu.Lock()
133 defer s.mu.Unlock()
134 delete(s.dirty, repo)
135}
136
137// takeDirty removes and returns at most n marked repositories.
138func (s *scheduler) takeDirty(n int) []domain.RepoName {
139 s.mu.Lock()
140 defer s.mu.Unlock()
141 var out []domain.RepoName
142 for r := range s.dirty {
143 if len(out) == n {
144 break
145 }
146 out = append(out, r)
147 delete(s.dirty, r)
148 }
149 return out
150}
151
152// prune forgets the repositories that are not in keep.
153func (s *scheduler) prune(keep []domain.RepoName) {
154 live := make(map[domain.RepoName]struct{}, len(keep))
155 for _, r := range keep {
156 live[r] = struct{}{}
157 }
158 s.mu.Lock()
159 defer s.mu.Unlock()
160 for r := range s.cache {
161 if _, ok := live[r]; !ok {
162 delete(s.cache, r)
163 }
164 }
165 for r := range s.dirty {
166 if _, ok := live[r]; !ok {
167 delete(s.dirty, r)
168 }
169 }
170}
171
172// CachedSchedules is the number of scheduled workflows the scheduler holds for a
173// repository, -1 when it holds nothing for it. For tests and diagnostics.
174func (c *CI) CachedSchedules(repo domain.RepoName) int {
175 if e := c.sched.get(repo); e != nil {
176 return len(e.files)
177 }
178 return -1
179}
180
181// SchedulePassed is the minute of the last pass of the scheduler that went
182// through every repository (zero before the first). For tests and diagnostics.
183func (c *CI) SchedulePassed() time.Time {
184 c.sched.mu.Lock()
185 defer c.sched.mu.Unlock()
186 return c.sched.passed
187}
188
189// markSchedulesDirty tells the scheduler that these updates may have moved the
190// default branch of the repository.
191func (c *CI) markSchedulesDirty(repo domain.RepoName, updates []domain.RefUpdate) {
192 for _, u := range updates {
193 if strings.HasPrefix(u.Ref, "refs/heads/") {
194 c.sched.markDirty(repo)
195 return
196 }
197 }
198}
199
200// ScheduleNow does one tick of the scheduler on the calling goroutine, with the
201// clock of the configuration. It is for tests (with a ScheduleTick too long to
202// fire) and operators' tooling; the server never needs to call it. It must not
203// run at the same time as the scheduler goroutine.
204func (c *CI) ScheduleNow(ctx context.Context) {
205 c.guardSchedule(func(context.Context) { c.scheduleTick(ctx) })
206}
207
208func (c *CI) scheduleLoop() {
209 defer c.wg.Done()
210 t := time.NewTicker(c.cfg.ScheduleTick)
211 defer t.Stop()
212 for {
213 select {
214 case <-c.ctx.Done():
215 return
216 case <-t.C:
217 }
218 c.guardSchedule(c.scheduleTick)
219 }
220}
221
222func (c *CI) guardSchedule(fn func(context.Context)) {
223 log := zerowrap.FromCtx(c.ctx)
224 defer func() {
225 if r := recover(); r != nil {
226 log.Error().Interface("panic", r).Bytes("stack", debug.Stack()).Msg("ci scheduler panicked")
227 }
228 }()
229 fn(c.ctx)
230}
231
232// scheduleTick does one tick of work: it checks the repositories whose default
233// branch may have moved, starts a pass when the clock reached a new minute, and
234// carries on the pass within the budget.
235func (c *CI) scheduleTick(ctx context.Context) {
236 s := c.sched
237 budget := c.cfg.ScheduleBudget
238 for _, repo := range s.takeDirty(budget) {
239 budget--
240 if s.get(repo) != nil && !c.scheduleCurrent(ctx, repo) {
241 s.drop(repo)
242 }
243 }
244 if minute := c.now().Truncate(time.Minute); minute.After(s.last) {
245 // A pass still going belongs to a minute that is over: it is dropped, and
246 // so is every minute between.
247 s.last = minute
248 s.pass = c.startPass(ctx, minute)
249 }
250 p := s.pass
251 for p != nil && p.next < len(p.repos) && budget > 0 && ctx.Err() == nil {
252 repo := p.repos[p.next]
253 p.next++
254 s.cursor = repo
255 budget--
256 if c.cfg.ScheduleVisit != nil {
257 c.cfg.ScheduleVisit(repo)
258 }
259 if err := c.scheduleRepo(ctx, repo, p.minute); err != nil && ctx.Err() == nil {
260 ctxLog(ctx).Error().Err(err).Str("repo", repo.String()).Msg("ci schedule failed")
261 }
262 }
263 if p != nil && p.next >= len(p.repos) {
264 s.mu.Lock()
265 s.passed = p.minute
266 s.mu.Unlock()
267 s.pass, s.cursor = nil, domain.RepoName{}
268 }
269}
270
271// startPass lists the repositories to look at for a minute and forgets the
272// ones that are gone. Without a runner CI is off and nothing is looked at.
273func (c *CI) startPass(ctx context.Context, minute time.Time) *schedulePass {
274 p := &schedulePass{minute: minute}
275 repos, err := c.cfg.Meta.List(ctx)
276 if err != nil {
277 ctxLog(ctx).Error().Err(err).Msg("ci schedule: list the repositories")
278 return p
279 }
280 names := make([]domain.RepoName, 0, len(repos))
281 for _, r := range repos {
282 names = append(names, r.Name)
283 }
284 c.sched.prune(names)
285 if runners, err := c.cfg.Store.ListRunners(ctx); err != nil || len(runners) == 0 {
286 return p
287 }
288 // Start after the repository the dropped pass stopped at (the list is in
289 // name order), and wrap around.
290 start := 0
291 if !c.sched.cursor.IsZero() {
292 for start < len(names) && names[start].String() <= c.sched.cursor.String() {
293 start++
294 }
295 }
296 p.repos = append(slices.Clone(names[start:]), names[:start]...)
297 return p
298}
299
300// scheduleCurrent reports whether the cached schedules of a repository were
301// read at the tip of its default branch as it is now.
302func (c *CI) scheduleCurrent(ctx context.Context, repo domain.RepoName) bool {
303 e := c.sched.get(repo)
304 if e == nil {
305 return false
306 }
307 tip, ok := c.defaultTip(ctx, repo)
308 return ok && tip == e.tip
309}
310
311// defaultTip is the latest commit of the default branch.
312func (c *CI) defaultTip(ctx context.Context, repo domain.RepoName) (domain.ObjectID, bool) {
313 def, exists, err := c.cfg.Repos.DefaultBranch(ctx, repo)
314 if err != nil || !exists {
315 return domain.ObjectID{}, false
316 }
317 tip, err := c.cfg.Git.BranchHead(ctx, repo, def)
318 if err != nil {
319 return domain.ObjectID{}, false
320 }
321 return tip, true
322}
323
324// loadSchedules reads the cron entries of the workflows at the tip of the
325// default branch. A workflow that does not parse is skipped (the push of the
326// file reported it).
327func (c *CI) loadSchedules(ctx context.Context, repo domain.RepoName) (*scheduleEntry, error) {
328 tip, ok := c.defaultTip(ctx, repo)
329 if !ok {
330 return &scheduleEntry{}, nil // no repository, no default branch, nothing to schedule
331 }
332 e := &scheduleEntry{tip: tip}
333 files, err := c.workflowFiles(ctx, repo, tip)
334 if err != nil {
335 return nil, err
336 }
337 for _, f := range files {
338 if f.problem != "" {
339 continue
340 }
341 wf, perrs := workflow.Parse(f.path, f.content)
342 if len(perrs) > 0 || wf.On == nil {
343 continue
344 }
345 ev := wf.On.Event(eventSchedule)
346 if ev == nil {
347 continue
348 }
349 sw := scheduledWorkflow{path: f.path, yaml: f.content}
350 for _, sc := range ev.Schedule {
351 cron, err := domain.ParseCron(sc.Cron)
352 if err != nil {
353 continue
354 }
355 sw.crons = append(sw.crons, scheduledCron{spec: sc.Cron, cron: cron})
356 }
357 if len(sw.crons) > 0 {
358 e.files = append(e.files, sw)
359 }
360 }
361 return e, nil
362}
363
364// dueFor returns, for each workflow of the entry that has an entry matching the
365// minute, the text of the first one that does.
366func dueFor(e *scheduleEntry, minute time.Time) (idx []int, specs []string) {
367 for i, f := range e.files {
368 for _, sc := range f.crons {
369 if sc.cron.Matches(minute) {
370 idx, specs = append(idx, i), append(specs, sc.spec)
371 break
372 }
373 }
374 }
375 return idx, specs
376}
377
378// scheduleRepo starts the scheduled workflows of a repository that are due at
379// the minute.
380func (c *CI) scheduleRepo(ctx context.Context, repo domain.RepoName, minute time.Time) error {
381 // The cache is of the commit the run will be on: it is read again when the
382 // default branch moved (a push marked it, or the default branch was changed).
383 tip, ok := c.defaultTip(ctx, repo)
384 if !ok {
385 c.sched.drop(repo)
386 return nil
387 }
388 e := c.sched.get(repo)
389 if e == nil || e.tip != tip {
390 var err error
391 if e, err = c.loadSchedules(ctx, repo); err != nil {
392 return fmt.Errorf("read the schedules: %w", err)
393 }
394 c.sched.put(repo, e)
395 }
396 idx, specs := dueFor(e, minute)
397 if len(idx) == 0 {
398 return nil
399 }
400 meta, err := c.cfg.Meta.Get(ctx, repo)
401 if err != nil {
402 c.sched.drop(repo)
403 return nil // deleted meanwhile
404 }
405 def, _, err := c.cfg.Repos.DefaultBranch(ctx, repo)
406 if err != nil {
407 return nil
408 }
409 commits, _, err := c.cfg.Git.ListRange(ctx, repo, domain.ObjectID{}, e.tip, 1)
410 if err != nil || len(commits) == 0 {
411 return fmt.Errorf("read the head commit of %s: %w", def, err)
412 }
413 var errs []error
414 for n, i := range idx {
415 errs = append(errs, c.fireSchedule(ctx, meta, def, e, commits[0], e.files[i], specs[n]))
416 }
417 return errors.Join(errs...)
418}
419
420// fireSchedule starts the run of one scheduled workflow.
421func (c *CI) fireSchedule(ctx context.Context, repo domain.Repo, def string, e *scheduleEntry, head domain.Commit, sw scheduledWorkflow, spec string) error {
422 wf, perrs := workflow.Parse(sw.path, sw.yaml)
423 if len(perrs) > 0 {
424 return nil
425 }
426 facts := pushFacts{
427 update: domain.RefUpdate{Ref: "refs/heads/" + def, Old: domain.ZeroOID, New: e.tip.String()}, commit: e.tip, head: head,
428 defaultBranch: def, custom: &customEvent{name: eventSchedule, cron: spec},
429 }
430 // Whoever moved the default branch to this commit, with their trust
431 // (ci_tips.go); the owner's name and no trust when that is not known.
432 actor, pusher, trusted := c.scheduleActor(ctx, repo.Name, def, e.tip)
433 t := ciTrigger{repo: repo.Name, actor: actor, pusher: pusher, trusted: trusted, at: c.now()}
434 if problem := c.matrixProblem(ctx, repo, t, facts, sw.path, wf); problem.text != "" {
435 name := sw.path
436 if problem.named {
437 name = workflowName(wf, sw.path)
438 }
439 return c.startInvalid(ctx, repo, t, facts, sw.path, name, problem.text)
440 }
441 _, err := c.launchRun(ctx, repo, t, facts, sw.path, string(sw.yaml), wf)
442 return err
443}