fix(executor): stop problem matchers when the step is cancelled

The output manager gets the cancellation channel of the step. A match in
progress is abandoned, no further matcher or attempt runs, and a timeout the
cancellation caused is not counted, so the final drain of a step with several
catastrophic matchers no longer holds it past timeout-minutes.

commit: 3d4e205848bfa2ceb541054c711da14ff568fc57
author: brice · 8d ago · 2026-10-14 02:42 UTC
parent: 9911deb
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

3 files · +99 -6

M internal/executor/exec_step.go +1 -1


@@ -192,7 +192,7 @@ func (j job) processEnv(stepEnv ordered) []string {
192	192		// step.
193	193		func (j job) startProcess(s stepRun, plan ShellPlan, dir string, argv []string, stepCancel <-chan struct{}) stepResult {
194	194		num := s.num
195		-	om := newOutputManager(s)
	195	+	om := newOutputManager(s, stepCancel)
196	196		defer om.dispose()
197	197		res, err := runProcess(procSpec{
198	198		Argv: argv, Dir: dir, Env: j.processEnv(s.env), Credential: j.cfg.Credential,


@@ -192,7 +192,7 @@ func (j job) processEnv(stepEnv ordered) []string {
192		// step.	192		// step.
193		func (j job) startProcess(s stepRun, plan ShellPlan, dir string, argv []string, stepCancel <-chan struct{}) stepResult {	193		func (j job) startProcess(s stepRun, plan ShellPlan, dir string, argv []string, stepCancel <-chan struct{}) stepResult {
194		num := s.num	194		num := s.num
195	-	om := newOutputManager(s)	195	+	om := newOutputManager(s, stepCancel)
196		defer om.dispose()	196		defer om.dispose()
197		res, err := runProcess(procSpec{	197		res, err := runProcess(procSpec{
198		Argv: argv, Dir: dir, Env: j.processEnv(s.env), Credential: j.cfg.Credential,	198		Argv: argv, Dir: dir, Env: j.processEnv(s.env), Credential: j.cfg.Credential,

M internal/executor/outputmgr.go +62 -5


@@ -114,7 +114,10 @@ func matcherTimeout(getenv func(string) string) time.Duration {
114	114		}
115	115
116	116		type outputManager struct {
117		-	s *stepRun
	117	+	s *stepRun
	118	+	// cancel is closed when the step is cancelled or times out: the matchers
	119	+	// then stop (see stopped). Nil means never.
	120	+	cancel <-chan struct{}
118	121		root *matcherRoot
119	122		timeout time.Duration
120	123

@@ -125,8 +128,14 @@ type outputManager struct {
125	128		}
126	129
127	130		// newOutputManager subscribes to the matchers of the job.
128		-	func newOutputManager(s stepRun) outputManager {
129		-	o := &outputManager{s: s, root: s.j.matchers, timeout: matcherTimeout(s.j.getenv), dirMap: map[string]string{}, hasDir: map[string]bool{}}
	131	+	//
	132	+	// cancel is the cancellation channel of the step (the one the process gets):
	133	+	// once it is closed no matcher runs any more. The runner has no such rule, but
	134	+	// its worker waits for the output of a step with the step's cancellation
	135	+	// token, and a pile of lines against catastrophic patterns (1 s each, 3
	136	+	// attempts, per matcher) would hold a step past its timeout for minutes.
	137	+	func newOutputManager(s stepRun, cancel <-chan struct{}) outputManager {
	138	+	o := &outputManager{s: s, cancel: cancel, root: s.j.matchers, timeout: matcherTimeout(s.j.getenv), dirMap: map[string]string{}, hasDir: map[string]bool{}}
130	139		o.mu.Lock()
131	140		o.root.subscribe(o)
132	141		for _, c := range o.root.snapshot() {

@@ -138,6 +147,48 @@ func newOutputManager(s stepRun) outputManager {
138	147		return o
139	148		}
140	149
	150	+	// errStopped is the result of a match that was abandoned because the step was
	151	+	// cancelled.
	152	+	var errStopped = msgError("matching stopped: the step was cancelled")
	153	+
	154	+	// runMatch is m.match, abandoned when the step is cancelled while it runs: the
	155	+	// engine cannot be interrupted, so it goes on in its goroutine until its own
	156	+	// timeout, but the step does not wait for it and the matcher is not used again
	157	+	// (stopped is true from then on).
	158	+	func (o outputManager) runMatch(m issueMatcher, line string) (*issueMatch, error) {
	159	+	if o.cancel == nil {
	160	+	return m.match(line)
	161	+	}
	162	+	type result struct {
	163	+	m *issueMatch
	164	+	err error
	165	+	}
	166	+	ch := make(chan result, 1)
	167	+	go func() {
	168	+	r, err := m.match(line)
	169	+	ch <- result{r, err}
	170	+	}()
	171	+	select {
	172	+	case r := <-ch:
	173	+	return r.m, r.err
	174	+	case <-o.cancel:
	175	+	return nil, errStopped
	176	+	}
	177	+	}
	178	+
	179	+	// stopped reports whether the step was cancelled.
	180	+	func (o *outputManager) stopped() bool {
	181	+	if o.cancel == nil {
	182	+	return false
	183	+	}
	184	+	select {
	185	+	case <-o.cancel:
	186	+	return true
	187	+	default:
	188	+	return false
	189	+	}
	190	+	}
	191	+
141	192		func (o *outputManager) dispose() { o.root.unsubscribe(o) }
142	193
143	194		// matcherChanged is OnMatcherChanged: the matcher goes first, the others of

@@ -188,7 +239,7 @@ func (o *outputManager) onData(line string) {
188	239		o.mu.Lock()
189	240		matchers := append([]*issueMatcher(nil), o.matchers...)
190	241		o.mu.Unlock()
191		-	if len(matchers) > 0 {
	242	+	if len(matchers) > 0 && !o.stopped() {
192	243		stripped := line
193	244		if strings.Contains(line, colorCodePrefix) {
194	245		stripped, _ = colorCodeRegex.Replace(line, "", -1, -1)

@@ -196,11 +247,17 @@ func (o *outputManager) onData(line string) {
196	247		for _, m := range matchers {
197	248		var match *issueMatch
198	249		for attempt := 1; attempt <= maxMatcherAttempts; attempt++ {
199		-	mt, err := m.match(stripped)
	250	+	if o.stopped() {
	251	+	break // the lines that are left are only logged
	252	+	}
	253	+	mt, err := o.runMatch(m, stripped)
200	254		if err == nil {
201	255		match = mt
202	256		break
203	257		}
	258	+	if o.stopped() {
	259	+	break // a timeout that the cancellation caused counts for nothing
	260	+	}
204	261		if attempt < maxMatcherAttempts {
205	262		s.Debug(fmt.Sprintf("Timeout processing issue matcher '%s' against line '%s'. Exception: System.Text.RegularExpressions.RegexMatchTimeoutException: %s", m.owner, stripped, err.Error()))
206	263		} else {


@@ -114,7 +114,10 @@ func matcherTimeout(getenv func(string) string) time.Duration {
114		}	114		}
115			115
116		type outputManager struct {	116		type outputManager struct {
117	-	s *stepRun	117	+	s *stepRun
			118	+	// cancel is closed when the step is cancelled or times out: the matchers
			119	+	// then stop (see stopped). Nil means never.
			120	+	cancel <-chan struct{}
118		root *matcherRoot	121		root *matcherRoot
119		timeout time.Duration	122		timeout time.Duration
120			123

@@ -125,8 +128,14 @@ type outputManager struct {
125		}	128		}
126			129
127		// newOutputManager subscribes to the matchers of the job.	130		// newOutputManager subscribes to the matchers of the job.
128	-	func newOutputManager(s stepRun) outputManager {	131	+	//
129	-	o := &outputManager{s: s, root: s.j.matchers, timeout: matcherTimeout(s.j.getenv), dirMap: map[string]string{}, hasDir: map[string]bool{}}	132	+	// cancel is the cancellation channel of the step (the one the process gets):
			133	+	// once it is closed no matcher runs any more. The runner has no such rule, but
			134	+	// its worker waits for the output of a step with the step's cancellation
			135	+	// token, and a pile of lines against catastrophic patterns (1 s each, 3
			136	+	// attempts, per matcher) would hold a step past its timeout for minutes.
			137	+	func newOutputManager(s stepRun, cancel <-chan struct{}) outputManager {
			138	+	o := &outputManager{s: s, cancel: cancel, root: s.j.matchers, timeout: matcherTimeout(s.j.getenv), dirMap: map[string]string{}, hasDir: map[string]bool{}}
130		o.mu.Lock()	139		o.mu.Lock()
131		o.root.subscribe(o)	140		o.root.subscribe(o)
132		for _, c := range o.root.snapshot() {	141		for _, c := range o.root.snapshot() {

@@ -138,6 +147,48 @@ func newOutputManager(s stepRun) outputManager {
138		return o	147		return o
139		}	148		}
140			149
			150	+	// errStopped is the result of a match that was abandoned because the step was
			151	+	// cancelled.
			152	+	var errStopped = msgError("matching stopped: the step was cancelled")
			153	+
			154	+	// runMatch is m.match, abandoned when the step is cancelled while it runs: the
			155	+	// engine cannot be interrupted, so it goes on in its goroutine until its own
			156	+	// timeout, but the step does not wait for it and the matcher is not used again
			157	+	// (stopped is true from then on).
			158	+	func (o outputManager) runMatch(m issueMatcher, line string) (*issueMatch, error) {
			159	+	if o.cancel == nil {
			160	+	return m.match(line)
			161	+	}
			162	+	type result struct {
			163	+	m *issueMatch
			164	+	err error
			165	+	}
			166	+	ch := make(chan result, 1)
			167	+	go func() {
			168	+	r, err := m.match(line)
			169	+	ch <- result{r, err}
			170	+	}()
			171	+	select {
			172	+	case r := <-ch:
			173	+	return r.m, r.err
			174	+	case <-o.cancel:
			175	+	return nil, errStopped
			176	+	}
			177	+	}
			178	+
			179	+	// stopped reports whether the step was cancelled.
			180	+	func (o *outputManager) stopped() bool {
			181	+	if o.cancel == nil {
			182	+	return false
			183	+	}
			184	+	select {
			185	+	case <-o.cancel:
			186	+	return true
			187	+	default:
			188	+	return false
			189	+	}
			190	+	}
			191	+
141		func (o *outputManager) dispose() { o.root.unsubscribe(o) }	192		func (o *outputManager) dispose() { o.root.unsubscribe(o) }
142			193
143		// matcherChanged is OnMatcherChanged: the matcher goes first, the others of	194		// matcherChanged is OnMatcherChanged: the matcher goes first, the others of

@@ -188,7 +239,7 @@ func (o *outputManager) onData(line string) {
188		o.mu.Lock()	239		o.mu.Lock()
189		matchers := append([]*issueMatcher(nil), o.matchers...)	240		matchers := append([]*issueMatcher(nil), o.matchers...)
190		o.mu.Unlock()	241		o.mu.Unlock()
191	-	if len(matchers) > 0 {	242	+	if len(matchers) > 0 && !o.stopped() {
192		stripped := line	243		stripped := line
193		if strings.Contains(line, colorCodePrefix) {	244		if strings.Contains(line, colorCodePrefix) {
194		stripped, _ = colorCodeRegex.Replace(line, "", -1, -1)	245		stripped, _ = colorCodeRegex.Replace(line, "", -1, -1)

@@ -196,11 +247,17 @@ func (o *outputManager) onData(line string) {
196		for _, m := range matchers {	247		for _, m := range matchers {
197		var match *issueMatch	248		var match *issueMatch
198		for attempt := 1; attempt <= maxMatcherAttempts; attempt++ {	249		for attempt := 1; attempt <= maxMatcherAttempts; attempt++ {
199	-	mt, err := m.match(stripped)	250	+	if o.stopped() {
			251	+	break // the lines that are left are only logged
			252	+	}
			253	+	mt, err := o.runMatch(m, stripped)
200		if err == nil {	254		if err == nil {
201		match = mt	255		match = mt
202		break	256		break
203		}	257		}
			258	+	if o.stopped() {
			259	+	break // a timeout that the cancellation caused counts for nothing
			260	+	}
204		if attempt < maxMatcherAttempts {	261		if attempt < maxMatcherAttempts {
205		s.Debug(fmt.Sprintf("Timeout processing issue matcher '%s' against line '%s'. Exception: System.Text.RegularExpressions.RegexMatchTimeoutException: %s", m.owner, stripped, err.Error()))	262		s.Debug(fmt.Sprintf("Timeout processing issue matcher '%s' against line '%s'. Exception: System.Text.RegularExpressions.RegexMatchTimeoutException: %s", m.owner, stripped, err.Error()))
206		} else {	263		} else {

M internal/executor/results_test.go +36 -0


@@ -338,3 +338,39 @@ func TestSaveStateIsKeptPerStep(t *testing.T) {
338	338		t.Errorf("%v", got)
339	339		}
340	340		}
	341	+
	342	+	// Several catastrophic matchers and a step timeout: the step ends at its
	343	+	// timeout, not after every matcher has used its 3 attempts on every line
	344	+	// (GO-004). Without the cancellation of the matchers the 8 matchers x 3
	345	+	// attempts x 1 s of the first line alone hold the step for 24 s.
	346	+	func TestCatastrophicMatchersDoNotHoldAStepPastItsTimeout(t *testing.T) {
	347	+	needTools(t, "bash")
	348	+	var matchers []string
	349	+	var adds []string
	350	+	for i := range 8 {
	351	+	matchers = append(matchers, fmt.Sprintf(`{"owner":"bad%d","pattern":[{"regexp":"^((a+)+)$","message":1}]}`, i))
	352	+	}
	353	+	adds = append(adds, `echo '{"problemMatcher":[`+strings.Join(matchers, ",")+`]}' > bad.json`)
	354	+	start := time.Now()
	355	+	rr, _ := runResults(t, wfHead+` steps:
	356	+	- name: slow lines
	357	+	timeout-minutes: 1
	358	+	run: \|
	359	+	`+adds[0]+`
	360	+	echo "::add-matcher::bad.json"
	361	+	for i in 1 2 3 4 5 6; do echo "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa!"; done
	362	+	sleep 30
	363	+	`, func(c Config) { c.StepTimeoutSkew = -(60time.Second - 1500*time.Millisecond) })
	364	+	took := time.Since(start)
	365	+	if took > 8*time.Second {
	366	+	t.Errorf("the step overran its timeout: %v", took)
	367	+	}
	368	+	log := logOf(rr.step(t, "slow lines"))
	369	+	if !strings.Contains(log, "The action 'slow lines' has timed out after 1 minutes.") {
	370	+	t.Errorf("no timeout message:\n%s", log)
	371	+	}
	372	+	// the lines that were not matched are still logged
	373	+	if strings.Count(log, "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa!") < 6 {
	374	+	t.Errorf("lines lost after the cancellation:\n%s", log)
	375	+	}
	376	+	}


@@ -338,3 +338,39 @@ func TestSaveStateIsKeptPerStep(t *testing.T) {
338	t.Errorf("%v", got)	338		t.Errorf("%v", got)
339	}	339		}
340	}	340		}
		341	+
		342	+	// Several catastrophic matchers and a step timeout: the step ends at its
		343	+	// timeout, not after every matcher has used its 3 attempts on every line
		344	+	// (GO-004). Without the cancellation of the matchers the 8 matchers x 3
		345	+	// attempts x 1 s of the first line alone hold the step for 24 s.
		346	+	func TestCatastrophicMatchersDoNotHoldAStepPastItsTimeout(t *testing.T) {
		347	+	needTools(t, "bash")
		348	+	var matchers []string
		349	+	var adds []string
		350	+	for i := range 8 {
		351	+	matchers = append(matchers, fmt.Sprintf(`{"owner":"bad%d","pattern":[{"regexp":"^((a+)+)$","message":1}]}`, i))
		352	+	}
		353	+	adds = append(adds, `echo '{"problemMatcher":[`+strings.Join(matchers, ",")+`]}' > bad.json`)
		354	+	start := time.Now()
		355	+	rr, _ := runResults(t, wfHead+` steps:
		356	+	- name: slow lines
		357	+	timeout-minutes: 1
		358	+	run: \|
		359	+	`+adds[0]+`
		360	+	echo "::add-matcher::bad.json"
		361	+	for i in 1 2 3 4 5 6; do echo "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa!"; done
		362	+	sleep 30
		363	+	`, func(c Config) { c.StepTimeoutSkew = -(60time.Second - 1500*time.Millisecond) })
		364	+	took := time.Since(start)
		365	+	if took > 8*time.Second {
		366	+	t.Errorf("the step overran its timeout: %v", took)
		367	+	}
		368	+	log := logOf(rr.step(t, "slow lines"))
		369	+	if !strings.Contains(log, "The action 'slow lines' has timed out after 1 minutes.") {
		370	+	t.Errorf("no timeout message:\n%s", log)
		371	+	}
		372	+	// the lines that were not matched are still logged
		373	+	if strings.Count(log, "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa!") < 6 {
		374	+	t.Errorf("lines lost after the cancellation:\n%s", log)
		375	+	}
		376	+	}