package ci
import (
"fmt"
"log"
"slices"
"strings"
"sync"
"time"
"craftdesign.group/quarry/internal/cronexpr"
"gopkg.in/yaml.v3"
)
// Scheduled runs are quarry's (RFC 0019's `schedule`, RFC 0021 §runs):
// the engine passes on: through uninterpreted and has no trigger that
// carries a check-in, so the forge reads each repository's default
// branch tip once a minute and launches what is due through the same
// path every other run takes. That launch registers the tip's definition
// just before it starts, so a schedule always runs the default branch's
// file, whatever a branch registered last; GitHub's rule.
//
// Cron lines are GitHub's: five fields, UTC. A minute missed while the
// forge was down is not run later, as GitHub does not.
// minScheduleGap is the fewest minutes between two fires of one
// workflow, GitHub's floor; maxScheduleLines bounds a workflow's lines.
const (
minScheduleGap = 5
maxScheduleLines = 10
maxCronLength = 128
)
// schedulesOf reads a ci.yml's on.schedule cron lines. A line that does
// not parse, a workflow whose lines together fire more often than every
// five minutes, or more than ten lines, is refused with a position,
// like every other key quarry reads. It runs where a workflow is
// compiled and when a default tip's schedules are first read, never per
// tick.
func schedulesOf(spec []byte) ([]string, error) {
doc, ok := yamlDoc(spec)
if !ok {
return nil, nil
}
on := value(doc, "on")
if on == nil || on.Kind != yaml.MappingNode {
return nil, nil
}
list := value(on, "schedule")
if list == nil {
return nil, nil
}
if list.Kind != yaml.SequenceNode {
return nil, fmt.Errorf("line %d, column %d: on.schedule is a list of {cron: \"...\"}", list.Line, list.Column)
}
if len(list.Content) > maxScheduleLines {
return nil, fmt.Errorf("line %d, column %d: on.schedule has %d lines; a workflow may have at most %d",
list.Line, list.Column, len(list.Content), maxScheduleLines)
}
var out []string
minutes := map[int]bool{}
for _, item := range list.Content {
c := value(item, "cron")
if item.Kind != yaml.MappingNode || c == nil || c.Kind != yaml.ScalarNode || len(item.Content) != 2 {
return nil, fmt.Errorf("line %d, column %d: on.schedule: each entry is {cron: \"...\"}", item.Line, item.Column)
}
expr, err := parseCron(c.Value)
if err != nil {
return nil, fmt.Errorf("line %d, column %d: on.schedule: %v", c.Line, c.Column, err)
}
for _, m := range expr.Minutes {
minutes[m] = true
}
if gap := smallestGap(minutes); gap < minScheduleGap {
return nil, fmt.Errorf("line %d, column %d: on.schedule: with %q this workflow fires %d minute(s) apart; the shortest interval a schedule may have is %d minutes",
c.Line, c.Column, c.Value, gap, minScheduleGap)
}
out = append(out, c.Value)
}
return out, nil
}
// smallestGap is the least cyclic distance, in minutes, between two
// minutes of the hour a workflow may fire on. It is judged on the
// minute field alone, whatever the hours and days say: a day the other
// fields allow at all (a leap day, say) fires on every minute the field
// names, so no sample of the calendar can stand in for it.
func smallestGap(minutes map[int]bool) int {
var ms []int
for m := range minutes {
ms = append(ms, m)
}
if len(ms) < 2 {
return 60
}
slices.Sort(ms)
least := 60 - ms[len(ms)-1] + ms[0]
for i := 1; i < len(ms); i++ {
least = min(least, ms[i]-ms[i-1])
}
return least
}
func parseCron(line string) (*cronexpr.CronExpr, error) {
if len(line) > maxCronLength {
return nil, fmt.Errorf("a cron line is at most %d characters", maxCronLength)
}
if len(strings.Fields(line)) != 5 {
return nil, fmt.Errorf("%q is not a cron line of five fields (minute hour day month weekday, UTC)", line)
}
expr, err := cronexpr.ParseCron(line)
if err != nil {
return nil, fmt.Errorf("%q: %v", line, err)
}
return expr, nil
}
// ScheduleStore is what firing needs beyond Store's.
type ScheduleStore interface {
Store
// ClaimSchedule takes one scheduled minute of one pipeline, true
// for the first claimant only: what makes a minute fire at most
// once across ticks, restarts and a lease handover. The claim comes
// before the launch, so a launch that fails loses its minute.
ClaimSchedule(repo, pipeline string, minute time.Time) (bool, error)
// ScheduleRunning reports a scheduled run of pipeline still going.
ScheduleRunning(repo, pipeline string) bool
// ScheduleFailed records why a claimed minute did not start, for the
// schedule's row on the CI page.
ScheduleFailed(repo, pipeline string, minute time.Time, why string)
}
// Scheduled is one scheduled workflow as the default branch tip says.
// Error is why it does not fire, and Next is then zero.
type Scheduled struct {
Pipeline string `json:"pipeline"`
Crons []string `json:"crons"`
Next time.Time `json:"next"`
Error string `json:"error,omitempty"`
// LastFailure is the last claimed minute that did not start, and why.
LastFailure string `json:"lastFailure,omitempty"`
LastFailureAt time.Time `json:"lastFailureAt,omitempty"`
}
// scheduleEntry is one workflow's parsed schedule at a tip.
type scheduleEntry struct {
pipeline string
crons []string
exprs []*cronexpr.CronExpr
err error
}
type scheduleTip struct {
c Checkin
defs definitions
entries []scheduleEntry
}
// ScheduleCache holds each repository's schedules as of its default
// tip, read again only when the tip moves: the tick and the CI page ask
// every minute and on every view, and the answer changes with a push.
type ScheduleCache struct {
mu sync.Mutex
tips map[string]scheduleTip
}
func NewScheduleCache() *ScheduleCache { return &ScheduleCache{tips: map[string]scheduleTip{}} }
// at is repo's schedules as of its default tip now. A workflow that
// cannot be read or is refused is logged once, when its tip is first
// read, and kept with its error.
func (sc *ScheduleCache) at(repo string, r Repo) (scheduleTip, error) {
branch := r.DefaultBranch()
tip := r.BranchTip(branch)
if tip == "" {
return scheduleTip{}, nil
}
sc.mu.Lock()
cached, ok := sc.tips[repo]
sc.mu.Unlock()
if ok && cached.c.Hash == tip {
return cached, nil
}
c := Checkin{Hash: tip, Branch: branch}
defs, err := pipelinesAt(r, c)
if err != nil {
return scheduleTip{}, err
}
st := scheduleTip{c: c, defs: defs}
for _, p := range defs.paths {
if strings.HasSuffix(p, ".json") {
continue
}
def, err := defs.read(p)
if err != nil {
return scheduleTip{}, transient(fmt.Errorf("read %s: %w", p, err))
}
crons, err := schedulesOf(def)
if err != nil {
log.Printf("ci: %s %s at %.12s: schedule not fired: %v", repo, p, tip, err)
st.entries = append(st.entries, scheduleEntry{pipeline: p, err: err})
continue
}
if len(crons) == 0 {
continue
}
e := scheduleEntry{pipeline: p, crons: crons}
for _, line := range crons {
expr, _ := parseCron(line) // schedulesOf parsed it
e.exprs = append(e.exprs, expr)
}
st.entries = append(st.entries, e)
}
sc.mu.Lock()
sc.tips[repo] = st
sc.mu.Unlock()
return st, nil
}
// Schedules lists repo's scheduled workflows at its default tip, each
// with its next fire after now, or the error that keeps it from firing.
func (sc *ScheduleCache) Schedules(repo string, r Repo, now time.Time) ([]Scheduled, error) {
st, err := sc.at(repo, r)
if err != nil {
return nil, err
}
var out []Scheduled
for _, e := range st.entries {
s := Scheduled{Pipeline: e.pipeline, Crons: e.crons}
if e.err != nil {
s.Error = e.err.Error()
} else {
s.Next = nextFire(e.exprs, now)
}
out = append(out, s)
}
return out, nil
}
// due reports whether any of exprs fires in minute (UTC).
func due(exprs []*cronexpr.CronExpr, minute time.Time) bool {
m := minute.UTC().Truncate(time.Minute)
for _, e := range exprs {
if e.Matches(m) {
return true
}
}
return false
}
// nextFire is the first minute after now any of exprs fires, within a
// year; zero beyond.
func nextFire(exprs []*cronexpr.CronExpr, now time.Time) time.Time {
var next time.Time
for _, e := range exprs {
if n := e.NextN(now.UTC(), 1); len(n) == 1 && (next.IsZero() || n[0].Before(next)) {
next = n[0]
}
}
return next
}
// FireSchedules launches repo's default-tip workflows due in any of
// minutes, each at most once per minute and never beside its own
// scheduled run still going. lead is asked before every claim: a node
// that lost the fleet lease mid-round claims nothing more. A workflow
// that is refused, or whose claim or launch fails, is skipped alone.
// The runs have no starter: a run token is minted for them only where
// no starter letter is required.
func FireSchedules(e *Engine, st ScheduleStore, sc *ScheduleCache, repo string, r Repo, cloneURL string, minutes []time.Time, lead func() bool) error {
if !e.Enabled() {
return nil
}
tip, err := sc.at(repo, r)
if err != nil || tip.c.Hash == "" {
return err
}
for _, minute := range minutes {
minute = minute.UTC().Truncate(time.Minute)
for _, en := range tip.entries {
if en.err != nil || !due(en.exprs, minute) {
continue
}
if !lead() {
return nil
}
won, err := st.ClaimSchedule(repo, en.pipeline, minute)
if err != nil {
log.Printf("ci: %s %s: claiming the %s schedule: %v", repo, en.pipeline, minute.Format("15:04"), err)
continue
}
if !won {
continue
}
if st.ScheduleRunning(repo, en.pipeline) {
log.Printf("ci: %s %s: skipped the %s schedule: its previous scheduled run is still going", repo, en.pipeline, minute.Format("15:04"))
continue
}
if err := launch(e, st, repo, r, tip.c, cloneURL, tip.defs.only([]string{en.pipeline}), scopeSchedule); err != nil {
log.Printf("ci: %s %s: the %s schedule did not start: %v", repo, en.pipeline, minute.Format("15:04"), err)
st.ScheduleFailed(repo, en.pipeline, minute, err.Error())
}
}
}
return nil
}
// ScheduleCatchUp bounds how many minutes one tick walks after a stall.
const ScheduleCatchUp = 10
// ScheduleTick is the scheduler's walk over minutes: each minute since
// the last tick, on the lease holder, never one from before it started
// or before it last held the lease (no backfill, as GitHub does not).
type ScheduleTick struct{ last time.Time }
func NewScheduleTick(now time.Time) *ScheduleTick {
return &ScheduleTick{last: now.UTC().Truncate(time.Minute).Add(-time.Minute)}
}
// Minutes is the minutes this tick at now should consider.
func (t *ScheduleTick) Minutes(now time.Time, leading bool) []time.Time {
cur := now.UTC().Truncate(time.Minute)
if !leading {
t.last = cur
return nil
}
from := t.last.Add(time.Minute)
if cur.Sub(from) >= ScheduleCatchUp*time.Minute {
from = cur.Add(-(ScheduleCatchUp - 1) * time.Minute)
}
var out []time.Time
for m := from; !m.After(cur); m = m.Add(time.Minute) {
out = append(out, m)
}
if !cur.Before(t.last) {
t.last = cur
}
return out
}
// RunnerFor reports whether a live worker drains repo's group. An
// engine whose workers report no groups at all (one predating
// dagnats#719) cannot say, and is answered yes.
func RunnerFor(ws []Worker, repo string, now time.Time) bool {
group, grouped := GroupFor(repo), false
for _, w := range ws {
if len(w.WorkerGroups) > 0 {
grouped = true
}
if seen, err := time.Parse(time.RFC3339, w.LastSeen); err == nil && now.Sub(seen) > workerStale {
continue
}
if slices.Contains(w.WorkerGroups, group) {
return true
}
}
return !grouped
}