feat: refactor schedule tasks management in actions service

- Remove `robfig/cron/v3 v3.0.1` from `go.mod`
- Remove `newSchedule()` and `resetSchedule()` functions from `services/actions/init.go`
- Change return type of `CreateScheduleTask()` from `(int, error)` to `error` in `services/actions/notifier_helper.go`
- Add `services/actions/schedule_tasks.go` file
- Add `StartScheduleTasks()` and `startTasks()` functions to `services/actions/schedule_tasks.go`
- Register `start_schedule_tasks` task in `services/cron/tasks_actions.go`

Signed-off-by: Bo-Yi Wu <appleboy.tw@gmail.com>

Author: appleboy

go.mod

@@ -89,7 +89,6 @@ require (
 	github.com/pquerna/otp v1.4.0
 	github.com/prometheus/client_golang v1.14.0
 	github.com/quasoft/websspi v1.1.2
-	github.com/robfig/cron/v3 v3.0.1
 	github.com/santhosh-tekuri/jsonschema/v5 v5.2.0
 	github.com/sergi/go-diff v1.3.1
 	github.com/shurcooL/vfsgen v0.0.0-20200824052919-0d455de96546

go.sum

@@ -1055,8 +1055,6 @@ github.com/rivo/uniseg v0.4.4/go.mod h1:FN3SvrM+Zdj16jyLfmOkMNblXMcoc8DfTHruCPUc
 github.com/robertkrimen/godocdown v0.0.0-20130622164427-0bfa04905481/go.mod h1:C9WhFzY47SzYBIvzFqSvHIR6ROgDo4TtdTuRaOMjF/s=
 github.com/robfig/cron v1.2.0 h1:ZjScXvvxeQ63Dbyxy76Fj3AT3Ut0aKsyd2/tl3DTMuQ=
 github.com/robfig/cron v1.2.0/go.mod h1:JGuDeoQd7Z6yL4zQhZ3OPEVHB7fL6Ka6skscFHfmt2k=
-github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
-github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
 github.com/rogpeppe/fastuuid v0.0.0-20150106093220-6724a57986af/go.mod h1:XWv6SoW27p1b0cqNHllgS5HIMJraePCO15w5zCzIWYg=
 github.com/rogpeppe/go-internal v1.1.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
 github.com/rogpeppe/go-internal v1.2.2/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=

services/actions/init.go

@@ -19,8 +19,4 @@ func Init() {
 	go graceful.GetManager().RunWithShutdownFns(jobEmitterQueue.Run)
 
 	notification.RegisterNotifier(NewNotifier())
-
-	// initial all schedule task
-	newSchedule()
-	resetSchedule()
 }

services/actions/notifier_helper.go

@@ -105,39 +105,36 @@ func (input *notifyInput) Notify(ctx context.Context) {
 	}
 }
 
-func CreateScheduleTask(ctx context.Context, cron *actions_model.ActionSchedule, spec string) (int, error) {
-	return schedule.AddFunc(spec, func() {
-		run := &actions_model.ActionRun{
-			Title:         cron.Title,
-			RepoID:        cron.RepoID,
-			OwnerID:       cron.OwnerID,
-			WorkflowID:    cron.WorkflowID,
-			TriggerUserID: cron.TriggerUserID,
-			Ref:           cron.Ref,
-			CommitSHA:     cron.CommitSHA,
-			Event:         cron.Event,
-			EventPayload:  cron.EventPayload,
-			Status:        actions_model.StatusWaiting,
-		}
-		jobs, err := jobparser.Parse(cron.Content)
-		if err != nil {
-			log.Error("jobparser.Parse: %v", err)
-			return
-		}
-		if err := actions_model.InsertRun(ctx, run, jobs); err != nil {
-			log.Error("InsertRun: %v", err)
-			return
-		}
-		if jobs, _, err := actions_model.FindRunJobs(ctx, actions_model.FindRunJobOptions{RunID: run.ID}); err != nil {
-			log.Error("FindRunJobs: %v", err)
-		} else {
-			for _, job := range jobs {
-				if err := CreateCommitStatus(ctx, job); err != nil {
-					log.Error("CreateCommitStatus: %v", err)
-				}
+func CreateScheduleTask(ctx context.Context, cron *actions_model.ActionSchedule, spec string) error {
+	run := &actions_model.ActionRun{
+		Title:         cron.Title,
+		RepoID:        cron.RepoID,
+		OwnerID:       cron.OwnerID,
+		WorkflowID:    cron.WorkflowID,
+		TriggerUserID: cron.TriggerUserID,
+		Ref:           cron.Ref,
+		CommitSHA:     cron.CommitSHA,
+		Event:         cron.Event,
+		EventPayload:  cron.EventPayload,
+		Status:        actions_model.StatusWaiting,
+	}
+	jobs, err := jobparser.Parse(cron.Content)
+	if err != nil {
+		return err
+	}
+	if err := actions_model.InsertRun(ctx, run, jobs); err != nil {
+		return err
+	}
+	if jobs, _, err := actions_model.FindRunJobs(ctx, actions_model.FindRunJobOptions{RunID: run.ID}); err != nil {
+		return err
+	} else {
+		for _, job := range jobs {
+			if err := CreateCommitStatus(ctx, job); err != nil {
+				return err
 			}
 		}
-	})
+	}
+	return nil
 }
 
 func handleSchedules(
@@ -156,10 +153,6 @@ func handleSchedules(
 		log.Error("FindCrons: %v", err)
 	}
 
-	for _, row := range rows {
-		schedule.Remove(row.EntryIDs)
-	}
-
 	if len(rows) > 0 {
 		if err := actions_model.DeleteScheduleTaskByRepo(ctx, input.Repo.ID); err != nil {
 			log.Error("DeleteCronTaskByRepo: %v", err)
@@ -209,15 +202,12 @@ func handleSchedules(
 
 	if len(crons) > 0 {
 		for _, cron := range crons {
-			entryIDs := []int{}
 			for _, spec := range cron.Specs {
-				id, err := CreateScheduleTask(ctx, cron, spec)
+				err := CreateScheduleTask(ctx, cron, spec)
 				if err != nil {
 					continue
 				}
-				entryIDs = append(entryIDs, id)
 			}
-			cron.EntryIDs = entryIDs
 		}
 
 		if err := actions_model.CreateScheduleTask(ctx, crons); err != nil {

services/actions/schedule.go

@@ -0,0 +1,51 @@
+// Copyright 2023 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package actions
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	actions_model "code.gitea.io/gitea/models/actions"
+	"code.gitea.io/gitea/modules/log"
+
+	"github.com/gogs/cron"
+)
+
+// StartScheduleTasks start the task
+func StartScheduleTasks(ctx context.Context) error {
+	return startTasks(ctx, actions_model.FindSpecOptions{
+		GetAll: true,
+	})
+}
+
+func startTasks(ctx context.Context, opts actions_model.FindSpecOptions) error {
+	specs, count, err := actions_model.FindSpecs(ctx, opts)
+	if err != nil {
+		return fmt.Errorf("find specs: %w", err)
+	}
+	if count == 0 {
+		return nil
+	}
+
+	now := time.Now()
+	for _, row := range specs {
+		schedule, err := cron.Parse(row.Spec)
+		if err != nil {
+			log.Error("ParseSpec: %v", err)
+			continue
+		}
+
+		next := schedule.Next(now)
+		if next.Sub(now) <= 60 {
+			if err := CreateScheduleTask(ctx, row.Schedule, row.Spec); err != nil {
+				log.Error("CreateScheduleTask: %v", err)
+			}
+		}
+
+	}
+
+	return nil
+}

services/actions/schedule_tasks.go

@@ -18,6 +18,17 @@ func initActionsTasks() {
 	registerStopZombieTasks()
 	registerStopEndlessTasks()
 	registerCancelAbandonedJobs()
+	registerScheduleTasks()
+}
+
+func registerScheduleTasks() {
+	RegisterTaskFatal("start_schedule_tasks", &BaseConfig{
+		Enabled:    true,
+		RunAtStart: true,
+		Schedule:   "@every 1m",
+	}, func(ctx context.Context, _ *user_model.User, cfg Config) error {
+		return actions_service.StartScheduleTasks(ctx)
+	})
 }
 
 func registerStopZombieTasks() {