fix(obs): apply UDP network fixtures

commit: 3d59ccb0e5fe76166cc478e308274789ea2dd514
author: brice · 3mo ago · 2026-07-12 23:42 UTC
parent: 1ffeb55
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

3 files · +346 -53

M cmd/vev-perf-harness/main.go +57 -24


@@ -167,15 +167,17 @@ func closeLaunchedRoles(processes []launchedRole) error {
167	167		// cliLauncher is deliberately limited to executing the supplied public vev
168	168		// binary. Scenario orchestration never reaches into an in-process daemon.
169	169		type cliLauncher struct {
170		-	bin string
171		-	mu sync.Mutex
172		-	peers map[string]peerRoute
	170	+	bin string
	171	+	netemFactory udpNetemFactory
	172	+	mu sync.Mutex
	173	+	peers map[string]peerRoute
173	174		}
174	175
175	176		type peerRoute struct {
176	177		mapping processMapping
177	178		command roleCommand
178	179		pidPath string
	180	+	netem udpNetem
179	181		}
180	182
181	183		type preparedPeer struct{ route peerRoute }

@@ -183,24 +185,26 @@ type preparedPeer struct{ route peerRoute }
183	185		func (p preparedPeer) Warmup([]byte) error { return nil }
184	186		func (p preparedPeer) Measure([]byte, func() error, func() error) error { return nil }
185	187		func (p preparedPeer) Close() error {
186		-	if p.route.pidPath == "" {
187		-	return nil
188		-	}
189		-	b, err := os.ReadFile(p.route.pidPath)
190		-	if errors.Is(err, os.ErrNotExist) {
191		-	return nil
192		-	}
193		-	if err != nil {
194		-	return err
195		-	}
196		-	pid, err := strconv.Atoi(strings.TrimSpace(string(b)))
197		-	if err != nil \|\| pid <= 0 {
198		-	return fmt.Errorf("invalid owned peer pid: %q", b)
	188	+	var errs []error
	189	+	if p.route.pidPath != "" {
	190	+	b, err := os.ReadFile(p.route.pidPath)
	191	+	if err == nil {
	192	+	pid, parseErr := strconv.Atoi(strings.TrimSpace(string(b)))
	193	+	if parseErr != nil \|\| pid <= 0 {
	194	+	errs = append(errs, fmt.Errorf("invalid owned peer pid: %q", b))
	195	+	} else if killErr := syscall.Kill(pid, syscall.SIGTERM); killErr != nil && !errors.Is(killErr, syscall.ESRCH) {
	196	+	errs = append(errs, killErr)
	197	+	}
	198	+	} else if !errors.Is(err, os.ErrNotExist) {
	199	+	errs = append(errs, err)
	200	+	}
199	201		}
200		-	if err := syscall.Kill(pid, syscall.SIGTERM); err != nil && !errors.Is(err, syscall.ESRCH) {
201		-	return err
	202	+	if p.route.netem != nil {
	203	+	if err := p.route.netem.Close(); err != nil {
	204	+	errs = append(errs, err)
	205	+	}
202	206		}
203		-	return nil
	207	+	return errors.Join(errs...)
204	208		}
205	209
206	210		type terminalOutput interface {

@@ -319,9 +323,27 @@ func (l *cliLauncher) preparePeer(m processMapping, role roleCommand) (launchedP
319	323		route := peerRoute{mapping: m, command: role}
320	324		if m.Role == "udp_peer" {
321	325		route.pidPath = filepath.Join(runDir, "udp-peer.pid")
	326	+	factory := l.netemFactory
	327	+	if factory == nil {
	328	+	factory = newUDPNetem
	329	+	}
	330	+	netem, err := factory(udpNetemConfig{
	331	+	RTT: time.Duration(role.Transport.RTTMS) * time.Millisecond,
	332	+	LossPercent: role.Transport.LossPercent,
	333	+	TargetPath: filepath.Join(runDir, "udp-peer.target"),
	334	+	})
	335	+	if err != nil {
	336	+	return nil, fmt.Errorf("start harness UDP netem: %w", err)
	337	+	}
	338	+	route.netem = netem
322	339		}
323	340		shim := filepath.Join(runDir, "ssh")
324	341		ready := filepath.Join(runDir, "udp-peer.ready")
	342	+	target := filepath.Join(runDir, "udp-peer.target")
	343	+	netemPort := 0
	344	+	if route.netem != nil {
	345	+	netemPort = route.netem.Port()
	346	+	}
325	347		stderr := filepath.Join(runDir, "udp-peer.stderr")
326	348		// The bootstrap protocol requires the SSH command to exit after forwarding
327	349		// the readiness line. The long-lived proxy is owned by pidPath and receives

@@ -333,20 +355,28 @@ case "$*" in
333	355		exec env VEV_PERF_TRACE=%[1]q VEV_PERF_PROCESS_ID=%[2]q XDG_RUNTIME_DIR=%[3]q XDG_STATE_HOME=%[4]q TERM=xterm-256color %[5]q _stdio %[6]q
334	356		;;
335	357		"_udp-bootstrap")
336		-	rm -f %[7]q %[8]q
337		-	env VEV_PERF_TRACE=%[1]q VEV_PERF_PROCESS_ID=%[2]q XDG_RUNTIME_DIR=%[3]q XDG_STATE_HOME=%[4]q TERM=xterm-256color VEV_PERF_UDP_RTT_MS=%[9]q VEV_PERF_UDP_LOSS_PERCENT=%[10]q %[5]q _udp-proxy %[6]q >%[7]q 2>%[8]q &
	358	+	rm -f %[7]q %[8]q %[9]q
	359	+	env VEV_PERF_TRACE=%[1]q VEV_PERF_PROCESS_ID=%[2]q XDG_RUNTIME_DIR=%[3]q XDG_STATE_HOME=%[4]q TERM=xterm-256color %[5]q _udp-proxy %[6]q >%[7]q 2>%[8]q &
338	360		peer=$!
339		-	printf '%%s\\n' "$peer" > %[11]q
	361	+	printf '%%s\n' "$peer" > %[11]q
340	362		i=0
341	363		while [ ! -s %[7]q ] && [ "$i" -lt 1000 ]; do sleep 0.01; i=$((i+1)); done
342	364		test -s %[7]q
343		-	head -n 1 %[7]q
	365	+	line=$(head -n 1 %[7]q)
	366	+	printf '%%s\n' "$line" > %[9]q
	367	+	set -- $line
	368	+	test "$1" = VEV-UDP
	369	+	test -n "$3"
	370	+	printf 'VEV-UDP %%s %%s\n' %[10]d "$3"
344	371		exit 0
345	372		;;
346	373		*) echo 'vev harness ssh seam rejected non-vev command' >&2; exit 64 ;;
347	374		esac
348		-	`, m.TracePath, m.ProcessID, filepath.Join(runDir, "runtime"), filepath.Join(runDir, "state"), l.bin, role.Args[1], ready, stderr, strconv.Itoa(role.Transport.RTTMS), strconv.Itoa(role.Transport.LossPercent), route.pidPath)
	375	+	`, m.TracePath, m.ProcessID, filepath.Join(runDir, "runtime"), filepath.Join(runDir, "state"), l.bin, role.Args[1], ready, stderr, target, netemPort, route.pidPath)
349	376		if err := os.WriteFile(shim, []byte(body), 0o700); err != nil {
	377	+	if route.netem != nil {
	378	+	_ = route.netem.Close()
	379	+	}
350	380		return nil, err
351	381		}
352	382		l.mu.Lock()

@@ -355,6 +385,9 @@ esac
355	385		}
356	386		if _, exists := l.peers[runDir]; exists {
357	387		l.mu.Unlock()
	388	+	if route.netem != nil {
	389	+	_ = route.netem.Close()
	390	+	}
358	391		return nil, fmt.Errorf("duplicate transport peer for %s", runDir)
359	392		}
360	393		l.peers[runDir] = route

 // cliLauncher is deliberately limited to executing the supplied public vev
 // binary. Scenario orchestration never reaches into an in-process daemon.
 type cliLauncher struct {
-	bin   string
+	bin          string
-	mu    sync.Mutex
+	netemFactory udpNetemFactory
-	peers map[string]peerRoute
+	mu           sync.Mutex
+	peers        map[string]peerRoute
 }
 type peerRoute struct {
 	mapping processMapping
 	command roleCommand
 	pidPath string
+	netem   udpNetem
 }
 type preparedPeer struct{ route peerRoute }
 func (p preparedPeer) Warmup([]byte) error                              { return nil }
 func (p preparedPeer) Measure([]byte, func() error, func() error) error { return nil }
 func (p preparedPeer) Close() error {
-	if p.route.pidPath == "" {
+	var errs []error
-		return nil
+	if p.route.pidPath != "" {
-	}
+		b, err := os.ReadFile(p.route.pidPath)
-	b, err := os.ReadFile(p.route.pidPath)
+		if err == nil {
-	if errors.Is(err, os.ErrNotExist) {
+			pid, parseErr := strconv.Atoi(strings.TrimSpace(string(b)))
-		return nil
+			if parseErr != nil || pid <= 0 {
-	}
+				errs = append(errs, fmt.Errorf("invalid owned peer pid: %q", b))
-	if err != nil {
+			} else if killErr := syscall.Kill(pid, syscall.SIGTERM); killErr != nil && !errors.Is(killErr, syscall.ESRCH) {
-		return err
+				errs = append(errs, killErr)
-	}
+			}
-	pid, err := strconv.Atoi(strings.TrimSpace(string(b)))
+		} else if !errors.Is(err, os.ErrNotExist) {
-	if err != nil || pid <= 0 {
+			errs = append(errs, err)
-		return fmt.Errorf("invalid owned peer pid: %q", b)
+		}
 	}
-	if err := syscall.Kill(pid, syscall.SIGTERM); err != nil && !errors.Is(err, syscall.ESRCH) {
+	if p.route.netem != nil {
-		return err
+		if err := p.route.netem.Close(); err != nil {
+			errs = append(errs, err)
+		}
 	}
-	return nil
+	return errors.Join(errs...)
 }
 type terminalOutput interface {
 	route := peerRoute{mapping: m, command: role}
 	if m.Role == "udp_peer" {
 		route.pidPath = filepath.Join(runDir, "udp-peer.pid")
+		factory := l.netemFactory
+		if factory == nil {
+			factory = newUDPNetem
+		}
+		netem, err := factory(udpNetemConfig{
+			RTT:         time.Duration(role.Transport.RTTMS) * time.Millisecond,
+			LossPercent: role.Transport.LossPercent,
+			TargetPath:  filepath.Join(runDir, "udp-peer.target"),
+		})
+		if err != nil {
+			return nil, fmt.Errorf("start harness UDP netem: %w", err)
+		}
+		route.netem = netem
 	}
 	shim := filepath.Join(runDir, "ssh")
 	ready := filepath.Join(runDir, "udp-peer.ready")
+	target := filepath.Join(runDir, "udp-peer.target")
+	netemPort := 0
+	if route.netem != nil {
+		netemPort = route.netem.Port()
+	}
 	stderr := filepath.Join(runDir, "udp-peer.stderr")
 	// The bootstrap protocol requires the SSH command to exit after forwarding
 	// the readiness line. The long-lived proxy is owned by pidPath and receives
     exec env VEV_PERF_TRACE=%[1]q VEV_PERF_PROCESS_ID=%[2]q XDG_RUNTIME_DIR=%[3]q XDG_STATE_HOME=%[4]q TERM=xterm-256color %[5]q _stdio %[6]q
     ;;
   *"_udp-bootstrap"*)
-    rm -f %[7]q %[8]q
+    rm -f %[7]q %[8]q %[9]q
-    env VEV_PERF_TRACE=%[1]q VEV_PERF_PROCESS_ID=%[2]q XDG_RUNTIME_DIR=%[3]q XDG_STATE_HOME=%[4]q TERM=xterm-256color VEV_PERF_UDP_RTT_MS=%[9]q VEV_PERF_UDP_LOSS_PERCENT=%[10]q %[5]q _udp-proxy %[6]q >%[7]q 2>%[8]q &
+    env VEV_PERF_TRACE=%[1]q VEV_PERF_PROCESS_ID=%[2]q XDG_RUNTIME_DIR=%[3]q XDG_STATE_HOME=%[4]q TERM=xterm-256color %[5]q _udp-proxy %[6]q >%[7]q 2>%[8]q &
     peer=$!
-    printf '%%s\\n' "$peer" > %[11]q
+    printf '%%s\n' "$peer" > %[11]q
     i=0
     while [ ! -s %[7]q ] && [ "$i" -lt 1000 ]; do sleep 0.01; i=$((i+1)); done
     test -s %[7]q
-    head -n 1 %[7]q
+    line=$(head -n 1 %[7]q)
+    printf '%%s\n' "$line" > %[9]q
+    set -- $line
+    test "$1" = VEV-UDP
+    test -n "$3"
+    printf 'VEV-UDP %%s %%s\n' %[10]d "$3"
     exit 0
     ;;
   *) echo 'vev harness ssh seam rejected non-vev command' >&2; exit 64 ;;
 esac
-`, m.TracePath, m.ProcessID, filepath.Join(runDir, "runtime"), filepath.Join(runDir, "state"), l.bin, role.Args[1], ready, stderr, strconv.Itoa(role.Transport.RTTMS), strconv.Itoa(role.Transport.LossPercent), route.pidPath)
+`, m.TracePath, m.ProcessID, filepath.Join(runDir, "runtime"), filepath.Join(runDir, "state"), l.bin, role.Args[1], ready, stderr, target, netemPort, route.pidPath)
 	if err := os.WriteFile(shim, []byte(body), 0o700); err != nil {
+		if route.netem != nil {
+			_ = route.netem.Close()
+		}
 		return nil, err
 	}
 	l.mu.Lock()
 	}
 	if _, exists := l.peers[runDir]; exists {
 		l.mu.Unlock()
+		if route.netem != nil {
+			_ = route.netem.Close()
+		}
 		return nil, fmt.Errorf("duplicate transport peer for %s", runDir)
 	}
 	l.peers[runDir] = route

M cmd/vev-perf-harness/main_test.go +132 -29


@@ -4,8 +4,10 @@ import (
4	4		"bytes"
5	5		"encoding/json"
6	6		"errors"
	7	+	"net"
7	8		"os"
8	9		"path/filepath"
	10	+	"strconv"
9	11		"strings"
10	12		"testing"
11	13		"time"

@@ -16,6 +18,14 @@ type fakeClock struct{ tick int64 }
16	18		func (c *fakeClock) Now() int64 { c.tick += 10; return c.tick }
17	19		func (*fakeClock) Sleep(time.Duration) {}
18	20
	21	+	type fakeUDPNetem struct {
	22	+	port int
	23	+	closed bool
	24	+	}
	25	+
	26	+	func (n *fakeUDPNetem) Port() int { return n.port }
	27	+	func (n *fakeUDPNetem) Close() error { n.closed = true; return nil }
	28	+
19	29		type fakeProcess struct {
20	30		fail bool
21	31		measureErr error

@@ -395,38 +405,131 @@ func TestHarnessFakeRunnerRoutesClientToPeerAndCleansEveryRole(t *testing.T) {
395	405		}
396	406
397	407		func TestCLITransportSeamOwnsExclusivePeerTraceAndCleanup(t *testing.T) {
398		-	dir := t.TempDir()
399		-	if err := os.Chmod(dir, 0o700); err != nil {
400		-	t.Fatal(err)
401		-	}
402		-	l := &cliLauncher{bin: "/bin/true"}
403		-	m := processMapping{ProcessID: "udp-peer", TracePath: filepath.Join(dir, "udp.jsonl"), Role: "udp_peer"}
404		-	if err := os.WriteFile(m.TracePath, nil, 0o600); err != nil {
405		-	t.Fatal(err)
406		-	}
407		-	p, err := l.preparePeer(m, roleCommand{Args: []string{"_udp-proxy", "work"}, Transport: transport{ID: "udp_25ms", Kind: "udp", RTTMS: 25}})
408		-	if err != nil {
409		-	t.Fatal(err)
410		-	}
411		-	shim, err := os.ReadFile(filepath.Join(dir, "ssh"))
412		-	if err != nil {
413		-	t.Fatal(err)
	408	+	fixtures := []struct {
	409	+	name string
	410	+	tr transport
	411	+	rtt time.Duration
	412	+	loss int
	413	+	}{
	414	+	{"baseline", transport{ID: "udp_baseline", Kind: "udp"}, 0, 0},
	415	+	{"25ms", transport{ID: "udp_25ms", Kind: "udp", RTTMS: 25}, 25 * time.Millisecond, 0},
	416	+	{"100ms", transport{ID: "udp_100ms", Kind: "udp", RTTMS: 100}, 100 * time.Millisecond, 0},
	417	+	{"loss0", transport{ID: "udp_loss_0pct", Kind: "udp"}, 0, 0},
	418	+	{"loss1", transport{ID: "udp_loss_1pct", Kind: "udp", LossPercent: 1}, 0, 1},
414	419		}
415		-	text := string(shim)
416		-	for _, want := range []string{"_udp-proxy", m.TracePath, m.ProcessID, "VEV_PERF_UDP_RTT_MS=\"25\"", "udp-peer.pid"} {
417		-	if !strings.Contains(text, want) {
418		-	t.Errorf("seam does not retain %q:\n%s", want, text)
419		-	}
	420	+	for _, tc := range fixtures {
	421	+	t.Run(tc.name, func(t *testing.T) {
	422	+	dir := t.TempDir()
	423	+	if err := os.Chmod(dir, 0o700); err != nil {
	424	+	t.Fatal(err)
	425	+	}
	426	+	var configs []udpNetemConfig
	427	+	netem := &fakeUDPNetem{port: 45678}
	428	+	l := &cliLauncher{bin: "/bin/true", netemFactory: func(c udpNetemConfig) (udpNetem, error) {
	429	+	configs = append(configs, c)
	430	+	return netem, nil
	431	+	}}
	432	+	m := processMapping{ProcessID: "udp-peer", TracePath: filepath.Join(dir, "udp.jsonl"), Role: "udp_peer"}
	433	+	if err := os.WriteFile(m.TracePath, nil, 0o600); err != nil {
	434	+	t.Fatal(err)
	435	+	}
	436	+	p, err := l.preparePeer(m, roleCommand{Args: []string{"_udp-proxy", "work"}, Transport: tc.tr})
	437	+	if err != nil {
	438	+	t.Fatal(err)
	439	+	}
	440	+	if len(configs) != 1 \|\| configs[0].RTT != tc.rtt \|\| configs[0].LossPercent != tc.loss \|\| configs[0].TargetPath != filepath.Join(dir, "udp-peer.target") {
	441	+	t.Fatalf("fixture did not reach netem seam: %+v", configs)
	442	+	}
	443	+	shim, err := os.ReadFile(filepath.Join(dir, "ssh"))
	444	+	if err != nil {
	445	+	t.Fatal(err)
	446	+	}
	447	+	text := string(shim)
	448	+	for _, want := range []string{"_udp-proxy", m.TracePath, m.ProcessID, "udp-peer.target", "VEV-UDP %s %s\\n' 45678", "udp-peer.pid"} {
	449	+	if !strings.Contains(text, want) {
	450	+	t.Errorf("seam does not retain %q:\n%s", want, text)
	451	+	}
	452	+	}
	453	+	if strings.Contains(text, "VEV_PERF_UDP_") {
	454	+	t.Fatalf("ignored vev UDP environment was used instead of netem: %s", text)
	455	+	}
	456	+	// A nonexistent pid is already-cleaned-up; Close must still close the
	457	+	// harness-owned emulator.
	458	+	if err := p.Close(); err != nil {
	459	+	t.Fatal(err)
	460	+	}
	461	+	if !netem.closed {
	462	+	t.Fatal("harness netem was not cleaned up")
	463	+	}
	464	+	})
420	465		}
421		-	// A nonexistent pid is already-cleaned-up; Close must treat it as such.
422		-	if err := p.Close(); err != nil {
423		-	t.Fatal(err)
	466	+	}
	467	+
	468	+	func TestUDPNetemExecutesRTTAndLossFixtures(t *testing.T) {
	469	+	fixtures := []struct {
	470	+	name string
	471	+	rtt time.Duration
	472	+	loss int
	473	+	sent int
	474	+	want int
	475	+	}{
	476	+	{"25ms RTT", 25 * time.Millisecond, 0, 1, 1},
	477	+	{"100ms RTT", 100 * time.Millisecond, 0, 1, 1},
	478	+	{"zero loss", 0, 0, 10, 10},
	479	+	{"one percent loss", 0, 1, 100, 99},
424	480		}
425		-	l.mu.Lock()
426		-	route := l.peers[dir]
427		-	l.mu.Unlock()
428		-	if route.mapping.TracePath != m.TracePath \|\| route.mapping.ProcessID != m.ProcessID \|\| route.command.Transport.RTTMS != 25 {
429		-	t.Fatalf("peer route is not exclusively correlated: %+v", route)
	481	+	for _, tc := range fixtures {
	482	+	t.Run(tc.name, func(t *testing.T) {
	483	+	target, err := net.ListenPacket("udp", "127.0.0.1:0")
	484	+	if err != nil {
	485	+	t.Fatal(err)
	486	+	}
	487	+	defer target.Close()
	488	+	path := filepath.Join(t.TempDir(), "target")
	489	+	if err := os.WriteFile(path, []byte("VEV-UDP "+strconv.Itoa(target.LocalAddr().(*net.UDPAddr).Port)+" key\\n"), 0o600); err != nil {
	490	+	t.Fatal(err)
	491	+	}
	492	+	netem, err := newUDPNetem(udpNetemConfig{RTT: tc.rtt, LossPercent: tc.loss, TargetPath: path})
	493	+	if err != nil {
	494	+	t.Fatal(err)
	495	+	}
	496	+	defer netem.Close()
	497	+	client, err := net.ListenPacket("udp", "127.0.0.1:0")
	498	+	if err != nil {
	499	+	t.Fatal(err)
	500	+	}
	501	+	defer client.Close()
	502	+	started := time.Now()
	503	+	for i := 0; i < tc.sent; i++ {
	504	+	if _, err := client.WriteTo([]byte{byte(i)}, &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: netem.Port()}); err != nil {
	505	+	t.Fatal(err)
	506	+	}
	507	+	}
	508	+	if err := target.SetReadDeadline(time.Now().Add(2 * time.Second)); err != nil {
	509	+	t.Fatal(err)
	510	+	}
	511	+	got := 0
	512	+	buf := make([]byte, 32)
	513	+	for got < tc.want {
	514	+	if _, _, err := target.ReadFrom(buf); err != nil {
	515	+	t.Fatalf("received %d packets, want %d: %v", got, tc.want, err)
	516	+	}
	517	+	got++
	518	+	}
	519	+	if tc.rtt > 0 {
	520	+	if elapsed := time.Since(started); elapsed < tc.rtt/2 {
	521	+	t.Fatalf("one-way netem delay=%s, want at least %s", elapsed, tc.rtt/2)
	522	+	}
	523	+	}
	524	+	// Keep the deadline short: extra carriage would prove the exact 1%%
	525	+	// loss cycle was not enforced.
	526	+	if err := target.SetReadDeadline(time.Now().Add(20 * time.Millisecond)); err != nil {
	527	+	t.Fatal(err)
	528	+	}
	529	+	if _, _, err := target.ReadFrom(buf); tc.want != tc.sent && err == nil {
	530	+	t.Fatalf("received more than %d packets with %d%% loss", tc.want, tc.loss)
	531	+	}
	532	+	})
430	533		}
431	534		}
432	535


@@ -4,8 +4,10 @@ import (
4		"bytes"	4		"bytes"
5		"encoding/json"	5		"encoding/json"
6		"errors"	6		"errors"
			7	+	"net"
7		"os"	8		"os"
8		"path/filepath"	9		"path/filepath"
			10	+	"strconv"
9		"strings"	11		"strings"
10		"testing"	12		"testing"
11		"time"	13		"time"

@@ -16,6 +18,14 @@ type fakeClock struct{ tick int64 }
16		func (c *fakeClock) Now() int64 { c.tick += 10; return c.tick }	18		func (c *fakeClock) Now() int64 { c.tick += 10; return c.tick }
17		func (*fakeClock) Sleep(time.Duration) {}	19		func (*fakeClock) Sleep(time.Duration) {}
18			20
			21	+	type fakeUDPNetem struct {
			22	+	port int
			23	+	closed bool
			24	+	}
			25	+
			26	+	func (n *fakeUDPNetem) Port() int { return n.port }
			27	+	func (n *fakeUDPNetem) Close() error { n.closed = true; return nil }
			28	+
19		type fakeProcess struct {	29		type fakeProcess struct {
20		fail bool	30		fail bool
21		measureErr error	31		measureErr error

@@ -395,38 +405,131 @@ func TestHarnessFakeRunnerRoutesClientToPeerAndCleansEveryRole(t *testing.T) {
395		}	405		}
396			406
397		func TestCLITransportSeamOwnsExclusivePeerTraceAndCleanup(t *testing.T) {	407		func TestCLITransportSeamOwnsExclusivePeerTraceAndCleanup(t *testing.T) {
398	-	dir := t.TempDir()	408	+	fixtures := []struct {
399	-	if err := os.Chmod(dir, 0o700); err != nil {	409	+	name string
400	-	t.Fatal(err)	410	+	tr transport
401	-	}	411	+	rtt time.Duration
402	-	l := &cliLauncher{bin: "/bin/true"}	412	+	loss int
403	-	m := processMapping{ProcessID: "udp-peer", TracePath: filepath.Join(dir, "udp.jsonl"), Role: "udp_peer"}	413	+	}{
404	-	if err := os.WriteFile(m.TracePath, nil, 0o600); err != nil {	414	+	{"baseline", transport{ID: "udp_baseline", Kind: "udp"}, 0, 0},
405	-	t.Fatal(err)	415	+	{"25ms", transport{ID: "udp_25ms", Kind: "udp", RTTMS: 25}, 25 * time.Millisecond, 0},
406	-	}	416	+	{"100ms", transport{ID: "udp_100ms", Kind: "udp", RTTMS: 100}, 100 * time.Millisecond, 0},
407	-	p, err := l.preparePeer(m, roleCommand{Args: []string{"_udp-proxy", "work"}, Transport: transport{ID: "udp_25ms", Kind: "udp", RTTMS: 25}})	417	+	{"loss0", transport{ID: "udp_loss_0pct", Kind: "udp"}, 0, 0},
408	-	if err != nil {	418	+	{"loss1", transport{ID: "udp_loss_1pct", Kind: "udp", LossPercent: 1}, 0, 1},
409	-	t.Fatal(err)
410	-	}
411	-	shim, err := os.ReadFile(filepath.Join(dir, "ssh"))
412	-	if err != nil {
413	-	t.Fatal(err)
414		}	419		}
415	-	text := string(shim)	420	+	for _, tc := range fixtures {
416	-	for _, want := range []string{"_udp-proxy", m.TracePath, m.ProcessID, "VEV_PERF_UDP_RTT_MS=\"25\"", "udp-peer.pid"} {	421	+	t.Run(tc.name, func(t *testing.T) {
417	-	if !strings.Contains(text, want) {	422	+	dir := t.TempDir()
418	-	t.Errorf("seam does not retain %q:\n%s", want, text)	423	+	if err := os.Chmod(dir, 0o700); err != nil {
419	-	}	424	+	t.Fatal(err)
			425	+	}
			426	+	var configs []udpNetemConfig
			427	+	netem := &fakeUDPNetem{port: 45678}
			428	+	l := &cliLauncher{bin: "/bin/true", netemFactory: func(c udpNetemConfig) (udpNetem, error) {
			429	+	configs = append(configs, c)
			430	+	return netem, nil
			431	+	}}
			432	+	m := processMapping{ProcessID: "udp-peer", TracePath: filepath.Join(dir, "udp.jsonl"), Role: "udp_peer"}
			433	+	if err := os.WriteFile(m.TracePath, nil, 0o600); err != nil {
			434	+	t.Fatal(err)
			435	+	}
			436	+	p, err := l.preparePeer(m, roleCommand{Args: []string{"_udp-proxy", "work"}, Transport: tc.tr})
			437	+	if err != nil {
			438	+	t.Fatal(err)
			439	+	}
			440	+	if len(configs) != 1 \|\| configs[0].RTT != tc.rtt \|\| configs[0].LossPercent != tc.loss \|\| configs[0].TargetPath != filepath.Join(dir, "udp-peer.target") {
			441	+	t.Fatalf("fixture did not reach netem seam: %+v", configs)
			442	+	}
			443	+	shim, err := os.ReadFile(filepath.Join(dir, "ssh"))
			444	+	if err != nil {
			445	+	t.Fatal(err)
			446	+	}
			447	+	text := string(shim)
			448	+	for _, want := range []string{"_udp-proxy", m.TracePath, m.ProcessID, "udp-peer.target", "VEV-UDP %s %s\\n' 45678", "udp-peer.pid"} {
			449	+	if !strings.Contains(text, want) {
			450	+	t.Errorf("seam does not retain %q:\n%s", want, text)
			451	+	}
			452	+	}
			453	+	if strings.Contains(text, "VEV_PERF_UDP_") {
			454	+	t.Fatalf("ignored vev UDP environment was used instead of netem: %s", text)
			455	+	}
			456	+	// A nonexistent pid is already-cleaned-up; Close must still close the
			457	+	// harness-owned emulator.
			458	+	if err := p.Close(); err != nil {
			459	+	t.Fatal(err)
			460	+	}
			461	+	if !netem.closed {
			462	+	t.Fatal("harness netem was not cleaned up")
			463	+	}
			464	+	})
420		}	465		}
421	-	// A nonexistent pid is already-cleaned-up; Close must treat it as such.	466	+	}
422	-	if err := p.Close(); err != nil {	467	+
423	-	t.Fatal(err)	468	+	func TestUDPNetemExecutesRTTAndLossFixtures(t *testing.T) {
			469	+	fixtures := []struct {
			470	+	name string
			471	+	rtt time.Duration
			472	+	loss int
			473	+	sent int
			474	+	want int
			475	+	}{
			476	+	{"25ms RTT", 25 * time.Millisecond, 0, 1, 1},
			477	+	{"100ms RTT", 100 * time.Millisecond, 0, 1, 1},
			478	+	{"zero loss", 0, 0, 10, 10},
			479	+	{"one percent loss", 0, 1, 100, 99},
424		}	480		}
425	-	l.mu.Lock()	481	+	for _, tc := range fixtures {
426	-	route := l.peers[dir]	482	+	t.Run(tc.name, func(t *testing.T) {
427	-	l.mu.Unlock()	483	+	target, err := net.ListenPacket("udp", "127.0.0.1:0")
428	-	if route.mapping.TracePath != m.TracePath \|\| route.mapping.ProcessID != m.ProcessID \|\| route.command.Transport.RTTMS != 25 {	484	+	if err != nil {
429	-	t.Fatalf("peer route is not exclusively correlated: %+v", route)	485	+	t.Fatal(err)
			486	+	}
			487	+	defer target.Close()
			488	+	path := filepath.Join(t.TempDir(), "target")
			489	+	if err := os.WriteFile(path, []byte("VEV-UDP "+strconv.Itoa(target.LocalAddr().(*net.UDPAddr).Port)+" key\\n"), 0o600); err != nil {
			490	+	t.Fatal(err)
			491	+	}
			492	+	netem, err := newUDPNetem(udpNetemConfig{RTT: tc.rtt, LossPercent: tc.loss, TargetPath: path})
			493	+	if err != nil {
			494	+	t.Fatal(err)
			495	+	}
			496	+	defer netem.Close()
			497	+	client, err := net.ListenPacket("udp", "127.0.0.1:0")
			498	+	if err != nil {
			499	+	t.Fatal(err)
			500	+	}
			501	+	defer client.Close()
			502	+	started := time.Now()
			503	+	for i := 0; i < tc.sent; i++ {
			504	+	if _, err := client.WriteTo([]byte{byte(i)}, &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: netem.Port()}); err != nil {
			505	+	t.Fatal(err)
			506	+	}
			507	+	}
			508	+	if err := target.SetReadDeadline(time.Now().Add(2 * time.Second)); err != nil {
			509	+	t.Fatal(err)
			510	+	}
			511	+	got := 0
			512	+	buf := make([]byte, 32)
			513	+	for got < tc.want {
			514	+	if _, _, err := target.ReadFrom(buf); err != nil {
			515	+	t.Fatalf("received %d packets, want %d: %v", got, tc.want, err)
			516	+	}
			517	+	got++
			518	+	}
			519	+	if tc.rtt > 0 {
			520	+	if elapsed := time.Since(started); elapsed < tc.rtt/2 {
			521	+	t.Fatalf("one-way netem delay=%s, want at least %s", elapsed, tc.rtt/2)
			522	+	}
			523	+	}
			524	+	// Keep the deadline short: extra carriage would prove the exact 1%%
			525	+	// loss cycle was not enforced.
			526	+	if err := target.SetReadDeadline(time.Now().Add(20 * time.Millisecond)); err != nil {
			527	+	t.Fatal(err)
			528	+	}
			529	+	if _, _, err := target.ReadFrom(buf); tc.want != tc.sent && err == nil {
			530	+	t.Fatalf("received more than %d packets with %d%% loss", tc.want, tc.loss)
			531	+	}
			532	+	})
430		}	533		}
431		}	534		}
432			535

A cmd/vev-perf-harness/udp_netem.go +157 -0

@@ -0,0 +1,157 @@
	1	+	package main
	2	+
	3	+	import (
	4	+	"errors"
	5	+	"fmt"
	6	+	"net"
	7	+	"os"
	8	+	"strconv"
	9	+	"strings"
	10	+	"sync"
	11	+	"time"
	12	+	)
	13	+
	14	+	// udpNetem is a harness-owned, user-space UDP relay. It sits between the
	15	+	// public dgram client and vev's public _udp-proxy readiness address, so fixture
	16	+	// impairment is real network carriage rather than an ignored vev option.
	17	+	type udpNetem interface {
	18	+	Port() int
	19	+	Close() error
	20	+	}
	21	+
	22	+	type udpNetemFactory func(udpNetemConfig) (udpNetem, error)
	23	+
	24	+	type udpNetemConfig struct {
	25	+	// RTT is split equally across both relay directions.
	26	+	RTT time.Duration
	27	+	LossPercent int
	28	+	TargetPath string
	29	+	}
	30	+
	31	+	type udpNetemRelay struct {
	32	+	config udpNetemConfig
	33	+	conn net.PacketConn
	34	+	done chan struct{}
	35	+	wg sync.WaitGroup
	36	+	once sync.Once
	37	+
	38	+	mu sync.Mutex
	39	+	client, dst net.Addr
	40	+	packet uint64
	41	+	}
	42	+
	43	+	func newUDPNetem(config udpNetemConfig) (udpNetem, error) {
	44	+	if config.RTT < 0 \|\| config.LossPercent < 0 \|\| config.LossPercent > 100 \|\| config.TargetPath == "" {
	45	+	return nil, fmt.Errorf("invalid UDP netem fixture: rtt=%s loss=%d target=%q", config.RTT, config.LossPercent, config.TargetPath)
	46	+	}
	47	+	conn, err := net.ListenPacket("udp", "127.0.0.1:0")
	48	+	if err != nil {
	49	+	return nil, err
	50	+	}
	51	+	r := &udpNetemRelay{config: config, conn: conn, done: make(chan struct{})}
	52	+	r.wg.Add(1)
	53	+	go r.run()
	54	+	return r, nil
	55	+	}
	56	+
	57	+	func (r *udpNetemRelay) Port() int {
	58	+	return r.conn.LocalAddr().(*net.UDPAddr).Port
	59	+	}
	60	+
	61	+	func (r *udpNetemRelay) Close() error {
	62	+	var closeErr error
	63	+	r.once.Do(func() {
	64	+	close(r.done)
	65	+	closeErr = r.conn.Close()
	66	+	r.wg.Wait()
	67	+	})
	68	+	return closeErr
	69	+	}
	70	+
	71	+	func (r *udpNetemRelay) run() {
	72	+	defer r.wg.Done()
	73	+	buf := make([]byte, 64*1024)
	74	+	for {
	75	+	n, from, err := r.conn.ReadFrom(buf)
	76	+	if err != nil {
	77	+	return
	78	+	}
	79	+	to, ok := r.destination(from)
	80	+	if !ok \|\| r.drop() {
	81	+	continue
	82	+	}
	83	+	packet := append([]byte(nil), buf[:n]...)
	84	+	r.schedule(packet, to)
	85	+	}
	86	+	}
	87	+
	88	+	func (r *udpNetemRelay) destination(from net.Addr) (net.Addr, bool) {
	89	+	r.mu.Lock()
	90	+	defer r.mu.Unlock()
	91	+	if r.dst == nil {
	92	+	dst, err := udpNetemTarget(r.config.TargetPath)
	93	+	if err != nil {
	94	+	return nil, false
	95	+	}
	96	+	r.dst = dst
	97	+	}
	98	+	if sameUDPAddress(from, r.dst) {
	99	+	return r.client, r.client != nil
	100	+	}
	101	+	r.client = from
	102	+	return r.dst, true
	103	+	}
	104	+
	105	+	func (r *udpNetemRelay) drop() bool {
	106	+	r.mu.Lock()
	107	+	defer r.mu.Unlock()
	108	+	r.packet++
	109	+	// The deterministic cycle makes the configured percentage auditable: a 1%
	110	+	// fixture drops exactly one carriage packet in every 100 attempts.
	111	+	return r.config.LossPercent > 0 && int((r.packet-1)%100) < r.config.LossPercent
	112	+	}
	113	+
	114	+	func (r *udpNetemRelay) schedule(packet []byte, to net.Addr) {
	115	+	delay := r.config.RTT / 2
	116	+	r.wg.Add(1)
	117	+	go func() {
	118	+	defer r.wg.Done()
	119	+	if delay > 0 {
	120	+	timer := time.NewTimer(delay)
	121	+	defer timer.Stop()
	122	+	select {
	123	+	case <-timer.C:
	124	+	case <-r.done:
	125	+	return
	126	+	}
	127	+	}
	128	+	select {
	129	+	case <-r.done:
	130	+	return
	131	+	default:
	132	+	_, _ = r.conn.WriteTo(packet, to)
	133	+	}
	134	+	}()
	135	+	}
	136	+
	137	+	func udpNetemTarget(path string) (net.Addr, error) {
	138	+	line, err := os.ReadFile(path)
	139	+	if err != nil {
	140	+	return nil, err
	141	+	}
	142	+	fields := strings.Fields(string(line))
	143	+	if len(fields) != 3 \|\| fields[0] != "VEV-UDP" {
	144	+	return nil, errors.New("invalid UDP proxy readiness")
	145	+	}
	146	+	port, err := strconv.Atoi(fields[1])
	147	+	if err != nil \|\| port < 1 \|\| port > 65535 \|\| fields[2] == "" {
	148	+	return nil, errors.New("invalid UDP proxy readiness")
	149	+	}
	150	+	return &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: port}, nil
	151	+	}
	152	+
	153	+	func sameUDPAddress(a, b net.Addr) bool {
	154	+	ua, aOK := a.(*net.UDPAddr)
	155	+	ub, bOK := b.(*net.UDPAddr)
	156	+	return aOK && bOK && ua.Port == ub.Port && ua.IP.Equal(ub.IP)
	157	+	}

@@ -0,0 +1,157 @@
	1	+	package main
	2	+
	3	+	import (
	4	+	"errors"
	5	+	"fmt"
	6	+	"net"
	7	+	"os"
	8	+	"strconv"
	9	+	"strings"
	10	+	"sync"
	11	+	"time"
	12	+	)
	13	+
	14	+	// udpNetem is a harness-owned, user-space UDP relay. It sits between the
	15	+	// public dgram client and vev's public _udp-proxy readiness address, so fixture
	16	+	// impairment is real network carriage rather than an ignored vev option.
	17	+	type udpNetem interface {
	18	+	Port() int
	19	+	Close() error
	20	+	}
	21	+
	22	+	type udpNetemFactory func(udpNetemConfig) (udpNetem, error)
	23	+
	24	+	type udpNetemConfig struct {
	25	+	// RTT is split equally across both relay directions.
	26	+	RTT time.Duration
	27	+	LossPercent int
	28	+	TargetPath string
	29	+	}
	30	+
	31	+	type udpNetemRelay struct {
	32	+	config udpNetemConfig
	33	+	conn net.PacketConn
	34	+	done chan struct{}
	35	+	wg sync.WaitGroup
	36	+	once sync.Once
	37	+
	38	+	mu sync.Mutex
	39	+	client, dst net.Addr
	40	+	packet uint64
	41	+	}
	42	+
	43	+	func newUDPNetem(config udpNetemConfig) (udpNetem, error) {
	44	+	if config.RTT < 0 \|\| config.LossPercent < 0 \|\| config.LossPercent > 100 \|\| config.TargetPath == "" {
	45	+	return nil, fmt.Errorf("invalid UDP netem fixture: rtt=%s loss=%d target=%q", config.RTT, config.LossPercent, config.TargetPath)
	46	+	}
	47	+	conn, err := net.ListenPacket("udp", "127.0.0.1:0")
	48	+	if err != nil {
	49	+	return nil, err
	50	+	}
	51	+	r := &udpNetemRelay{config: config, conn: conn, done: make(chan struct{})}
	52	+	r.wg.Add(1)
	53	+	go r.run()
	54	+	return r, nil
	55	+	}
	56	+
	57	+	func (r *udpNetemRelay) Port() int {
	58	+	return r.conn.LocalAddr().(*net.UDPAddr).Port
	59	+	}
	60	+
	61	+	func (r *udpNetemRelay) Close() error {
	62	+	var closeErr error
	63	+	r.once.Do(func() {
	64	+	close(r.done)
	65	+	closeErr = r.conn.Close()
	66	+	r.wg.Wait()
	67	+	})
	68	+	return closeErr
	69	+	}
	70	+
	71	+	func (r *udpNetemRelay) run() {
	72	+	defer r.wg.Done()
	73	+	buf := make([]byte, 64*1024)
	74	+	for {
	75	+	n, from, err := r.conn.ReadFrom(buf)
	76	+	if err != nil {
	77	+	return
	78	+	}
	79	+	to, ok := r.destination(from)
	80	+	if !ok \|\| r.drop() {
	81	+	continue
	82	+	}
	83	+	packet := append([]byte(nil), buf[:n]...)
	84	+	r.schedule(packet, to)
	85	+	}
	86	+	}
	87	+
	88	+	func (r *udpNetemRelay) destination(from net.Addr) (net.Addr, bool) {
	89	+	r.mu.Lock()
	90	+	defer r.mu.Unlock()
	91	+	if r.dst == nil {
	92	+	dst, err := udpNetemTarget(r.config.TargetPath)
	93	+	if err != nil {
	94	+	return nil, false
	95	+	}
	96	+	r.dst = dst
	97	+	}
	98	+	if sameUDPAddress(from, r.dst) {
	99	+	return r.client, r.client != nil
	100	+	}
	101	+	r.client = from
	102	+	return r.dst, true
	103	+	}
	104	+
	105	+	func (r *udpNetemRelay) drop() bool {
	106	+	r.mu.Lock()
	107	+	defer r.mu.Unlock()
	108	+	r.packet++
	109	+	// The deterministic cycle makes the configured percentage auditable: a 1%
	110	+	// fixture drops exactly one carriage packet in every 100 attempts.
	111	+	return r.config.LossPercent > 0 && int((r.packet-1)%100) < r.config.LossPercent
	112	+	}
	113	+
	114	+	func (r *udpNetemRelay) schedule(packet []byte, to net.Addr) {
	115	+	delay := r.config.RTT / 2
	116	+	r.wg.Add(1)
	117	+	go func() {
	118	+	defer r.wg.Done()
	119	+	if delay > 0 {
	120	+	timer := time.NewTimer(delay)
	121	+	defer timer.Stop()
	122	+	select {
	123	+	case <-timer.C:
	124	+	case <-r.done:
	125	+	return
	126	+	}
	127	+	}
	128	+	select {
	129	+	case <-r.done:
	130	+	return
	131	+	default:
	132	+	_, _ = r.conn.WriteTo(packet, to)
	133	+	}
	134	+	}()
	135	+	}
	136	+
	137	+	func udpNetemTarget(path string) (net.Addr, error) {
	138	+	line, err := os.ReadFile(path)
	139	+	if err != nil {
	140	+	return nil, err
	141	+	}
	142	+	fields := strings.Fields(string(line))
	143	+	if len(fields) != 3 \|\| fields[0] != "VEV-UDP" {
	144	+	return nil, errors.New("invalid UDP proxy readiness")
	145	+	}
	146	+	port, err := strconv.Atoi(fields[1])
	147	+	if err != nil \|\| port < 1 \|\| port > 65535 \|\| fields[2] == "" {
	148	+	return nil, errors.New("invalid UDP proxy readiness")
	149	+	}
	150	+	return &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: port}, nil
	151	+	}
	152	+
	153	+	func sameUDPAddress(a, b net.Addr) bool {
	154	+	ua, aOK := a.(*net.UDPAddr)
	155	+	ub, bOK := b.(*net.UDPAddr)
	156	+	return aOK && bOK && ua.Port == ub.Port && ua.IP.Equal(ub.IP)
	157	+	}