fix(actions): run every due schedule exactly once per occurrence (#39078)

Signed-off-by: Lunny Xiao <xiaolunwen@gmail.com>
Co-authored-by: silverwind <me@silverwind.io>
Co-authored-by: Lunny Xiao <xiaolunwen@gmail.com>
Co-authored-by: wxiaoguang <wxiaoguang@gmail.com>

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 {

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
 

models/actions/schedule_spec_list.go

@@ -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
-}

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

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.

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)
+}