fix: a corrupted schedule task is disabled

The following errors are specific to the scheduled workflow being
handled. They do not imply the remaining scheduled workflows cannot
be scheduled successfully.

- Failure to create a scheduled task which is most likely caused
  by an invalid YAML file.
- Failure to parse the scheduling specs which can be caused by a
  number of formating errors.

Instead of returning on error, the corrupted workflow is disabled.

Also display more informative error messages so that the failed
workflow can be identified from the logged error.

models/repo/repo_unit.go

@@ -336,5 +336,8 @@ func getUnitsByRepoID(ctx context.Context, repoID int64) (units []*RepoUnit, err
 // UpdateRepoUnit updates the provided repo unit
 func UpdateRepoUnit(ctx context.Context, unit *RepoUnit) error {
 	_, err := db.GetEngine(ctx).ID(unit.ID).Update(unit)
-	return err
+	if err != nil {
+		return fmt.Errorf("UpdateRepoUnit: %v", err)
+	}
+	return nil
 }

services/actions/TestServiceActions_startTask/action_schedule.yml

@@ -0,0 +1,41 @@
+# A corrupted cron spec with a valid schedule workflow
+-
+  id: 1
+  title: schedule_title1
+  specs:
+    - '* * * * *'
+  repo_id: 4
+  owner_id: 2
+  workflow_id: 'workflow1.yml'
+  trigger_user_id: 2
+  ref: main
+  commit_sha: shashasha
+  event: "schedule"
+  event_payload: "fakepayload"
+  content: |
+    jobs:
+      job2:
+        runs-on: ubuntu-latest
+        steps:
+          - run: true
+
+# A valid cron spec with a corrupted schedule workflow
+-
+  id: 2
+  title: schedule_title2
+  specs:
+    - '* * * * *'
+  repo_id: 4
+  owner_id: 2
+  workflow_id: 'workflow2.yml'
+  trigger_user_id: 2
+  ref: main
+  commit_sha: shashasha
+  event: "schedule"
+  event_payload: "fakepayload"
+  content: |
+    jobs:
+      job2: { invalid yaml
+        runs-on: ubuntu-latest
+        steps:
+          - run: true

services/actions/TestServiceActions_startTask/action_schedule_spec.yml

@@ -0,0 +1,15 @@
+# A corrupted cron spec with a valid schedule workflow
+-
+  id: 1
+  repo_id: 4
+  schedule_id: 1
+  next: 1
+  spec: 'corrupted * *'
+
+# A valid cron spec with a corrupted schedule workflow
+-
+  id: 2
+  repo_id: 4
+  schedule_id: 2
+  next: 1
+  spec: '* * * * *'

services/actions/schedule_tasks.go

@@ -20,6 +20,7 @@ import (
 
 	"code.forgejo.org/forgejo/runner/v9/act/jobparser"
 	act_model "code.forgejo.org/forgejo/runner/v9/act/model"
+	"github.com/robfig/cron/v3"
 	"xorm.io/builder"
 )
 
@@ -83,20 +84,33 @@ func startTasks(ctx context.Context) error {
 				}
 				return fmt.Errorf("GetUnit: %w", err)
 			}
-			if cfg.ActionsConfig().IsWorkflowDisabled(row.Schedule.WorkflowID) {
+			actionConfig := cfg.ActionsConfig()
+			if actionConfig.IsWorkflowDisabled(row.Schedule.WorkflowID) {
 				continue
 			}
 
-			if err := CreateScheduleTask(ctx, row.Schedule); err != nil {
-				log.Error("CreateScheduleTask: %v", err)
-				return err
+			createAndSchedule := func(row *actions_model.ActionScheduleSpec) (cron.Schedule, error) {
+				if err := CreateScheduleTask(ctx, row.Schedule); err != nil {
+					return nil, fmt.Errorf("CreateScheduleTask: %v", err)
+				}
+
+				// Parse the spec
+				schedule, err := row.Parse()
+				if err != nil {
+					return nil, fmt.Errorf("Parse(Spec=%v): %v", row.Spec, err)
+				}
+				return schedule, nil
 			}
 
-			// Parse the spec
-			schedule, err := row.Parse()
+			schedule, err := createAndSchedule(row)
 			if err != nil {
-				log.Error("Parse: %v", err)
-				return err
+				log.Error("RepoID=%v WorkflowID=%v: %v", row.Schedule.RepoID, row.Schedule.WorkflowID, err)
+				actionConfig.DisableWorkflow(row.Schedule.WorkflowID)
+				if err := repo_model.UpdateRepoUnit(ctx, cfg); err != nil {
+					log.Error("RepoID=%v WorkflowID=%v: CreateScheduleTask: %v", row.Schedule.RepoID, row.Schedule.WorkflowID, err)
+					return err
+				}
+				continue
 			}
 
 			// Update the spec's next run time and previous run time

services/actions/schedule_tasks_test.go

@@ -7,7 +7,9 @@ import (
 	"testing"
 
 	actions_model "forgejo.org/models/actions"
+	"forgejo.org/models/db"
 	repo_model "forgejo.org/models/repo"
+	"forgejo.org/models/unit"
 	"forgejo.org/models/unittest"
 	webhook_module "forgejo.org/modules/webhook"
 
@@ -15,6 +17,58 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+func TestServiceActions_startTask(t *testing.T) {
+	defer unittest.OverrideFixtures("services/actions/TestServiceActions_startTask")()
+	require.NoError(t, unittest.PrepareTestDatabase())
+
+	// Load fixtures that are corrupted and create one valid scheduled workflow
+	repo := unittest.AssertExistsAndLoadBean(t, &repo_model.Repository{ID: 4})
+
+	workflowID := "some.yml"
+	schedules := []*actions_model.ActionSchedule{
+		{
+			Title:         "scheduletitle1",
+			RepoID:        repo.ID,
+			OwnerID:       repo.OwnerID,
+			WorkflowID:    workflowID,
+			TriggerUserID: repo.OwnerID,
+			Ref:           "branch",
+			CommitSHA:     "fakeSHA",
+			Event:         webhook_module.HookEventSchedule,
+			EventPayload:  "fakepayload",
+			Specs:         []string{"* * * * *"},
+			Content: []byte(
+				`
+jobs:
+  job2:
+    runs-on: ubuntu-latest
+    steps:
+      - run: true
+`),
+		},
+	}
+
+	require.Equal(t, 2, unittest.GetCount(t, actions_model.ActionScheduleSpec{}))
+	require.NoError(t, actions_model.CreateScheduleTask(t.Context(), schedules))
+	require.Equal(t, 3, unittest.GetCount(t, actions_model.ActionScheduleSpec{}))
+	_, err := db.GetEngine(db.DefaultContext).Exec("UPDATE `action_schedule_spec` SET next = 1")
+	require.NoError(t, err)
+
+	// After running startTasks an ActionRun row is created for the valid scheduled workflow
+	require.Empty(t, unittest.GetCount(t, actions_model.ActionRun{WorkflowID: workflowID}))
+	require.NoError(t, startTasks(t.Context()))
+	require.NotEmpty(t, unittest.GetCount(t, actions_model.ActionRun{WorkflowID: workflowID}))
+
+	// The invalid workflows loaded from the fixtures are disabled
+	repo = unittest.AssertExistsAndLoadBean(t, &repo_model.Repository{ID: 4})
+	actionUnit, err := repo.GetUnit(t.Context(), unit.TypeActions)
+	require.NoError(t, err)
+	actionConfig := actionUnit.ActionsConfig()
+	assert.True(t, actionConfig.IsWorkflowDisabled("workflow2.yml"))
+	assert.True(t, actionConfig.IsWorkflowDisabled("workflow1.yml"))
+	assert.False(t, actionConfig.IsWorkflowDisabled("some.yml"))
+}
+
 func TestCreateScheduleTask(t *testing.T) {
 	require.NoError(t, unittest.PrepareTestDatabase())
 	repo := unittest.AssertExistsAndLoadBean(t, &repo_model.Repository{ID: 2, OwnerID: 2})