test: wait for the ready line or the first frame instead of sleeping (G8)

commit: 8ec9ad20c0fa9a7089c6bec237eaa9c64028f097
author: brice · 10d ago · 2026-10-12 19:40 UTC
parent: 3530513
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +28 -11

M internal/executor/job_test.go +2 -1


@@ -278,9 +278,10 @@ func TestCancelSendsSIGINTThenStopsTheJob(t *testing.T) {
278	278		needTools(t, "bash")
279	279		c := NewCanceller()
280	280		started := make(chan struct{})
	281	+	// The script prints "ready" after it installed its trap: the cancel can come
	282	+	// at once.
281	283		go func() {
282	284		<-started
283		-	time.Sleep(300 * time.Millisecond)
284	285		c.Cancel(CancelUser)
285	286		}()
286	287		rec := newRecorder()

 	needTools(t, "bash")
 	c := NewCanceller()
 	started := make(chan struct{})
+	// The script prints "ready" after it installed its trap: the cancel can come
+	// at once.
 	go func() {
 		<-started
-		time.Sleep(300 * time.Millisecond)
 		c.Cancel(CancelUser)
 	}()
 	rec := newRecorder()

M internal/guest/jobexec_test.go +26 -10


@@ -18,7 +18,7 @@ import (
18	18		)
19	19
20	20		// runJobSession starts Exec on a unix socket and drives it with a job.
21		-	func runJobSession(t testing.T, yaml string, cancelAfter time.Duration, tweak func(ExecConfig)) (steps []string, logs []execpb.Log, res execpb.JobResult, exit *execpb.Exit) {
	21	+	func runJobSession(t testing.T, yaml string, cancelOn string, tweak func(ExecConfig)) (steps []string, logs []execpb.Log, res execpb.JobResult, exit *execpb.Exit) {
22	22		t.Helper()
23	23		sock := filepath.Join(t.TempDir(), "s")
24	24		ln, err := net.Listen("unix", sock)

@@ -45,11 +45,13 @@ func runJobSession(t *testing.T, yaml string, cancelAfter time.Duration, tweak f
45	45		t.Fatal(err)
46	46		}
47	47		defer func() { _ = conn.Close() }()
	48	+	// With cancelOn, the job is cancelled when it logs that line (the script
	49	+	// prints it once it is ready to be interrupted): no timer to guess.
48	50		var cancel chan *execpb.Cancel
49		-	if cancelAfter > 0 {
	51	+	if cancelOn != "" {
50	52		cancel = make(chan *execpb.Cancel, 1)
51		-	time.AfterFunc(cancelAfter, func() { cancel <- &execpb.Cancel{Reason: "test", Kind: execpb.CancelKind_CANCEL_KIND_USER} })
52	53		}
	54	+	var cancelOnce sync.Once
53	55		var mu sync.Mutex
54	56		_, exit, err = execstream.DriveJob(conn, &execpb.Run{Job: &execpb.JobSpec{
55	57		WorkflowPath: ".github/workflows/ci.yml", WorkflowYaml: yaml, JobKey: "j", Repo: "hello", RunnerName: "r", RunnerVersion: "v",

@@ -60,7 +62,12 @@ func runJobSession(t *testing.T, yaml string, cancelAfter time.Duration, tweak f
60	62		steps = append(steps, s.Name)
61	63		mu.Unlock()
62	64		},
63		-	Log: func(l *execpb.Log) { logs = append(logs, l) },
	65	+	Log: func(l *execpb.Log) {
	66	+	logs = append(logs, l)
	67	+	if cancelOn != "" && l.Text == cancelOn {
	68	+	cancelOnce.Do(func() { cancel <- &execpb.Cancel{Reason: "test", Kind: execpb.CancelKind_CANCEL_KIND_USER} })
	69	+	}
	70	+	},
64	71		JobResult: func(r *execpb.JobResult) { res = r },
65	72		})
66	73		if err != nil {

@@ -79,7 +86,7 @@ func requireBash(t *testing.T) {
79	86
80	87		func TestExecRunsAJob(t *testing.T) {
81	88		requireBash(t)
82		-	steps, logs, res, exit := runJobSession(t, "on: push\njobs:\n j:\n runs-on: x\n outputs:\n o: ${{ steps.s.outputs.v }}\n steps:\n - id: s\n run: \|\n echo hello-from-job\n echo v=42 >> \"$GITHUB_OUTPUT\"\n", 0, nil)
	89	+	steps, logs, res, exit := runJobSession(t, "on: push\njobs:\n j:\n runs-on: x\n outputs:\n o: ${{ steps.s.outputs.v }}\n steps:\n - id: s\n run: \|\n echo hello-from-job\n echo v=42 >> \"$GITHUB_OUTPUT\"\n", "", nil)
83	90		if res == nil \|\| res.Conclusion != execpb.StepConclusion_STEP_CONCLUSION_SUCCESS \|\| res.Outputs["o"] != "42" \|\| exit.Code != 0 {
84	91		t.Fatalf("res=%v exit=%v", res, exit)
85	92		}

@@ -99,7 +106,7 @@ func TestExecRunsAJob(t *testing.T) {
99	106
100	107		func TestExecJobFailureExitsNonZero(t *testing.T) {
101	108		requireBash(t)
102		-	_, _, res, exit := runJobSession(t, "on: push\njobs:\n j:\n runs-on: x\n steps:\n - run: exit 4\n", 0, nil)
	109	+	_, _, res, exit := runJobSession(t, "on: push\njobs:\n j:\n runs-on: x\n steps:\n - run: exit 4\n", "", nil)
103	110		if res.Conclusion != execpb.StepConclusion_STEP_CONCLUSION_FAILURE \|\| exit.Code == 0 {
104	111		t.Fatalf("res=%v exit=%v", res, exit)
105	112		}

@@ -107,7 +114,7 @@ func TestExecJobFailureExitsNonZero(t *testing.T) {
107	114
108	115		func TestExecJobLongLinesAreContinued(t *testing.T) {
109	116		requireBash(t)
110		-	_, logs, res, _ := runJobSession(t, "on: push\njobs:\n j:\n runs-on: x\n steps:\n - run: head -c 150000 /dev/zero \| tr '\\0' y; echo\n", 0, nil)
	117	+	_, logs, res, _ := runJobSession(t, "on: push\njobs:\n j:\n runs-on: x\n steps:\n - run: head -c 150000 /dev/zero \| tr '\\0' y; echo\n", "", nil)
111	118		if res.Conclusion != execpb.StepConclusion_STEP_CONCLUSION_SUCCESS {
112	119		t.Fatalf("%v", res)
113	120		}

@@ -130,7 +137,7 @@ func TestExecJobLongLinesAreContinued(t *testing.T) {
130	137		func TestExecJobCancel(t *testing.T) {
131	138		requireBash(t)
132	139		start := time.Now()
133		-	_, _, res, _ := runJobSession(t, "on: push\njobs:\n j:\n runs-on: x\n steps:\n - run: \|\n echo ready\n sleep 30\n", 700*time.Millisecond, nil)
	140	+	_, _, res, _ := runJobSession(t, "on: push\njobs:\n j:\n runs-on: x\n steps:\n - run: \|\n echo ready\n sleep 30\n", "ready", nil)
134	141		if res.Conclusion != execpb.StepConclusion_STEP_CONCLUSION_CANCELLED \|\| res.RunnerShutdown {
135	142		t.Fatalf("%v", res)
136	143		}

@@ -161,10 +168,19 @@ func TestExecJobHostGoneIsAShutdown(t *testing.T) {
161	168		t.Fatal(err)
162	169		}
163	170		_ = c.Send(&execpb.Frame{Body: &execpb.Frame_Run{Run: &execpb.Run{Job: &execpb.JobSpec{
164		-	WorkflowYaml: "on: push\njobs:\n j:\n runs-on: x\n steps:\n - run: sleep 60\n", JobKey: "j", Repo: "hello",
	171	+	WorkflowYaml: "on: push\njobs:\n j:\n runs-on: x\n steps:\n - run: echo running; sleep 60\n", JobKey: "j", Repo: "hello",
165	172		Contexts: `{"github":{"repository":"o/hello"}}`,
166	173		}}}})
167		-	time.Sleep(700 * time.Millisecond)
	174	+	// Close the stream once the step is running (its first line arrived).
	175	+	for {
	176	+	f, err := c.Recv()
	177	+	if err != nil {
	178	+	t.Fatal(err)
	179	+	}
	180	+	if l := f.GetLog(); l != nil && l.Text == "running" {
	181	+	break
	182	+	}
	183	+	}
168	184		_ = conn.Close()
169	185		select {
170	186		case <-done:

 )
 // runJobSession starts Exec on a unix socket and drives it with a job.
-func runJobSession(t *testing.T, yaml string, cancelAfter time.Duration, tweak func(*ExecConfig)) (steps []string, logs []*execpb.Log, res *execpb.JobResult, exit *execpb.Exit) {
+func runJobSession(t *testing.T, yaml string, cancelOn string, tweak func(*ExecConfig)) (steps []string, logs []*execpb.Log, res *execpb.JobResult, exit *execpb.Exit) {
 	t.Helper()
 	sock := filepath.Join(t.TempDir(), "s")
 	ln, err := net.Listen("unix", sock)
 		t.Fatal(err)
 	}
 	defer func() { _ = conn.Close() }()
+	// With cancelOn, the job is cancelled when it logs that line (the script
+	// prints it once it is ready to be interrupted): no timer to guess.
 	var cancel chan *execpb.Cancel
-	if cancelAfter > 0 {
+	if cancelOn != "" {
 		cancel = make(chan *execpb.Cancel, 1)
-		time.AfterFunc(cancelAfter, func() { cancel <- &execpb.Cancel{Reason: "test", Kind: execpb.CancelKind_CANCEL_KIND_USER} })
 	}
+	var cancelOnce sync.Once
 	var mu sync.Mutex
 	_, exit, err = execstream.DriveJob(conn, &execpb.Run{Job: &execpb.JobSpec{
 		WorkflowPath: ".github/workflows/ci.yml", WorkflowYaml: yaml, JobKey: "j", Repo: "hello", RunnerName: "r", RunnerVersion: "v",
 			steps = append(steps, s.Name)
 			mu.Unlock()
 		},
-		Log:       func(l *execpb.Log) { logs = append(logs, l) },
+		Log: func(l *execpb.Log) {
+			logs = append(logs, l)
+			if cancelOn != "" && l.Text == cancelOn {
+				cancelOnce.Do(func() { cancel <- &execpb.Cancel{Reason: "test", Kind: execpb.CancelKind_CANCEL_KIND_USER} })
+			}
+		},
 		JobResult: func(r *execpb.JobResult) { res = r },
 	})
 	if err != nil {
 func TestExecRunsAJob(t *testing.T) {
 	requireBash(t)
-	steps, logs, res, exit := runJobSession(t, "on: push\njobs:\n  j:\n    runs-on: x\n    outputs:\n      o: ${{ steps.s.outputs.v }}\n    steps:\n      - id: s\n        run: |\n          echo hello-from-job\n          echo v=42 >> \"$GITHUB_OUTPUT\"\n", 0, nil)
+	steps, logs, res, exit := runJobSession(t, "on: push\njobs:\n  j:\n    runs-on: x\n    outputs:\n      o: ${{ steps.s.outputs.v }}\n    steps:\n      - id: s\n        run: |\n          echo hello-from-job\n          echo v=42 >> \"$GITHUB_OUTPUT\"\n", "", nil)
 	if res == nil || res.Conclusion != execpb.StepConclusion_STEP_CONCLUSION_SUCCESS || res.Outputs["o"] != "42" || exit.Code != 0 {
 		t.Fatalf("res=%v exit=%v", res, exit)
 	}
 func TestExecJobFailureExitsNonZero(t *testing.T) {
 	requireBash(t)
-	_, _, res, exit := runJobSession(t, "on: push\njobs:\n  j:\n    runs-on: x\n    steps:\n      - run: exit 4\n", 0, nil)
+	_, _, res, exit := runJobSession(t, "on: push\njobs:\n  j:\n    runs-on: x\n    steps:\n      - run: exit 4\n", "", nil)
 	if res.Conclusion != execpb.StepConclusion_STEP_CONCLUSION_FAILURE || exit.Code == 0 {
 		t.Fatalf("res=%v exit=%v", res, exit)
 	}
 func TestExecJobLongLinesAreContinued(t *testing.T) {
 	requireBash(t)
-	_, logs, res, _ := runJobSession(t, "on: push\njobs:\n  j:\n    runs-on: x\n    steps:\n      - run: head -c 150000 /dev/zero | tr '\\0' y; echo\n", 0, nil)
+	_, logs, res, _ := runJobSession(t, "on: push\njobs:\n  j:\n    runs-on: x\n    steps:\n      - run: head -c 150000 /dev/zero | tr '\\0' y; echo\n", "", nil)
 	if res.Conclusion != execpb.StepConclusion_STEP_CONCLUSION_SUCCESS {
 		t.Fatalf("%v", res)
 	}
 func TestExecJobCancel(t *testing.T) {
 	requireBash(t)
 	start := time.Now()
-	_, _, res, _ := runJobSession(t, "on: push\njobs:\n  j:\n    runs-on: x\n    steps:\n      - run: |\n          echo ready\n          sleep 30\n", 700*time.Millisecond, nil)
+	_, _, res, _ := runJobSession(t, "on: push\njobs:\n  j:\n    runs-on: x\n    steps:\n      - run: |\n          echo ready\n          sleep 30\n", "ready", nil)
 	if res.Conclusion != execpb.StepConclusion_STEP_CONCLUSION_CANCELLED || res.RunnerShutdown {
 		t.Fatalf("%v", res)
 	}
 		t.Fatal(err)
 	}
 	_ = c.Send(&execpb.Frame{Body: &execpb.Frame_Run{Run: &execpb.Run{Job: &execpb.JobSpec{
-		WorkflowYaml: "on: push\njobs:\n  j:\n    runs-on: x\n    steps:\n      - run: sleep 60\n", JobKey: "j", Repo: "hello",
+		WorkflowYaml: "on: push\njobs:\n  j:\n    runs-on: x\n    steps:\n      - run: echo running; sleep 60\n", JobKey: "j", Repo: "hello",
 		Contexts: `{"github":{"repository":"o/hello"}}`,
 	}}}})
-	time.Sleep(700 * time.Millisecond)
+	// Close the stream once the step is running (its first line arrived).
+	for {
+		f, err := c.Recv()
+		if err != nil {
+			t.Fatal(err)
+		}
+		if l := f.GetLog(); l != nil && l.Text == "running" {
+			break
+		}
+	}
 	_ = conn.Close()
 	select {
 	case <-done: