7d099f2 fix(obs): bound netem scheduling and evidence writes · vev

M cmd/vev-perf-harness/evidence.go +30 -23


@@ -167,28 +167,25 @@ func mergeProcessTraces(mappings []processMapping) ([]span, error) {
167	167		for scan.Scan() {
168	168		var r traceRecord
169	169		if err := json.Unmarshal(scan.Bytes(), &r); err != nil {
170		-	_ = f.Close()
171		-	return nil, err
	170	+	return nil, closeFile(f, err)
172	171		}
173	172		if r.ProcessID != m.ProcessID \|\| r.Scenario != m.Scenario \|\| r.Run != uint64(m.Run) {
174		-	_ = f.Close()
175		-	return nil, errors.New("trace record identity does not match manifest")
	173	+	return nil, closeFile(f, errors.New("trace record identity does not match manifest"))
176	174		}
177	175		if r.Scenario == "" \|\| r.Run == 0 \|\| r.Component == "" \|\| r.Sequence == 0 \|\| r.RequestID == 0 \|\| r.Epoch == 0 {
178		-	_ = f.Close()
179		-	return nil, errors.New("trace record has invalid correlation fields")
	176	+	return nil, closeFile(f, errors.New("trace record has invalid correlation fields"))
180	177		}
181	178		if _, ok := known[r.ProcessID]; !ok {
182		-	_ = f.Close()
183		-	return nil, errors.New("unknown trace process_id")
	179	+	return nil, closeFile(f, errors.New("unknown trace process_id"))
184	180		}
185	181		records = append(records, r)
186	182		}
187	183		if err := scan.Err(); err != nil {
188		-	_ = f.Close()
	184	+	return nil, closeFile(f, err)
	185	+	}
	186	+	if err := f.Close(); err != nil {
189	187		return nil, err
190	188		}
191		-	_ = f.Close()
192	189		// A process shares one observer across concurrent components. Correlation
193	190		// IDs are allocated before a goroutine reaches its first mark, so a later
194	191		// ID can be serialized before an earlier one. Their first appearance is

@@ -271,14 +268,19 @@ func percentiles(samples []int64) distribution {
271	268		return distribution{len(v), at(50), at(95), at(99), v[len(v)-1]}
272	269		}
273	270
	271	+	func closeFile(f *os.File, err error) error {
	272	+	return errors.Join(err, f.Close())
	273	+	}
	274	+
274	275		func writeJSON(path string, v any) error {
275		-	f, e := os.OpenFile(path, os.O_WRONLY\|os.O_CREATE\|os.O_EXCL, 0o600)
276		-	if e != nil {
277		-	return e
	276	+	f, err := os.OpenFile(path, os.O_WRONLY\|os.O_CREATE\|os.O_EXCL, 0o600)
	277	+	if err != nil {
	278	+	return err
278	279		}
279		-	defer f.Close()
280		-	e = json.NewEncoder(f).Encode(v)
281		-	return e
	280	+	if err := json.NewEncoder(f).Encode(v); err != nil {
	281	+	return closeFile(f, err)
	282	+	}
	283	+	return f.Close()
282	284		}
283	285
284	286		func mustJSON(v any) string { b, _ := json.Marshal(v); return string(b) }

@@ -293,19 +295,24 @@ func safeName(s string) string {
293	295		}
294	296
295	297		func readJSONL(path string) ([]map[string]json.RawMessage, error) {
296		-	f, e := os.Open(path)
297		-	if e != nil {
298		-	return nil, e
	298	+	f, err := os.Open(path)
	299	+	if err != nil {
	300	+	return nil, err
299	301		}
300		-	defer f.Close()
301	302		var r []map[string]json.RawMessage
302	303		scan := bufio.NewScanner(f)
303	304		for scan.Scan() {
304	305		var v map[string]json.RawMessage
305		-	if e := json.Unmarshal(scan.Bytes(), &v); e != nil {
306		-	return nil, e
	306	+	if err := json.Unmarshal(scan.Bytes(), &v); err != nil {
	307	+	return nil, closeFile(f, err)
307	308		}
308	309		r = append(r, v)
309	310		}
310		-	return r, scan.Err()
	311	+	if err := scan.Err(); err != nil {
	312	+	return nil, closeFile(f, err)
	313	+	}
	314	+	if err := f.Close(); err != nil {
	315	+	return nil, err
	316	+	}
	317	+	return r, nil
311	318		}

 		for scan.Scan() {
 			var r traceRecord
 			if err := json.Unmarshal(scan.Bytes(), &r); err != nil {
-				_ = f.Close()
+				return nil, closeFile(f, err)
-				return nil, err
 			}
 			if r.ProcessID != m.ProcessID || r.Scenario != m.Scenario || r.Run != uint64(m.Run) {
-				_ = f.Close()
+				return nil, closeFile(f, errors.New("trace record identity does not match manifest"))
-				return nil, errors.New("trace record identity does not match manifest")
 			}
 			if r.Scenario == "" || r.Run == 0 || r.Component == "" || r.Sequence == 0 || r.RequestID == 0 || r.Epoch == 0 {
-				_ = f.Close()
+				return nil, closeFile(f, errors.New("trace record has invalid correlation fields"))
-				return nil, errors.New("trace record has invalid correlation fields")
 			}
 			if _, ok := known[r.ProcessID]; !ok {
-				_ = f.Close()
+				return nil, closeFile(f, errors.New("unknown trace process_id"))
-				return nil, errors.New("unknown trace process_id")
 			}
 			records = append(records, r)
 		}
 		if err := scan.Err(); err != nil {
-			_ = f.Close()
+			return nil, closeFile(f, err)
+		}
+		if err := f.Close(); err != nil {
 			return nil, err
 		}
-		_ = f.Close()
 		// A process shares one observer across concurrent components. Correlation
 		// IDs are allocated before a goroutine reaches its first mark, so a later
 		// ID can be serialized before an earlier one. Their first appearance is
 	return distribution{len(v), at(50), at(95), at(99), v[len(v)-1]}
 }
+func closeFile(f *os.File, err error) error {
+	return errors.Join(err, f.Close())
+}
 func writeJSON(path string, v any) error {
-	f, e := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600)
+	f, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600)
-	if e != nil {
+	if err != nil {
-		return e
+		return err
 	}
-	defer f.Close()
+	if err := json.NewEncoder(f).Encode(v); err != nil {
-	e = json.NewEncoder(f).Encode(v)
+		return closeFile(f, err)
-	return e
+	}
+	return f.Close()
 }
 func mustJSON(v any) string { b, _ := json.Marshal(v); return string(b) }
 }
 func readJSONL(path string) ([]map[string]json.RawMessage, error) {
-	f, e := os.Open(path)
+	f, err := os.Open(path)
-	if e != nil {
+	if err != nil {
-		return nil, e
+		return nil, err
 	}
-	defer f.Close()
 	var r []map[string]json.RawMessage
 	scan := bufio.NewScanner(f)
 	for scan.Scan() {
 		var v map[string]json.RawMessage
-		if e := json.Unmarshal(scan.Bytes(), &v); e != nil {
+		if err := json.Unmarshal(scan.Bytes(), &v); err != nil {
-			return nil, e
+			return nil, closeFile(f, err)
 		}
 		r = append(r, v)
 	}
-	return r, scan.Err()
+	if err := scan.Err(); err != nil {
+		return nil, closeFile(f, err)
+	}
+	if err := f.Close(); err != nil {
+		return nil, err
+	}
+	return r, nil
 }

M cmd/vev-perf-harness/options_manifest.go +10 -6


@@ -67,14 +67,18 @@ func resolvePathOptions(o options) (options, error) {
67	67		}
68	68
69	69		func readManifest(path string) (manifest, error) {
70		-	f, e := os.Open(path)
71		-	if e != nil {
72		-	return manifest{}, e
	70	+	f, err := os.Open(path)
	71	+	if err != nil {
	72	+	return manifest{}, err
73	73		}
74		-	defer f.Close()
75	74		var m manifest
76		-	e = json.NewDecoder(f).Decode(&m)
77		-	return m, e
	75	+	if err := json.NewDecoder(f).Decode(&m); err != nil {
	76	+	return manifest{}, closeFile(f, err)
	77	+	}
	78	+	if err := f.Close(); err != nil {
	79	+	return manifest{}, err
	80	+	}
	81	+	return m, nil
78	82		}
79	83
80	84		func validateManifest(m manifest) error {

 }
 func readManifest(path string) (manifest, error) {
-	f, e := os.Open(path)
+	f, err := os.Open(path)
-	if e != nil {
+	if err != nil {
-		return manifest{}, e
+		return manifest{}, err
 	}
-	defer f.Close()
 	var m manifest
-	e = json.NewDecoder(f).Decode(&m)
+	if err := json.NewDecoder(f).Decode(&m); err != nil {
-	return m, e
+		return manifest{}, closeFile(f, err)
+	}
+	if err := f.Close(); err != nil {
+		return manifest{}, err
+	}
+	return m, nil
 }
 func validateManifest(m manifest) error {

M cmd/vev-perf-harness/runner.go +12 -3

@@ -1,6 +1,7 @@
1	1		package main
2	2
3	3		import (
	4	+	"context"
4	5		"errors"
5	6		"fmt"
6	7		"io"

@@ -66,6 +67,8 @@ func createExclusiveRunDir(path string) error {
66	67		// recordedGitSHA prefers build provenance so an installed public harness does
67	68		// not depend on its launch directory being a checkout. The git fallback keeps
68	69		// go run evidence attributable as well.
	70	+	const gitSHACommandTimeout = time.Second
	71	+
69	72		func recordedGitSHA() string {
70	73		if info, ok := debug.ReadBuildInfo(); ok {
71	74		for _, setting := range info.Settings {

@@ -74,14 +77,16 @@ func recordedGitSHA() string {
74	77		}
75	78		}
76	79		}
77		-	out, err := exec.Command("git", "rev-parse", "HEAD").Output()
	80	+	ctx, cancel := context.WithTimeout(context.Background(), gitSHACommandTimeout)
	81	+	defer cancel()
	82	+	out, err := exec.CommandContext(ctx, "git", "rev-parse", "HEAD").Output()
78	83		if err == nil && strings.TrimSpace(string(out)) != "" {
79	84		return strings.TrimSpace(string(out))
80	85		}
81	86		return "unknown"
82	87		}
83	88
84		-	func run(args []string, h *harness) error {
	89	+	func run(args []string, h *harness) (err error) {
85	90		opt, err := parseOptions(args)
86	91		if err != nil {
87	92		return err

@@ -104,7 +109,11 @@ func run(args []string, h *harness) error {
104	109		if err != nil {
105	110		return err
106	111		}
107		-	defer raw.Close()
	112	+	defer func() {
	113	+	if closeErr := raw.Close(); closeErr != nil {
	114	+	err = errors.Join(err, closeErr)
	115	+	}
	116	+	}()
108	117		results := make([]runResult, 0, len(m.Scenarios)*opt.repetitions)
109	118		var localSpans []span
110	119		scenarios := m.Scenarios

 package main
 import (
+	"context"
 	"errors"
 	"fmt"
 	"io"
 // recordedGitSHA prefers build provenance so an installed public harness does
 // not depend on its launch directory being a checkout. The git fallback keeps
 // go run evidence attributable as well.
+const gitSHACommandTimeout = time.Second
 func recordedGitSHA() string {
 	if info, ok := debug.ReadBuildInfo(); ok {
 		for _, setting := range info.Settings {
 			}
 		}
 	}
-	out, err := exec.Command("git", "rev-parse", "HEAD").Output()
+	ctx, cancel := context.WithTimeout(context.Background(), gitSHACommandTimeout)
+	defer cancel()
+	out, err := exec.CommandContext(ctx, "git", "rev-parse", "HEAD").Output()
 	if err == nil && strings.TrimSpace(string(out)) != "" {
 		return strings.TrimSpace(string(out))
 	}
 	return "unknown"
 }
-func run(args []string, h *harness) error {
+func run(args []string, h *harness) (err error) {
 	opt, err := parseOptions(args)
 	if err != nil {
 		return err
 	if err != nil {
 		return err
 	}
-	defer raw.Close()
+	defer func() {
+		if closeErr := raw.Close(); closeErr != nil {
+			err = errors.Join(err, closeErr)
+		}
+	}()
 	results := make([]runResult, 0, len(m.Scenarios)*opt.repetitions)
 	var localSpans []span
 	scenarios := m.Scenarios

M cmd/vev-perf-harness/udp_netem.go +115 -27

@@ -1,6 +1,7 @@
1	1		package main
2	2
3	3		import (
	4	+	"container/heap"
4	5		"errors"
5	6		"fmt"
6	7		"net"

@@ -8,6 +9,7 @@ import (
8	9		"strconv"
9	10		"strings"
10	11		"sync"
	12	+	"sync/atomic"
11	13		"time"
12	14		)
13	15

@@ -28,6 +30,31 @@ type udpNetemConfig struct {
28	30		TargetPath string
29	31		}
30	32
	33	+	// udpNetemQueueCapacity bounds both packets waiting to be scheduled and
	34	+	// packets waiting for their delivery deadline. When it is full, the relay
	35	+	// drops the arriving packet and records it in queueOverflowDrops.
	36	+	const udpNetemQueueCapacity = 256
	37	+
	38	+	type scheduledPacket struct {
	39	+	packet []byte
	40	+	to net.Addr
	41	+	due time.Time
	42	+	}
	43	+
	44	+	type scheduledPackets []scheduledPacket
	45	+
	46	+	func (p scheduledPackets) Len() int { return len(p) }
	47	+	func (p scheduledPackets) Less(i, j int) bool { return p[i].due.Before(p[j].due) }
	48	+	func (p scheduledPackets) Swap(i, j int) { p[i], p[j] = p[j], p[i] }
	49	+	func (p scheduledPackets) Push(x any) { p = append(*p, x.(scheduledPacket)) }
	50	+	func (p *scheduledPackets) Pop() any {
	51	+	old := *p
	52	+	n := len(old)
	53	+	item := old[n-1]
	54	+	*p = old[:n-1]
	55	+	return item
	56	+	}
	57	+
31	58		type udpNetemRelay struct {
32	59		config udpNetemConfig
33	60		conn net.PacketConn

@@ -35,6 +62,12 @@ type udpNetemRelay struct {
35	62		wg sync.WaitGroup
36	63		once sync.Once
37	64
	65	+	// slots bounds queued and deadline-pending packets together. The reader and
	66	+	// the sole scheduler are the relay's only goroutines.
	67	+	slots chan struct{}
	68	+	queue chan scheduledPacket
	69	+	queueOverflowDrops atomic.Uint64
	70	+
38	71		mu sync.Mutex
39	72		client, dst net.Addr
40	73		packet uint64

@@ -48,9 +81,16 @@ func newUDPNetem(config udpNetemConfig) (udpNetem, error) {
48	81		if err != nil {
49	82		return nil, err
50	83		}
51		-	r := &udpNetemRelay{config: config, conn: conn, done: make(chan struct{})}
52		-	r.wg.Add(1)
	84	+	r := &udpNetemRelay{
	85	+	config: config,
	86	+	conn: conn,
	87	+	done: make(chan struct{}),
	88	+	slots: make(chan struct{}, udpNetemQueueCapacity),
	89	+	queue: make(chan scheduledPacket, udpNetemQueueCapacity),
	90	+	}
	91	+	r.wg.Add(2)
53	92		go r.run()
	93	+	go r.schedule()
54	94		return r, nil
55	95		}
56	96

@@ -80,8 +120,79 @@ func (r *udpNetemRelay) run() {
80	120		if !ok \|\| r.drop() {
81	121		continue
82	122		}
83		-	packet := append([]byte(nil), buf[:n]...)
84		-	r.schedule(packet, to)
	123	+	// Copy because PacketConn reuses buf on its next read.
	124	+	r.enqueue(scheduledPacket{packet: append([]byte(nil), buf[:n]...), to: to, due: time.Now().Add(r.config.RTT / 2)})
	125	+	}
	126	+	}
	127	+
	128	+	// enqueue implements the explicit overflow policy: never block the reader;
	129	+	// drop a packet if the bounded scheduler capacity has been reached.
	130	+	func (r *udpNetemRelay) enqueue(packet scheduledPacket) bool {
	131	+	select {
	132	+	case r.slots <- struct{}{}:
	133	+	case <-r.done:
	134	+	return false
	135	+	default:
	136	+	r.queueOverflowDrops.Add(1)
	137	+	return false
	138	+	}
	139	+	select {
	140	+	case r.queue <- packet:
	141	+	return true
	142	+	case <-r.done:
	143	+	<-r.slots
	144	+	return false
	145	+	}
	146	+	}
	147	+
	148	+	func (r *udpNetemRelay) schedule() {
	149	+	defer r.wg.Done()
	150	+	pending := scheduledPackets{}
	151	+	heap.Init(&pending)
	152	+	timer := time.NewTimer(time.Hour)
	153	+	if !timer.Stop() {
	154	+	<-timer.C
	155	+	}
	156	+	defer timer.Stop()
	157	+	var timerC <-chan time.Time
	158	+	resetTimer := func() {
	159	+	if timerC != nil && !timer.Stop() {
	160	+	select {
	161	+	case <-timer.C:
	162	+	default:
	163	+	}
	164	+	}
	165	+	if pending.Len() == 0 {
	166	+	timerC = nil
	167	+	return
	168	+	}
	169	+	delay := time.Until(pending[0].due)
	170	+	if delay < 0 {
	171	+	delay = 0
	172	+	}
	173	+	timer.Reset(delay)
	174	+	timerC = timer.C
	175	+	}
	176	+	for {
	177	+	resetTimer()
	178	+	select {
	179	+	case <-r.done:
	180	+	return
	181	+	case packet := <-r.queue:
	182	+	heap.Push(&pending, packet)
	183	+	case <-timerC:
	184	+	now := time.Now()
	185	+	for pending.Len() > 0 && !pending[0].due.After(now) {
	186	+	packet := heap.Pop(&pending).(scheduledPacket)
	187	+	select {
	188	+	case <-r.done:
	189	+	return
	190	+	default:
	191	+	_, _ = r.conn.WriteTo(packet.packet, packet.to)
	192	+	<-r.slots
	193	+	}
	194	+	}
	195	+	}
85	196		}
86	197		}
87	198

@@ -111,29 +222,6 @@ func (r *udpNetemRelay) drop() bool {
111	222		return r.config.LossPercent > 0 && int((r.packet-1)%100) < r.config.LossPercent
112	223		}
113	224
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	225		func udpNetemTarget(path string) (net.Addr, error) {
138	226		line, err := os.ReadFile(path)
139	227		if err != nil {

@@ -1,6 +1,7 @@
1		package main	1		package main
2			2
3		import (	3		import (
			4	+	"container/heap"
4		"errors"	5		"errors"
5		"fmt"	6		"fmt"
6		"net"	7		"net"

@@ -8,6 +9,7 @@ import (
8		"strconv"	9		"strconv"
9		"strings"	10		"strings"
10		"sync"	11		"sync"
			12	+	"sync/atomic"
11		"time"	13		"time"
12		)	14		)
13			15

@@ -28,6 +30,31 @@ type udpNetemConfig struct {
28		TargetPath string	30		TargetPath string
29		}	31		}
30			32
			33	+	// udpNetemQueueCapacity bounds both packets waiting to be scheduled and
			34	+	// packets waiting for their delivery deadline. When it is full, the relay
			35	+	// drops the arriving packet and records it in queueOverflowDrops.
			36	+	const udpNetemQueueCapacity = 256
			37	+
			38	+	type scheduledPacket struct {
			39	+	packet []byte
			40	+	to net.Addr
			41	+	due time.Time
			42	+	}
			43	+
			44	+	type scheduledPackets []scheduledPacket
			45	+
			46	+	func (p scheduledPackets) Len() int { return len(p) }
			47	+	func (p scheduledPackets) Less(i, j int) bool { return p[i].due.Before(p[j].due) }
			48	+	func (p scheduledPackets) Swap(i, j int) { p[i], p[j] = p[j], p[i] }
			49	+	func (p scheduledPackets) Push(x any) { p = append(*p, x.(scheduledPacket)) }
			50	+	func (p *scheduledPackets) Pop() any {
			51	+	old := *p
			52	+	n := len(old)
			53	+	item := old[n-1]
			54	+	*p = old[:n-1]
			55	+	return item
			56	+	}
			57	+
31		type udpNetemRelay struct {	58		type udpNetemRelay struct {
32		config udpNetemConfig	59		config udpNetemConfig
33		conn net.PacketConn	60		conn net.PacketConn

@@ -35,6 +62,12 @@ type udpNetemRelay struct {
35		wg sync.WaitGroup	62		wg sync.WaitGroup
36		once sync.Once	63		once sync.Once
37			64
			65	+	// slots bounds queued and deadline-pending packets together. The reader and
			66	+	// the sole scheduler are the relay's only goroutines.
			67	+	slots chan struct{}
			68	+	queue chan scheduledPacket
			69	+	queueOverflowDrops atomic.Uint64
			70	+
38		mu sync.Mutex	71		mu sync.Mutex
39		client, dst net.Addr	72		client, dst net.Addr
40		packet uint64	73		packet uint64

@@ -48,9 +81,16 @@ func newUDPNetem(config udpNetemConfig) (udpNetem, error) {
48		if err != nil {	81		if err != nil {
49		return nil, err	82		return nil, err
50		}	83		}
51	-	r := &udpNetemRelay{config: config, conn: conn, done: make(chan struct{})}	84	+	r := &udpNetemRelay{
52	-	r.wg.Add(1)	85	+	config: config,
			86	+	conn: conn,
			87	+	done: make(chan struct{}),
			88	+	slots: make(chan struct{}, udpNetemQueueCapacity),
			89	+	queue: make(chan scheduledPacket, udpNetemQueueCapacity),
			90	+	}
			91	+	r.wg.Add(2)
53		go r.run()	92		go r.run()
			93	+	go r.schedule()
54		return r, nil	94		return r, nil
55		}	95		}
56			96

@@ -80,8 +120,79 @@ func (r *udpNetemRelay) run() {
80		if !ok \|\| r.drop() {	120		if !ok \|\| r.drop() {
81		continue	121		continue
82		}	122		}
83	-	packet := append([]byte(nil), buf[:n]...)	123	+	// Copy because PacketConn reuses buf on its next read.
84	-	r.schedule(packet, to)	124	+	r.enqueue(scheduledPacket{packet: append([]byte(nil), buf[:n]...), to: to, due: time.Now().Add(r.config.RTT / 2)})
			125	+	}
			126	+	}
			127	+
			128	+	// enqueue implements the explicit overflow policy: never block the reader;
			129	+	// drop a packet if the bounded scheduler capacity has been reached.
			130	+	func (r *udpNetemRelay) enqueue(packet scheduledPacket) bool {
			131	+	select {
			132	+	case r.slots <- struct{}{}:
			133	+	case <-r.done:
			134	+	return false
			135	+	default:
			136	+	r.queueOverflowDrops.Add(1)
			137	+	return false
			138	+	}
			139	+	select {
			140	+	case r.queue <- packet:
			141	+	return true
			142	+	case <-r.done:
			143	+	<-r.slots
			144	+	return false
			145	+	}
			146	+	}
			147	+
			148	+	func (r *udpNetemRelay) schedule() {
			149	+	defer r.wg.Done()
			150	+	pending := scheduledPackets{}
			151	+	heap.Init(&pending)
			152	+	timer := time.NewTimer(time.Hour)
			153	+	if !timer.Stop() {
			154	+	<-timer.C
			155	+	}
			156	+	defer timer.Stop()
			157	+	var timerC <-chan time.Time
			158	+	resetTimer := func() {
			159	+	if timerC != nil && !timer.Stop() {
			160	+	select {
			161	+	case <-timer.C:
			162	+	default:
			163	+	}
			164	+	}
			165	+	if pending.Len() == 0 {
			166	+	timerC = nil
			167	+	return
			168	+	}
			169	+	delay := time.Until(pending[0].due)
			170	+	if delay < 0 {
			171	+	delay = 0
			172	+	}
			173	+	timer.Reset(delay)
			174	+	timerC = timer.C
			175	+	}
			176	+	for {
			177	+	resetTimer()
			178	+	select {
			179	+	case <-r.done:
			180	+	return
			181	+	case packet := <-r.queue:
			182	+	heap.Push(&pending, packet)
			183	+	case <-timerC:
			184	+	now := time.Now()
			185	+	for pending.Len() > 0 && !pending[0].due.After(now) {
			186	+	packet := heap.Pop(&pending).(scheduledPacket)
			187	+	select {
			188	+	case <-r.done:
			189	+	return
			190	+	default:
			191	+	_, _ = r.conn.WriteTo(packet.packet, packet.to)
			192	+	<-r.slots
			193	+	}
			194	+	}
			195	+	}
85		}	196		}
86		}	197		}
87			198

@@ -111,29 +222,6 @@ func (r *udpNetemRelay) drop() bool {
111		return r.config.LossPercent > 0 && int((r.packet-1)%100) < r.config.LossPercent	222		return r.config.LossPercent > 0 && int((r.packet-1)%100) < r.config.LossPercent
112		}	223		}
113			224
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) {	225		func udpNetemTarget(path string) (net.Addr, error) {
138		line, err := os.ReadFile(path)	226		line, err := os.ReadFile(path)
139		if err != nil {	227		if err != nil {

M cmd/vev-perf-harness/udp_netem_test.go +43 -2


@@ -9,6 +9,26 @@ import (
9	9		"time"
10	10		)
11	11
	12	+	func TestUDPNetemDropsWhenBoundedSchedulerIsFull(t *testing.T) {
	13	+	r := &udpNetemRelay{
	14	+	done: make(chan struct{}),
	15	+	slots: make(chan struct{}, udpNetemQueueCapacity),
	16	+	queue: make(chan scheduledPacket, udpNetemQueueCapacity),
	17	+	}
	18	+	packet := scheduledPacket{packet: []byte("bounded"), to: &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 1}}
	19	+	for range udpNetemQueueCapacity {
	20	+	if !r.enqueue(packet) {
	21	+	t.Fatal("enqueue() dropped packet before scheduler capacity was full")
	22	+	}
	23	+	}
	24	+	if r.enqueue(packet) {
	25	+	t.Fatal("enqueue() accepted packet beyond scheduler capacity")
	26	+	}
	27	+	if got := r.queueOverflowDrops.Load(); got != 1 {
	28	+	t.Fatalf("queue overflow drops=%d, want 1", got)
	29	+	}
	30	+	}
	31	+
12	32		func TestUDPNetemExecutesRTTAndLossFixtures(t *testing.T) {
13	33		fixtures := []struct {
14	34		name string

@@ -54,15 +74,36 @@ func TestUDPNetemExecutesRTTAndLossFixtures(t *testing.T) {
54	74		}
55	75		got := 0
56	76		buf := make([]byte, 32)
	77	+	var relayAddr net.Addr
57	78		for got < tc.want {
58		-	if _, _, err := target.ReadFrom(buf); err != nil {
	79	+	if _, relayAddr, err = target.ReadFrom(buf); err != nil {
59	80		t.Fatalf("received %d packets, want %d: %v", got, tc.want, err)
60	81		}
61	82		got++
62	83		}
63	84		if tc.rtt > 0 {
64	85		if elapsed := time.Since(started); elapsed < tc.rtt/2 {
65		-	t.Fatalf("one-way netem delay=%s, want at least %s", elapsed, tc.rtt/2)
	86	+	t.Fatalf("client-to-target netem delay=%s, want at least %s", elapsed, tc.rtt/2)
	87	+	}
	88	+	}
	89	+	if tc.loss == 0 {
	90	+	// The relay learns the client from the first carriage packet. Replying
	91	+	// through that same relay address verifies the target-to-client half
	92	+	// of the configured RTT as well.
	93	+	started = time.Now()
	94	+	if _, err := target.WriteTo([]byte("reply"), relayAddr); err != nil {
	95	+	t.Fatal(err)
	96	+	}
	97	+	if err := client.SetReadDeadline(time.Now().Add(2 * time.Second)); err != nil {
	98	+	t.Fatal(err)
	99	+	}
	100	+	if _, _, err := client.ReadFrom(buf); err != nil {
	101	+	t.Fatalf("relay reply: %v", err)
	102	+	}
	103	+	if tc.rtt > 0 {
	104	+	if elapsed := time.Since(started); elapsed < tc.rtt/2 {
	105	+	t.Fatalf("target-to-client netem delay=%s, want at least %s", elapsed, tc.rtt/2)
	106	+	}
66	107		}
67	108		}
68	109		// Keep the deadline short: extra carriage would prove the exact 1%%


@@ -9,6 +9,26 @@ import (
9		"time"	9		"time"
10		)	10		)
11			11
			12	+	func TestUDPNetemDropsWhenBoundedSchedulerIsFull(t *testing.T) {
			13	+	r := &udpNetemRelay{
			14	+	done: make(chan struct{}),
			15	+	slots: make(chan struct{}, udpNetemQueueCapacity),
			16	+	queue: make(chan scheduledPacket, udpNetemQueueCapacity),
			17	+	}
			18	+	packet := scheduledPacket{packet: []byte("bounded"), to: &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 1}}
			19	+	for range udpNetemQueueCapacity {
			20	+	if !r.enqueue(packet) {
			21	+	t.Fatal("enqueue() dropped packet before scheduler capacity was full")
			22	+	}
			23	+	}
			24	+	if r.enqueue(packet) {
			25	+	t.Fatal("enqueue() accepted packet beyond scheduler capacity")
			26	+	}
			27	+	if got := r.queueOverflowDrops.Load(); got != 1 {
			28	+	t.Fatalf("queue overflow drops=%d, want 1", got)
			29	+	}
			30	+	}
			31	+
12		func TestUDPNetemExecutesRTTAndLossFixtures(t *testing.T) {	32		func TestUDPNetemExecutesRTTAndLossFixtures(t *testing.T) {
13		fixtures := []struct {	33		fixtures := []struct {
14		name string	34		name string

@@ -54,15 +74,36 @@ func TestUDPNetemExecutesRTTAndLossFixtures(t *testing.T) {
54		}	74		}
55		got := 0	75		got := 0
56		buf := make([]byte, 32)	76		buf := make([]byte, 32)
			77	+	var relayAddr net.Addr
57		for got < tc.want {	78		for got < tc.want {
58	-	if _, _, err := target.ReadFrom(buf); err != nil {	79	+	if _, relayAddr, err = target.ReadFrom(buf); err != nil {
59		t.Fatalf("received %d packets, want %d: %v", got, tc.want, err)	80		t.Fatalf("received %d packets, want %d: %v", got, tc.want, err)
60		}	81		}
61		got++	82		got++
62		}	83		}
63		if tc.rtt > 0 {	84		if tc.rtt > 0 {
64		if elapsed := time.Since(started); elapsed < tc.rtt/2 {	85		if elapsed := time.Since(started); elapsed < tc.rtt/2 {
65	-	t.Fatalf("one-way netem delay=%s, want at least %s", elapsed, tc.rtt/2)	86	+	t.Fatalf("client-to-target netem delay=%s, want at least %s", elapsed, tc.rtt/2)
			87	+	}
			88	+	}
			89	+	if tc.loss == 0 {
			90	+	// The relay learns the client from the first carriage packet. Replying
			91	+	// through that same relay address verifies the target-to-client half
			92	+	// of the configured RTT as well.
			93	+	started = time.Now()
			94	+	if _, err := target.WriteTo([]byte("reply"), relayAddr); err != nil {
			95	+	t.Fatal(err)
			96	+	}
			97	+	if err := client.SetReadDeadline(time.Now().Add(2 * time.Second)); err != nil {
			98	+	t.Fatal(err)
			99	+	}
			100	+	if _, _, err := client.ReadFrom(buf); err != nil {
			101	+	t.Fatalf("relay reply: %v", err)
			102	+	}
			103	+	if tc.rtt > 0 {
			104	+	if elapsed := time.Since(started); elapsed < tc.rtt/2 {
			105	+	t.Fatalf("target-to-client netem delay=%s, want at least %s", elapsed, tc.rtt/2)
			106	+	}
66		}	107		}
67		}	108		}
68		// Keep the deadline short: extra carriage would prove the exact 1%%	109		// Keep the deadline short: extra carriage would prove the exact 1%%

M internal/adapters/dgram/transport_observability_test.go +12 -2


@@ -95,28 +95,38 @@ func assertDgramFailedSpanPairs(t *testing.T, path string) {
95	95		type spanKey struct {
96	96		processID, scenario string
97	97		run, sequence, requestID, epoch uint64
	98	+	kind string
98	99		}
99	100		starts := make(map[spanKey]mark)
	101	+	pairs := map[string]int{}
100	102		for _, line := range strings.Split(strings.TrimSpace(string(data)), "\n") {
101	103		var m mark
102	104		if err := json.Unmarshal([]byte(line), &m); err != nil {
103	105		t.Fatalf("Unmarshal(trace): %v", err)
104	106		}
105		-	key := spanKey{m.ProcessID, m.Scenario, m.Run, m.Sequence, m.RequestID, m.Epoch}
	107	+	key := spanKey{m.ProcessID, m.Scenario, m.Run, m.Sequence, m.RequestID, m.Epoch, m.Kind}
106	108		switch m.Kind {
107	109		case "adapter_receive_start", "adapter_send_start":
	110	+	if _, exists := starts[key]; exists {
	111	+	t.Fatalf("duplicate adapter start: %#v", m)
	112	+	}
108	113		starts[key] = m
109	114		case "adapter_receive_end", "adapter_send_end":
	115	+	key.kind = strings.TrimSuffix(m.Kind, "_end") + "_start"
110	116		start, ok := starts[key]
111		-	if !ok \|\| strings.TrimSuffix(start.Kind, "_start") != strings.TrimSuffix(m.Kind, "_end") \|\| m.Valid {
	117	+	if !ok \|\| m.Valid {
112	118		t.Fatalf("unmatched or valid failed end: start=%#v end=%#v", start, m)
113	119		}
	120	+	pairs[strings.TrimSuffix(m.Kind, "_end")]++
114	121		delete(starts, key)
115	122		}
116	123		}
117	124		if len(starts) != 0 {
118	125		t.Fatalf("unmatched adapter starts: %#v", starts)
119	126		}
	127	+	if pairs["adapter_send"] != 1 \|\| pairs["adapter_receive"] != 1 \|\| len(pairs) != 2 {
	128	+	t.Fatalf("failed adapter span pairs=%v, want exactly one send and one receive", pairs)
	129	+	}
120	130		}
121	131
122	132		func TestTransportObservabilityDgramKeepsBehaviorClockSeparate(t *testing.T) {

 	type spanKey struct {
 		processID, scenario             string
 		run, sequence, requestID, epoch uint64
+		kind                            string
 	}
 	starts := make(map[spanKey]mark)
+	pairs := map[string]int{}
 	for _, line := range strings.Split(strings.TrimSpace(string(data)), "\n") {
 		var m mark
 		if err := json.Unmarshal([]byte(line), &m); err != nil {
 			t.Fatalf("Unmarshal(trace): %v", err)
 		}
-		key := spanKey{m.ProcessID, m.Scenario, m.Run, m.Sequence, m.RequestID, m.Epoch}
+		key := spanKey{m.ProcessID, m.Scenario, m.Run, m.Sequence, m.RequestID, m.Epoch, m.Kind}
 		switch m.Kind {
 		case "adapter_receive_start", "adapter_send_start":
+			if _, exists := starts[key]; exists {
+				t.Fatalf("duplicate adapter start: %#v", m)
+			}
 			starts[key] = m
 		case "adapter_receive_end", "adapter_send_end":
+			key.kind = strings.TrimSuffix(m.Kind, "_end") + "_start"
 			start, ok := starts[key]
-			if !ok || strings.TrimSuffix(start.Kind, "_start") != strings.TrimSuffix(m.Kind, "_end") || m.Valid {
+			if !ok || m.Valid {
 				t.Fatalf("unmatched or valid failed end: start=%#v end=%#v", start, m)
 			}
+			pairs[strings.TrimSuffix(m.Kind, "_end")]++
 			delete(starts, key)
 		}
 	}
 	if len(starts) != 0 {
 		t.Fatalf("unmatched adapter starts: %#v", starts)
 	}
+	if pairs["adapter_send"] != 1 || pairs["adapter_receive"] != 1 || len(pairs) != 2 {
+		t.Fatalf("failed adapter span pairs=%v, want exactly one send and one receive", pairs)
+	}
 }
 func TestTransportObservabilityDgramKeepsBehaviorClockSeparate(t *testing.T) {