fix(ci): free the group of a replaced run whose record lost it (#45)
- commit
- 0d0b8b2f060498ce8d805730c6b1f2f075c20f76
- author
- brice · · 2026-10-16 10:23 UTC
- committer
- bnema · · 2026-10-16 10:23 UTC
- parent
- cc3aa02
- signature
- SSH Unverified
SHA256:VOCA5lYARHcbKDZ7GhTV45VmcjWnJ6ZQIBazj17LivAThe key is unknown
2 files · +59 -1
M internal/app/ci_concurrency_test.go +47 -0
| @@ -802,6 +802,53 @@ func TestCIConcurrencyReplacedRunWithLostWritesIsCancelled(t *testing.T) { | |||
| 802 | 802 | c.waitFor("its mark goes with it", func() bool { return c.markState("Deploy", bEntry) == "" }) | |
| 803 | 803 | } | |
| 804 | 804 | ||
| 805 | + | // As above, but only B's own writes fail: C's record knows it is pending, so | |
| 806 | + | // the reconciliation does not plan C again, and no other decision on the group | |
| 807 | + | // prunes B's mark. The cancel that ends B frees the group that decided it, | |
| 808 | + | // although B's own record names no group. | |
| 809 | + | func TestCIConcurrencyCancelOfARunUnawareOfItsGroupDropsItsMark(t *testing.T) { | |
| 810 | + | c, runner, fs := flakyConcurrencyHarness(t, -1) | |
| 811 | + | r1 := c.pushN(1, workflowAt(wfGroup, "")) | |
| 812 | + | runner.await() | |
| 813 | + | ||
| 814 | + | bID := r1.GetId() + 1 // run ids are taken in sequence | |
| 815 | + | fs.match = func(w ports.CIWrite) bool { | |
| 816 | + | for _, r := range w.Runs { | |
| 817 | + | if r.ID == bID && r.Concurrency.State == domain.CIConcurrencyPending { | |
| 818 | + | return true | |
| 819 | + | } | |
| 820 | + | } | |
| 821 | + | for _, j := range w.Jobs { | |
| 822 | + | if j.RunID == bID && j.Ended() && j.Conclusion == domain.CICancelled { | |
| 823 | + | return true | |
| 824 | + | } | |
| 825 | + | } | |
| 826 | + | return false | |
| 827 | + | } | |
| 828 | + | fs.fails.Store(1 << 20) | |
| 829 | + | b := c.pushN(2, workflowAt(wfGroup, "")) | |
| 830 | + | require.Equal(t, bID, b.GetId()) | |
| 831 | + | bEntry := domain.CIConcurrencyEntry{RunID: bID, Attempt: 1} | |
| 832 | + | c.waitFor("B is named by the group", func() bool { return c.markState("Deploy", bEntry) == domain.CIConcurrencyPending }) | |
| 833 | + | ||
| 834 | + | r3 := c.pushN(3, workflowAt(wfGroup, "")) | |
| 835 | + | c.waitFor("B is marked replaced", func() bool { return c.markState("Deploy", bEntry) == domain.CIConcurrencyReplaced }) | |
| 836 | + | c.waitJobs("C knows it waits", r3, func(js []*pb.RunJob) bool { | |
| 837 | + | return len(js) == 1 && js[0].GetStatus() == pb.CiStatus_CI_STATUS_PENDING | |
| 838 | + | }) | |
| 839 | + | run, err := c.forge.store.CI().GetRun(c.ctx, c.h("proj"), bID) | |
| 840 | + | require.NoError(t, err) | |
| 841 | + | require.Empty(t, run.Concurrency.Group, "B's own record names no group") | |
| 842 | + | ||
| 843 | + | fs.fails.Store(0) | |
| 844 | + | c.forge.CI.ReconcileConcurrency(c.ctx) | |
| 845 | + | c.waitFor("B ends cancelled", func() bool { return c.conclusionOf(b) == pb.CiConclusion_CI_CONCLUSION_CANCELLED }) | |
| 846 | + | c.waitFor("its mark goes with it", func() bool { return c.markState("Deploy", bEntry) == "" }) | |
| 847 | + | rec := c.groupRecord("Deploy") | |
| 848 | + | require.Len(t, rec.Pending, 1) | |
| 849 | + | assert.Equal(t, r3.GetId(), rec.Pending[0].RunID, "C still waits") | |
| 850 | + | } | |
| 851 | + | ||
| 805 | 852 | // A grouped job B is replaced by C, the cancel never lands, C is promoted when | |
| 806 | 853 | // the running job ends, and an unrelated sibling of B completes and advances B: | |
| 807 | 854 | // B's grouped job must not take the free place, it ends cancelled and never runs. | |
| @@ -802,6 +802,53 @@ func TestCIConcurrencyReplacedRunWithLostWritesIsCancelled(t *testing.T) { | |||||
| 802 | c.waitFor("its mark goes with it", func() bool { return c.markState("Deploy", bEntry) == "" }) | 802 | c.waitFor("its mark goes with it", func() bool { return c.markState("Deploy", bEntry) == "" }) | ||
| 803 | } | 803 | } | ||
| 804 | 804 | ||||
| 805 | + | // As above, but only B's own writes fail: C's record knows it is pending, so | |||
| 806 | + | // the reconciliation does not plan C again, and no other decision on the group | |||
| 807 | + | // prunes B's mark. The cancel that ends B frees the group that decided it, | |||
| 808 | + | // although B's own record names no group. | |||
| 809 | + | func TestCIConcurrencyCancelOfARunUnawareOfItsGroupDropsItsMark(t *testing.T) { | |||
| 810 | + | c, runner, fs := flakyConcurrencyHarness(t, -1) | |||
| 811 | + | r1 := c.pushN(1, workflowAt(wfGroup, "")) | |||
| 812 | + | runner.await() | |||
| 813 | + | ||||
| 814 | + | bID := r1.GetId() + 1 // run ids are taken in sequence | |||
| 815 | + | fs.match = func(w ports.CIWrite) bool { | |||
| 816 | + | for _, r := range w.Runs { | |||
| 817 | + | if r.ID == bID && r.Concurrency.State == domain.CIConcurrencyPending { | |||
| 818 | + | return true | |||
| 819 | + | } | |||
| 820 | + | } | |||
| 821 | + | for _, j := range w.Jobs { | |||
| 822 | + | if j.RunID == bID && j.Ended() && j.Conclusion == domain.CICancelled { | |||
| 823 | + | return true | |||
| 824 | + | } | |||
| 825 | + | } | |||
| 826 | + | return false | |||
| 827 | + | } | |||
| 828 | + | fs.fails.Store(1 << 20) | |||
| 829 | + | b := c.pushN(2, workflowAt(wfGroup, "")) | |||
| 830 | + | require.Equal(t, bID, b.GetId()) | |||
| 831 | + | bEntry := domain.CIConcurrencyEntry{RunID: bID, Attempt: 1} | |||
| 832 | + | c.waitFor("B is named by the group", func() bool { return c.markState("Deploy", bEntry) == domain.CIConcurrencyPending }) | |||
| 833 | + | ||||
| 834 | + | r3 := c.pushN(3, workflowAt(wfGroup, "")) | |||
| 835 | + | c.waitFor("B is marked replaced", func() bool { return c.markState("Deploy", bEntry) == domain.CIConcurrencyReplaced }) | |||
| 836 | + | c.waitJobs("C knows it waits", r3, func(js []*pb.RunJob) bool { | |||
| 837 | + | return len(js) == 1 && js[0].GetStatus() == pb.CiStatus_CI_STATUS_PENDING | |||
| 838 | + | }) | |||
| 839 | + | run, err := c.forge.store.CI().GetRun(c.ctx, c.h("proj"), bID) | |||
| 840 | + | require.NoError(t, err) | |||
| 841 | + | require.Empty(t, run.Concurrency.Group, "B's own record names no group") | |||
| 842 | + | ||||
| 843 | + | fs.fails.Store(0) | |||
| 844 | + | c.forge.CI.ReconcileConcurrency(c.ctx) | |||
| 845 | + | c.waitFor("B ends cancelled", func() bool { return c.conclusionOf(b) == pb.CiConclusion_CI_CONCLUSION_CANCELLED }) | |||
| 846 | + | c.waitFor("its mark goes with it", func() bool { return c.markState("Deploy", bEntry) == "" }) | |||
| 847 | + | rec := c.groupRecord("Deploy") | |||
| 848 | + | require.Len(t, rec.Pending, 1) | |||
| 849 | + | assert.Equal(t, r3.GetId(), rec.Pending[0].RunID, "C still waits") | |||
| 850 | + | } | |||
| 851 | + | ||||
| 805 | // A grouped job B is replaced by C, the cancel never lands, C is promoted when | 852 | // A grouped job B is replaced by C, the cancel never lands, C is promoted when | ||
| 806 | // the running job ends, and an unrelated sibling of B completes and advances B: | 853 | // the running job ends, and an unrelated sibling of B completes and advances B: | ||
| 807 | // B's grouped job must not take the free place, it ends cancelled and never runs. | 854 | // B's grouped job must not take the free place, it ends cancelled and never runs. | ||
M internal/usecase/ci_concurrency.go +12 -1
| @@ -552,7 +552,18 @@ func (c *CI) cancelEntry(ctx context.Context, e domain.CIConcurrencyEntry, group | |||
| 552 | 552 | rs.cancelForGroup(&rs.jobs[i], msg) | |
| 553 | 553 | } | |
| 554 | 554 | } | |
| 555 | - | return rs.flush(ctx) | |
| 555 | + | if err := rs.flush(ctx); err != nil { | |
| 556 | + | return err | |
| 557 | + | } | |
| 558 | + | // The flush frees the group the run's own record names. A run whose | |
| 559 | + | // writes of its group were lost names none: free the group that | |
| 560 | + | // decided the cancel, and the mark with it. | |
| 561 | + | if rs.run.Status == domain.CICompleted && domain.ConcurrencyKey(rs.run.Concurrency.Group) != domain.ConcurrencyKey(group) { | |
| 562 | + | if err := c.releaseEntry(ctx, rs.run.Repo, group, e); err != nil && !errors.Is(err, domain.ErrNotFound) { | |
| 563 | + | return err | |
| 564 | + | } | |
| 565 | + | } | |
| 566 | + | return nil | |
| 556 | 567 | } | |
| 557 | 568 | j := rs.find(e.JobID) | |
| 558 | 569 | if j == nil || !j.Planned { | |
| @@ -552,7 +552,18 @@ func (c *CI) cancelEntry(ctx context.Context, e domain.CIConcurrencyEntry, group | |||||
| 552 | rs.cancelForGroup(&rs.jobs[i], msg) | 552 | rs.cancelForGroup(&rs.jobs[i], msg) | ||
| 553 | } | 553 | } | ||
| 554 | } | 554 | } | ||
| 555 | - | return rs.flush(ctx) | 555 | + | if err := rs.flush(ctx); err != nil { |
| 556 | + | return err | |||
| 557 | + | } | |||
| 558 | + | // The flush frees the group the run's own record names. A run whose | |||
| 559 | + | // writes of its group were lost names none: free the group that | |||
| 560 | + | // decided the cancel, and the mark with it. | |||
| 561 | + | if rs.run.Status == domain.CICompleted && domain.ConcurrencyKey(rs.run.Concurrency.Group) != domain.ConcurrencyKey(group) { | |||
| 562 | + | if err := c.releaseEntry(ctx, rs.run.Repo, group, e); err != nil && !errors.Is(err, domain.ErrNotFound) { | |||
| 563 | + | return err | |||
| 564 | + | } | |||
| 565 | + | } | |||
| 566 | + | return nil | |||
| 556 | } | 567 | } | ||
| 557 | j := rs.find(e.JobID) | 568 | j := rs.find(e.JobID) | ||
| 558 | if j == nil || !j.Planned { | 569 | if j == nil || !j.Planned { | ||