package spindle
import (
"context"
"errors"
"io"
"log/slog"
"slices"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/bluesky-social/indigo/atproto/syntax"
"tangled.org/core/api/org_tangled"
"tangled.org/core/spindle/db"
"tangled.org/core/spindle/knotclient"
"tangled.org/core/spindle/models"
"tangled.org/core/workflow"
)
func scheduleTestRepo() db.Repo {
return db.Repo{
Knot: "knot.test",
Owner: syntax.DID("did:plc:owner"),
Rkey: syntax.RecordKey("repo"),
RepoDid: syntax.DID("did:plc:repo"),
}
}
func scheduleTestSnapshot() db.ScheduledRepo {
return db.ScheduledRepo{
RepoDid: scheduleTestRepo().RepoDid.String(),
Branch: "main",
SHA: "sha",
}
}
func mustCron(t *testing.T, expression, name string) scheduledWorkflow {
t.Helper()
repo := scheduleTestRepo()
schedule, err := workflow.ParseCron(expression, "", repo.RepoDid.String()+"\x00build.yml")
if err != nil {
t.Fatalf("ParseCron(%q, UTC): %v", expression, err)
}
return scheduledWorkflow{
repo: repo,
scheduledRepo: scheduleTestSnapshot(),
name: name,
expression: expression,
schedule: schedule,
}
}
func newTestScheduler(t *testing.T) *pipelineScheduler {
t.Helper()
d, _ := newTestSpindleDB(t)
return &pipelineScheduler{
db: d,
l: slog.New(slog.NewTextHandler(io.Discard, nil)),
byRepo: make(map[string][]*scheduledWorkflow),
minimumInterval: 5 * time.Minute,
concurrency: 1,
now: time.Now,
}
}
// a released claim can be retaken, a completed one can't
func claimable(t *testing.T, s *pipelineScheduler, entry scheduledWorkflow, at time.Time) bool {
t.Helper()
ok, err := s.db.ClaimScheduleRun(context.Background(), entry.repo.RepoDid.String(), entry.name, at, s.minimumInterval)
if err != nil {
t.Fatal(err)
}
return ok
}
func TestPipelineSchedulerDueUsesUTCMinuteAndDeduplicatesWorkflows(t *testing.T) {
scheduler := newTestScheduler(t)
scheduler.replaceCachedSchedules(
scheduleTestRepo().RepoDid.String(),
[]scheduledWorkflow{mustCron(t, "0 9 * * 1-5", "build.yml"), mustCron(t, "0 9 * * *", "build.yml"), mustCron(t, "0 12 * * *", "noon.yml")},
time.Date(2026, time.August, 10, 8, 59, 0, 0, time.UTC),
)
at := time.Date(2026, time.August, 10, 11, 0, 47, 0, time.FixedZone("local", 2*60*60))
due := scheduler.due(at)
group, ok := due[scheduleTestRepo().RepoDid.String()]
if !ok {
t.Fatal("expected repository to have due workflows")
}
var names []string
for name := range group.workflows {
names = append(names, name)
}
slices.Sort(names)
if !slices.Equal(names, []string{"build.yml"}) {
t.Fatalf("due workflows = %v, want [build.yml]", names)
}
}
func TestPipelineSchedulerDispatchDueClaimsOccurrenceOnce(t *testing.T) {
d, _ := newTestSpindleDB(t)
entry := mustCron(t, "0 9 * * *", "build.yml")
at := time.Date(2026, time.August, 10, 9, 0, 0, 0, time.UTC)
var calls int
dispatch := func(_ context.Context, repo db.Repo, _ db.ScheduledRepo, workflows []string, scheduledAt time.Time) (models.PipelineId, error) {
calls++
if repo.RepoDid != entry.repo.RepoDid || !slices.Equal(workflows, []string{"build.yml"}) || !scheduledAt.Equal(at) {
t.Fatalf("unexpected dispatch: repo=%s workflows=%v scheduledAt=%s", repo.RepoDid, workflows, scheduledAt)
}
return models.PipelineId("pipeline"), nil
}
for range 2 {
scheduler := newTestScheduler(t)
scheduler.db = d
scheduler.dispatch = dispatch
scheduler.replaceCachedSchedules(entry.repo.RepoDid.String(), []scheduledWorkflow{entry}, at.Add(-time.Minute))
scheduler.dispatchDue(context.Background(), at)
}
if calls != 1 {
t.Fatalf("dispatch calls = %d, want 1", calls)
}
}
func TestPipelineSchedulerDispatchTimeoutReleasesClaim(t *testing.T) {
entry := mustCron(t, "0 9 * * *", "build.yml")
at := time.Date(2026, time.August, 10, 9, 0, 0, 0, time.UTC)
scheduler := newTestScheduler(t)
scheduler.dispatchTimeout = 25 * time.Millisecond
scheduler.dispatch = func(ctx context.Context, _ db.Repo, _ db.ScheduledRepo, _ []string, _ time.Time) (models.PipelineId, error) {
<-ctx.Done()
return models.PipelineId(""), ctx.Err()
}
scheduler.replaceCachedSchedules(entry.repo.RepoDid.String(), []scheduledWorkflow{entry}, at.Add(-time.Minute))
start := time.Now()
scheduler.dispatchDue(context.Background(), at)
if elapsed := time.Since(start); elapsed > 5*time.Second {
t.Fatalf("dispatchDue returned after %s; want the dispatch timeout to unblock it", elapsed)
}
if !claimable(t, scheduler, entry, at) {
t.Fatal("timed-out dispatch left its schedule claim behind")
}
}
func TestPipelineSchedulerBoundsConcurrentRepositoryDispatches(t *testing.T) {
at := time.Date(2026, time.August, 10, 9, 0, 0, 0, time.UTC)
started := make(chan struct{}, 2)
release := make(chan struct{})
var mu sync.Mutex
active, maxActive := 0, 0
scheduler := newTestScheduler(t)
scheduler.concurrency = 2
scheduler.dispatch = func(_ context.Context, repo db.Repo, _ db.ScheduledRepo, _ []string, _ time.Time) (models.PipelineId, error) {
mu.Lock()
active++
maxActive = max(maxActive, active)
mu.Unlock()
started <- struct{}{}
<-release
mu.Lock()
active--
mu.Unlock()
return models.PipelineId("pipeline"), nil
}
for _, did := range []string{"did:plc:repoa", "did:plc:repob"} {
entry := mustCron(t, "0 9 * * *", "build.yml")
entry.repo.RepoDid = syntax.DID(did)
entry.scheduledRepo.RepoDid = did
scheduler.replaceCachedSchedules(did, []scheduledWorkflow{entry}, at.Add(-time.Minute))
}
done := make(chan struct{})
go func() {
scheduler.dispatchDue(context.Background(), at)
close(done)
}()
for range 2 {
select {
case <-started:
case <-time.After(time.Second):
t.Fatal("repository dispatches did not start concurrently")
}
}
mu.Lock()
gotMaxActive := maxActive
mu.Unlock()
if gotMaxActive != 2 {
t.Fatalf("maximum concurrent dispatches = %d, want 2", gotMaxActive)
}
close(release)
select {
case <-done:
case <-time.After(time.Second):
t.Fatal("dispatchDue did not wait for repository dispatches")
}
}
func TestPipelineSchedulerReleasesClaimWhenDispatchIsCanceled(t *testing.T) {
at := time.Date(2026, time.August, 10, 9, 0, 0, 0, time.UTC)
entry := mustCron(t, "0 9 * * *", "build.yml")
started := make(chan struct{})
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
scheduler := newTestScheduler(t)
scheduler.dispatch = func(ctx context.Context, _ db.Repo, _ db.ScheduledRepo, _ []string, _ time.Time) (models.PipelineId, error) {
close(started)
<-ctx.Done()
return models.PipelineId(""), ctx.Err()
}
scheduler.replaceCachedSchedules(entry.repo.RepoDid.String(), []scheduledWorkflow{entry}, at.Add(-time.Minute))
done := make(chan struct{})
go func() {
scheduler.dispatchDue(ctx, at)
close(done)
}()
<-started
cancel()
select {
case <-done:
case <-time.After(time.Second):
t.Fatal("canceled dispatch did not finish")
}
if !claimable(t, scheduler, entry, at) {
t.Fatal("canceled dispatch left its schedule claim")
}
}
func TestPipelineSchedulerEnforcesMinimumRunIntervalAcrossSchedules(t *testing.T) {
at := time.Date(2026, time.August, 10, 9, 0, 0, 0, time.UTC)
var entries []scheduledWorkflow
for _, expression := range []string{"0 9 * * *", "5 9 * * *", "10 9 * * *"} {
entries = append(entries, mustCron(t, expression, "build.yml"))
}
var dispatched []time.Time
scheduler := newTestScheduler(t)
scheduler.minimumInterval = 10 * time.Minute
scheduler.dispatch = func(_ context.Context, _ db.Repo, _ db.ScheduledRepo, _ []string, scheduledAt time.Time) (models.PipelineId, error) {
dispatched = append(dispatched, scheduledAt)
return models.PipelineId("pipeline"), nil
}
scheduler.replaceCachedSchedules(scheduleTestRepo().RepoDid.String(), entries, at.Add(-time.Minute))
scheduler.dispatchDue(context.Background(), at)
scheduler.dispatchDue(context.Background(), at.Add(5*time.Minute))
scheduler.dispatchDue(context.Background(), at.Add(10*time.Minute))
want := []time.Time{at, at.Add(10 * time.Minute)}
if !slices.Equal(dispatched, want) {
t.Fatalf("scheduled dispatches = %v, want %v", dispatched, want)
}
}
func TestPipelineSchedulerDispatchFailureClaimLifecycle(t *testing.T) {
tests := []struct {
name string
pipelineID models.PipelineId
wantReleased bool
}{
{name: "before pipeline creation", wantReleased: true},
{name: "after pipeline creation", pipelineID: models.PipelineId("pipeline")},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
entry := mustCron(t, "0 9 * * *", "build.yml")
at := time.Date(2026, time.August, 10, 9, 0, 0, 0, time.UTC)
scheduler := newTestScheduler(t)
scheduler.dispatch = func(context.Context, db.Repo, db.ScheduledRepo, []string, time.Time) (models.PipelineId, error) {
return test.pipelineID, errors.New("temporary failure")
}
scheduler.replaceCachedSchedules(entry.repo.RepoDid.String(), []scheduledWorkflow{entry}, at.Add(-time.Minute))
scheduler.dispatchDue(context.Background(), at)
if reclaimed := claimable(t, scheduler, entry, at); reclaimed != test.wantReleased {
t.Fatalf("schedule claim reclaimed = %v, want %v", reclaimed, test.wantReleased)
}
if test.wantReleased {
return
}
var completed int
if err := scheduler.db.QueryRow(
`select count(*) from schedule_runs where repo_did = ? and workflow = ? and scheduled_at = ? and pipeline_id = ?`,
entry.repo.RepoDid.String(), entry.name, at.Unix(), test.pipelineID.String(),
).Scan(&completed); err != nil {
t.Fatal(err)
}
if completed != 1 {
t.Fatal("pipeline creation error left its schedule claim incomplete")
}
})
}
}
func TestPipelineSchedulerRestoresPersistedSchedulesWithoutLoadingRepository(t *testing.T) {
ctx := context.Background()
scheduler := newTestScheduler(t)
repo := scheduleTestRepo()
if err := scheduler.db.AddRepo(repo); err != nil {
t.Fatalf("AddRepo: %v", err)
}
if err := scheduler.db.ReplaceWorkflowSchedules(ctx, db.ScheduledRepo{
RepoDid: repo.RepoDid.String(),
Branch: "main",
SHA: "persisted-sha",
}, []db.WorkflowSchedule{{
RepoDid: repo.RepoDid.String(),
Workflow: "build.yml",
Expression: "30 5 * * 1-5",
Timezone: "America/New_York",
}}, time.Now()); err != nil {
t.Fatalf("ReplaceWorkflowSchedules: %v", err)
}
at := time.Date(2026, time.August, 10, 9, 30, 0, 0, time.UTC)
scheduler.now = func() time.Time { return at.Add(-time.Minute) }
if err := scheduler.restoreSchedules(ctx); err != nil {
t.Fatalf("restoreSchedules: %v", err)
}
due := scheduler.due(at)
group, ok := due[repo.RepoDid.String()]
if !ok {
t.Fatal("persisted schedule was not restored")
}
if group.scheduledRepo.Branch != "main" || group.scheduledRepo.SHA != "persisted-sha" {
t.Fatalf("restored repository snapshot = %#v, want main/persisted-sha", group.scheduledRepo)
}
}
type countingSchedule struct {
next time.Time
calls int
}
func (s *countingSchedule) Next(time.Time) time.Time {
s.calls++
return s.next
}
func TestPipelineSchedulerDoesNotEvaluateNonDueSchedulesEachMinute(t *testing.T) {
now := time.Date(2026, time.August, 10, 9, 0, 0, 0, time.UTC)
entries := make([]scheduledWorkflow, 10_000)
schedules := make([]*countingSchedule, len(entries))
for i := range entries {
schedules[i] = &countingSchedule{next: now.Add(24 * time.Hour)}
entries[i] = scheduledWorkflow{
repo: scheduleTestRepo(),
scheduledRepo: scheduleTestSnapshot(),
name: "build.yml",
schedule: schedules[i],
}
}
scheduler := newTestScheduler(t)
scheduler.replaceCachedSchedules(scheduleTestRepo().RepoDid.String(), entries, now)
for _, schedule := range schedules {
schedule.calls = 0
}
if due := scheduler.due(now.Add(time.Minute)); len(due) != 0 {
t.Fatalf("due repositories = %d, want 0", len(due))
}
for i, schedule := range schedules {
if schedule.calls != 0 {
t.Fatalf("schedule %d evaluated %d times while not due", i, schedule.calls)
}
}
}
type sequenceSchedule struct {
next []time.Time
}
func (s *sequenceSchedule) Next(time.Time) time.Time {
if len(s.next) == 0 {
return time.Time{}
}
next := s.next[0]
s.next = s.next[1:]
return next
}
func TestPipelineSchedulerDropsSchedulesThatDoNotAdvance(t *testing.T) {
at := time.Date(2026, time.August, 10, 9, 0, 0, 0, time.UTC)
tests := []struct {
name string
next []time.Time
due bool
}{
{name: "zero initial occurrence"},
{name: "zero after due occurrence", next: []time.Time{at, {}}, due: true},
{name: "same occurrence repeats", next: []time.Time{at, at}, due: true},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
entry := scheduledWorkflow{
repo: scheduleTestRepo(),
scheduledRepo: scheduleTestSnapshot(),
name: "build.yml",
schedule: &sequenceSchedule{next: test.next},
}
scheduler := newTestScheduler(t)
scheduler.replaceCachedSchedules(entry.repo.RepoDid.String(), []scheduledWorkflow{entry}, at.Add(-time.Minute))
due := scheduler.due(at)
if got := len(due) > 0; got != test.due {
t.Fatalf("due = %v, want %v", got, test.due)
}
if scheduler.queue.Len() != 0 {
t.Fatalf("queue length = %d, want 0 after non-advancing occurrence", scheduler.queue.Len())
}
})
}
}
type slowFirstLoad struct {
started chan struct{}
gate chan struct{}
release func()
calls atomic.Int32
entries []scheduledWorkflow
}
func (l *slowFirstLoad) load(context.Context, db.Repo, time.Time) ([]scheduledWorkflow, db.ScheduledRepo, error) {
call := int(l.calls.Add(1))
if call == 1 {
close(l.started)
<-l.gate
}
return []scheduledWorkflow{l.entries[min(call, len(l.entries))-1]}, scheduleTestSnapshot(), nil
}
func newRefreshTest(t *testing.T, entries ...scheduledWorkflow) (*pipelineScheduler, *slowFirstLoad, db.Repo) {
t.Helper()
scheduler := newTestScheduler(t)
repo := scheduleTestRepo()
if err := scheduler.db.AddRepo(repo); err != nil {
t.Fatalf("AddRepo: %v", err)
}
slow := &slowFirstLoad{started: make(chan struct{}), gate: make(chan struct{}), entries: entries}
var once sync.Once
slow.release = func() { once.Do(func() { close(slow.gate) }) }
t.Cleanup(slow.release)
scheduler.load = slow.load
return scheduler, slow, repo
}
func refreshAsync(s *pipelineScheduler, repo db.Repo) <-chan error {
return goErr(func() error { return s.RefreshRepo(context.Background(), repo) })
}
func mustRefresh(t *testing.T, done <-chan error) {
t.Helper()
select {
case err := <-done:
if err != nil {
t.Fatalf("RefreshRepo: %v", err)
}
case <-time.After(10 * time.Second):
t.Fatal("RefreshRepo did not return")
}
}
func persistedWorkflows(t *testing.T, s *pipelineScheduler) []string {
t.Helper()
definitions, err := s.db.WorkflowSchedules(context.Background())
if err != nil {
t.Fatalf("WorkflowSchedules: %v", err)
}
var names []string
for _, definition := range definitions {
names = append(names, definition.Workflow)
}
return names
}
func requireNothingInflight(t *testing.T, s *pipelineScheduler) {
t.Helper()
s.refreshMu.Lock()
defer s.refreshMu.Unlock()
if len(s.inflight) != 0 {
t.Fatalf("%d repos still tracked once no refresh is running", len(s.inflight))
}
}
func waitForQueuedRefresh(t *testing.T, s *pipelineScheduler, repo db.Repo) {
t.Helper()
eventually(t, 2*time.Second, "the second refresh to queue behind the first", func() bool {
s.refreshMu.Lock()
defer s.refreshMu.Unlock()
e := s.inflight[repo.RepoDid.String()]
return e != nil && e.users == 2
})
}
func TestPipelineSchedulerSerializesRepositoryRefreshes(t *testing.T) {
scheduler, slow, repo := newRefreshTest(t, mustCron(t, "0 8 * * *", "old.yml"), mustCron(t, "0 9 * * *", "new.yml"))
firstDone := refreshAsync(scheduler, repo)
<-slow.started
secondDone := refreshAsync(scheduler, repo)
time.Sleep(time.Second)
if got := slow.calls.Load(); got != 1 {
t.Errorf("%d loads started before the first refresh completed, want 1", got)
}
slow.release()
mustRefresh(t, firstDone)
mustRefresh(t, secondDone)
if got := persistedWorkflows(t, scheduler); !slices.Equal(got, []string{"new.yml"}) {
t.Fatalf("persisted schedules = %v, want only newest refresh", got)
}
}
func TestPipelineSchedulerRemovalDoesNotWaitForInflightRefresh(t *testing.T) {
scheduler, slow, repo := newRefreshTest(t, mustCron(t, "0 9 * * *", "build.yml"))
refreshDone := refreshAsync(scheduler, repo)
<-slow.started
removeDone := make(chan error, 1)
go func() { removeDone <- scheduler.RemoveRepo(context.Background(), repo.RepoDid.String()) }()
// a clone can take minutes, and removal can't wait on it
select {
case err := <-removeDone:
if err != nil {
t.Fatalf("RemoveRepo: %v", err)
}
case <-time.After(time.Second):
t.Fatal("repository removal waited for its schedule refresh that was still in flight")
}
slow.release()
mustRefresh(t, refreshDone)
if got := persistedWorkflows(t, scheduler); len(got) != 0 {
t.Fatalf("persisted schedules after removal = %v, want none, a load that finished after the removal has to be thrown away", got)
}
}
func TestAReloadAfterARemovalKeepsItsResultAndNothingStaysInFlight(t *testing.T) {
scheduler, slow, repo := newRefreshTest(t, mustCron(t, "0 8 * * *", "old.yml"), mustCron(t, "0 9 * * *", "new.yml"))
firstDone := refreshAsync(scheduler, repo)
<-slow.started
if err := scheduler.RemoveRepo(context.Background(), repo.RepoDid.String()); err != nil {
t.Fatalf("RemoveRepo: %v", err)
}
secondDone := refreshAsync(scheduler, repo)
time.Sleep(50 * time.Millisecond)
slow.release()
mustRefresh(t, firstDone)
mustRefresh(t, secondDone)
if got := persistedWorkflows(t, scheduler); !slices.Equal(got, []string{"new.yml"}) {
t.Fatalf("persisted schedules = %v, want only what the reload after the removal found", got)
}
requireNothingInflight(t, scheduler)
}
func TestARefreshQueuedBehindACloneWhenTheRepoIsRemovedThrowsItsResultAway(t *testing.T) {
scheduler, slow, repo := newRefreshTest(t, mustCron(t, "0 8 * * *", "wf.yml"))
firstDone := refreshAsync(scheduler, repo)
<-slow.started
secondDone := refreshAsync(scheduler, repo)
waitForQueuedRefresh(t, scheduler, repo)
if err := scheduler.RemoveRepo(context.Background(), repo.RepoDid.String()); err != nil {
t.Fatalf("RemoveRepo: %v", err)
}
slow.release()
mustRefresh(t, firstDone)
mustRefresh(t, secondDone)
if got := persistedWorkflows(t, scheduler); len(got) != 0 {
t.Fatalf("persisted schedules = %v, want none for a repo removed before the queued refresh ran", got)
}
}
func TestRefreshesAskedForWhileOneIsAlreadyQueuedAreNotQueuedAgain(t *testing.T) {
scheduler, slow, repo := newRefreshTest(t, mustCron(t, "0 8 * * *", "wf.yml"))
firstDone := refreshAsync(scheduler, repo)
<-slow.started
secondDone := refreshAsync(scheduler, repo)
waitForQueuedRefresh(t, scheduler, repo)
for range 20 {
select {
case err := <-refreshAsync(scheduler, repo):
if err != nil {
t.Fatalf("RefreshRepo: %v", err)
}
case <-time.After(time.Second):
t.Fatal("a refresh asked for while one is queued waited behind the clone instead of being covered by it")
}
}
slow.release()
mustRefresh(t, firstDone)
mustRefresh(t, secondDone)
if got := slow.calls.Load(); got != 2 {
t.Fatalf("%d schedule loads, want the running one and the queued one", got)
}
}
func TestFlapsOfOneRepoLeaveOneWaiterBehindTheCloneNotOnePerRemoval(t *testing.T) {
scheduler, slow, repo := newRefreshTest(t, mustCron(t, "0 8 * * *", "wf.yml"))
firstDone := refreshAsync(scheduler, repo)
<-slow.started
const returns = 20
flaps := make(chan error, returns)
for range returns {
if err := scheduler.RemoveRepo(context.Background(), repo.RepoDid.String()); err != nil {
t.Fatalf("RemoveRepo: %v", err)
}
go func() { flaps <- scheduler.RefreshRepo(context.Background(), repo) }()
time.Sleep(time.Millisecond)
}
for range returns - 1 {
select {
case err := <-flaps:
if err != nil {
t.Fatalf("RefreshRepo: %v", err)
}
case <-time.After(2 * time.Second):
t.Fatal("refreshes of a repo that was removed again are still parked behind the clone, one per removal")
}
}
slow.release()
mustRefresh(t, firstDone)
mustRefresh(t, flaps)
if got := slow.calls.Load(); got != 2 {
t.Fatalf("%d schedule loads, want the running one and the newest request's", got)
}
if got := persistedWorkflows(t, scheduler); !slices.Equal(got, []string{"wf.yml"}) {
t.Fatalf("persisted schedules = %v, want what the newest request loaded", got)
}
requireNothingInflight(t, scheduler)
}
func TestDefaultBranchComesFromTheRefsRecord(t *testing.T) {
env := newEnsureEnv(t)
env.serve("repo")
if _, err := env.s.defaultBranch(context.Background(), env.k, testRepoDid); err == nil {
t.Fatal("a repo without a refs record has no default branch yet")
}
env.knot.put(testRepoDid, org_tangled.GitRefsNSID, knotclient.RefsRkey, org_tangled.GitRefs{
LexiconTypeID: org_tangled.GitRefsNSID,
DefaultBranch: "trunk",
})
got, err := env.s.defaultBranch(context.Background(), env.k, testRepoDid)
if err != nil || got != "trunk" {
t.Fatalf("default branch = %q, %v, want trunk", got, err)
}
}