From 3c0bfe9463c8c069dc550c74ad00365170290969 Mon Sep 17 00:00:00 2001 From: bircni Date: Sat, 29 Aug 2026 08:44:51 +0200 Subject: [PATCH] fix(actions): run every due schedule exactly once per occurrence (#39078) Signed-off-by: Lunny Xiao Co-authored-by: silverwind Co-authored-by: Lunny Xiao Co-authored-by: wxiaoguang --- models/actions/schedule.go | 22 ++-- models/actions/schedule_spec.go | 2 +- models/actions/schedule_spec_list.go | 97 ---------------- services/actions/notifier_helper.go | 2 +- services/actions/schedule_tasks.go | 141 ++++++++++++------------ services/actions/schedule_tasks_test.go | 50 +++++++++ 6 files changed, 132 insertions(+), 182 deletions(-) delete mode 100644 models/actions/schedule_spec_list.go diff --git a/models/actions/schedule.go b/models/actions/schedule.go index 91171cc8827..79679203489 100644 --- a/models/actions/schedule.go +++ b/models/actions/schedule.go @@ -40,17 +40,8 @@ func init() { db.RegisterModel(new(ActionSchedule)) } -// GetSchedulesMapByIDs returns the schedules by given id slice. -func GetSchedulesMapByIDs(ctx context.Context, ids []int64) (map[int64]*ActionSchedule, error) { - schedules := make(map[int64]*ActionSchedule, len(ids)) - if len(ids) == 0 { - return schedules, nil - } - return schedules, db.GetEngine(ctx).In("id", ids).Find(&schedules) -} - -// CreateScheduleTask creates new schedule task. -func CreateScheduleTask(ctx context.Context, rows []*ActionSchedule) error { +// CreateScheduleTaskBySchedules creates new schedule task. +func CreateScheduleTaskBySchedules(ctx context.Context, rows []*ActionSchedule) error { // Return early if there are no rows to insert if len(rows) == 0 { return nil @@ -74,13 +65,16 @@ func CreateScheduleTask(ctx context.Context, rows []*ActionSchedule) error { ScheduleID: row.ID, Spec: spec, } - // Parse the spec and check for errors schedule, err := specRow.Parse() if err != nil { - continue // skip to the next spec if there's an error + continue } - specRow.Next = timeutil.TimeStamp(schedule.Next(now).Unix()) + next := schedule.Next(now) + if next.IsZero() { + continue // the spec parses but can never occur, like "0 0 30 2 *" + } + specRow.Next = timeutil.TimeStamp(next.Unix()) // Insert the new schedule spec row if err = db.Insert(ctx, specRow); err != nil { diff --git a/models/actions/schedule_spec.go b/models/actions/schedule_spec.go index 3e127cec8c8..41e8a2ef8f8 100644 --- a/models/actions/schedule_spec.go +++ b/models/actions/schedule_spec.go @@ -26,7 +26,7 @@ type ActionScheduleSpec struct { // Next time the job will run, or the zero time if Cron has not been // started or this entry's schedule is unsatisfiable Next timeutil.TimeStamp `xorm:"index"` - // Prev is the last time this job was run, or the zero time if never. + // Prev is the occurrence this spec was last processed for, or the zero time if never. Prev timeutil.TimeStamp Spec string diff --git a/models/actions/schedule_spec_list.go b/models/actions/schedule_spec_list.go deleted file mode 100644 index 81994471f02..00000000000 --- a/models/actions/schedule_spec_list.go +++ /dev/null @@ -1,97 +0,0 @@ -// Copyright 2023 The Gitea Authors. All rights reserved. -// SPDX-License-Identifier: MIT - -package actions - -import ( - "context" - - "gitea.dev/models/db" - repo_model "gitea.dev/models/repo" - "gitea.dev/modules/container" - - "xorm.io/builder" -) - -type SpecList []*ActionScheduleSpec - -func (specs SpecList) GetScheduleIDs() []int64 { - return container.FilterSlice(specs, func(spec *ActionScheduleSpec) (int64, bool) { - return spec.ScheduleID, true - }) -} - -func (specs SpecList) LoadSchedules(ctx context.Context) error { - scheduleIDs := specs.GetScheduleIDs() - schedules, err := GetSchedulesMapByIDs(ctx, scheduleIDs) - if err != nil { - return err - } - for _, spec := range specs { - spec.Schedule = schedules[spec.ScheduleID] - } - - repoIDs := specs.GetRepoIDs() - repos, err := repo_model.GetRepositoriesMapByIDs(ctx, repoIDs) - if err != nil { - return err - } - for _, spec := range specs { - spec.Repo = repos[spec.RepoID] - } - - return nil -} - -func (specs SpecList) GetRepoIDs() []int64 { - return container.FilterSlice(specs, func(spec *ActionScheduleSpec) (int64, bool) { - return spec.RepoID, true - }) -} - -func (specs SpecList) LoadRepos(ctx context.Context) error { - repoIDs := specs.GetRepoIDs() - repos, err := repo_model.GetRepositoriesMapByIDs(ctx, repoIDs) - if err != nil { - return err - } - for _, spec := range specs { - spec.Repo = repos[spec.RepoID] - } - return nil -} - -type FindSpecOptions struct { - db.ListOptions - RepoID int64 - Next int64 -} - -func (opts FindSpecOptions) ToConds() builder.Cond { - cond := builder.NewCond() - if opts.RepoID > 0 { - cond = cond.And(builder.Eq{"repo_id": opts.RepoID}) - } - - if opts.Next > 0 { - cond = cond.And(builder.Lte{"next": opts.Next}) - } - - return cond -} - -func (opts FindSpecOptions) ToOrders() string { - return "`id` DESC" -} - -func FindSpecs(ctx context.Context, opts FindSpecOptions) (SpecList, int64, error) { - specs, total, err := db.FindAndCount[ActionScheduleSpec](ctx, opts) - if err != nil { - return nil, 0, err - } - - if err := SpecList(specs).LoadSchedules(ctx); err != nil { - return nil, 0, err - } - return specs, total, nil -} diff --git a/services/actions/notifier_helper.go b/services/actions/notifier_helper.go index 980b50fc88d..700750ccf2b 100644 --- a/services/actions/notifier_helper.go +++ b/services/actions/notifier_helper.go @@ -595,7 +595,7 @@ func handleSchedules( crons = append(crons, run) } - return actions_model.CreateScheduleTask(ctx, crons) + return actions_model.CreateScheduleTaskBySchedules(ctx, crons) } // DetectAndHandleSchedules detects the schedule workflows on the default branch and create schedule tasks diff --git a/services/actions/schedule_tasks.go b/services/actions/schedule_tasks.go index 0decf7b9689..0656767c985 100644 --- a/services/actions/schedule_tasks.go +++ b/services/actions/schedule_tasks.go @@ -22,6 +22,8 @@ import ( "gitea.dev/modules/timeutil" webhook_module "gitea.dev/modules/webhook" "gitea.dev/services/convert" + + "xorm.io/builder" ) // StartScheduleTasks start the task @@ -29,84 +31,85 @@ func StartScheduleTasks(ctx context.Context) error { return startTasks(ctx) } -// startTasks retrieves specifications in pages, creates a schedule task for each specification, -// and updates the specification's next run time and previous run time. -// The function returns an error if there's an issue with finding or updating the specifications. +// startTasks starts every due spec and returns an error if any of them failed. func startTasks(ctx context.Context) error { - // Set the page size - pageSize := 50 - - // Retrieve specs in pages until all specs have been retrieved + var failed int now := time.Now() - for page := 1; ; page++ { - // Retrieve the specs for the current page - specs, _, err := actions_model.FindSpecs(ctx, actions_model.FindSpecOptions{ - ListOptions: db.ListOptions{ - Page: page, - PageSize: pageSize, - }, - Next: now.Unix(), + err := db.Iterate(ctx, + builder.And(builder.Gt{"next": 0}, builder.Lte{"next": now.Unix()}), + func(ctx context.Context, row *actions_model.ActionScheduleSpec) error { + // one failing spec must not abort the pass, or a single broken workflow stops every other schedule + if err := startTask(ctx, row, now); err != nil { + failed++ + log.Error("start schedule spec %d (repo %d, schedule %d): %v", row.ID, row.RepoID, row.ScheduleID, err) + } + return nil }) - if err != nil { - return fmt.Errorf("find specs: %w", err) - } - - if err := specs.LoadRepos(ctx); err != nil { - return fmt.Errorf("LoadRepos: %w", err) - } - - // Loop through each spec and create a schedule task for it - for _, row := range specs { - if row.Repo.IsArchived { - // Skip if the repo is archived - continue - } - - cfg, err := row.Repo.GetUnit(ctx, unit.TypeActions) - if err != nil { - if repo_model.IsErrUnitTypeNotExist(err) { - // Skip the actions unit of this repo is disabled. - continue - } - return fmt.Errorf("GetUnit: %w", err) - } - if cfg.ActionsConfig().IsWorkflowDisabled(row.Schedule.WorkflowID) { - continue - } - - if err := CreateScheduleTask(ctx, row); err != nil { - log.Error("CreateScheduleTask: %v", err) - return err - } - - // Parse the spec - schedule, err := row.Parse() - if err != nil { - log.Error("Parse: %v", err) - return err - } - - // Update the spec's next run time and previous run time - row.Prev = row.Next - row.Next = timeutil.TimeStamp(schedule.Next(now.Add(1 * time.Minute)).Unix()) - if err := actions_model.UpdateScheduleSpec(ctx, row, "prev", "next"); err != nil { - log.Error("UpdateScheduleSpec: %v", err) - return err - } - } - - // Stop if all specs have been retrieved - if len(specs) < pageSize { - break - } + if err != nil { + return fmt.Errorf("iterate specs: %w", err) } + // surfaces as an admin notice through the cron task, once per occurrence rather than once per pass + if failed > 0 { + return fmt.Errorf("%d schedule(s) could not be started", failed) + } return nil } -// CreateScheduleTask creates a scheduled task from a cron action schedule spec. +// startTask advances the spec to its next occurrence before creating the run, so a failing workflow +// retries on its own schedule instead of on every pass, and a failed update cannot duplicate the run. +func startTask(ctx context.Context, row *actions_model.ActionScheduleSpec, now time.Time) error { + cronSchedule, err := row.Parse() + if err != nil { + return fmt.Errorf("parse %q: %w", row.Spec, err) + } + row.Prev = row.Next + row.Next = timeutil.TimeStamp(cronSchedule.Next(now.Add(time.Minute)).Unix()) + if err := actions_model.UpdateScheduleSpec(ctx, row, "prev", "next"); err != nil { + return fmt.Errorf("update spec: %w", err) + } + + // a spec whose schedule or repo row is gone is skipped, not reported on every occurrence + schedule, exist, err := db.GetByID[actions_model.ActionSchedule](ctx, row.ScheduleID) + if err != nil { + return fmt.Errorf("get schedule %d: %w", row.ScheduleID, err) + } else if !exist { + return nil + } + repo, exist, err := db.GetByID[repo_model.Repository](ctx, row.RepoID) + if err != nil { + return fmt.Errorf("get repo %d: %w", row.RepoID, err) + } else if !exist { + return nil + } + row.Schedule, row.Repo = schedule, repo + + // only archived repos are skipped; mirrors keep their schedules because a mirror is a normal repo + // for Actions, and nightly builds or scans of the mirrored code are a common reason to run one + if row.Repo.IsArchived { + return nil + } + + cfg, err := row.Repo.GetUnit(ctx, unit.TypeActions) + if err != nil { + if repo_model.IsErrUnitTypeNotExist(err) { + return nil + } + return fmt.Errorf("GetUnit: %w", err) + } + if cfg.ActionsConfig().IsWorkflowDisabled(row.Schedule.WorkflowID) { + return nil + } + + if err := CreateScheduleTaskBySpec(ctx, row); err != nil { + return fmt.Errorf("create run for %s workflow %q: %w", row.Repo.FullName(), row.Schedule.WorkflowID, err) + } + return nil +} + +// CreateScheduleTaskBySpec creates a scheduled task from a cron action schedule spec. // It creates an action run based on the schedule, inserts it into the database, and creates commit statuses for each job. -func CreateScheduleTask(ctx context.Context, spec *actions_model.ActionScheduleSpec) error { +func CreateScheduleTaskBySpec(ctx context.Context, spec *actions_model.ActionScheduleSpec) error { cron := spec.Schedule // Scheduled runs carry no webhook payload; synthesize what github.event.* expects. diff --git a/services/actions/schedule_tasks_test.go b/services/actions/schedule_tasks_test.go index d1e154a8a0f..1bfb60ff680 100644 --- a/services/actions/schedule_tasks_test.go +++ b/services/actions/schedule_tasks_test.go @@ -5,9 +5,15 @@ package actions import ( "testing" + "time" + actions_model "gitea.dev/models/actions" + "gitea.dev/models/db" + "gitea.dev/models/unittest" "gitea.dev/modules/json" api "gitea.dev/modules/structs" + "gitea.dev/modules/timeutil" + webhook_module "gitea.dev/modules/webhook" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -73,3 +79,47 @@ func TestWithScheduleInEventPayload(t *testing.T) { assert.Equal(t, payload, updated) }) } + +func TestStartTasks(t *testing.T) { + require.NoError(t, unittest.PrepareTestDatabase()) + + insertSchedule := func(repoID, ownerID int64, workflowID, cronSpec, content string, next timeutil.TimeStamp) *actions_model.ActionScheduleSpec { + schedule := &actions_model.ActionSchedule{ + Title: workflowID, + RepoID: repoID, + OwnerID: ownerID, + WorkflowID: workflowID, + TriggerUserID: 1, + Ref: "refs/heads/master", + CommitSHA: "c2d72f548424103f01ee1dc02889c1e2bff816b0", + Event: webhook_module.HookEventSchedule, + EventPayload: "{}", + Content: []byte(content), + } + require.NoError(t, db.Insert(t.Context(), schedule)) + spec := &actions_model.ActionScheduleSpec{RepoID: repoID, ScheduleID: schedule.ID, Spec: cronSpec, Next: next} + require.NoError(t, db.Insert(t.Context(), spec)) + return spec + } + + due := timeutil.TimeStamp(time.Now().Add(-time.Minute).Unix()) + validWorkflow := "jobs:\n job:\n runs-on: ubuntu-latest\n steps:\n - run: true\n" + + // specs are processed by ascending id, so the broken one runs first and used to abort the whole pass + broken := insertSchedule(1, 2, "broken.yml", "@every 1m", "this: [is: not: a: workflow", due) + valid := insertSchedule(4, 5, "valid.yml", "@every 1m", validWorkflow, due) + never := insertSchedule(4, 5, "never.yml", "0 0 30 2 *", validWorkflow, timeutil.TimeStamp(time.Time{}.Unix())) + + require.ErrorContains(t, startTasks(t.Context()), "1 schedule(s) could not be started") + + assert.Equal(t, 1, unittest.GetCount(t, &actions_model.ActionRun{RepoID: 4, WorkflowID: "valid.yml"})) + assert.Equal(t, 0, unittest.GetCount(t, &actions_model.ActionRun{RepoID: 1, WorkflowID: "broken.yml"})) + assert.Equal(t, 0, unittest.GetCount(t, &actions_model.ActionRun{RepoID: 4, WorkflowID: "never.yml"})) + + // the broken spec moves on too, so it does not fail again on every pass + for _, spec := range []*actions_model.ActionScheduleSpec{broken, valid} { + updated := unittest.AssertExistsAndLoadBean(t, &actions_model.ActionScheduleSpec{ID: spec.ID}) + assert.Greater(t, updated.Next, spec.Next) + } + assert.Equal(t, never.Next, unittest.AssertExistsAndLoadBean(t, &actions_model.ActionScheduleSpec{ID: never.ID}).Next) +}