package ci
import (
"encoding/json"
"errors"
"fmt"
"net/http"
"net/http/httptest"
"strings"
"sync"
"testing"
"time"
"craftdesign.group/quarry/internal/cronexpr"
)
func schedSpec(lines ...string) []byte {
var b strings.Builder
b.WriteString("on:\n schedule:\n")
for _, l := range lines {
b.WriteString(" - cron: '" + l + "'\n")
}
b.WriteString("jobs:\n t:\n steps:\n - run: x\n")
return []byte(b.String())
}
// THE 5-MINUTE FLOOR HOLDS FOR THE WORKFLOW, NOT LINE BY LINE, AND FOR
// EVERY DAY, NOT A SAMPLE: the minute sets of all lines merge, and the
// cyclic gap between consecutive minutes is at least five.
func TestTheFloorHoldsAcrossLinesAndDays(t *testing.T) {
var offsets []string
for i := 0; i < 5; i++ {
offsets = append(offsets, fmt.Sprintf("%d-59/5 * * * *", i))
}
for name, lines := range map[string][]string{
"two lines two minutes apart": {"0 * * * *", "2 * * * *"},
"offset */5 lines every minute": offsets,
"every minute on a leap day": {"* * 29 2 *"},
"step under five": {"*/3 4 * * *"},
"wrapping across the hour": {"58 * * * *", "1 * * * *"},
"03:17 beside 03:15 on weekdays": {"17 3 * * *", "*/15 * * * 1-5"},
"wrapping across midnight": {"58 23 * * *", "1 0 * * *"},
"eleven lines": {"0 0 * * *", "5 0 * * *", "10 0 * * *", "15 0 * * *", "20 0 * * *", "25 0 * * *", "30 0 * * *", "35 0 * * *", "40 0 * * *", "45 0 * * *", "50 0 * * *"},
} {
if _, err := schedulesOf(schedSpec(lines...)); err == nil || !strings.Contains(err.Error(), "line ") {
t.Errorf("FAIL: %s %v was accepted (%v); want a refusal with a position", name, lines, err)
}
}
for name, lines := range map[string][]string{
"hourly and half past": {"0 * * * *", "30 * * * *"},
"every five": {"*/5 * * * *"},
"leap day once": {"0 12 29 2 *"},
"wrap of exactly five": {"57 * * * *", "2 * * * *"},
"an hour apart": {"0 3 * * *", "2 4 * * *"},
} {
if _, err := schedulesOf(schedSpec(lines...)); err != nil {
t.Errorf("FAIL: %s %v was refused: %v", name, lines, err)
}
}
if _, err := schedulesOf(schedSpec(strings.Repeat("1", 300) + " * * * *")); err == nil {
t.Errorf("FAIL: an over-long cron line was accepted")
}
t.Log("ok: the floor is the workflow's minute set, cyclic, every day; lines capped; an over-long line refused, never a panic")
}
type countingRepo struct {
*fakeRepo
mu sync.Mutex
reads int
}
func (c *countingRepo) Read(p, v string) ([]byte, error) {
c.mu.Lock()
c.reads++
c.mu.Unlock()
return c.fakeRepo.Read(p, v)
}
type failStore struct {
*fakeStore
scheduleRig
failClaim string
failed []string
}
func (f *failStore) ClaimSchedule(repo, p string, m time.Time) (bool, error) {
if p == f.failClaim {
return false, errors.New("database is locked")
}
return f.scheduleRig.ClaimSchedule(repo, p, m)
}
func (f *failStore) ScheduleFailed(repo, p string, m time.Time, why string) {
f.failed = append(f.failed, p+": "+why)
}
// ONE BAD WORKFLOW STOPS ONLY ITSELF; THE SCHEDULES ARE READ ONCE PER
// DEFAULT TIP; A LOST LEASE STOPS CLAIMS; A REFUSED SCHEDULE IS LISTED
// AS NOT FIRING, WITH ITS ERROR.
func TestSchedulesAreReadOncePerTipAndFailAlone(t *testing.T) {
var mu sync.Mutex
started := map[string]int{}
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
mu.Lock()
defer mu.Unlock()
switch {
case r.URL.Path == "/v1/ci/compile":
var body struct{ Name string }
json.NewDecoder(r.Body).Decode(&body)
fmt.Fprintf(w, `{"workflow":{"name":%q},"registered":true,"on":{"schedule":[]}}`, body.Name)
case r.URL.Path == "/runs" && r.Method == http.MethodPost:
var body struct{ Input map[string]any }
json.NewDecoder(r.Body).Decode(&body)
started[fmt.Sprint(body.Input["pipeline"])]++
w.WriteHeader(http.StatusCreated)
fmt.Fprintf(w, `{"run_id":"r%d"}`, len(started))
case r.URL.Path == "/runs":
fmt.Fprint(w, `[]`)
}
}))
defer srv.Close()
e := New(srv.URL)
repo := &countingRepo{fakeRepo: &fakeRepo{
files: []string{"ci/a.yml", "ci/bad.yml", "ci/b.yml", "ci/c.yml"},
blobs: map[string][]byte{
"ci/a.yml": schedSpec("0 * * * *"),
"ci/bad.yml": schedSpec("* * * * *"),
"ci/b.yml": schedSpec("0 * * * *"),
"ci/c.yml": schedSpec("0 * * * *"),
},
tips: map[string]string{"trunk": "t1"},
}}
st := &failStore{fakeStore: &fakeStore{mark: map[string]string{}},
scheduleRig: scheduleRig{claims: map[string]bool{}, running: map[string]bool{}, mu: &sync.Mutex{}}, failClaim: "ci/b.yml"}
cache := NewScheduleCache()
base := time.Date(2026, 10, 3, 3, 0, 0, 0, time.UTC)
minutes := []time.Time{base, base.Add(time.Minute), base.Add(2 * time.Minute)}
if err := FireSchedules(e, st, cache, "alpha", repo, "u", minutes, always); err != nil {
t.Errorf("FAIL: one bad workflow failed the repository: %v", err)
}
if started["ci/a.yml"] != 1 || started["ci/c.yml"] != 1 || started["ci/bad.yml"] != 0 || started["ci/b.yml"] != 0 {
t.Errorf("FAIL: started %v; want a and c once, the refused and the unclaimable none", started)
}
readsAfterFirst := repo.reads
if err := FireSchedules(e, st, cache, "alpha", repo, "u", []time.Time{base.Add(time.Hour)}, always); err != nil {
t.Fatal(err)
}
// launch reads the two it starts; the schedules themselves are not read again
if repo.reads-readsAfterFirst > 2 {
t.Errorf("FAIL: an unchanged tip was read %d more times; want only the launches' own reads", repo.reads-readsAfterFirst)
}
listed, err := cache.Schedules("alpha", repo, base)
if err != nil {
t.Fatal(err)
}
var bad *Scheduled
for i := range listed {
if listed[i].Pipeline == "ci/bad.yml" {
bad = &listed[i]
}
}
if bad == nil || bad.Error == "" || !bad.Next.IsZero() {
t.Errorf("FAIL: the refused schedule is listed as %+v; want not firing, with its error", bad)
}
// The lease lost between claims: nothing more is claimed.
n := len(st.claims)
if err := FireSchedules(e, st, cache, "alpha", repo, "u", []time.Time{base.Add(2 * time.Hour)}, func() bool { return false }); err != nil {
t.Fatal(err)
}
if len(st.claims) != n {
t.Errorf("FAIL: claims were taken without the lease")
}
t.Log("ok: a bad or unclaimable workflow skips alone; one read per tip; no lease, no claim; a refused schedule is listed with its error")
}
// THE TICK LOOP: NO BACKFILL AT START, EVERY MINUTE SINCE THE LAST ONE
// WHILE LEADING, NOTHING WHILE NOT, CATCH-UP BOUNDED.
func TestTheScheduleTickWalksMinutes(t *testing.T) {
at := func(h, m, s int) time.Time { return time.Date(2026, 10, 3, h, m, s, 0, time.UTC) }
tk := NewScheduleTick(at(3, 0, 20))
if got := tk.Minutes(at(3, 0, 35), true); len(got) != 1 || !got[0].Equal(at(3, 0, 0)) {
t.Errorf("FAIL: the first tick walked %v; want the current minute only", got)
}
if got := tk.Minutes(at(3, 0, 50), true); len(got) != 0 {
t.Errorf("FAIL: a second tick in the same minute walked %v", got)
}
if got := tk.Minutes(at(3, 3, 5), true); len(got) != 3 {
t.Errorf("FAIL: a slow tick walked %v; want the three minutes since", got)
}
if got := tk.Minutes(at(3, 5, 0), false); len(got) != 0 {
t.Errorf("FAIL: a node without the lease walked %v", got)
}
if got := tk.Minutes(at(3, 6, 0), true); len(got) != 1 {
t.Errorf("FAIL: regaining the lease walked %v; want only the current minute", got)
}
if got := tk.Minutes(at(4, 0, 0), true); len(got) != ScheduleCatchUp {
t.Errorf("FAIL: an hour's stall walked %d minutes; want the bound %d", len(got), ScheduleCatchUp)
}
t.Log("ok: no backfill, minute by minute while leading, nothing while not, bounded catch-up")
}
// A PARSER PANIC IS A REFUSAL WITH THE LINE'S POSITION, NEVER A CRASH;
// A LINE THAT NEVER FIRES IS REFUSED.
func TestAParserPanicIsARefusal(t *testing.T) {
was := parseCronFn
parseCronFn = func(string) (*cronexpr.CronExpr, error) { panic("index out of range") }
defer func() { parseCronFn = was }()
if _, err := schedulesOf(schedSpec("0 3 * * *")); err == nil || !strings.Contains(err.Error(), "line 3, column 13") {
t.Errorf("FAIL: a parser panic gave %v; want a refusal at the line", err)
}
parseCronFn = was
for _, line := range []string{"0,,30 3 * * *", ",5 * * * *", "5, * * * *", "0 3 * * 1,"} {
if _, err := schedulesOf(schedSpec(line)); err == nil || !strings.Contains(err.Error(), "line 3") {
t.Errorf("FAIL: %q gave %v; want a refusal with a position", line, err)
}
}
if _, err := schedulesOf(schedSpec("0 0 31 2 *")); err == nil || !strings.Contains(err.Error(), "never") {
t.Errorf("FAIL: a line that never fires gave %v; want a refusal", err)
}
if _, err := schedulesOf(schedSpec("0 0 29 2 *")); err != nil {
t.Errorf("FAIL: a leap-day line was refused: %v", err)
}
t.Log("ok: parser panics and never-firing lines are refusals")
}
// A schedule that starts clears its last failure; a repository gone is
// evicted from the cache.
func TestASuccessClearsTheFailureAndGoneReposAreEvicted(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch {
case r.URL.Path == "/v1/ci/compile":
fmt.Fprint(w, `{"workflow":{"name":"x"},"registered":true,"on":{"schedule":[]}}`)
case r.URL.Path == "/runs" && r.Method == http.MethodPost:
w.WriteHeader(http.StatusCreated)
fmt.Fprint(w, `{"run_id":"r1"}`)
case r.URL.Path == "/runs":
fmt.Fprint(w, `[]`)
}
}))
defer srv.Close()
repo := &fakeRepo{files: []string{"ci/a.yml"}, blobs: map[string][]byte{"ci/a.yml": schedSpec("0 * * * *")}, tips: map[string]string{"trunk": "t1"}}
st := &failStore{fakeStore: &fakeStore{mark: map[string]string{}},
scheduleRig: scheduleRig{claims: map[string]bool{}, running: map[string]bool{}, mu: &sync.Mutex{}}}
cache := NewScheduleCache()
if err := FireSchedules(New(srv.URL), st, cache, "alpha", repo, "u", []time.Time{time.Date(2026, 10, 3, 3, 0, 0, 0, time.UTC)}, always); err != nil {
t.Fatal(err)
}
if len(st.failed) != 1 || st.failed[0] != "ci/a.yml: " {
t.Errorf("FAIL: a started schedule recorded %v; want its failure cleared", st.failed)
}
cache.Retain([]string{"beta"})
cache.mu.Lock()
_, kept := cache.tips["alpha"]
cache.mu.Unlock()
if kept {
t.Errorf("FAIL: a repository no longer served stayed in the cache")
}
t.Log("ok: success clears the last failure; gone repositories are evicted")
}