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