test(perf): drive the ssh peer through the broker mux

commit: db8139bb55f42e142c3fe38cd3fa093f9e87a9ac
author: brice · 29d ago · 2026-09-23 13:45 UTC
parent: 0328584
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

6 files · +17 -17

M cmd/vev-perf-harness/evidence_test.go +2 -2


@@ -25,7 +25,7 @@ func TestCLITransportSeamOwnsExclusivePeerTraceAndCleanup(t *testing.T) {
25	25		if err := os.WriteFile(m.TracePath, nil, 0o600); err != nil {
26	26		t.Fatal(err)
27	27		}
28		-	p, err := l.preparePeer(m, roleCommand{Args: []string{"_stdio"}, Transport: transport{ID: "ssh_stdio", Kind: "ssh_stdio"}})
	28	+	p, err := l.preparePeer(m, roleCommand{Args: []string{"_broker-mux-stdio", "--production"}, Transport: transport{ID: "ssh_stdio", Kind: "ssh_stdio"}})
29	29		if err != nil {
30	30		t.Fatal(err)
31	31		}

@@ -40,7 +40,7 @@ func TestCLITransportSeamOwnsExclusivePeerTraceAndCleanup(t *testing.T) {
40	40		}
41	41		}
42	42		if strings.Contains(text, "VEV_PERF_UDP_") {
43		-	t.Fatalf("ignored vev UDP environment was used instead of the _stdio seam: %s", text)
	43	+	t.Fatalf("ignored vev UDP environment was used instead of the broker mux stdio seam: %s", text)
44	44		}
45	45		// Close must be a no-op for a peer with no owned background process.
46	46		if err := p.Close(); err != nil {


@@ -25,7 +25,7 @@ func TestCLITransportSeamOwnsExclusivePeerTraceAndCleanup(t *testing.T) {
25		if err := os.WriteFile(m.TracePath, nil, 0o600); err != nil {	25		if err := os.WriteFile(m.TracePath, nil, 0o600); err != nil {
26		t.Fatal(err)	26		t.Fatal(err)
27		}	27		}
28	-	p, err := l.preparePeer(m, roleCommand{Args: []string{"_stdio"}, Transport: transport{ID: "ssh_stdio", Kind: "ssh_stdio"}})	28	+	p, err := l.preparePeer(m, roleCommand{Args: []string{"_broker-mux-stdio", "--production"}, Transport: transport{ID: "ssh_stdio", Kind: "ssh_stdio"}})
29		if err != nil {	29		if err != nil {
30		t.Fatal(err)	30		t.Fatal(err)
31		}	31		}

@@ -40,7 +40,7 @@ func TestCLITransportSeamOwnsExclusivePeerTraceAndCleanup(t *testing.T) {
40		}	40		}
41		}	41		}
42		if strings.Contains(text, "VEV_PERF_UDP_") {	42		if strings.Contains(text, "VEV_PERF_UDP_") {
43	-	t.Fatalf("ignored vev UDP environment was used instead of the _stdio seam: %s", text)	43	+	t.Fatalf("ignored vev UDP environment was used instead of the broker mux stdio seam: %s", text)
44		}	44		}
45		// Close must be a no-op for a peer with no owned background process.	45		// Close must be a no-op for a peer with no owned background process.
46		if err := p.Close(); err != nil {	46		if err := p.Close(); err != nil {

M cmd/vev-perf-harness/hangup_test.go +3 -3


@@ -137,7 +137,7 @@ func TestClientTracePairsAfterPTYHangup(t *testing.T) {
137	137
138	138		// TestClientGracefulShutdownDrainsTracedStdioDescendant proves the launcher's
139	139		// normal client stop exits the attached shell before sending any signal. The
140		-	// client then closes and waits for its ssh transport, giving the traced _stdio
	140	+	// client then closes and waits for its ssh transport, giving the traced broker mux stdio
141	141		// descendant time to serialize its blocked receive's end mark and exit before
142	142		// the client is reaped.
143	143		func TestClientGracefulShutdownDrainsTracedStdioDescendant(t *testing.T) {

@@ -185,8 +185,8 @@ func TestClientGracefulShutdownDrainsTracedStdioDescendant(t *testing.T) {
185	185		case "daemon":
186	186		command = routeRoleArgs(s, pm, transport{})
187	187		case "ssh_stdio_peer":
188		-	// The transport-neutral _stdio mode requests an ephemeral session.
189		-	command = roleCommand{Args: []string{"_stdio"}, Transport: transport{ID: "ssh_stdio", Kind: "ssh_stdio"}}
	188	+	// The broker mux stdio helper carries the remote daemon connection.
	189	+	command = roleCommand{Args: []string{"_broker-mux-stdio", "--production"}, Transport: transport{ID: "ssh_stdio", Kind: "ssh_stdio"}}
190	190		case "client":
191	191		command = roleCommand{Args: []string{"attach", "harness@127.0.0.1"}}
192	192		}


@@ -137,7 +137,7 @@ func TestClientTracePairsAfterPTYHangup(t *testing.T) {
137			137
138		// TestClientGracefulShutdownDrainsTracedStdioDescendant proves the launcher's	138		// TestClientGracefulShutdownDrainsTracedStdioDescendant proves the launcher's
139		// normal client stop exits the attached shell before sending any signal. The	139		// normal client stop exits the attached shell before sending any signal. The
140	-	// client then closes and waits for its ssh transport, giving the traced _stdio	140	+	// client then closes and waits for its ssh transport, giving the traced broker mux stdio
141		// descendant time to serialize its blocked receive's end mark and exit before	141		// descendant time to serialize its blocked receive's end mark and exit before
142		// the client is reaped.	142		// the client is reaped.
143		func TestClientGracefulShutdownDrainsTracedStdioDescendant(t *testing.T) {	143		func TestClientGracefulShutdownDrainsTracedStdioDescendant(t *testing.T) {

@@ -185,8 +185,8 @@ func TestClientGracefulShutdownDrainsTracedStdioDescendant(t *testing.T) {
185		case "daemon":	185		case "daemon":
186		command = routeRoleArgs(s, pm, transport{})	186		command = routeRoleArgs(s, pm, transport{})
187		case "ssh_stdio_peer":	187		case "ssh_stdio_peer":
188	-	// The transport-neutral _stdio mode requests an ephemeral session.	188	+	// The broker mux stdio helper carries the remote daemon connection.
189	-	command = roleCommand{Args: []string{"_stdio"}, Transport: transport{ID: "ssh_stdio", Kind: "ssh_stdio"}}	189	+	command = roleCommand{Args: []string{"_broker-mux-stdio", "--production"}, Transport: transport{ID: "ssh_stdio", Kind: "ssh_stdio"}}
190		case "client":	190		case "client":
191		command = roleCommand{Args: []string{"attach", "harness@127.0.0.1"}}	191		command = roleCommand{Args: []string{"attach", "harness@127.0.0.1"}}
192		}	192		}

M cmd/vev-perf-harness/launcher.go +7 -7


@@ -303,7 +303,7 @@ func (l *cliLauncher) Launch(m processMapping, role roleCommand) (launchedProces
303	303		p.pty, p.output, p.chunks, p.done = master, &capturedTerminalOutput{file: output}, make(chan []byte, 32), make(chan struct{})
304	304		// Stop workload generation through the attached shell before any signal.
305	305		// A normal shell exit makes the daemon detach the client; the client then
306		-	// closes and waits for its ssh transport, allowing a traced _stdio
	306	+	// closes and waits for its ssh transport, allowing a traced broker mux stdio
307	307		// descendant to flush end marks and exit. Process-group signals are reserved
308	308		// for timeout escalation because they would kill the descendant concurrently.
309	309		p.gracefulShutdown = func() { _, _ = master.Write([]byte("exit\n")) }

@@ -379,7 +379,7 @@ func (l *cliLauncher) releaseRuntime(runDir string) error {
379	379
380	380		// preparePeer installs a per-run ssh command seam. The ordinary public client
381	381		// still executes its documented remote attach command; its ssh child invokes
382		-	// only the parsed public _stdio entrypoint. This makes the peer
	382	+	// only the broker mux stdio helper. This makes the peer
383	383		// carrying client traffic the declared, exclusively traced role rather than a
384	384		// separately launched but unused process.
385	385		func (l *cliLauncher) preparePeer(m processMapping, role roleCommand) (launchedProcess, error) {

@@ -410,8 +410,8 @@ func (l *cliLauncher) preparePeer(m processMapping, role roleCommand) (launchedP
410	410		if err := safedir.EnsurePrivate(filepath.Join(runDir, "state")); err != nil {
411	411		return nil, fail(err)
412	412		}
413		-	if len(role.Args) != 1 \|\| role.Args[0] != "_stdio" {
414		-	return nil, fail(fmt.Errorf("unsupported public peer command %q", role.Args))
	413	+	if len(role.Args) != 2 \|\| role.Args[0] != "_broker-mux-stdio" \|\| role.Args[1] != "--production" {
	414	+	return nil, fail(fmt.Errorf("unsupported broker mux peer command %q", role.Args))
415	415		}
416	416		route := peerRoute{mapping: m, command: role, runtimeDir: filepath.Join(runDir, "peer-runtime"), stateDir: filepath.Join(runDir, "peer-state")}
417	417		if err := safedir.EnsurePrivate(route.runtimeDir); err != nil {

@@ -421,7 +421,7 @@ func (l *cliLauncher) preparePeer(m processMapping, role roleCommand) (launchedP
421	421		return nil, fail(err)
422	422		}
423	423		shim := filepath.Join(runDir, "ssh")
424		-	// The SSH command seam forwards the public _stdio entrypoint with the
	424	+	// The SSH command seam forwards the broker mux stdio helper with the
425	425		// peer's unique trace identity, so the peer carrying client traffic is the
426	426		// declared, exclusively traced role.
427	427		body := fmt.Sprintf(`#!/bin/sh

@@ -770,7 +770,7 @@ func (p *cliProcess) Close() error {
770	770		waitForExit()
771	771		}
772	772		// 3. Client role: stop the shell normally. Session teardown detaches the
773		-	// client, which closes and waits for ssh/_stdio before it exits.
	773	+	// client, which closes and waits for ssh/broker mux stdio before it exits.
774	774		if p.gracefulShutdown != nil && !reaped() {
775	775		p.gracefulShutdown()
776	776		waitForExit()

@@ -882,7 +882,7 @@ func (p *cliProcess) forceProcessGroupCleanup() {
882	882		}
883	883		if p.cmd != nil && p.cmd.Process != nil {
884	884		// The client is a session/process-group leader; forced cleanup must
885		-	// include its ssh seam/_stdio descendants.
	885	+	// include its ssh seam/broker mux stdio descendants.
886	886		_ = syscall.Kill(-p.cmd.Process.Pid, syscall.SIGTERM)
887	887		}
888	888		}


@@ -303,7 +303,7 @@ func (l *cliLauncher) Launch(m processMapping, role roleCommand) (launchedProces
303		p.pty, p.output, p.chunks, p.done = master, &capturedTerminalOutput{file: output}, make(chan []byte, 32), make(chan struct{})	303		p.pty, p.output, p.chunks, p.done = master, &capturedTerminalOutput{file: output}, make(chan []byte, 32), make(chan struct{})
304		// Stop workload generation through the attached shell before any signal.	304		// Stop workload generation through the attached shell before any signal.
305		// A normal shell exit makes the daemon detach the client; the client then	305		// A normal shell exit makes the daemon detach the client; the client then
306	-	// closes and waits for its ssh transport, allowing a traced _stdio	306	+	// closes and waits for its ssh transport, allowing a traced broker mux stdio
307		// descendant to flush end marks and exit. Process-group signals are reserved	307		// descendant to flush end marks and exit. Process-group signals are reserved
308		// for timeout escalation because they would kill the descendant concurrently.	308		// for timeout escalation because they would kill the descendant concurrently.
309		p.gracefulShutdown = func() { _, _ = master.Write([]byte("exit\n")) }	309		p.gracefulShutdown = func() { _, _ = master.Write([]byte("exit\n")) }

@@ -379,7 +379,7 @@ func (l *cliLauncher) releaseRuntime(runDir string) error {
379			379
380		// preparePeer installs a per-run ssh command seam. The ordinary public client	380		// preparePeer installs a per-run ssh command seam. The ordinary public client
381		// still executes its documented remote attach command; its ssh child invokes	381		// still executes its documented remote attach command; its ssh child invokes
382	-	// only the parsed public _stdio entrypoint. This makes the peer	382	+	// only the broker mux stdio helper. This makes the peer
383		// carrying client traffic the declared, exclusively traced role rather than a	383		// carrying client traffic the declared, exclusively traced role rather than a
384		// separately launched but unused process.	384		// separately launched but unused process.
385		func (l *cliLauncher) preparePeer(m processMapping, role roleCommand) (launchedProcess, error) {	385		func (l *cliLauncher) preparePeer(m processMapping, role roleCommand) (launchedProcess, error) {

@@ -410,8 +410,8 @@ func (l *cliLauncher) preparePeer(m processMapping, role roleCommand) (launchedP
410		if err := safedir.EnsurePrivate(filepath.Join(runDir, "state")); err != nil {	410		if err := safedir.EnsurePrivate(filepath.Join(runDir, "state")); err != nil {
411		return nil, fail(err)	411		return nil, fail(err)
412		}	412		}
413	-	if len(role.Args) != 1 \|\| role.Args[0] != "_stdio" {	413	+	if len(role.Args) != 2 \|\| role.Args[0] != "_broker-mux-stdio" \|\| role.Args[1] != "--production" {
414	-	return nil, fail(fmt.Errorf("unsupported public peer command %q", role.Args))	414	+	return nil, fail(fmt.Errorf("unsupported broker mux peer command %q", role.Args))
415		}	415		}
416		route := peerRoute{mapping: m, command: role, runtimeDir: filepath.Join(runDir, "peer-runtime"), stateDir: filepath.Join(runDir, "peer-state")}	416		route := peerRoute{mapping: m, command: role, runtimeDir: filepath.Join(runDir, "peer-runtime"), stateDir: filepath.Join(runDir, "peer-state")}
417		if err := safedir.EnsurePrivate(route.runtimeDir); err != nil {	417		if err := safedir.EnsurePrivate(route.runtimeDir); err != nil {

@@ -421,7 +421,7 @@ func (l *cliLauncher) preparePeer(m processMapping, role roleCommand) (launchedP
421		return nil, fail(err)	421		return nil, fail(err)
422		}	422		}
423		shim := filepath.Join(runDir, "ssh")	423		shim := filepath.Join(runDir, "ssh")
424	-	// The SSH command seam forwards the public _stdio entrypoint with the	424	+	// The SSH command seam forwards the broker mux stdio helper with the
425		// peer's unique trace identity, so the peer carrying client traffic is the	425		// peer's unique trace identity, so the peer carrying client traffic is the
426		// declared, exclusively traced role.	426		// declared, exclusively traced role.
427		body := fmt.Sprintf(`#!/bin/sh	427		body := fmt.Sprintf(`#!/bin/sh

@@ -770,7 +770,7 @@ func (p *cliProcess) Close() error {
770		waitForExit()	770		waitForExit()
771		}	771		}
772		// 3. Client role: stop the shell normally. Session teardown detaches the	772		// 3. Client role: stop the shell normally. Session teardown detaches the
773	-	// client, which closes and waits for ssh/_stdio before it exits.	773	+	// client, which closes and waits for ssh/broker mux stdio before it exits.
774		if p.gracefulShutdown != nil && !reaped() {	774		if p.gracefulShutdown != nil && !reaped() {
775		p.gracefulShutdown()	775		p.gracefulShutdown()
776		waitForExit()	776		waitForExit()

@@ -882,7 +882,7 @@ func (p *cliProcess) forceProcessGroupCleanup() {
882		}	882		}
883		if p.cmd != nil && p.cmd.Process != nil {	883		if p.cmd != nil && p.cmd.Process != nil {
884		// The client is a session/process-group leader; forced cleanup must	884		// The client is a session/process-group leader; forced cleanup must
885	-	// include its ssh seam/_stdio descendants.	885	+	// include its ssh seam/broker mux stdio descendants.
886		_ = syscall.Kill(-p.cmd.Process.Pid, syscall.SIGTERM)	886		_ = syscall.Kill(-p.cmd.Process.Pid, syscall.SIGTERM)
887		}	887		}
888		}	888		}

M cmd/vev-perf-harness/launcher_test.go +2 -2


@@ -415,7 +415,7 @@ func TestPreparePeerReleasesOwnedRuntimeOnPreparationFailure(t *testing.T) {
415	415		Role: "ssh_stdio_peer",
416	416		TracePath: filepath.Join(runDir, "peer.jsonl"),
417	417		}, roleCommand{Args: []string{"invalid"}})
418		-	if err == nil \|\| !strings.Contains(err.Error(), "unsupported public peer command") {
	418	+	if err == nil \|\| !strings.Contains(err.Error(), "unsupported broker mux peer command") {
419	419		t.Fatalf("preparePeer error = %v", err)
420	420		}
421	421		if runtimeDir := launcher.runtimes[runDir]; runtimeDir != "" {

@@ -603,7 +603,7 @@ func TestHarnessUsesPublicRoleCommandsAndPTYWorkloads(t *testing.T) {
603	603		}{
604	604		{"daemon", []string{"_broker-production-serve"}},
605	605		{"client", []string{"new", "perf-s-001"}},
606		-	{"ssh_stdio_peer", []string{"_stdio"}},
	606	+	{"ssh_stdio_peer", []string{"_broker-mux-stdio", "--production"}},
607	607		} {
608	608		t.Run(tc.role, func(t *testing.T) {
609	609		got := roleArgs(scenario{ID: "s", Transport: "local"}, processMapping{Role: tc.role, Run: 1})


@@ -415,7 +415,7 @@ func TestPreparePeerReleasesOwnedRuntimeOnPreparationFailure(t *testing.T) {
415		Role: "ssh_stdio_peer",	415		Role: "ssh_stdio_peer",
416		TracePath: filepath.Join(runDir, "peer.jsonl"),	416		TracePath: filepath.Join(runDir, "peer.jsonl"),
417		}, roleCommand{Args: []string{"invalid"}})	417		}, roleCommand{Args: []string{"invalid"}})
418	-	if err == nil \|\| !strings.Contains(err.Error(), "unsupported public peer command") {	418	+	if err == nil \|\| !strings.Contains(err.Error(), "unsupported broker mux peer command") {
419		t.Fatalf("preparePeer error = %v", err)	419		t.Fatalf("preparePeer error = %v", err)
420		}	420		}
421		if runtimeDir := launcher.runtimes[runDir]; runtimeDir != "" {	421		if runtimeDir := launcher.runtimes[runDir]; runtimeDir != "" {

@@ -603,7 +603,7 @@ func TestHarnessUsesPublicRoleCommandsAndPTYWorkloads(t *testing.T) {
603		}{	603		}{
604		{"daemon", []string{"_broker-production-serve"}},	604		{"daemon", []string{"_broker-production-serve"}},
605		{"client", []string{"new", "perf-s-001"}},	605		{"client", []string{"new", "perf-s-001"}},
606	-	{"ssh_stdio_peer", []string{"_stdio"}},	606	+	{"ssh_stdio_peer", []string{"_broker-mux-stdio", "--production"}},
607		} {	607		} {
608		t.Run(tc.role, func(t *testing.T) {	608		t.Run(tc.role, func(t *testing.T) {
609		got := roleArgs(scenario{ID: "s", Transport: "local"}, processMapping{Role: tc.role, Run: 1})	609		got := roleArgs(scenario{ID: "s", Transport: "local"}, processMapping{Role: tc.role, Run: 1})

M cmd/vev-perf-harness/runner.go +1 -1


@@ -398,7 +398,7 @@ func routeRoleArgs(s scenario, m processMapping, selected transport) roleCommand
398	398		return roleCommand{Args: remote}
399	399		}
400	400		case "ssh_stdio_peer":
401		-	return roleCommand{Args: []string{"_stdio"}, Transport: transport{ID: s.Transport, Kind: "ssh_stdio"}}
	401	+	return roleCommand{Args: []string{"_broker-mux-stdio", "--production"}, Transport: transport{ID: s.Transport, Kind: "ssh_stdio"}}
402	402		}
403	403		return roleCommand{}
404	404		}


@@ -398,7 +398,7 @@ func routeRoleArgs(s scenario, m processMapping, selected transport) roleCommand
398		return roleCommand{Args: remote}	398		return roleCommand{Args: remote}
399		}	399		}
400		case "ssh_stdio_peer":	400		case "ssh_stdio_peer":
401	-	return roleCommand{Args: []string{"_stdio"}, Transport: transport{ID: s.Transport, Kind: "ssh_stdio"}}	401	+	return roleCommand{Args: []string{"_broker-mux-stdio", "--production"}, Transport: transport{ID: s.Transport, Kind: "ssh_stdio"}}
402		}	402		}
403		return roleCommand{}	403		return roleCommand{}
404		}	404		}

M cmd/vev-perf-harness/runner_test.go +2 -2


@@ -254,7 +254,7 @@ func TestHarnessRoutesEveryRemoteFixtureThroughItsDeclaredPeer(t *testing.T) {
254	254		if !equalStrings(client.Args, []string{"attach", "harness@127.0.0.1"}) {
255	255		t.Fatalf("client did not use ephemeral public remote attach: %q", client.Args)
256	256		}
257		-	if !equalStrings(peer.Args, []string{"_stdio"}) {
	257	+	if !equalStrings(peer.Args, []string{"_broker-mux-stdio", "--production"}) {
258	258		t.Fatalf("ssh peer command=%q", peer.Args)
259	259		}
260	260		})

@@ -265,7 +265,7 @@ func TestHarnessFakeRunnerRoutesClientToPeerAndCleansEveryRole(t *testing.T) {
265	265		for _, tc := range []struct {
266	266		name, transport, peer, peerCommand string
267	267		}{
268		-	{"ssh", "ssh_stdio", "ssh_stdio_peer", "_stdio"},
	268	+	{"ssh", "ssh_stdio", "ssh_stdio_peer", "_broker-mux-stdio"},
269	269		} {
270	270		t.Run(tc.name, func(t *testing.T) {
271	271		dir := t.TempDir()


@@ -254,7 +254,7 @@ func TestHarnessRoutesEveryRemoteFixtureThroughItsDeclaredPeer(t *testing.T) {
254		if !equalStrings(client.Args, []string{"attach", "harness@127.0.0.1"}) {	254		if !equalStrings(client.Args, []string{"attach", "harness@127.0.0.1"}) {
255		t.Fatalf("client did not use ephemeral public remote attach: %q", client.Args)	255		t.Fatalf("client did not use ephemeral public remote attach: %q", client.Args)
256		}	256		}
257	-	if !equalStrings(peer.Args, []string{"_stdio"}) {	257	+	if !equalStrings(peer.Args, []string{"_broker-mux-stdio", "--production"}) {
258		t.Fatalf("ssh peer command=%q", peer.Args)	258		t.Fatalf("ssh peer command=%q", peer.Args)
259		}	259		}
260		})	260		})

@@ -265,7 +265,7 @@ func TestHarnessFakeRunnerRoutesClientToPeerAndCleansEveryRole(t *testing.T) {
265		for _, tc := range []struct {	265		for _, tc := range []struct {
266		name, transport, peer, peerCommand string	266		name, transport, peer, peerCommand string
267		}{	267		}{
268	-	{"ssh", "ssh_stdio", "ssh_stdio_peer", "_stdio"},	268	+	{"ssh", "ssh_stdio", "ssh_stdio_peer", "_broker-mux-stdio"},
269		} {	269		} {
270		t.Run(tc.name, func(t *testing.T) {	270		t.Run(tc.name, func(t *testing.T) {
271		dir := t.TempDir()	271		dir := t.TempDir()