fix(obs): flush spans before harness cleanup

commit: a244e217eab3750a66f6d7c6aaf7ee7cc1b3813a
author: brice · 3mo ago · 2026-07-12 23:19 UTC
parent: e99dffb
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +83 -20

M cmd/vev-perf-harness/main.go +37 -20


@@ -193,13 +193,15 @@ type terminalOutput interface {
193	193		}
194	194
195	195		type cliProcess struct {
196		-	cmd *exec.Cmd
197		-	pty io.ReadWriteCloser
198		-	output terminalOutput
199		-	chunks chan []byte
200		-	done chan struct{}
201		-	closed sync.Once
202		-	waitErr chan error
	196	+	cmd *exec.Cmd
	197	+	pty io.ReadWriteCloser
	198	+	output terminalOutput
	199	+	chunks chan []byte
	200	+	done chan struct{}
	201	+	closed sync.Once
	202	+	waitErr chan error
	203	+	waitTimeout func() <-chan time.Time
	204	+	forceCleanup func()
203	205		}
204	206
205	207		func (l *cliLauncher) Launch(m processMapping, role roleCommand) (launchedProcess, error) {

@@ -456,32 +458,47 @@ func (p *cliProcess) Close() error {
456	458		if p.done != nil {
457	459		close(p.done)
458	460		}
459		-	if p.cmd != nil && p.cmd.Process != nil {
460		-	// The client is a session/process-group leader; terminate its ssh
461		-	// seam/_stdio descendants as well as the terminal client itself.
462		-	_ = syscall.Kill(-p.cmd.Process.Pid, syscall.SIGTERM)
463		-	}
	461	+	// Closing the client PTY causes its stdin pump to detach, then close the
	462	+	// transport and wait for its adapter receive end mark. Do this before a
	463	+	// signal so process-local spans are complete in the trace.
464	464		if p.pty != nil {
465	465		_ = p.pty.Close()
466	466		}
467		-	if p.output != nil {
468		-	result = p.output.Close()
469		-	}
470	467		if p.waitErr != nil {
471	468		select {
472	469		case <-p.waitErr:
473		-	// SIGTERM is the harness's normal deterministic cleanup path.
474		-	case <-time.After(2 * time.Second):
475		-	if p.cmd != nil && p.cmd.Process != nil {
476		-	_ = p.cmd.Process.Kill()
477		-	}
	470	+	// The client exited through its normal detach path.
	471	+	case <-p.closeDeadline():
	472	+	p.forceProcessGroupCleanup()
478	473		<-p.waitErr
479	474		}
480	475		}
	476	+	if p.output != nil {
	477	+	result = p.output.Close()
	478	+	}
481	479		})
482	480		return result
483	481		}
484	482
	483	+	func (p *cliProcess) closeDeadline() <-chan time.Time {
	484	+	if p.waitTimeout != nil {
	485	+	return p.waitTimeout()
	486	+	}
	487	+	return time.After(2 * time.Second)
	488	+	}
	489	+
	490	+	func (p *cliProcess) forceProcessGroupCleanup() {
	491	+	if p.forceCleanup != nil {
	492	+	p.forceCleanup()
	493	+	return
	494	+	}
	495	+	if p.cmd != nil && p.cmd.Process != nil {
	496	+	// The client is a session/process-group leader; forced cleanup must
	497	+	// include its ssh seam/_stdio descendants.
	498	+	_ = syscall.Kill(-p.cmd.Process.Pid, syscall.SIGTERM)
	499	+	}
	500	+	}
	501	+
485	502		// openPTY is the small Linux PTY boundary needed to drive the public terminal
486	503		// client. No daemon or adapter implementation is imported by this command.
487	504		func openPTY() (os.File, os.File, error) {

 }
 type cliProcess struct {
-	cmd     *exec.Cmd
+	cmd          *exec.Cmd
-	pty     io.ReadWriteCloser
+	pty          io.ReadWriteCloser
-	output  terminalOutput
+	output       terminalOutput
-	chunks  chan []byte
+	chunks       chan []byte
-	done    chan struct{}
+	done         chan struct{}
-	closed  sync.Once
+	closed       sync.Once
-	waitErr chan error
+	waitErr      chan error
+	waitTimeout  func() <-chan time.Time
+	forceCleanup func()
 }
 func (l *cliLauncher) Launch(m processMapping, role roleCommand) (launchedProcess, error) {
 		if p.done != nil {
 			close(p.done)
 		}
-		if p.cmd != nil && p.cmd.Process != nil {
+		// Closing the client PTY causes its stdin pump to detach, then close the
-			// The client is a session/process-group leader; terminate its ssh
+		// transport and wait for its adapter receive end mark. Do this before a
-			// seam/_stdio descendants as well as the terminal client itself.
+		// signal so process-local spans are complete in the trace.
-			_ = syscall.Kill(-p.cmd.Process.Pid, syscall.SIGTERM)
-		}
 		if p.pty != nil {
 			_ = p.pty.Close()
 		}
-		if p.output != nil {
-			result = p.output.Close()
-		}
 		if p.waitErr != nil {
 			select {
 			case <-p.waitErr:
-				// SIGTERM is the harness's normal deterministic cleanup path.
+				// The client exited through its normal detach path.
-			case <-time.After(2 * time.Second):
+			case <-p.closeDeadline():
-				if p.cmd != nil && p.cmd.Process != nil {
+				p.forceProcessGroupCleanup()
-					_ = p.cmd.Process.Kill()
-				}
 				<-p.waitErr
 			}
 		}
+		if p.output != nil {
+			result = p.output.Close()
+		}
 	})
 	return result
 }
+func (p *cliProcess) closeDeadline() <-chan time.Time {
+	if p.waitTimeout != nil {
+		return p.waitTimeout()
+	}
+	return time.After(2 * time.Second)
+}
+func (p *cliProcess) forceProcessGroupCleanup() {
+	if p.forceCleanup != nil {
+		p.forceCleanup()
+		return
+	}
+	if p.cmd != nil && p.cmd.Process != nil {
+		// The client is a session/process-group leader; forced cleanup must
+		// include its ssh seam/_stdio descendants.
+		_ = syscall.Kill(-p.cmd.Process.Pid, syscall.SIGTERM)
+	}
+}
 // openPTY is the small Linux PTY boundary needed to drive the public terminal
 // client. No daemon or adapter implementation is imported by this command.
 func openPTY() (*os.File, *os.File, error) {

M cmd/vev-perf-harness/main_test.go +46 -0


@@ -158,6 +158,52 @@ type fakeOutput struct {
158	158		func (o *fakeOutput) Sync() error { o.syncs++; return nil }
159	159		func (*fakeOutput) Close() error { return nil }
160	160
	161	+	type orderedPTY struct{ close func() }
	162	+
	163	+	func (*orderedPTY) Read([]byte) (int, error) { return 0, os.ErrClosed }
	164	+	func (*orderedPTY) Write(b []byte) (int, error) { return len(b), nil }
	165	+	func (p *orderedPTY) Close() error {
	166	+	p.close()
	167	+	return nil
	168	+	}
	169	+
	170	+	type orderedOutput struct{ close func() }
	171	+
	172	+	func (*orderedOutput) Write(b []byte) (int, error) { return len(b), nil }
	173	+	func (*orderedOutput) Sync() error { return nil }
	174	+	func (o *orderedOutput) Close() error {
	175	+	o.close()
	176	+	return nil
	177	+	}
	178	+
	179	+	func TestCLIProcessCloseGracefulDetachCompletesSpanBeforeForcedCleanup(t *testing.T) {
	180	+	var order []string
	181	+	waitErr := make(chan error, 1)
	182	+	timeout := make(chan time.Time)
	183	+	p := &cliProcess{
	184	+	pty: &orderedPTY{close: func() {
	185	+	order = append(order, "pty_closed")
	186	+	// The graceful client detach has closed its transport and emitted its
	187	+	// adapter receive-end mark before cmd.Wait reports completion.
	188	+	order = append(order, "adapter_receive_end")
	189	+	waitErr <- nil
	190	+	}},
	191	+	output: &orderedOutput{close: func() { order = append(order, "output_closed") }},
	192	+	waitErr: waitErr,
	193	+	waitTimeout: func() <-chan time.Time { return timeout },
	194	+	forceCleanup: func() {
	195	+	order = append(order, "forced_process_group_cleanup")
	196	+	},
	197	+	}
	198	+
	199	+	if err := p.Close(); err != nil {
	200	+	t.Fatal(err)
	201	+	}
	202	+	if !equalStrings(order, []string{"pty_closed", "adapter_receive_end", "output_closed"}) {
	203	+	t.Fatalf("cleanup order=%q", order)
	204	+	}
	205	+	}
	206	+
161	207		func TestHarnessDoesNotInheritNestedSession(t *testing.T) {
162	208		env := withoutEnv([]string{"VEV=session=old", "PATH=/bin", "VEV_PERF_TRACE=old"}, "VEV")
163	209		if !equalStrings(env, []string{"PATH=/bin", "VEV_PERF_TRACE=old"}) {


@@ -158,6 +158,52 @@ type fakeOutput struct {
158	func (o *fakeOutput) Sync() error { o.syncs++; return nil }	158		func (o *fakeOutput) Sync() error { o.syncs++; return nil }
159	func (*fakeOutput) Close() error { return nil }	159		func (*fakeOutput) Close() error { return nil }
160		160
		161	+	type orderedPTY struct{ close func() }
		162	+
		163	+	func (*orderedPTY) Read([]byte) (int, error) { return 0, os.ErrClosed }
		164	+	func (*orderedPTY) Write(b []byte) (int, error) { return len(b), nil }
		165	+	func (p *orderedPTY) Close() error {
		166	+	p.close()
		167	+	return nil
		168	+	}
		169	+
		170	+	type orderedOutput struct{ close func() }
		171	+
		172	+	func (*orderedOutput) Write(b []byte) (int, error) { return len(b), nil }
		173	+	func (*orderedOutput) Sync() error { return nil }
		174	+	func (o *orderedOutput) Close() error {
		175	+	o.close()
		176	+	return nil
		177	+	}
		178	+
		179	+	func TestCLIProcessCloseGracefulDetachCompletesSpanBeforeForcedCleanup(t *testing.T) {
		180	+	var order []string
		181	+	waitErr := make(chan error, 1)
		182	+	timeout := make(chan time.Time)
		183	+	p := &cliProcess{
		184	+	pty: &orderedPTY{close: func() {
		185	+	order = append(order, "pty_closed")
		186	+	// The graceful client detach has closed its transport and emitted its
		187	+	// adapter receive-end mark before cmd.Wait reports completion.
		188	+	order = append(order, "adapter_receive_end")
		189	+	waitErr <- nil
		190	+	}},
		191	+	output: &orderedOutput{close: func() { order = append(order, "output_closed") }},
		192	+	waitErr: waitErr,
		193	+	waitTimeout: func() <-chan time.Time { return timeout },
		194	+	forceCleanup: func() {
		195	+	order = append(order, "forced_process_group_cleanup")
		196	+	},
		197	+	}
		198	+
		199	+	if err := p.Close(); err != nil {
		200	+	t.Fatal(err)
		201	+	}
		202	+	if !equalStrings(order, []string{"pty_closed", "adapter_receive_end", "output_closed"}) {
		203	+	t.Fatalf("cleanup order=%q", order)
		204	+	}
		205	+	}
		206	+
161	func TestHarnessDoesNotInheritNestedSession(t *testing.T) {	207		func TestHarnessDoesNotInheritNestedSession(t *testing.T) {
162	env := withoutEnv([]string{"VEV=session=old", "PATH=/bin", "VEV_PERF_TRACE=old"}, "VEV")	208		env := withoutEnv([]string{"VEV=session=old", "PATH=/bin", "VEV_PERF_TRACE=old"}, "VEV")
163	if !equalStrings(env, []string{"PATH=/bin", "VEV_PERF_TRACE=old"}) {	209		if !equalStrings(env, []string{"PATH=/bin", "VEV_PERF_TRACE=old"}) {