ce2fb4f fix(executor): check the cleanup budget before every cleanup command (GO-008) · ermite-runner

M internal/executor/containers_cleanup_test.go +44 -0


@@ -334,3 +334,47 @@ func TestInvalidEnvNamesAreRefused(t *testing.T) {
334	334		t.Errorf("docker exec called %d times", n)
335	335		}
336	336		}
	337	+
	338	+	// GO-008: the budget of the whole cleanup is checked before every command. With
	339	+	// a budget spent by the first stalled removal, the others (and the retry of the
	340	+	// fallback) are not started: they would each cost a signal escalation.
	341	+	func TestCleanupBudgetIsCheckedBeforeEveryCommand(t *testing.T) {
	342	+	f := newFakeCtr(t)
	343	+	f.set(t, "hang_for", "rm --force")
	344	+	f.set(t, "hang_ignore", "1") // SIGINT is ignored: a started command costs TermAfter
	345	+	svcs := ""
	346	+	for _, n := range []string{"a", "b", "c", "d", "e", "f", "g", "h"} {
	347	+	svcs += " " + n + ":\n image: redis:7\n"
	348	+	}
	349	+	start := time.Now()
	350	+	rec, res := ctrRun(t, f, " container: ubuntu:24.04\n services:\n"+svcs, " - run: echo ok\n", func(c *Config) {
	351	+	c.Actions.Containers = ContainerTimeouts{Command: time.Minute, Pull: time.Minute, CleanupCommand: time.Minute, CleanupBudget: time.Second}
	352	+	})
	353	+	total := time.Since(start)
	354	+	if res.Conclusion == ConclusionSuccess {
	355	+	t.Fatalf("an unfinished cleanup must be visible: %+v", res)
	356	+	}
	357	+	stop := rec.step(t, "Stop containers")
	358	+	took := stop.Completed.Sub(stop.Started)
	359	+	// The budget (1 s) and one escalation (TermAfter 200 ms in the tests, plus the kill delay).
	360	+	if took > 1900time.Millisecond \|\| total > 12time.Second {
	361	+	t.Errorf("Stop containers took %s (run %s), want about the budget plus one escalation", took, total)
	362	+	}
	363	+	rms := 0
	364	+	for _, c := range f.calls(t) {
	365	+	if c[0] == "rm" {
	366	+	rms++
	367	+	}
	368	+	}
	369	+	if rms != 1 {
	370	+	t.Errorf("%d removals started, want 1 (the budget was spent by it)", rms)
	371	+	}
	372	+	log := logOf(stop)
	373	+	if !strings.Contains(log, "##[error]The cleanup of the containers ran out of time (1s): stopped: docker rm --force ") \|\|
	374	+	!strings.Contains(log, "ran out of time (1s): not run: docker ") {
	375	+	t.Errorf("log:\n%s", log)
	376	+	}
	377	+	if strings.Count(log, "not run: docker network rm") != 1 {
	378	+	t.Errorf("the network removal must be reported as not run once:\n%s", log)
	379	+	}
	380	+	}


@@ -334,3 +334,47 @@ func TestInvalidEnvNamesAreRefused(t *testing.T) {
334	t.Errorf("docker exec called %d times", n)	334		t.Errorf("docker exec called %d times", n)
335	}	335		}
336	}	336		}
		337	+
		338	+	// GO-008: the budget of the whole cleanup is checked before every command. With
		339	+	// a budget spent by the first stalled removal, the others (and the retry of the
		340	+	// fallback) are not started: they would each cost a signal escalation.
		341	+	func TestCleanupBudgetIsCheckedBeforeEveryCommand(t *testing.T) {
		342	+	f := newFakeCtr(t)
		343	+	f.set(t, "hang_for", "rm --force")
		344	+	f.set(t, "hang_ignore", "1") // SIGINT is ignored: a started command costs TermAfter
		345	+	svcs := ""
		346	+	for _, n := range []string{"a", "b", "c", "d", "e", "f", "g", "h"} {
		347	+	svcs += " " + n + ":\n image: redis:7\n"
		348	+	}
		349	+	start := time.Now()
		350	+	rec, res := ctrRun(t, f, " container: ubuntu:24.04\n services:\n"+svcs, " - run: echo ok\n", func(c *Config) {
		351	+	c.Actions.Containers = ContainerTimeouts{Command: time.Minute, Pull: time.Minute, CleanupCommand: time.Minute, CleanupBudget: time.Second}
		352	+	})
		353	+	total := time.Since(start)
		354	+	if res.Conclusion == ConclusionSuccess {
		355	+	t.Fatalf("an unfinished cleanup must be visible: %+v", res)
		356	+	}
		357	+	stop := rec.step(t, "Stop containers")
		358	+	took := stop.Completed.Sub(stop.Started)
		359	+	// The budget (1 s) and one escalation (TermAfter 200 ms in the tests, plus the kill delay).
		360	+	if took > 1900time.Millisecond \|\| total > 12time.Second {
		361	+	t.Errorf("Stop containers took %s (run %s), want about the budget plus one escalation", took, total)
		362	+	}
		363	+	rms := 0
		364	+	for _, c := range f.calls(t) {
		365	+	if c[0] == "rm" {
		366	+	rms++
		367	+	}
		368	+	}
		369	+	if rms != 1 {
		370	+	t.Errorf("%d removals started, want 1 (the budget was spent by it)", rms)
		371	+	}
		372	+	log := logOf(stop)
		373	+	if !strings.Contains(log, "##[error]The cleanup of the containers ran out of time (1s): stopped: docker rm --force ") \|\|
		374	+	!strings.Contains(log, "ran out of time (1s): not run: docker ") {
		375	+	t.Errorf("log:\n%s", log)
		376	+	}
		377	+	if strings.Count(log, "not run: docker network rm") != 1 {
		378	+	t.Errorf("the network removal must be reported as not run once:\n%s", log)
		379	+	}
		380	+	}

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


@@ -30,7 +30,7 @@ func newFakeCtr(t testing.T) fakeCtr {
30	30		D=` + d + `
31	31		{ echo "--"; for a in "$@"; do printf "%s\n" "$a"; done; echo "ENV:CVAR=$CVAR:EMPTYV=${EMPTYV-unset}:PATH=$PATH"; } >> $D/calls
32	32		[ "$1" = "--config" ] && { CFG=$2; shift 2; }
33		-	hang() { trap 'exit 130' INT TERM; while :; do sleep 0.05; done; }
	33	+	hang() { if [ -f $D/hang_ignore ]; then trap '' INT; trap 'exit 143' TERM; else trap 'exit 130' INT TERM; fi; while :; do sleep 0.05; done; }
34	34		[ -f $D/hang_$1 ] && hang
35	35		if [ -f $D/hang_for ]; then case "$" in "$(cat $D/hang_for)"*) hang ;; esac; fi
36	36		case "$1" in


@@ -30,7 +30,7 @@ func newFakeCtr(t testing.T) fakeCtr {
30		D=` + d + `	30		D=` + d + `
31		{ echo "--"; for a in "$@"; do printf "%s\n" "$a"; done; echo "ENV:CVAR=$CVAR:EMPTYV=${EMPTYV-unset}:PATH=$PATH"; } >> $D/calls	31		{ echo "--"; for a in "$@"; do printf "%s\n" "$a"; done; echo "ENV:CVAR=$CVAR:EMPTYV=${EMPTYV-unset}:PATH=$PATH"; } >> $D/calls
32		[ "$1" = "--config" ] && { CFG=$2; shift 2; }	32		[ "$1" = "--config" ] && { CFG=$2; shift 2; }
33	-	hang() { trap 'exit 130' INT TERM; while :; do sleep 0.05; done; }	33	+	hang() { if [ -f $D/hang_ignore ]; then trap '' INT; trap 'exit 143' TERM; else trap 'exit 130' INT TERM; fi; while :; do sleep 0.05; done; }
34		[ -f $D/hang_$1 ] && hang	34		[ -f $D/hang_$1 ] && hang
35		if [ -f $D/hang_for ]; then case "$" in "$(cat $D/hang_for)"*) hang ;; esac; fi	35		if [ -f $D/hang_for ]; then case "$" in "$(cat $D/hang_for)"*) hang ;; esac; fi
36		case "$1" in	36		case "$1" in

M internal/executor/containers_init.go +4 -0


@@ -32,6 +32,10 @@ type containerSet struct {
32	32		networkTry string // network name, set before docker network create
33	33		netRemoved bool // the network is gone
34	34		dirs []string // docker --config directories that could not be removed
	35	+	// deadline is the end of the cleanup budget, fixed by the first cleanup of
	36	+	// the job: "Stop containers" and the fallback share it, so the cleanup of a
	37	+	// job is bounded as a whole.
	38	+	deadline time.Time
35	39		}
36	40
37	41		func (cs *containerSet) tryNetwork(name string) { cs.mu.Lock(); cs.networkTry = name; cs.mu.Unlock() }


@@ -32,6 +32,10 @@ type containerSet struct {
32	networkTry string // network name, set before docker network create	32		networkTry string // network name, set before docker network create
33	netRemoved bool // the network is gone	33		netRemoved bool // the network is gone
34	dirs []string // docker --config directories that could not be removed	34		dirs []string // docker --config directories that could not be removed
		35	+	// deadline is the end of the cleanup budget, fixed by the first cleanup of
		36	+	// the job: "Stop containers" and the fallback share it, so the cleanup of a
		37	+	// job is bounded as a whole.
		38	+	deadline time.Time
35	}	39		}
36		40
37	func (cs *containerSet) tryNetwork(name string) { cs.mu.Lock(); cs.networkTry = name; cs.mu.Unlock() }	41		func (cs *containerSet) tryNetwork(name string) { cs.mu.Lock(); cs.networkTry = name; cs.mu.Unlock() }

M internal/executor/containers_stop.go +42 -14


@@ -151,29 +151,29 @@ func (j job) cleanupContainers(s stepRun) (errs []error) {
151	151		cs.cleanupMu.Lock()
152	152		defer cs.cleanupMu.Unlock()
153	153		t := j.timeouts()
154		-	budget, budgetOut, stop := bound(nil, t.CleanupBudget)
	154	+	cs.mu.Lock()
	155	+	if cs.deadline.IsZero() {
	156	+	cs.deadline = time.Now().Add(t.CleanupBudget)
	157	+	}
	158	+	remaining := time.Until(cs.deadline)
	159	+	cs.mu.Unlock()
	160	+	budget, budgetOut, stop := bound(nil, max(remaining, time.Nanosecond))
155	161		defer stop()
	162	+	defer func() { errs = dedupeErrors(errs) }()
156	163		out := func(line string) {
157	164		if s != nil {
158	165		j.out(s.num, line)
159	166		}
160	167		}
161	168		run := func(c dockerCmd) (int, error) {
162		-	select {
163		-	case <-budget:
164		-	if budgetOut() {
165		-	return 0, ghErrorf("The cleanup of the containers ran out of time (%s)", t.CleanupBudget)
166		-	}
167		-	default:
168		-	}
169		-	_, code, err := j.dockerDo(c, dockerOpts{s: s, cancel: budget, all: true, timeout: t.CleanupCommand})
	169	+	_, code, err := j.dockerDoCleanup(c, s, budget, budgetOut, t, t.CleanupCommand)
170	170		return code, err
171	171		}
172	172		// rm runs `docker rm --force target`. It reports whether the target is
173	173		// gone: removed, or docker says it does not exist. Any other failure leaves
174	174		// it pending, for the fallback of the job to try again.
175	175		rm := func(kind string, c dockerCmd) (gone bool) {
176		-	lines, code, err := j.dockerDoCleanup(c, s, budget, t.CleanupCommand)
	176	+	lines, code, err := j.dockerDoCleanup(c, s, budget, budgetOut, t, t.CleanupCommand)
177	177		switch {
178	178		case err != nil:
179	179		errs = append(errs, err)

@@ -197,7 +197,7 @@ func (j job) cleanupContainers(s stepRun) (errs []error) {
197	197		}
198	198		if len(unresolved) > 0 {
199	199		ok := true
200		-	ids, code, err := j.dockerDoCleanup(j.newDockerCmd("ps", raw("--all", "--quiet", "--no-trunc"), quotedOpt("--filter", "label="+j.dockerLabel())), nil, budget, t.CleanupCommand)
	200	+	ids, code, err := j.dockerDoCleanup(j.newDockerCmd("ps", raw("--all", "--quiet", "--no-trunc"), quotedOpt("--filter", "label="+j.dockerLabel())), nil, budget, budgetOut, t, t.CleanupCommand)
201	201		switch {
202	202		case err != nil:
203	203		errs, ok = append(errs, err), false

@@ -290,7 +290,35 @@ func absent(lines []string) bool {
290	290		}
291	291
292	292		// dockerDoCleanup runs a command of the cleanup: bounded by its own limit and
293		-	// by the budget, its output (stdout and stderr) returned, logged to s if any.
294		-	func (j job) dockerDoCleanup(c dockerCmd, s stepRun, budget <-chan struct{}, limit time.Duration) ([]string, int, error) {
295		-	return j.dockerDo(c, dockerOpts{s: s, cancel: budget, all: true, timeout: limit, errToOut: true})
	293	+	// by the budget of the whole cleanup, its output (stdout and stderr) returned,
	294	+	// logged to s if any. The budget is checked here, before every command: once
	295	+	// it is spent nothing is started (a command started with a closed cancel would
	296	+	// still cost a signal escalation) and the object stays pending.
	297	+	func (j job) dockerDoCleanup(c dockerCmd, s stepRun, budget <-chan struct{}, budgetOut func() bool, t ContainerTimeouts, limit time.Duration) ([]string, int, error) {
	298	+	spent := func() error {
	299	+	return ghErrorf("The cleanup of the containers ran out of time (%s): not run: docker %s", t.CleanupBudget, j.dockerArgString(c))
	300	+	}
	301	+	select {
	302	+	case <-budget:
	303	+	return nil, 0, spent()
	304	+	default:
	305	+	}
	306	+	lines, code, err := j.dockerDo(c, dockerOpts{s: s, cancel: budget, all: true, timeout: limit, errToOut: true})
	307	+	if err == errCancelled && budgetOut() {
	308	+	return lines, code, ghErrorf("The cleanup of the containers ran out of time (%s): stopped: docker %s", t.CleanupBudget, j.dockerArgString(c))
	309	+	}
	310	+	return lines, code, err
	311	+	}
	312	+
	313	+	// dedupeErrors keeps the first of the errors that read the same.
	314	+	func dedupeErrors(errs []error) []error {
	315	+	seen := map[string]bool{}
	316	+	var out []error
	317	+	for _, e := range errs {
	318	+	if !seen[e.Error()] {
	319	+	seen[e.Error()] = true
	320	+	out = append(out, e)
	321	+	}
	322	+	}
	323	+	return out
296	324		}

 	cs.cleanupMu.Lock()
 	defer cs.cleanupMu.Unlock()
 	t := j.timeouts()
-	budget, budgetOut, stop := bound(nil, t.CleanupBudget)
+	cs.mu.Lock()
+	if cs.deadline.IsZero() {
+		cs.deadline = time.Now().Add(t.CleanupBudget)
+	}
+	remaining := time.Until(cs.deadline)
+	cs.mu.Unlock()
+	budget, budgetOut, stop := bound(nil, max(remaining, time.Nanosecond))
 	defer stop()
+	defer func() { errs = dedupeErrors(errs) }()
 	out := func(line string) {
 		if s != nil {
 			j.out(s.num, line)
 		}
 	}
 	run := func(c dockerCmd) (int, error) {
-		select {
+		_, code, err := j.dockerDoCleanup(c, s, budget, budgetOut, t, t.CleanupCommand)
-		case <-budget:
-			if budgetOut() {
-				return 0, ghErrorf("The cleanup of the containers ran out of time (%s)", t.CleanupBudget)
-			}
-		default:
-		}
-		_, code, err := j.dockerDo(c, dockerOpts{s: s, cancel: budget, all: true, timeout: t.CleanupCommand})
 		return code, err
 	}
 	// rm runs `docker rm --force target`. It reports whether the target is
 	// gone: removed, or docker says it does not exist. Any other failure leaves
 	// it pending, for the fallback of the job to try again.
 	rm := func(kind string, c dockerCmd) (gone bool) {
-		lines, code, err := j.dockerDoCleanup(c, s, budget, t.CleanupCommand)
+		lines, code, err := j.dockerDoCleanup(c, s, budget, budgetOut, t, t.CleanupCommand)
 		switch {
 		case err != nil:
 			errs = append(errs, err)
 	}
 	if len(unresolved) > 0 {
 		ok := true
-		ids, code, err := j.dockerDoCleanup(j.newDockerCmd("ps", raw("--all", "--quiet", "--no-trunc"), quotedOpt("--filter", "label="+j.dockerLabel())), nil, budget, t.CleanupCommand)
+		ids, code, err := j.dockerDoCleanup(j.newDockerCmd("ps", raw("--all", "--quiet", "--no-trunc"), quotedOpt("--filter", "label="+j.dockerLabel())), nil, budget, budgetOut, t, t.CleanupCommand)
 		switch {
 		case err != nil:
 			errs, ok = append(errs, err), false
 }
 // dockerDoCleanup runs a command of the cleanup: bounded by its own limit and
-// by the budget, its output (stdout and stderr) returned, logged to s if any.
+// by the budget of the whole cleanup, its output (stdout and stderr) returned,
-func (j *job) dockerDoCleanup(c dockerCmd, s *stepRun, budget <-chan struct{}, limit time.Duration) ([]string, int, error) {
+// logged to s if any. The budget is checked here, before every command: once
-	return j.dockerDo(c, dockerOpts{s: s, cancel: budget, all: true, timeout: limit, errToOut: true})
+// it is spent nothing is started (a command started with a closed cancel would
+// still cost a signal escalation) and the object stays pending.
+func (j *job) dockerDoCleanup(c dockerCmd, s *stepRun, budget <-chan struct{}, budgetOut func() bool, t ContainerTimeouts, limit time.Duration) ([]string, int, error) {
+	spent := func() error {
+		return ghErrorf("The cleanup of the containers ran out of time (%s): not run: docker %s", t.CleanupBudget, j.dockerArgString(c))
+	}
+	select {
+	case <-budget:
+		return nil, 0, spent()
+	default:
+	}
+	lines, code, err := j.dockerDo(c, dockerOpts{s: s, cancel: budget, all: true, timeout: limit, errToOut: true})
+	if err == errCancelled && budgetOut() {
+		return lines, code, ghErrorf("The cleanup of the containers ran out of time (%s): stopped: docker %s", t.CleanupBudget, j.dockerArgString(c))
+	}
+	return lines, code, err
+}
+// dedupeErrors keeps the first of the errors that read the same.
+func dedupeErrors(errs []error) []error {
+	seen := map[string]bool{}
+	var out []error
+	for _, e := range errs {
+		if !seen[e.Error()] {
+			seen[e.Error()] = true
+			out = append(out, e)
+		}
+	}
+	return out
 }