fix(ci): end a cancelled job whose assignment was not acknowledged
A fail-fast or owner cancel that arrived before the first heartbeat was lost when the assignment was cancelled: the job went back to the queue and ran.
- commit
- 9728ecc6a5786423d9605fcf7e08bc797acf0f9e
- author
- brice · · 2026-10-12 19:09 UTC
- parent
- ce785bf
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
2 files · +43 -0
M internal/app/ci_flow_test.go +33 -0
| @@ -594,3 +594,36 @@ func TestCINoRunnerNoRunsNoChecks(t *testing.T) { | |||
| 594 | 594 | c.push("main", map[string]string{"a.txt": "a\n"}, "") | |
| 595 | 595 | c.waitFor("a run once a runner exists", func() bool { return len(c.runs()) == 1 }) | |
| 596 | 596 | } | |
| 597 | + | ||
| 598 | + | // A job that fail-fast cancelled while its assignment was not acknowledged is | |
| 599 | + | // over when the assignment is cancelled: it is not given to a runner again. | |
| 600 | + | func TestCICancelledUnacknowledgedAssignmentIsNotAssignedAgain(t *testing.T) { | |
| 601 | + | c := newCIHarness(t, func(cfg *usecase.CIConfig) { | |
| 602 | + | cfg.AckTimeout, cfg.Tick, cfg.PollTimeout = 300*time.Millisecond, 20*time.Millisecond, 100*time.Millisecond | |
| 603 | + | }) | |
| 604 | + | runner := c.registerRunner("alpha", "ubuntu-latest") | |
| 605 | + | c.push("main", map[string]string{".github/workflows/m.yml": `on: push | |
| 606 | + | jobs: | |
| 607 | + | build: | |
| 608 | + | runs-on: ubuntu-latest | |
| 609 | + | strategy: | |
| 610 | + | matrix: | |
| 611 | + | n: [1, 2] | |
| 612 | + | steps: | |
| 613 | + | - run: echo ${{ matrix.n }} | |
| 614 | + | `}, "") | |
| 615 | + | a, b := runner.await(), runner.await() // both assigned, neither acknowledged yet | |
| 616 | + | assert.NotEqual(t, a.GetJobId(), b.GetJobId()) | |
| 617 | + | ||
| 618 | + | runner.runJob(a, pb.CiConclusion_CI_CONCLUSION_FAILURE, nil, "x") // fail-fast asks b to stop | |
| 619 | + | ||
| 620 | + | c.waitFor("the run ends", func() bool { return c.runs()[0].GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED }) | |
| 621 | + | for _, j := range c.getRun(c.runs()[0].GetId()).GetJobs() { | |
| 622 | + | if j.GetId() == b.GetJobId() { | |
| 623 | + | assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_CANCELLED, j.GetConclusion()) | |
| 624 | + | } | |
| 625 | + | } | |
| 626 | + | again, err := runner.fetch("") | |
| 627 | + | require.NoError(t, err) | |
| 628 | + | assert.Nil(t, again, "a cancelled job is not given again") | |
| 629 | + | } | |
| @@ -594,3 +594,36 @@ func TestCINoRunnerNoRunsNoChecks(t *testing.T) { | |||||
| 594 | c.push("main", map[string]string{"a.txt": "a\n"}, "") | 594 | c.push("main", map[string]string{"a.txt": "a\n"}, "") | ||
| 595 | c.waitFor("a run once a runner exists", func() bool { return len(c.runs()) == 1 }) | 595 | c.waitFor("a run once a runner exists", func() bool { return len(c.runs()) == 1 }) | ||
| 596 | } | 596 | } | ||
| 597 | + | ||||
| 598 | + | // A job that fail-fast cancelled while its assignment was not acknowledged is | |||
| 599 | + | // over when the assignment is cancelled: it is not given to a runner again. | |||
| 600 | + | func TestCICancelledUnacknowledgedAssignmentIsNotAssignedAgain(t *testing.T) { | |||
| 601 | + | c := newCIHarness(t, func(cfg *usecase.CIConfig) { | |||
| 602 | + | cfg.AckTimeout, cfg.Tick, cfg.PollTimeout = 300*time.Millisecond, 20*time.Millisecond, 100*time.Millisecond | |||
| 603 | + | }) | |||
| 604 | + | runner := c.registerRunner("alpha", "ubuntu-latest") | |||
| 605 | + | c.push("main", map[string]string{".github/workflows/m.yml": `on: push | |||
| 606 | + | jobs: | |||
| 607 | + | build: | |||
| 608 | + | runs-on: ubuntu-latest | |||
| 609 | + | strategy: | |||
| 610 | + | matrix: | |||
| 611 | + | n: [1, 2] | |||
| 612 | + | steps: | |||
| 613 | + | - run: echo ${{ matrix.n }} | |||
| 614 | + | `}, "") | |||
| 615 | + | a, b := runner.await(), runner.await() // both assigned, neither acknowledged yet | |||
| 616 | + | assert.NotEqual(t, a.GetJobId(), b.GetJobId()) | |||
| 617 | + | ||||
| 618 | + | runner.runJob(a, pb.CiConclusion_CI_CONCLUSION_FAILURE, nil, "x") // fail-fast asks b to stop | |||
| 619 | + | ||||
| 620 | + | c.waitFor("the run ends", func() bool { return c.runs()[0].GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED }) | |||
| 621 | + | for _, j := range c.getRun(c.runs()[0].GetId()).GetJobs() { | |||
| 622 | + | if j.GetId() == b.GetJobId() { | |||
| 623 | + | assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_CANCELLED, j.GetConclusion()) | |||
| 624 | + | } | |||
| 625 | + | } | |||
| 626 | + | again, err := runner.fetch("") | |||
| 627 | + | require.NoError(t, err) | |||
| 628 | + | assert.Nil(t, again, "a cancelled job is not given again") | |||
| 629 | + | } | |||
M internal/usecase/ci_watch.go +10 -0
| @@ -198,6 +198,16 @@ func (c *CI) cancelAssignment(ctx context.Context, rs *runState, j *domain.CIJob | |||
| 198 | 198 | } | |
| 199 | 199 | } | |
| 200 | 200 | c.mu.Unlock() | |
| 201 | + | if j.CancelReason != domain.CICancelNone { | |
| 202 | + | // Cancelled (by the owner, fail-fast or the time budget) before the | |
| 203 | + | // runner answered: it is over, not queued again for another runner. | |
| 204 | + | conclusion := domain.CICancelled | |
| 205 | + | if j.CancelReason == domain.CICancelTimeout { | |
| 206 | + | conclusion = domain.CITimedOut | |
| 207 | + | } | |
| 208 | + | rs.end(j, conclusion, "") | |
| 209 | + | return | |
| 210 | + | } | |
| 201 | 211 | if j.AssignCancels >= domain.CIMaxAssignCancels { | |
| 202 | 212 | rs.end(j, domain.CIFailure, "The job was given to a runner "+itoa(int64(domain.CIMaxAssignCancels))+" times and none of them acknowledged it.") | |
| 203 | 213 | return | |
| @@ -198,6 +198,16 @@ func (c *CI) cancelAssignment(ctx context.Context, rs *runState, j *domain.CIJob | |||||
| 198 | } | 198 | } | ||
| 199 | } | 199 | } | ||
| 200 | c.mu.Unlock() | 200 | c.mu.Unlock() | ||
| 201 | + | if j.CancelReason != domain.CICancelNone { | |||
| 202 | + | // Cancelled (by the owner, fail-fast or the time budget) before the | |||
| 203 | + | // runner answered: it is over, not queued again for another runner. | |||
| 204 | + | conclusion := domain.CICancelled | |||
| 205 | + | if j.CancelReason == domain.CICancelTimeout { | |||
| 206 | + | conclusion = domain.CITimedOut | |||
| 207 | + | } | |||
| 208 | + | rs.end(j, conclusion, "") | |||
| 209 | + | return | |||
| 210 | + | } | |||
| 201 | if j.AssignCancels >= domain.CIMaxAssignCancels { | 211 | if j.AssignCancels >= domain.CIMaxAssignCancels { | ||
| 202 | rs.end(j, domain.CIFailure, "The job was given to a runner "+itoa(int64(domain.CIMaxAssignCancels))+" times and none of them acknowledged it.") | 212 | rs.end(j, domain.CIFailure, "The job was given to a runner "+itoa(int64(domain.CIMaxAssignCancels))+" times and none of them acknowledged it.") | ||
| 203 | return | 213 | return | ||