Skip to content

Commit 96d6938

Browse files
Zettat123Zettat123
committed
fix bugs and tests
1 parent 63eaf15 commit 96d6938

File tree

4 files changed

+69
-64
lines changed

4 files changed

+69
-64
lines changed

‎models/actions/run.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -404,7 +404,7 @@ func ShouldBlockRunByConcurrency(ctx context.Context, actionRun *ActionRun) (boo
404404
concurrentRuns, err := db.Find[ActionRun](ctx, &FindRunOptions{
405405
RepoID: actionRun.RepoID,
406406
ConcurrencyGroup: actionRun.ConcurrencyGroup,
407-
Status: []Status{StatusWaiting, StatusRunning},
407+
Status: []Status{StatusRunning},
408408
})
409409
if err != nil {
410410
return false, fmt.Errorf("find running and waiting runs: %w", err)

‎models/actions/run_job.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -222,7 +222,7 @@ func ShouldBlockJobByConcurrency(ctx context.Context, job *ActionRunJob) (bool,
222222
concurrentJobsNum, err := db.Count[ActionRunJob](ctx, FindRunJobOptions{
223223
RepoID: job.RepoID,
224224
ConcurrencyGroup: job.ConcurrencyGroup,
225-
Statuses: []Status{StatusRunning, StatusWaiting},
225+
Statuses: []Status{StatusRunning},
226226
})
227227
if err != nil {
228228
return false, fmt.Errorf("count running and waiting jobs: %w", err)

‎services/actions/job_emitter.go‎

Lines changed: 14 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -285,12 +285,17 @@ func (r *jobStatusResolver) resolve(ctx context.Context) map[int64]actions_model
285285
}
286286
}
287287
if allDone {
288-
// evaluate concurrency
289-
if err := evaluateConcurrencyForJobWithNeeds(ctx, r.jobMap[id], r.vars); err != nil {
288+
// check concurrency
289+
blockedByJobConcurrency, err := checkConcurrencyForJobWithNeeds(ctx, r.jobMap[id], r.vars)
290+
if err != nil {
290291
log.Error("Check job %d concurrency: %v. This job will stay blocked.", id, err)
291292
continue
292293
}
293294

295+
if blockedByJobConcurrency {
296+
continue
297+
}
298+
294299
if err := CancelJobsByJobConcurrency(ctx, r.jobMap[id]); err != nil {
295300
log.Error("Cancel previous jobs for job %d: %v", id, err)
296301
}
@@ -304,7 +309,6 @@ func (r *jobStatusResolver) resolve(ctx context.Context) map[int64]actions_model
304309
_, wfJob := wfJobs[0].Job()
305310
hasIf = len(wfJob.If.Value) > 0
306311
}
307-
308312
if hasIf {
309313
// act_runner will check the "if" condition
310314
ret[id] = actions_model.StatusWaiting
@@ -319,18 +323,18 @@ func (r *jobStatusResolver) resolve(ctx context.Context) map[int64]actions_model
319323
return ret
320324
}
321325

322-
func evaluateConcurrencyForJobWithNeeds(ctx context.Context, actionRunJob *actions_model.ActionRunJob, vars map[string]string) error {
326+
func checkConcurrencyForJobWithNeeds(ctx context.Context, actionRunJob *actions_model.ActionRunJob, vars map[string]string) (bool, error) {
323327
if actionRunJob.RawConcurrencyGroup == "" {
324-
return nil
328+
return false, nil
325329
}
326330
if err := actionRunJob.LoadAttributes(ctx); err != nil {
327-
return err
331+
return false, err
328332
}
329333

330334
if !actionRunJob.IsConcurrencyEvaluated {
331335
taskNeeds, err := FindTaskNeeds(ctx, actionRunJob)
332336
if err != nil {
333-
return fmt.Errorf("find task needs: %w", err)
337+
return false, fmt.Errorf("find task needs: %w", err)
334338
}
335339
jobResults := make(map[string]*jobparser.JobResult, len(taskNeeds))
336340
for jobID, taskNeed := range taskNeeds {
@@ -343,14 +347,14 @@ func evaluateConcurrencyForJobWithNeeds(ctx context.Context, actionRunJob *actio
343347

344348
actionRunJob.ConcurrencyGroup, actionRunJob.ConcurrencyCancel, err = EvaluateJobConcurrency(ctx, actionRunJob.Run, actionRunJob, vars, jobResults)
345349
if err != nil {
346-
return fmt.Errorf("evaluate job concurrency: %w", err)
350+
return false, fmt.Errorf("evaluate job concurrency: %w", err)
347351
}
348352
actionRunJob.IsConcurrencyEvaluated = true
349353

350354
if _, err := actions_model.UpdateRunJob(ctx, actionRunJob, nil, "concurrency_group", "concurrency_cancel", "is_concurrency_evaluated"); err != nil {
351-
return fmt.Errorf("update run job: %w", err)
355+
return false, fmt.Errorf("update run job: %w", err)
352356
}
353357
}
354358

355-
return nil
359+
return actions_model.ShouldBlockJobByConcurrency(ctx, actionRunJob)
356360
}

‎tests/integration/actions_concurrency_test.go‎

Lines changed: 53 additions & 52 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,6 @@ import (
88
"fmt"
99
"net/http"
1010
"net/url"
11-
"slices"
1211
"testing"
1312
"time"
1413

@@ -91,14 +90,10 @@ jobs:
9190
steps:
9291
- run: echo 'job from workflow3'
9392
`
93+
// push workflow1
9494
opts1 := getWorkflowCreateFileOptions(user2, repo.DefaultBranch, "create "+wf1TreePath, wf1FileContent)
9595
createWorkflowFile(t, token, user2.Name, repo.Name, wf1TreePath, opts1)
96-
opts2 := getWorkflowCreateFileOptions(user2, repo.DefaultBranch, "create "+wf2TreePath, wf2FileContent)
97-
createWorkflowFile(t, token, user2.Name, repo.Name, wf2TreePath, opts2)
98-
opts3 := getWorkflowCreateFileOptions(user2, repo.DefaultBranch, "create "+wf3TreePath, wf3FileContent)
99-
createWorkflowFile(t, token, user2.Name, repo.Name, wf3TreePath, opts3)
100-
101-
// fetch and exec workflow1, workflow2 and workflow3 are blocked
96+
// fetch and exec workflow1
10297
task := runner.fetchTask(t)
10398
_, _, run := getTaskAndJobAndRunByTaskID(t, task.Id)
10499
assert.Equal(t, "workflow-main-abc123-user2", run.ConcurrencyGroup)
@@ -108,23 +103,30 @@ jobs:
108103
result: runnerv1.Result_RESULT_SUCCESS,
109104
})
110105

111-
// fetch workflow2 or workflow3
112-
workflowNames := []string{"concurrent-workflow-2.yml", "concurrent-workflow-3.yml"}
106+
// push workflow2
107+
opts2 := getWorkflowCreateFileOptions(user2, repo.DefaultBranch, "create "+wf2TreePath, wf2FileContent)
108+
createWorkflowFile(t, token, user2.Name, repo.Name, wf2TreePath, opts2)
109+
// fetch workflow2
113110
task = runner.fetchTask(t)
114111
_, _, run = getTaskAndJobAndRunByTaskID(t, task.Id)
115-
assert.Contains(t, workflowNames, run.WorkflowID)
116-
workflowNames = slices.DeleteFunc(workflowNames, func(wfn string) bool { return wfn == run.WorkflowID })
117112
assert.Equal(t, "workflow-main-abc123-user2", run.ConcurrencyGroup)
113+
assert.Equal(t, "concurrent-workflow-2.yml", run.WorkflowID)
114+
115+
// push workflow3
116+
opts3 := getWorkflowCreateFileOptions(user2, repo.DefaultBranch, "create "+wf3TreePath, wf3FileContent)
117+
createWorkflowFile(t, token, user2.Name, repo.Name, wf3TreePath, opts3)
118118
runner.fetchNoTask(t)
119+
120+
// exec workflow2
119121
runner.execTask(t, task, &mockTaskOutcome{
120122
result: runnerv1.Result_RESULT_SUCCESS,
121123
})
122124

123-
// fetch the last workflow (workflow2 or workflow3)
125+
// fetch and exec workflow3
124126
task = runner.fetchTask(t)
125127
_, _, run = getTaskAndJobAndRunByTaskID(t, task.Id)
126128
assert.Equal(t, "workflow-main-abc123-user2", run.ConcurrencyGroup)
127-
assert.Equal(t, workflowNames[0], run.WorkflowID)
129+
assert.Equal(t, "concurrent-workflow-3.yml", run.WorkflowID)
128130
runner.fetchNoTask(t)
129131
runner.execTask(t, task, &mockTaskOutcome{
130132
result: runnerv1.Result_RESULT_SUCCESS,
@@ -425,43 +427,56 @@ on:
425427
paths:
426428
- '.gitea/workflows/concurrent-workflow-1.yml'
427429
jobs:
428-
job1:
430+
wf1-job:
429431
runs-on: ${{ matrix.os }}-runner
430432
strategy:
431433
matrix:
432434
os: [windows, linux]
433435
concurrency:
434436
group: job-os-${{ matrix.os }}
435437
steps:
436-
- run: echo 'job1'
437-
job2:
438+
- run: echo 'wf1'
439+
`
440+
441+
wf2TreePath := ".gitea/workflows/concurrent-workflow-2.yml"
442+
wf2FileContent := `name: concurrent-workflow-2
443+
on:
444+
push:
445+
paths:
446+
- '.gitea/workflows/concurrent-workflow-2.yml'
447+
jobs:
448+
wf2-job:
438449
runs-on: ${{ matrix.os }}-runner
439450
strategy:
440451
matrix:
441452
os: [darwin, windows, linux]
442453
concurrency:
443454
group: job-os-${{ matrix.os }}
444455
steps:
445-
- run: echo 'job2'
456+
- run: echo 'wf2'
446457
`
447458

448-
opts1 := getWorkflowCreateFileOptions(user2, repo.DefaultBranch, "create %s"+wf1TreePath, wf1FileContent)
459+
opts1 := getWorkflowCreateFileOptions(user2, repo.DefaultBranch, "create "+wf1TreePath, wf1FileContent)
449460
createWorkflowFile(t, token, user2.Name, repo.Name, wf1TreePath, opts1)
450461

451462
job1WinTask := windowsRunner.fetchTask(t)
452463
job1LinuxTask := linuxRunner.fetchTask(t)
453464
windowsRunner.fetchNoTask(t)
454465
linuxRunner.fetchNoTask(t)
455-
job2DarwinTask := darwinRunner.fetchTask(t)
456466
_, job1WinJob, _ := getTaskAndJobAndRunByTaskID(t, job1WinTask.Id)
457-
assert.Equal(t, "job1 (windows)", job1WinJob.Name)
467+
assert.Equal(t, "wf1-job (windows)", job1WinJob.Name)
458468
assert.Equal(t, "job-os-windows", job1WinJob.ConcurrencyGroup)
459469
_, job1LinuxJob, _ := getTaskAndJobAndRunByTaskID(t, job1LinuxTask.Id)
460-
assert.Equal(t, "job1 (linux)", job1LinuxJob.Name)
470+
assert.Equal(t, "wf1-job (linux)", job1LinuxJob.Name)
461471
assert.Equal(t, "job-os-linux", job1LinuxJob.ConcurrencyGroup)
472+
473+
opts2 := getWorkflowCreateFileOptions(user2, repo.DefaultBranch, "create "+wf2TreePath, wf2FileContent)
474+
createWorkflowFile(t, token, user2.Name, repo.Name, wf2TreePath, opts2)
475+
job2DarwinTask := darwinRunner.fetchTask(t)
462476
_, job2DarwinJob, _ := getTaskAndJobAndRunByTaskID(t, job2DarwinTask.Id)
463-
assert.Equal(t, "job2 (darwin)", job2DarwinJob.Name)
477+
assert.Equal(t, "wf2-job (darwin)", job2DarwinJob.Name)
464478
assert.Equal(t, "job-os-darwin", job2DarwinJob.ConcurrencyGroup)
479+
465480
windowsRunner.execTask(t, job1WinTask, &mockTaskOutcome{
466481
result: runnerv1.Result_RESULT_SUCCESS,
467482
})
@@ -472,10 +487,10 @@ jobs:
472487
job2WinTask := windowsRunner.fetchTask(t)
473488
job2LinuxTask := linuxRunner.fetchTask(t)
474489
_, job2WinJob, _ := getTaskAndJobAndRunByTaskID(t, job2WinTask.Id)
475-
assert.Equal(t, "job2 (windows)", job2WinJob.Name)
490+
assert.Equal(t, "wf2-job (windows)", job2WinJob.Name)
476491
assert.Equal(t, "job-os-windows", job2WinJob.ConcurrencyGroup)
477492
_, job2LinuxJob, _ := getTaskAndJobAndRunByTaskID(t, job2LinuxTask.Id)
478-
assert.Equal(t, "job2 (linux)", job2LinuxJob.Name)
493+
assert.Equal(t, "wf2-job (linux)", job2LinuxJob.Name)
479494
assert.Equal(t, "job-os-linux", job2LinuxJob.ConcurrencyGroup)
480495
})
481496
}
@@ -701,27 +716,20 @@ jobs:
701716
})
702717
_ = session.MakeRequest(t, req, http.StatusOK)
703718

719+
run2_2 := unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRun{ID: run2.ID})
720+
assert.Equal(t, actions_model.StatusWaiting, run2_2.Status)
721+
704722
req = NewRequestWithValues(t, "POST", fmt.Sprintf("/%s/%s/actions/runs/%d/rerun", user2.Name, apiRepo.Name, run2.Index+1), map[string]string{
705723
"_csrf": GetUserCSRFToken(t, session),
706724
})
707725
_ = session.MakeRequest(t, req, http.StatusOK)
708726

709727
task6 := runner.fetchTask(t)
710-
_, _, run2_2 := getTaskAndJobAndRunByTaskID(t, task6.Id)
711-
assert.Equal(t, "workflow-dispatch-v1.22", run2_2.ConcurrencyGroup)
712-
713-
runner.fetchNoTask(t) // cannot fetch task because task2 is not completed
714-
715-
runner.execTask(t, task6, &mockTaskOutcome{
716-
result: runnerv1.Result_RESULT_SUCCESS,
717-
})
718-
719-
task7 := runner.fetchTask(t)
720-
_, _, run3 := getTaskAndJobAndRunByTaskID(t, task7.Id)
728+
_, _, run3 := getTaskAndJobAndRunByTaskID(t, task6.Id)
721729
assert.Equal(t, "workflow-dispatch-v1.22", run3.ConcurrencyGroup)
722-
runner.execTask(t, task7, &mockTaskOutcome{
723-
result: runnerv1.Result_RESULT_SUCCESS,
724-
})
730+
731+
run2_2 = unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRun{ID: run2_2.ID})
732+
assert.Equal(t, actions_model.StatusCancelled, run2_2.Status) // cancelled by run3
725733
})
726734
}
727735

@@ -852,27 +860,20 @@ jobs:
852860
})
853861
_ = session.MakeRequest(t, req, http.StatusOK)
854862

863+
run2_2 := unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRun{ID: run2.ID})
864+
assert.Equal(t, actions_model.StatusWaiting, run2_2.Status)
865+
855866
req = NewRequestWithValues(t, "POST", fmt.Sprintf("/%s/%s/actions/runs/%d/jobs/%d/rerun", user2.Name, apiRepo.Name, run2.Index+1, 1), map[string]string{
856867
"_csrf": GetUserCSRFToken(t, session),
857868
})
858869
_ = session.MakeRequest(t, req, http.StatusOK)
859870

860871
task6 := runner.fetchTask(t)
861-
_, _, run2_2 := getTaskAndJobAndRunByTaskID(t, task6.Id)
862-
assert.Equal(t, "workflow-dispatch-v1.22", run2_2.ConcurrencyGroup)
863-
864-
runner.fetchNoTask(t) // cannot fetch task because task2 is not completed
865-
866-
runner.execTask(t, task6, &mockTaskOutcome{
867-
result: runnerv1.Result_RESULT_SUCCESS,
868-
})
869-
870-
task7 := runner.fetchTask(t)
871-
_, _, run3 := getTaskAndJobAndRunByTaskID(t, task7.Id)
872+
_, _, run3 := getTaskAndJobAndRunByTaskID(t, task6.Id)
872873
assert.Equal(t, "workflow-dispatch-v1.22", run3.ConcurrencyGroup)
873-
runner.execTask(t, task7, &mockTaskOutcome{
874-
result: runnerv1.Result_RESULT_SUCCESS,
875-
})
874+
875+
run2_2 = unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRun{ID: run2_2.ID})
876+
assert.Equal(t, actions_model.StatusCancelled, run2_2.Status) // cancelled by run3
876877
})
877878
}
878879

0 commit comments

Comments
 (0)