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

models/actions/schedule.go

@@ -74,13 +74,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

@@ -49,33 +49,21 @@ func (specs SpecList) GetRepoIDs() []int64 {
 	})
 }
 
-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
+	Next     int64
+	BeforeID int64
 }
 
 func (opts FindSpecOptions) ToConds() builder.Cond {
-	cond := builder.NewCond()
-	if opts.RepoID > 0 {
-		cond = cond.And(builder.Eq{"repo_id": opts.RepoID})
-	}
+	cond := builder.NewCond().And(builder.Gt{"next": 0}) // an unsatisfiable spec stores a negative "next" and would stay due forever
 
 	if opts.Next > 0 {
 		cond = cond.And(builder.Lte{"next": opts.Next})
 	}
+	if opts.BeforeID > 0 {
+		cond = cond.And(builder.Lt{"id": opts.BeforeID})
+	}
 
 	return cond
 }

services/actions/schedule_tasks.go

@@ -5,6 +5,7 @@ package actions
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"maps"
 	"time"
@@ -29,78 +30,78 @@ 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
+	const pageSize = 50
 
-	// Retrieve specs in pages until all specs have been retrieved
 	now := time.Now()
-	for page := 1; ; page++ {
-		// Retrieve the specs for the current page
+	var failed int
+	var beforeID int64
+	for {
 		specs, _, err := actions_model.FindSpecs(ctx, actions_model.FindSpecOptions{
-			ListOptions: db.ListOptions{
-				Page:     page,
-				PageSize: pageSize,
-			},
-			Next: now.Unix(),
+			ListOptions: db.ListOptions{Page: 1, PageSize: pageSize},
+			Next:        now.Unix(),
+			BeforeID:    beforeID,
 		})
 		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
+			// 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)
 			}
+		}
 
-			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 len(specs) < pageSize {
+			break
+		}
+		beforeID = specs[len(specs)-1].ID // keyset, because advanced specs drop out of the "next <= now" filter
+	}
 
-			if err := CreateScheduleTask(ctx, row); err != nil {
-				log.Error("CreateScheduleTask: %v", err)
-				return 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
+}
 
-			// Parse the spec
-			schedule, err := row.Parse()
-			if err != nil {
-				log.Error("Parse: %v", err)
-				return err
-			}
+// 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 {
+	schedule, err := row.Parse()
+	if err != nil {
+		return fmt.Errorf("parse %q: %w", row.Spec, err)
+	}
+	row.Prev = row.Next
+	row.Next = timeutil.TimeStamp(schedule.Next(now.Add(time.Minute)).Unix())
+	if err := actions_model.UpdateScheduleSpec(ctx, row, "prev", "next"); err != nil {
+		return fmt.Errorf("update spec: %w", 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
-			}
-		}
+	if row.Repo == nil || row.Schedule == nil {
+		return errors.New("repo or schedule row is gone")
+	}
+	if row.Repo.IsArchived {
+		return nil
+	}
 
-		// Stop if all specs have been retrieved
-		if len(specs) < pageSize {
-			break
+	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 := CreateScheduleTask(ctx, row); err != nil {
+		return fmt.Errorf("create run for %s workflow %q: %w", row.Repo.FullName(), row.Schedule.WorkflowID, err)
+	}
 	return nil
 }
 

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 descending id, so the broken one runs first and used to abort the whole pass
+	valid := insertSchedule(4, 5, "valid.yml", "@every 1m", validWorkflow, due)
+	broken := insertSchedule(1, 2, "broken.yml", "@every 1m", "this: [is: not: a: workflow", 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{valid, broken} {
+		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)
+}