Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659package 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'tfunc 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) }}