loading up the forgejo repo on tangled to test page performance
feat: move StopTask, CancelPreviousJobs and CleanRepoScheduleTasks to services/actions
This enables all action run state changes (from a not done to a done
state) to also send a notification.
Moved these:
- models/actions/task.go|423 col 6| func StopTask(ctx context.Context, taskID int64, status Status) error {
- models/actions/run.go|190 col 6| func CancelPreviousJobs(ctx context.Context, repoID int64, ref, workflowID string, event webhook_module.HookEventType) error {
- models/actions/schedule.go|122 col 6| func CleanRepoScheduleTasks(ctx context.Context, repo *repo_model.Repository, cancelPreviousJobs bool) error {
+156
-156
-69
models/actions/run.go
-69
models/actions/run.go
···
185
185
return err
186
186
}
187
187
188
-
// CancelPreviousJobs cancels all previous jobs of the same repository, reference, workflow, and event.
189
-
// It's useful when a new run is triggered, and all previous runs needn't be continued anymore.
190
-
func CancelPreviousJobs(ctx context.Context, repoID int64, ref, workflowID string, event webhook_module.HookEventType) error {
191
-
// Find all runs in the specified repository, reference, and workflow with non-final status
192
-
runs, total, err := db.FindAndCount[ActionRun](ctx, FindRunOptions{
193
-
RepoID: repoID,
194
-
Ref: ref,
195
-
WorkflowID: workflowID,
196
-
TriggerEvent: event,
197
-
Status: []Status{StatusRunning, StatusWaiting, StatusBlocked},
198
-
})
199
-
if err != nil {
200
-
return err
201
-
}
202
-
203
-
// If there are no runs found, there's no need to proceed with cancellation, so return nil.
204
-
if total == 0 {
205
-
return nil
206
-
}
207
-
208
-
// Iterate over each found run and cancel its associated jobs.
209
-
for _, run := range runs {
210
-
// Find all jobs associated with the current run.
211
-
jobs, err := db.Find[ActionRunJob](ctx, FindRunJobOptions{
212
-
RunID: run.ID,
213
-
})
214
-
if err != nil {
215
-
return err
216
-
}
217
-
218
-
// Iterate over each job and attempt to cancel it.
219
-
for _, job := range jobs {
220
-
// Skip jobs that are already in a terminal state (completed, cancelled, etc.).
221
-
status := job.Status
222
-
if status.IsDone() {
223
-
continue
224
-
}
225
-
226
-
// If the job has no associated task (probably an error), set its status to 'Cancelled' and stop it.
227
-
if job.TaskID == 0 {
228
-
job.Status = StatusCancelled
229
-
job.Stopped = timeutil.TimeStampNow()
230
-
231
-
// Update the job's status and stopped time in the database.
232
-
n, err := UpdateRunJob(ctx, job, builder.Eq{"task_id": 0}, "status", "stopped")
233
-
if err != nil {
234
-
return err
235
-
}
236
-
237
-
// If the update affected 0 rows, it means the job has changed in the meantime, so we need to try again.
238
-
if n == 0 {
239
-
return fmt.Errorf("job has changed, try again")
240
-
}
241
-
242
-
// Continue with the next job.
243
-
continue
244
-
}
245
-
246
-
// If the job has an associated task, try to stop the task, effectively cancelling the job.
247
-
if err := StopTask(ctx, job.TaskID, StatusCancelled); err != nil {
248
-
return err
249
-
}
250
-
}
251
-
}
252
-
253
-
// Return nil to indicate successful cancellation of all running and waiting jobs.
254
-
return nil
255
-
}
256
-
257
188
// InsertRun inserts a run
258
189
// The title will be cut off at 255 characters if it's longer than 255 characters.
259
190
func InsertRun(ctx context.Context, run *ActionRun, jobs []*jobparser.SingleWorkflow) error {
-22
models/actions/schedule.go
-22
models/actions/schedule.go
···
5
5
6
6
import (
7
7
"context"
8
-
"fmt"
9
8
"time"
10
9
11
10
"forgejo.org/models/db"
···
117
116
}
118
117
119
118
return committer.Commit()
120
-
}
121
-
122
-
func CleanRepoScheduleTasks(ctx context.Context, repo *repo_model.Repository, cancelPreviousJobs bool) error {
123
-
// If actions disabled when there is schedule task, this will remove the outdated schedule tasks
124
-
// There is no other place we can do this because the app.ini will be changed manually
125
-
if err := DeleteScheduleTaskByRepo(ctx, repo.ID); err != nil {
126
-
return fmt.Errorf("DeleteCronTaskByRepo: %v", err)
127
-
}
128
-
if cancelPreviousJobs {
129
-
// cancel running cron jobs of this repository and delete old schedules
130
-
if err := CancelPreviousJobs(
131
-
ctx,
132
-
repo.ID,
133
-
repo.DefaultBranch,
134
-
"",
135
-
webhook_module.HookEventSchedule,
136
-
); err != nil {
137
-
return fmt.Errorf("CancelPreviousJobs: %v", err)
138
-
}
139
-
}
140
-
return nil
141
119
}
142
120
143
121
type FindScheduleOptions struct {
-51
models/actions/task.go
-51
models/actions/task.go
···
420
420
return task, nil
421
421
}
422
422
423
-
func StopTask(ctx context.Context, taskID int64, status Status) error {
424
-
if !status.IsDone() {
425
-
return fmt.Errorf("cannot stop task with status %v", status)
426
-
}
427
-
e := db.GetEngine(ctx)
428
-
429
-
task := &ActionTask{}
430
-
if has, err := e.ID(taskID).Get(task); err != nil {
431
-
return err
432
-
} else if !has {
433
-
return util.ErrNotExist
434
-
}
435
-
if task.Status.IsDone() {
436
-
return nil
437
-
}
438
-
439
-
now := timeutil.TimeStampNow()
440
-
task.Status = status
441
-
task.Stopped = now
442
-
if _, err := UpdateRunJob(ctx, &ActionRunJob{
443
-
ID: task.JobID,
444
-
Status: task.Status,
445
-
Stopped: task.Stopped,
446
-
}, nil); err != nil {
447
-
return err
448
-
}
449
-
450
-
if err := UpdateTask(ctx, task, "status", "stopped"); err != nil {
451
-
return err
452
-
}
453
-
454
-
if err := task.LoadAttributes(ctx); err != nil {
455
-
return err
456
-
}
457
-
458
-
for _, step := range task.Steps {
459
-
if !step.Status.IsDone() {
460
-
step.Status = status
461
-
if step.Started == 0 {
462
-
step.Started = now
463
-
}
464
-
step.Stopped = now
465
-
}
466
-
if _, err := e.ID(step.ID).Update(step); err != nil {
467
-
return err
468
-
}
469
-
}
470
-
471
-
return nil
472
-
}
473
-
474
423
func FindOldTasksToExpire(ctx context.Context, olderThan timeutil.TimeStamp, limit int) ([]*ActionTask, error) {
475
424
e := db.GetEngine(ctx)
476
425
+1
-2
routers/api/v1/repo/repo.go
+1
-2
routers/api/v1/repo/repo.go
···
11
11
"strings"
12
12
"time"
13
13
14
-
actions_model "forgejo.org/models/actions"
15
14
activities_model "forgejo.org/models/activities"
16
15
"forgejo.org/models/db"
17
16
"forgejo.org/models/organization"
···
1065
1064
ctx.Error(http.StatusInternalServerError, "ArchiveRepoState", err)
1066
1065
return err
1067
1066
}
1068
-
if err := actions_model.CleanRepoScheduleTasks(ctx, repo, true); err != nil {
1067
+
if err := actions_service.CleanRepoScheduleTasks(ctx, repo, true); err != nil {
1069
1068
log.Error("CleanRepoScheduleTasks for archived repo %s/%s: %v", ctx.Repo.Owner.Name, repo.Name, err)
1070
1069
}
1071
1070
log.Trace("Repository was archived: %s/%s", ctx.Repo.Owner.Name, repo.Name)
+1
-1
routers/web/repo/actions/view.go
+1
-1
routers/web/repo/actions/view.go
+1
-2
routers/web/repo/setting/setting.go
+1
-2
routers/web/repo/setting/setting.go
···
14
14
"time"
15
15
16
16
"forgejo.org/models"
17
-
actions_model "forgejo.org/models/actions"
18
17
"forgejo.org/models/db"
19
18
"forgejo.org/models/organization"
20
19
quota_model "forgejo.org/models/quota"
···
1034
1033
return
1035
1034
}
1036
1035
1037
-
if err := actions_model.CleanRepoScheduleTasks(ctx, repo, true); err != nil {
1036
+
if err := actions_service.CleanRepoScheduleTasks(ctx, repo, true); err != nil {
1038
1037
log.Error("CleanRepoScheduleTasks for archived repo %s/%s: %v", ctx.Repo.Owner.Name, repo.Name, err)
1039
1038
}
1040
1039
+1
-1
services/actions/clear_tasks.go
+1
-1
services/actions/clear_tasks.go
···
41
41
jobs := make([]*actions_model.ActionRunJob, 0, len(tasks))
42
42
for _, task := range tasks {
43
43
if err := db.WithTx(ctx, func(ctx context.Context) error {
44
-
if err := actions_model.StopTask(ctx, task.ID, actions_model.StatusFailure); err != nil {
44
+
if err := StopTask(ctx, task.ID, actions_model.StatusFailure); err != nil {
45
45
return err
46
46
}
47
47
if err := task.LoadJob(ctx); err != nil {
+3
-3
services/actions/notifier_helper.go
+3
-3
services/actions/notifier_helper.go
···
139
139
return nil
140
140
}
141
141
if unit_model.TypeActions.UnitGlobalDisabled() {
142
-
if err := actions_model.CleanRepoScheduleTasks(ctx, input.Repo, true); err != nil {
142
+
if err := CleanRepoScheduleTasks(ctx, input.Repo, true); err != nil {
143
143
log.Error("CleanRepoScheduleTasks: %v", err)
144
144
}
145
145
return nil
···
373
373
// cancel running jobs if the event is push or pull_request_sync
374
374
if run.Event == webhook_module.HookEventPush ||
375
375
run.Event == webhook_module.HookEventPullRequestSync {
376
-
if err := actions_model.CancelPreviousJobs(
376
+
if err := CancelPreviousJobs(
377
377
ctx,
378
378
run.RepoID,
379
379
run.Ref,
···
504
504
log.Error("CountSchedules: %v", err)
505
505
return err
506
506
} else if count > 0 {
507
-
if err := actions_model.CleanRepoScheduleTasks(ctx, input.Repo, false); err != nil {
507
+
if err := CleanRepoScheduleTasks(ctx, input.Repo, false); err != nil {
508
508
log.Error("CleanRepoScheduleTasks: %v", err)
509
509
}
510
510
}
+92
-1
services/actions/schedule_tasks.go
+92
-1
services/actions/schedule_tasks.go
···
17
17
webhook_module "forgejo.org/modules/webhook"
18
18
19
19
"github.com/nektos/act/pkg/jobparser"
20
+
"xorm.io/builder"
20
21
)
21
22
22
23
// StartScheduleTasks start the task
···
55
56
// cancel running jobs if the event is push
56
57
if row.Schedule.Event == webhook_module.HookEventPush {
57
58
// cancel running jobs of the same workflow
58
-
if err := actions_model.CancelPreviousJobs(
59
+
if err := CancelPreviousJobs(
59
60
ctx,
60
61
row.RepoID,
61
62
row.Schedule.Ref,
···
152
153
// Return nil if no errors occurred
153
154
return nil
154
155
}
156
+
157
+
// CancelPreviousJobs cancels all previous jobs of the same repository, reference, workflow, and event.
158
+
// It's useful when a new run is triggered, and all previous runs needn't be continued anymore.
159
+
func CancelPreviousJobs(ctx context.Context, repoID int64, ref, workflowID string, event webhook_module.HookEventType) error {
160
+
// Find all runs in the specified repository, reference, and workflow with non-final status
161
+
runs, total, err := db.FindAndCount[actions_model.ActionRun](ctx, actions_model.FindRunOptions{
162
+
RepoID: repoID,
163
+
Ref: ref,
164
+
WorkflowID: workflowID,
165
+
TriggerEvent: event,
166
+
Status: []actions_model.Status{actions_model.StatusRunning, actions_model.StatusWaiting, actions_model.StatusBlocked},
167
+
})
168
+
if err != nil {
169
+
return err
170
+
}
171
+
172
+
// If there are no runs found, there's no need to proceed with cancellation, so return nil.
173
+
if total == 0 {
174
+
return nil
175
+
}
176
+
177
+
// Iterate over each found run and cancel its associated jobs.
178
+
for _, run := range runs {
179
+
// Find all jobs associated with the current run.
180
+
jobs, err := db.Find[actions_model.ActionRunJob](ctx, actions_model.FindRunJobOptions{
181
+
RunID: run.ID,
182
+
})
183
+
if err != nil {
184
+
return err
185
+
}
186
+
187
+
// Iterate over each job and attempt to cancel it.
188
+
for _, job := range jobs {
189
+
// Skip jobs that are already in a terminal state (completed, cancelled, etc.).
190
+
status := job.Status
191
+
if status.IsDone() {
192
+
continue
193
+
}
194
+
195
+
// If the job has no associated task (probably an error), set its status to 'Cancelled' and stop it.
196
+
if job.TaskID == 0 {
197
+
job.Status = actions_model.StatusCancelled
198
+
job.Stopped = timeutil.TimeStampNow()
199
+
200
+
// Update the job's status and stopped time in the database.
201
+
n, err := actions_model.UpdateRunJob(ctx, job, builder.Eq{"task_id": 0}, "status", "stopped")
202
+
if err != nil {
203
+
return err
204
+
}
205
+
206
+
// If the update affected 0 rows, it means the job has changed in the meantime, so we need to try again.
207
+
if n == 0 {
208
+
return fmt.Errorf("job has changed, try again")
209
+
}
210
+
211
+
// Continue with the next job.
212
+
continue
213
+
}
214
+
215
+
// If the job has an associated task, try to stop the task, effectively cancelling the job.
216
+
if err := StopTask(ctx, job.TaskID, actions_model.StatusCancelled); err != nil {
217
+
return err
218
+
}
219
+
}
220
+
}
221
+
222
+
// Return nil to indicate successful cancellation of all running and waiting jobs.
223
+
return nil
224
+
}
225
+
226
+
func CleanRepoScheduleTasks(ctx context.Context, repo *repo_model.Repository, cancelPreviousJobs bool) error {
227
+
// If actions disabled when there is schedule task, this will remove the outdated schedule tasks
228
+
// There is no other place we can do this because the app.ini will be changed manually
229
+
if err := actions_model.DeleteScheduleTaskByRepo(ctx, repo.ID); err != nil {
230
+
return fmt.Errorf("DeleteCronTaskByRepo: %v", err)
231
+
}
232
+
if cancelPreviousJobs {
233
+
// cancel running cron jobs of this repository and delete old schedules
234
+
if err := CancelPreviousJobs(
235
+
ctx,
236
+
repo.ID,
237
+
repo.DefaultBranch,
238
+
"",
239
+
webhook_module.HookEventSchedule,
240
+
); err != nil {
241
+
return fmt.Errorf("CancelPreviousJobs: %v", err)
242
+
}
243
+
}
244
+
return nil
245
+
}
+53
services/actions/task.go
+53
services/actions/task.go
···
10
10
actions_model "forgejo.org/models/actions"
11
11
"forgejo.org/models/db"
12
12
secret_model "forgejo.org/models/secret"
13
+
"forgejo.org/modules/timeutil"
14
+
"forgejo.org/modules/util"
13
15
14
16
runnerv1 "code.gitea.io/actions-proto-go/runner/v1"
15
17
"google.golang.org/protobuf/types/known/structpb"
···
105
107
}
106
108
return ret, nil
107
109
}
110
+
111
+
func StopTask(ctx context.Context, taskID int64, status actions_model.Status) error {
112
+
if !status.IsDone() {
113
+
return fmt.Errorf("cannot stop task with status %v", status)
114
+
}
115
+
e := db.GetEngine(ctx)
116
+
117
+
task := &actions_model.ActionTask{}
118
+
if has, err := e.ID(taskID).Get(task); err != nil {
119
+
return err
120
+
} else if !has {
121
+
return util.ErrNotExist
122
+
}
123
+
if task.Status.IsDone() {
124
+
return nil
125
+
}
126
+
127
+
now := timeutil.TimeStampNow()
128
+
task.Status = status
129
+
task.Stopped = now
130
+
if _, err := actions_model.UpdateRunJob(ctx, &actions_model.ActionRunJob{
131
+
ID: task.JobID,
132
+
Status: task.Status,
133
+
Stopped: task.Stopped,
134
+
}, nil); err != nil {
135
+
return err
136
+
}
137
+
138
+
if err := actions_model.UpdateTask(ctx, task, "status", "stopped"); err != nil {
139
+
return err
140
+
}
141
+
142
+
if err := task.LoadAttributes(ctx); err != nil {
143
+
return err
144
+
}
145
+
146
+
for _, step := range task.Steps {
147
+
if !step.Status.IsDone() {
148
+
step.Status = status
149
+
if step.Started == 0 {
150
+
step.Started = now
151
+
}
152
+
step.Stopped = now
153
+
}
154
+
if _, err := e.ID(step.ID).Update(step); err != nil {
155
+
return err
156
+
}
157
+
}
158
+
159
+
return nil
160
+
}
+3
-2
services/repository/branch.go
+3
-2
services/repository/branch.go
···
28
28
"forgejo.org/modules/timeutil"
29
29
"forgejo.org/modules/util"
30
30
webhook_module "forgejo.org/modules/webhook"
31
+
actions_service "forgejo.org/services/actions"
31
32
notify_service "forgejo.org/services/notify"
32
33
pull_service "forgejo.org/services/pull"
33
34
files_service "forgejo.org/services/repository/files"
···
377
378
log.Error("DeleteCronTaskByRepo: %v", err)
378
379
}
379
380
// cancel running cron jobs of this repository and delete old schedules
380
-
if err := actions_model.CancelPreviousJobs(
381
+
if err := actions_service.CancelPreviousJobs(
381
382
ctx,
382
383
repo.ID,
383
384
from,
···
578
579
log.Error("DeleteCronTaskByRepo: %v", err)
579
580
}
580
581
// cancel running cron jobs of this repository and delete old schedules
581
-
if err := actions_model.CancelPreviousJobs(
582
+
if err := actions_service.CancelPreviousJobs(
582
583
ctx,
583
584
repo.ID,
584
585
oldDefaultBranchName,
+1
-2
services/repository/setting.go
+1
-2
services/repository/setting.go
···
7
7
"context"
8
8
"slices"
9
9
10
-
actions_model "forgejo.org/models/actions"
11
10
"forgejo.org/models/db"
12
11
repo_model "forgejo.org/models/repo"
13
12
"forgejo.org/models/unit"
···
29
28
}
30
29
31
30
if slices.Contains(deleteUnitTypes, unit.TypeActions) {
32
-
if err := actions_model.CleanRepoScheduleTasks(ctx, repo, true); err != nil {
31
+
if err := actions_service.CleanRepoScheduleTasks(ctx, repo, true); err != nil {
33
32
log.Error("CleanRepoScheduleTasks: %v", err)
34
33
}
35
34
}