update

Author: Zettat123

models/actions/run.go

@@ -47,6 +47,7 @@ type ActionRun struct {
 	Status            Status                       `xorm:"index"`
 	Version           int                          `xorm:"version default 0"` // Status could be updated concomitantly, so an optimistic lock is needed
 	ConcurrencyGroup  string
+	ConcurrencyCancel bool
 	// Started and Stopped is used for recording last run time, if rerun happened, they will be reset to 0
 	Started timeutil.TimeStamp
 	Stopped timeutil.TimeStamp
@@ -270,7 +271,7 @@ func CancelPreviousJobsWithOpts(ctx context.Context, opts *FindRunOptions) error
 
 // InsertRun inserts a run
 // The title will be cut off at 255 characters if it's longer than 255 characters.
-func InsertRun(ctx context.Context, run *ActionRun, jobs []*jobparser.SingleWorkflow, blockedByConcurrency bool) error {
+func InsertRun(ctx context.Context, run *ActionRun, jobs []*jobparser.SingleWorkflow) error {
 	ctx, committer, err := db.TxContext(ctx)
 	if err != nil {
 		return err
@@ -284,6 +285,32 @@ func InsertRun(ctx context.Context, run *ActionRun, jobs []*jobparser.SingleWork
 	run.Index = index
 	run.Title, _ = util.SplitStringAtByteN(run.Title, 255)
 
+	blockedByWorkflowConcurrency := false
+	if len(run.ConcurrencyGroup) > 0 {
+		if run.ConcurrencyCancel {
+			if err := CancelPreviousJobsWithOpts(ctx, &FindRunOptions{
+				RepoID:           run.RepoID,
+				ConcurrencyGroup: run.ConcurrencyGroup,
+				Status:           []Status{StatusRunning, StatusWaiting, StatusBlocked},
+			}); err != nil {
+				return err
+			}
+		} else {
+			waitingConcurrentRunsNum, err := db.Count[ActionRun](ctx, &FindRunOptions{
+				RepoID:           run.RepoID,
+				ConcurrencyGroup: run.ConcurrencyGroup,
+				Status:           []Status{StatusWaiting},
+			})
+			if err != nil {
+				return err
+			}
+			blockedByWorkflowConcurrency = waitingConcurrentRunsNum > 0
+		}
+	}
+	if blockedByWorkflowConcurrency {
+		run.Status = StatusBlocked
+	}
+
 	if err := db.Insert(ctx, run); err != nil {
 		return err
 	}
@@ -310,13 +337,13 @@ func InsertRun(ctx context.Context, run *ActionRun, jobs []*jobparser.SingleWork
 		}
 		payload, _ := v.Marshal()
 		status := StatusWaiting
-		if len(needs) > 0 || run.NeedApproval {
+		if len(needs) > 0 || run.NeedApproval || blockedByWorkflowConcurrency {
 			status = StatusBlocked
 		} else {
 			hasWaiting = true
 		}
 		job.Name, _ = util.SplitStringAtByteN(job.Name, 255)
-		runJobs = append(runJobs, &ActionRunJob{
+		runJob := &ActionRunJob{
 			RunID:             run.ID,
 			RepoID:            run.RepoID,
 			OwnerID:           run.OwnerID,
@@ -328,7 +355,12 @@ func InsertRun(ctx context.Context, run *ActionRun, jobs []*jobparser.SingleWork
 			Needs:             needs,
 			RunsOn:            job.RunsOn(),
 			Status:            status,
-		})
+		}
+		if job.RawConcurrency != nil {
+			runJob.RawConcurrencyGroup = job.RawConcurrency.Group
+			runJob.RawConcurrencyCancel = job.RawConcurrency.CancelInProgress
+		}
+		runJobs = append(runJobs, runJob)
 	}
 	if err := db.Insert(ctx, runJobs); err != nil {
 		return err

models/actions/run_job.go

@@ -33,10 +33,16 @@ type ActionRunJob struct {
 	RunsOn            []string `xorm:"JSON TEXT"`
 	TaskID            int64    // the latest task of the job
 	Status            Status   `xorm:"index"`
-	Started           timeutil.TimeStamp
-	Stopped           timeutil.TimeStamp
-	Created           timeutil.TimeStamp `xorm:"created"`
-	Updated           timeutil.TimeStamp `xorm:"updated index"`
+
+	RawConcurrencyGroup  string // raw concurrency.group
+	RawConcurrencyCancel string // raw concurrency.cancel-in-progress
+	ConcurrencyGroup     string // interpolated concurrency.group
+	ConcurrencyCancel    bool   // interpolated concurrency.cancel-in-progress
+
+	Started timeutil.TimeStamp
+	Stopped timeutil.TimeStamp
+	Created timeutil.TimeStamp `xorm:"created"`
+	Updated timeutil.TimeStamp `xorm:"updated index"`
 }
 
 func init() {

modules/actions/workflows.go

@@ -21,10 +21,10 @@ import (
 )
 
 type DetectedWorkflow struct {
-	EntryName    string
-	TriggerEvent *jobparser.Event
-	Content      []byte
-	Concurrency  *jobparser.UninterpolatedConcurrency
+	EntryName      string
+	TriggerEvent   *jobparser.Event
+	Content        []byte
+	RawConcurrency *model.RawConcurrency
 }
 
 func init() {
@@ -96,17 +96,12 @@ func GetEventsFromContent(content []byte) ([]*jobparser.Event, error) {
 	return events, nil
 }
 
-func GetConcurrencyFromContent(content []byte) (*jobparser.UninterpolatedConcurrency, error) {
+func GetConcurrencyFromContent(content []byte) (*model.RawConcurrency, error) {
 	workflow, err := model.ReadWorkflow(bytes.NewReader(content))
 	if err != nil {
 		return nil, err
 	}
-	uc, err := jobparser.ParseRawConcurrency(&workflow.RawConcurrency)
-	if err != nil {
-		return nil, err
-	}
-
-	return uc, nil
+	return workflow.RawConcurrency, nil
 }
 
 func DetectWorkflows(
@@ -153,10 +148,10 @@ func DetectWorkflows(
 				}
 			} else if detectMatched(gitRepo, commit, triggedEvent, payload, evt) {
 				dwf := &DetectedWorkflow{
-					EntryName:    entry.Name(),
-					TriggerEvent: evt,
-					Content:      content,
-					Concurrency:  concurrency,
+					EntryName:      entry.Name(),
+					TriggerEvent:   evt,
+					Content:        content,
+					RawConcurrency: concurrency,
 				}
 				workflows = append(workflows, dwf)
 			}

routers/web/repo/actions/view.go

@@ -546,6 +546,7 @@ func Approve(ctx *context_module.Context) {
 		}
 		for _, job := range jobs {
 			if len(job.Needs) == 0 && job.Status.IsBlocked() {
+				// TODO: check concurrency
 				job.Status = actions_model.StatusWaiting
 				_, err := actions_model.UpdateRunJob(ctx, job, nil, "status")
 				if err != nil {
@@ -899,7 +900,7 @@ func Run(ctx *context_module.Context) {
 	}
 
 	// Insert the action run and its associated jobs into the database
-	if err := actions_model.InsertRun(ctx, run, workflows, false); err != nil {
+	if err := actions_model.InsertRun(ctx, run, workflows); err != nil {
 		ctx.ServerError("workflow", err)
 		return
 	}

services/actions/concurrency.go

@@ -333,14 +333,12 @@ func handleWorkflows(
 		}
 
 		// check workflow concurrency
-		wfGitCtx := jobparser.ToGitContext(GenerateGitContext(run, nil))
-		wfConcurrencyGroup, wfConcurrencyCancel := jobparser.InterpolateWorkflowConcurrency(dwf.Concurrency, wfGitCtx, vars)
-		if len(wfConcurrencyGroup) > 0 {
-			run.ConcurrencyGroup = wfConcurrencyGroup
-			if wfConcurrencyCancel {
-				if err := CancelActionRunByConcurrency(ctx, run); err != nil {
-					log.Error("CancelActionRunByConcurrency: %v", err)
-				}
+		if dwf.RawConcurrency != nil {
+			wfGitCtx := jobparser.ToGitContext(GenerateGitContext(run, nil))
+			wfConcurrencyGroup, wfConcurrencyCancel := jobparser.InterpolateWorkflowConcurrency(dwf.RawConcurrency, wfGitCtx, vars)
+			if len(wfConcurrencyGroup) > 0 {
+				run.ConcurrencyGroup = wfConcurrencyGroup
+				run.ConcurrencyCancel = wfConcurrencyCancel
 			}
 		}
 
@@ -350,21 +348,7 @@ func handleWorkflows(
 			continue
 		}
 
-		// cancel running jobs if the event is push or pull_request_sync
-		if run.Event == webhook_module.HookEventPush ||
-			run.Event == webhook_module.HookEventPullRequestSync {
-			if err := actions_model.CancelPreviousJobs(
-				ctx,
-				run.RepoID,
-				run.Ref,
-				run.WorkflowID,
-				run.Event,
-			); err != nil {
-				log.Error("CancelPreviousJobs: %v", err)
-			}
-		}
-
-		if err := actions_model.InsertRun(ctx, run, jobs, !wfConcurrencyCancel); err != nil {
+		if err := actions_model.InsertRun(ctx, run, jobs); err != nil {
 			log.Error("InsertRun: %v", err)
 			continue
 		}

services/actions/notifier_helper.go

@@ -145,7 +145,7 @@ func CreateScheduleTask(ctx context.Context, cron *actions_model.ActionSchedule)
 	}
 
 	// Insert the action run and its associated jobs into the database
-	if err := actions_model.InsertRun(ctx, run, workflows, false); err != nil {
+	if err := actions_model.InsertRun(ctx, run, workflows); err != nil {
 		return err
 	}