Prevent duplicate actions email

routers/web/repo/actions/view.go

@@ -429,6 +429,12 @@ func Rerun(ctx *context_module.Context) {
 			ctx.ServerError("UpdateRun", err)
 			return
 		}
+
+		if err := run.LoadAttributes(ctx); err != nil {
+			ctx.ServerError("run.LoadAttributes", err)
+			return
+		}
+		notify_service.WorkflowRunStatusUpdate(ctx, run.Repo, run.TriggerUser, run)
 	}
 
 	job, jobs := getRunJobs(ctx, runIndex, jobIndex)
@@ -485,7 +491,6 @@ func rerunJob(ctx *context_module.Context, job *actions_model.ActionRunJob, shou
 	}
 
 	actions_service.CreateCommitStatus(ctx, job)
-	actions_service.NotifyWorkflowRunStatusUpdateWithReload(ctx, job)
 	notify_service.WorkflowJobStatusUpdate(ctx, job.Run.Repo, job.Run.TriggerUser, job, nil)
 
 	return nil
@@ -560,9 +565,8 @@ func Cancel(ctx *context_module.Context) {
 	if len(updatedjobs) > 0 {
 		job := updatedjobs[0]
 		actions_service.NotifyWorkflowRunStatusUpdateWithReload(ctx, job)
-		notify_service.WorkflowRunStatusUpdate(ctx, job.Run.Repo, job.Run.TriggerUser, job.Run)
 	}
-	ctx.JSON(http.StatusOK, struct{}{})
+	ctx.JSONOK()
 }
 
 func Approve(ctx *context_module.Context) {
@@ -606,15 +610,14 @@ func Approve(ctx *context_module.Context) {
 	if len(updatedjobs) > 0 {
 		job := updatedjobs[0]
 		actions_service.NotifyWorkflowRunStatusUpdateWithReload(ctx, job)
-		notify_service.WorkflowRunStatusUpdate(ctx, job.Run.Repo, job.Run.TriggerUser, job.Run)
 	}
 
 	for _, job := range updatedjobs {
 		_ = job.LoadAttributes(ctx)
 		notify_service.WorkflowJobStatusUpdate(ctx, job.Run.Repo, job.Run.TriggerUser, job, nil)
 	}
 
-	ctx.JSON(http.StatusOK, struct{}{})
+	ctx.JSONOK()
 }
 
 func Delete(ctx *context_module.Context) {

services/actions/clear_tasks.go

@@ -42,10 +42,8 @@ func notifyWorkflowJobStatusUpdate(ctx context.Context, jobs []*actions_model.Ac
 			_ = job.LoadAttributes(ctx)
 			notify_service.WorkflowJobStatusUpdate(ctx, job.Run.Repo, job.Run.TriggerUser, job, nil)
 		}
-		if len(jobs) > 0 {
-			job := jobs[0]
-			notify_service.WorkflowRunStatusUpdate(ctx, job.Run.Repo, job.Run.TriggerUser, job.Run)
-		}
+		job := jobs[0]
+		notify_service.WorkflowRunStatusUpdate(ctx, job.Run.Repo, job.Run.TriggerUser, job.Run)
 	}
 }
 
@@ -101,7 +99,7 @@ func stopTasks(ctx context.Context, opts actions_model.FindTaskOptions) error {
 	return nil
 }
 
-// CancelAbandonedJobs cancels the jobs which have waiting status, but haven't been picked by a runner for a long time
+// CancelAbandonedJobs cancels jobs that have not been picked by any runner for a long time
 func CancelAbandonedJobs(ctx context.Context) error {
 	jobs, err := db.Find[actions_model.ActionRunJob](ctx, actions_model.FindRunJobOptions{
 		Statuses:      []actions_model.Status{actions_model.StatusWaiting, actions_model.StatusBlocked},
@@ -113,24 +111,40 @@ func CancelAbandonedJobs(ctx context.Context) error {
 	}
 
 	now := timeutil.TimeStampNow()
+
+	// Collect one job per run to send workflow run status update
+	updatedRuns := map[int64]*actions_model.ActionRunJob{}
+
 	for _, job := range jobs {
 		job.Status = actions_model.StatusCancelled
 		job.Stopped = now
 		updated := false
 		if err := db.WithTx(ctx, func(ctx context.Context) error {
 			n, err := actions_model.UpdateRunJob(ctx, job, nil, "status", "stopped")
-			updated = err == nil && n > 0
-			return err
+			if err != nil {
+				return err
+			}
+			if err := job.LoadAttributes(ctx); err != nil {
+				return err
+			}
+			updated = n > 0
+			if updated && job.Run.Status.IsDone() {
+				updatedRuns[job.RunID] = job
+			}
+			return nil
 		}); err != nil {
 			log.Warn("cancel abandoned job %v: %v", job.ID, err)
 			// go on
 		}
 		CreateCommitStatus(ctx, job)
 		if updated {
-			NotifyWorkflowRunStatusUpdateWithReload(ctx, job)
 			notify_service.WorkflowJobStatusUpdate(ctx, job.Run.Repo, job.Run.TriggerUser, job, nil)
 		}
 	}
 
+	for _, job := range updatedRuns {
+		notify_service.WorkflowRunStatusUpdate(ctx, job.Run.Repo, job.Run.TriggerUser, job.Run)
+	}
+
 	return nil
 }

services/mailer/mail_workflow_run.go

@@ -33,7 +33,18 @@ func generateMessageIDForActionsWorkflowRunStatusEmail(repo *repo_model.Reposito
 	return fmt.Sprintf("<%s/actions/runs/%d@%s>", repo.FullName(), run.Index, setting.Domain)
 }
 
-func composeAndSendActionsWorkflowRunStatusEmail(ctx context.Context, repo *repo_model.Repository, run *actions_model.ActionRun, sender *user_model.User, recipients []*user_model.User) {
+func composeAndSendActionsWorkflowRunStatusEmail(ctx context.Context, repo *repo_model.Repository, run *actions_model.ActionRun, sender *user_model.User, recipients []*user_model.User) error {
+	jobs, err := actions_model.GetRunJobsByRunID(ctx, run.ID)
+	if err != nil {
+		return err
+	}
+	for _, job := range jobs {
+		if !job.Status.IsDone() {
+			log.Debug("composeAndSendActionsWorkflowRunStatusEmail: A job is not done. Will not compose and send actions email.")
+			return nil
+		}
+	}
+
 	subject := "Run"
 	switch run.Status {
 	case actions_model.StatusFailure:
@@ -48,11 +59,6 @@ func composeAndSendActionsWorkflowRunStatusEmail(ctx context.Context, repo *repo
 	messageID := generateMessageIDForActionsWorkflowRunStatusEmail(repo, run)
 	metadataHeaders := generateMetadataHeaders(repo)
 
-	jobs, err := actions_model.GetRunJobsByRunID(ctx, run.ID)
-	if err != nil {
-		log.Error("GetRunJobsByRunID: %v", err)
-		return
-	}
 	sort.SliceStable(jobs, func(i, j int) bool {
 		si, sj := jobs[i].Status, jobs[j].Status
 		/*
@@ -111,11 +117,11 @@ func composeAndSendActionsWorkflowRunStatusEmail(ctx context.Context, repo *repo
 			"Jobs":          convertedJobs,
 			"locale":        locale,
 		}); err != nil {
-			log.Error("ExecuteTemplate [%s]: %v", tplWorkflowRun, err)
-			return
+			return err
 		}
 		msgs := make([]*sender_service.Message, 0, len(tos))
 		for _, rec := range tos {
+			log.Trace("Sending actions email to %s (UID: %d)", rec.Name, rec.ID)
 			msg := sender_service.NewMessageFrom(
 				rec.Email,
 				displayName,
@@ -135,14 +141,16 @@ func composeAndSendActionsWorkflowRunStatusEmail(ctx context.Context, repo *repo
 		}
 		SendAsync(msgs...)
 	}
+
+	return nil
 }
 
-func MailActionsTrigger(ctx context.Context, sender *user_model.User, repo *repo_model.Repository, run *actions_model.ActionRun) {
+func MailActionsTrigger(ctx context.Context, sender *user_model.User, repo *repo_model.Repository, run *actions_model.ActionRun) error {
 	if setting.MailService == nil {
-		return
+		return nil
 	}
-	if run.Status.IsSkipped() {
-		return
+	if !run.Status.IsDone() || run.Status.IsSkipped() {
+		return nil
 	}
 
 	recipients := make([]*user_model.User, 0)
@@ -151,15 +159,16 @@ func MailActionsTrigger(ctx context.Context, sender *user_model.User, repo *repo
 		notifyPref, err := user_model.GetUserSetting(ctx, sender.ID,
 			user_model.SettingsKeyEmailNotificationGiteaActions, user_model.SettingEmailNotificationGiteaActionsFailureOnly)
 		if err != nil {
-			log.Error("GetUserSetting: %v", err)
-			return
+			return err
 		}
 		if notifyPref == user_model.SettingEmailNotificationGiteaActionsAll || !run.Status.IsSuccess() && notifyPref != user_model.SettingEmailNotificationGiteaActionsDisabled {
 			recipients = append(recipients, sender)
 		}
 	}
 
 	if len(recipients) > 0 {
-		composeAndSendActionsWorkflowRunStatusEmail(ctx, repo, run, sender, recipients)
+		log.Debug("MailActionsTrigger: Initiate email composition")
+		return composeAndSendActionsWorkflowRunStatusEmail(ctx, repo, run, sender, recipients)
 	}
+	return nil
 }

services/mailer/notify.go

@@ -208,8 +208,7 @@ func (m *mailNotifier) RepoPendingTransfer(ctx context.Context, doer, newOwner *
 }
 
 func (m *mailNotifier) WorkflowRunStatusUpdate(ctx context.Context, repo *repo_model.Repository, sender *user_model.User, run *actions_model.ActionRun) {
-	if !run.Status.IsDone() {
-		return
+	if err := MailActionsTrigger(ctx, sender, repo, run); err != nil {
+		log.Error("MailActionsTrigger: %v", err)
 	}
-	MailActionsTrigger(ctx, sender, repo, run)
 }