d4ed0f5 fix: address UDP transport review findings · vev

M internal/adapters/dgram/control.go +18 -7


@@ -40,6 +40,12 @@ func (t *Transport) queueACK(seq uint64) {
40	40		case t.ackWake <- struct{}{}:
41	41		default:
42	42		}
	43	+	t.mu.Lock()
	44	+	afterACKQueued := t.afterACKQueued
	45	+	t.mu.Unlock()
	46	+	if afterACKQueued != nil {
	47	+	afterACKQueued()
	48	+	}
43	49		}
44	50
45	51		// queueControl is deliberately lossy when the bounded control queue is full:

@@ -93,15 +99,20 @@ func (t *Transport) ackScheduleLoop() {
93	99		}
94	100		}()
95	101		scheduleACK := func() {
96		-	if ackTimerC != nil {
97		-	return
	102	+	if ackTimerC == nil {
	103	+	if ackTimer == nil {
	104	+	ackTimer = t.clock.NewTimer(maxACKDelay)
	105	+	} else {
	106	+	ackTimer.Reset(maxACKDelay)
	107	+	}
	108	+	ackTimerC = ackTimer.C()
98	109		}
99		-	if ackTimer == nil {
100		-	ackTimer = t.clock.NewTimer(maxACKDelay)
101		-	} else {
102		-	ackTimer.Reset(maxACKDelay)
	110	+	t.mu.Lock()
	111	+	afterACKScheduled := t.afterACKScheduled
	112	+	t.mu.Unlock()
	113	+	if afterACKScheduled != nil {
	114	+	afterACKScheduled()
103	115		}
104		-	ackTimerC = ackTimer.C()
105	116		}
106	117		dispatchACK := func() {
107	118		seq, ok := t.takeACK()

 	case t.ackWake <- struct{}{}:
 	default:
 	}
+	t.mu.Lock()
+	afterACKQueued := t.afterACKQueued
+	t.mu.Unlock()
+	if afterACKQueued != nil {
+		afterACKQueued()
+	}
 }
 // queueControl is deliberately lossy when the bounded control queue is full:
 		}
 	}()
 	scheduleACK := func() {
-		if ackTimerC != nil {
+		if ackTimerC == nil {
-			return
+			if ackTimer == nil {
+				ackTimer = t.clock.NewTimer(maxACKDelay)
+			} else {
+				ackTimer.Reset(maxACKDelay)
+			}
+			ackTimerC = ackTimer.C()
 		}
-		if ackTimer == nil {
+		t.mu.Lock()
-			ackTimer = t.clock.NewTimer(maxACKDelay)
+		afterACKScheduled := t.afterACKScheduled
-		} else {
+		t.mu.Unlock()
-			ackTimer.Reset(maxACKDelay)
+		if afterACKScheduled != nil {
+			afterACKScheduled()
 		}
-		ackTimerC = ackTimer.C()
 	}
 	dispatchACK := func() {
 		seq, ok := t.takeACK()

M internal/adapters/dgram/network_sim_test.go +123 -10


@@ -94,16 +94,20 @@ func (l simulatedLink) enqueue(p fakePC, peer *fakePC, b []byte) bool {
94	94		return true
95	95		}
96	96
97		-	func (l simulatedLink) flush(peer fakePC) {
	97	+	func (l simulatedLink) flush(peer fakePC) int {
98	98		l.mu.Lock()
99	99		defer l.mu.Unlock()
100	100		now := l.clock.Now()
	101	+	delivered := 0
101	102		for len(l.queue) > 0 && !l.queue[0].due.After(now) {
102	103		q := l.queue[0]
103	104		l.queue = l.queue[1:]
104	105		l.queueBytes -= len(q.b)
105		-	l.deliverLocked(peer, q.packet)
	106	+	if l.deliverLocked(peer, q.packet) {
	107	+	delivered++
	108	+	}
106	109		}
	110	+	return delivered
107	111		}
108	112
109	113		func (l *simulatedLink) queuedBytes() int {

@@ -179,6 +183,7 @@ func (p *fakePC) ReadFrom(b []byte) (int, net.Addr, error) {
179	183		func (p *fakePC) WriteTo(b []byte, addr net.Addr) (int, error) {
180	184		p.mu.Lock()
181	185		closed := p.closed
	186	+	afterWrite := p.afterWrite
182	187		p.mu.Unlock()
183	188		if closed {
184	189		return 0, errors.New("closed")

@@ -204,11 +209,29 @@ func (p *fakePC) WriteTo(b []byte, addr net.Addr) (int, error) {
204	209		}
205	210		peer.mu.Unlock()
206	211		}
207		-	if p.afterWrite != nil {
208		-	p.afterWrite()
	212	+	if afterWrite != nil {
	213	+	afterWrite()
209	214		}
210	215		return len(b), nil
211	216		}
	217	+	func (p *fakePC) replaceAfterWrite(afterWrite func()) {
	218	+	p.mu.Lock()
	219	+	defer p.mu.Unlock()
	220	+	p.afterWrite = afterWrite
	221	+	}
	222	+
	223	+	func (p *fakePC) addAfterWrite(afterWrite func()) {
	224	+	p.mu.Lock()
	225	+	defer p.mu.Unlock()
	226	+	previous := p.afterWrite
	227	+	p.afterWrite = func() {
	228	+	if previous != nil {
	229	+	previous()
	230	+	}
	231	+	afterWrite()
	232	+	}
	233	+	}
	234	+
212	235		func (p *fakePC) Close() error {
213	236		p.mu.Lock()
214	237		defer p.mu.Unlock()

@@ -255,6 +278,15 @@ func awaitResult[T any](t *testing.T, ch <-chan T, context string) T {
255	278		}
256	279		}
257	280
	281	+	func flushAndAwait(t testing.T, link simulatedLink, peer *fakePC, processed <-chan struct{}) int {
	282	+	t.Helper()
	283	+	delivered := link.flush(peer)
	284	+	for range delivered {
	285	+	awaitSignal(t, processed, "flushed packet processing")
	286	+	}
	287	+	return delivered
	288	+	}
	289	+
258	290		// TestTransportFloodClassification drives authenticated datagrams through real
259	291		// transports. The metrics are transport state, not simulator bookkeeping.
260	292		func TestTransportFloodClassification(t *testing.T) {

@@ -356,19 +388,65 @@ func TestTransportFloodClassification(t *testing.T) {
356	388		defer closeFloodTransports(a, b)
357	389		waitForManualTimers(t, clk, 6)
358	390		var contacted atomic.Bool
	391	+	processed := make(chan struct{})
	392	+	primerWritten := make(chan struct{}, 1)
	393	+	primerACKScheduled := make(chan struct{}, 1)
	394	+	primerAcknowledged := make(chan struct{}, 1)
	395	+	ackQueued := make(chan struct{}, 128)
	396	+	ackScheduled := make(chan struct{}, 128)
	397	+	ackWritten := make(chan struct{}, 128)
	398	+	aPC.replaceAfterWrite(func() {
	399	+	select {
	400	+	case primerWritten <- struct{}{}:
	401	+	default:
	402	+	}
	403	+	})
	404	+	bPC.replaceAfterWrite(func() {
	405	+	select {
	406	+	case primerAcknowledged <- struct{}{}:
	407	+	default:
	408	+	}
	409	+	select {
	410	+	case ackWritten <- struct{}{}:
	411	+	default:
	412	+	}
	413	+	})
	414	+	a.mu.Lock()
	415	+	a.afterPacketProcessed = func() { processed <- struct{}{} }
	416	+	a.mu.Unlock()
359	417		b.mu.Lock()
360	418		b.afterAuthenticatedPacket = func() { contacted.Store(true) }
	419	+	b.afterPacketProcessed = func() { processed <- struct{}{} }
	420	+	b.afterACKScheduled = func() {
	421	+	select {
	422	+	case primerACKScheduled <- struct{}{}:
	423	+	default:
	424	+	}
	425	+	select {
	426	+	case ackScheduled <- struct{}{}:
	427	+	default:
	428	+	}
	429	+	}
	430	+	b.afterACKQueued = func() { ackQueued <- struct{}{} }
361	431		b.mu.Unlock()
362	432		primer := ports.MarshalOutput(ports.Output{BaseStateNum: 0, NewStateNum: 1, Data: []byte("primer")})
363	433		if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: primer}); err != nil {
364	434		t.Fatal(err)
365	435		}
	436	+	awaitSignal(t, primerWritten, "primer write")
366	437		primerDrained := false
	438	+	awaitingPrimerACK := false
367	439		for range 40 {
368	440		clk.advance(25 * time.Millisecond)
369		-	forward.flush(bPC)
370		-	reverse.flush(aPC)
371		-	time.Sleep(100 * time.Microsecond)
	441	+	if awaitingPrimerACK {
	442	+	awaitSignal(t, primerAcknowledged, "primer ACK write")
	443	+	awaitingPrimerACK = false
	444	+	}
	445	+	if flushAndAwait(t, forward, bPC, processed) > 0 {
	446	+	awaitSignal(t, primerACKScheduled, "primer ACK scheduling")
	447	+	awaitingPrimerACK = true
	448	+	}
	449	+	flushAndAwait(t, reverse, aPC, processed)
372	450		a.mu.Lock()
373	451		primerDrained = len(a.pending) == 0
374	452		a.mu.Unlock()

@@ -379,9 +457,22 @@ func TestTransportFloodClassification(t *testing.T) {
379	457		if !primerDrained {
380	458		t.Fatal("clean RTT primer did not drain")
381	459		}
	460	+	<-ackQueued
	461	+	<-ackScheduled
382	462		a.mu.Lock()
383	463		initialCwnd := a.congestion.cwndBytes
384	464		a.mu.Unlock()
	465	+	dataPaceStarted := make(chan struct{}, 128)
	466	+	dataWritten := make(chan struct{}, 1)
	467	+	a.mu.Lock()
	468	+	a.beforeDataPace = func() { dataPaceStarted <- struct{}{} }
	469	+	a.mu.Unlock()
	470	+	aPC.addAfterWrite(func() {
	471	+	select {
	472	+	case dataWritten <- struct{}{}:
	473	+	default:
	474	+	}
	475	+	})
385	476		for state := 1; state < outputCount; state++ {
386	477		payload := ports.MarshalOutput(ports.Output{
387	478		BaseStateNum: uint64(state),

@@ -392,8 +483,11 @@ func TestTransportFloodClassification(t *testing.T) {
392	483		t.Fatal(err)
393	484		}
394	485		}
	486	+	awaitSignal(t, dataPaceStarted, "output pacing")
	487	+	awaitSignal(t, dataWritten, "first paced output write")
395	488
396	489		var maxContactAge time.Duration
	490	+	pendingACKWrites := 0
397	491		lossReducedCwnd := false
398	492		latestState := uint64(0)
399	493		probed := false

@@ -401,9 +495,28 @@ func TestTransportFloodClassification(t *testing.T) {
401	495		probeWho := ""
402	496		for range 2000 {
403	497		clk.advance(25 * time.Millisecond)
404		-	forward.flush(bPC)
405		-	reverse.flush(aPC)
406		-	time.Sleep(100 * time.Microsecond)
	498	+	for range pendingACKWrites {
	499	+	awaitSignal(t, ackWritten, "cumulative ACK write")
	500	+	}
	501	+	pendingACKWrites = 0
	502	+	flushAndAwait(t, forward, bPC, processed)
	503	+	for {
	504	+	select {
	505	+	case <-ackQueued:
	506	+	awaitSignal(t, ackScheduled, "cumulative ACK scheduling")
	507	+	pendingACKWrites++
	508	+	default:
	509	+	goto ackRecordsDrained
	510	+	}
	511	+	}
	512	+	ackRecordsDrained:
	513	+	flushAndAwait(t, reverse, aPC, processed)
	514	+	a.mu.Lock()
	515	+	queuedOutput := len(a.outputQueue) > 0
	516	+	a.mu.Unlock()
	517	+	if queuedOutput {
	518	+	awaitSignal(t, dataPaceStarted, "output pacing")
	519	+	}
407	520
408	521		if forward.queuedBytes() > queueLimit \|\| reverse.queuedBytes() > queueLimit {
409	522		t.Fatalf("queue exceeded bound: forward=%d reverse=%d limit=%d", forward.queuedBytes(), reverse.queuedBytes(), queueLimit)


@@ -94,16 +94,20 @@ func (l simulatedLink) enqueue(p fakePC, peer *fakePC, b []byte) bool {
94		return true	94		return true
95		}	95		}
96			96
97	-	func (l simulatedLink) flush(peer fakePC) {	97	+	func (l simulatedLink) flush(peer fakePC) int {
98		l.mu.Lock()	98		l.mu.Lock()
99		defer l.mu.Unlock()	99		defer l.mu.Unlock()
100		now := l.clock.Now()	100		now := l.clock.Now()
			101	+	delivered := 0
101		for len(l.queue) > 0 && !l.queue[0].due.After(now) {	102		for len(l.queue) > 0 && !l.queue[0].due.After(now) {
102		q := l.queue[0]	103		q := l.queue[0]
103		l.queue = l.queue[1:]	104		l.queue = l.queue[1:]
104		l.queueBytes -= len(q.b)	105		l.queueBytes -= len(q.b)
105	-	l.deliverLocked(peer, q.packet)	106	+	if l.deliverLocked(peer, q.packet) {
			107	+	delivered++
			108	+	}
106		}	109		}
			110	+	return delivered
107		}	111		}
108			112
109		func (l *simulatedLink) queuedBytes() int {	113		func (l *simulatedLink) queuedBytes() int {

@@ -179,6 +183,7 @@ func (p *fakePC) ReadFrom(b []byte) (int, net.Addr, error) {
179		func (p *fakePC) WriteTo(b []byte, addr net.Addr) (int, error) {	183		func (p *fakePC) WriteTo(b []byte, addr net.Addr) (int, error) {
180		p.mu.Lock()	184		p.mu.Lock()
181		closed := p.closed	185		closed := p.closed
			186	+	afterWrite := p.afterWrite
182		p.mu.Unlock()	187		p.mu.Unlock()
183		if closed {	188		if closed {
184		return 0, errors.New("closed")	189		return 0, errors.New("closed")

@@ -204,11 +209,29 @@ func (p *fakePC) WriteTo(b []byte, addr net.Addr) (int, error) {
204		}	209		}
205		peer.mu.Unlock()	210		peer.mu.Unlock()
206		}	211		}
207	-	if p.afterWrite != nil {	212	+	if afterWrite != nil {
208	-	p.afterWrite()	213	+	afterWrite()
209		}	214		}
210		return len(b), nil	215		return len(b), nil
211		}	216		}
			217	+	func (p *fakePC) replaceAfterWrite(afterWrite func()) {
			218	+	p.mu.Lock()
			219	+	defer p.mu.Unlock()
			220	+	p.afterWrite = afterWrite
			221	+	}
			222	+
			223	+	func (p *fakePC) addAfterWrite(afterWrite func()) {
			224	+	p.mu.Lock()
			225	+	defer p.mu.Unlock()
			226	+	previous := p.afterWrite
			227	+	p.afterWrite = func() {
			228	+	if previous != nil {
			229	+	previous()
			230	+	}
			231	+	afterWrite()
			232	+	}
			233	+	}
			234	+
212		func (p *fakePC) Close() error {	235		func (p *fakePC) Close() error {
213		p.mu.Lock()	236		p.mu.Lock()
214		defer p.mu.Unlock()	237		defer p.mu.Unlock()

@@ -255,6 +278,15 @@ func awaitResult[T any](t *testing.T, ch <-chan T, context string) T {
255		}	278		}
256		}	279		}
257			280
			281	+	func flushAndAwait(t testing.T, link simulatedLink, peer *fakePC, processed <-chan struct{}) int {
			282	+	t.Helper()
			283	+	delivered := link.flush(peer)
			284	+	for range delivered {
			285	+	awaitSignal(t, processed, "flushed packet processing")
			286	+	}
			287	+	return delivered
			288	+	}
			289	+
258		// TestTransportFloodClassification drives authenticated datagrams through real	290		// TestTransportFloodClassification drives authenticated datagrams through real
259		// transports. The metrics are transport state, not simulator bookkeeping.	291		// transports. The metrics are transport state, not simulator bookkeeping.
260		func TestTransportFloodClassification(t *testing.T) {	292		func TestTransportFloodClassification(t *testing.T) {

@@ -356,19 +388,65 @@ func TestTransportFloodClassification(t *testing.T) {
356		defer closeFloodTransports(a, b)	388		defer closeFloodTransports(a, b)
357		waitForManualTimers(t, clk, 6)	389		waitForManualTimers(t, clk, 6)
358		var contacted atomic.Bool	390		var contacted atomic.Bool
			391	+	processed := make(chan struct{})
			392	+	primerWritten := make(chan struct{}, 1)
			393	+	primerACKScheduled := make(chan struct{}, 1)
			394	+	primerAcknowledged := make(chan struct{}, 1)
			395	+	ackQueued := make(chan struct{}, 128)
			396	+	ackScheduled := make(chan struct{}, 128)
			397	+	ackWritten := make(chan struct{}, 128)
			398	+	aPC.replaceAfterWrite(func() {
			399	+	select {
			400	+	case primerWritten <- struct{}{}:
			401	+	default:
			402	+	}
			403	+	})
			404	+	bPC.replaceAfterWrite(func() {
			405	+	select {
			406	+	case primerAcknowledged <- struct{}{}:
			407	+	default:
			408	+	}
			409	+	select {
			410	+	case ackWritten <- struct{}{}:
			411	+	default:
			412	+	}
			413	+	})
			414	+	a.mu.Lock()
			415	+	a.afterPacketProcessed = func() { processed <- struct{}{} }
			416	+	a.mu.Unlock()
359		b.mu.Lock()	417		b.mu.Lock()
360		b.afterAuthenticatedPacket = func() { contacted.Store(true) }	418		b.afterAuthenticatedPacket = func() { contacted.Store(true) }
			419	+	b.afterPacketProcessed = func() { processed <- struct{}{} }
			420	+	b.afterACKScheduled = func() {
			421	+	select {
			422	+	case primerACKScheduled <- struct{}{}:
			423	+	default:
			424	+	}
			425	+	select {
			426	+	case ackScheduled <- struct{}{}:
			427	+	default:
			428	+	}
			429	+	}
			430	+	b.afterACKQueued = func() { ackQueued <- struct{}{} }
361		b.mu.Unlock()	431		b.mu.Unlock()
362		primer := ports.MarshalOutput(ports.Output{BaseStateNum: 0, NewStateNum: 1, Data: []byte("primer")})	432		primer := ports.MarshalOutput(ports.Output{BaseStateNum: 0, NewStateNum: 1, Data: []byte("primer")})
363		if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: primer}); err != nil {	433		if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: primer}); err != nil {
364		t.Fatal(err)	434		t.Fatal(err)
365		}	435		}
			436	+	awaitSignal(t, primerWritten, "primer write")
366		primerDrained := false	437		primerDrained := false
			438	+	awaitingPrimerACK := false
367		for range 40 {	439		for range 40 {
368		clk.advance(25 * time.Millisecond)	440		clk.advance(25 * time.Millisecond)
369	-	forward.flush(bPC)	441	+	if awaitingPrimerACK {
370	-	reverse.flush(aPC)	442	+	awaitSignal(t, primerAcknowledged, "primer ACK write")
371	-	time.Sleep(100 * time.Microsecond)	443	+	awaitingPrimerACK = false
			444	+	}
			445	+	if flushAndAwait(t, forward, bPC, processed) > 0 {
			446	+	awaitSignal(t, primerACKScheduled, "primer ACK scheduling")
			447	+	awaitingPrimerACK = true
			448	+	}
			449	+	flushAndAwait(t, reverse, aPC, processed)
372		a.mu.Lock()	450		a.mu.Lock()
373		primerDrained = len(a.pending) == 0	451		primerDrained = len(a.pending) == 0
374		a.mu.Unlock()	452		a.mu.Unlock()

@@ -379,9 +457,22 @@ func TestTransportFloodClassification(t *testing.T) {
379		if !primerDrained {	457		if !primerDrained {
380		t.Fatal("clean RTT primer did not drain")	458		t.Fatal("clean RTT primer did not drain")
381		}	459		}
			460	+	<-ackQueued
			461	+	<-ackScheduled
382		a.mu.Lock()	462		a.mu.Lock()
383		initialCwnd := a.congestion.cwndBytes	463		initialCwnd := a.congestion.cwndBytes
384		a.mu.Unlock()	464		a.mu.Unlock()
			465	+	dataPaceStarted := make(chan struct{}, 128)
			466	+	dataWritten := make(chan struct{}, 1)
			467	+	a.mu.Lock()
			468	+	a.beforeDataPace = func() { dataPaceStarted <- struct{}{} }
			469	+	a.mu.Unlock()
			470	+	aPC.addAfterWrite(func() {
			471	+	select {
			472	+	case dataWritten <- struct{}{}:
			473	+	default:
			474	+	}
			475	+	})
385		for state := 1; state < outputCount; state++ {	476		for state := 1; state < outputCount; state++ {
386		payload := ports.MarshalOutput(ports.Output{	477		payload := ports.MarshalOutput(ports.Output{
387		BaseStateNum: uint64(state),	478		BaseStateNum: uint64(state),

@@ -392,8 +483,11 @@ func TestTransportFloodClassification(t *testing.T) {
392		t.Fatal(err)	483		t.Fatal(err)
393		}	484		}
394		}	485		}
			486	+	awaitSignal(t, dataPaceStarted, "output pacing")
			487	+	awaitSignal(t, dataWritten, "first paced output write")
395			488
396		var maxContactAge time.Duration	489		var maxContactAge time.Duration
			490	+	pendingACKWrites := 0
397		lossReducedCwnd := false	491		lossReducedCwnd := false
398		latestState := uint64(0)	492		latestState := uint64(0)
399		probed := false	493		probed := false

@@ -401,9 +495,28 @@ func TestTransportFloodClassification(t *testing.T) {
401		probeWho := ""	495		probeWho := ""
402		for range 2000 {	496		for range 2000 {
403		clk.advance(25 * time.Millisecond)	497		clk.advance(25 * time.Millisecond)
404	-	forward.flush(bPC)	498	+	for range pendingACKWrites {
405	-	reverse.flush(aPC)	499	+	awaitSignal(t, ackWritten, "cumulative ACK write")
406	-	time.Sleep(100 * time.Microsecond)	500	+	}
			501	+	pendingACKWrites = 0
			502	+	flushAndAwait(t, forward, bPC, processed)
			503	+	for {
			504	+	select {
			505	+	case <-ackQueued:
			506	+	awaitSignal(t, ackScheduled, "cumulative ACK scheduling")
			507	+	pendingACKWrites++
			508	+	default:
			509	+	goto ackRecordsDrained
			510	+	}
			511	+	}
			512	+	ackRecordsDrained:
			513	+	flushAndAwait(t, reverse, aPC, processed)
			514	+	a.mu.Lock()
			515	+	queuedOutput := len(a.outputQueue) > 0
			516	+	a.mu.Unlock()
			517	+	if queuedOutput {
			518	+	awaitSignal(t, dataPaceStarted, "output pacing")
			519	+	}
407			520
408		if forward.queuedBytes() > queueLimit \|\| reverse.queuedBytes() > queueLimit {	521		if forward.queuedBytes() > queueLimit \|\| reverse.queuedBytes() > queueLimit {
409		t.Fatalf("queue exceeded bound: forward=%d reverse=%d limit=%d", forward.queuedBytes(), reverse.queuedBytes(), queueLimit)	522		t.Fatalf("queue exceeded bound: forward=%d reverse=%d limit=%d", forward.queuedBytes(), reverse.queuedBytes(), queueLimit)

M internal/adapters/dgram/output_pacer.go +1 -1


@@ -35,7 +35,7 @@ func (t *Transport) outputPaceLoop() {
35	35		copy(t.outputQueue, t.outputQueue[1:])
36	36		t.outputQueue = t.outputQueue[:len(t.outputQueue)-1]
37	37		t.mu.Unlock()
38		-	if err := t.queueDataJob(dataSendJob(q)); err != nil {
	38	+	if err := t.queueDataJob(q); err != nil {
39	39		t.removeQueuedPending([]queuedSend{q}, err)
40	40		t.outboundMu.Unlock()
41	41		return


@@ -35,7 +35,7 @@ func (t *Transport) outputPaceLoop() {
35		copy(t.outputQueue, t.outputQueue[1:])	35		copy(t.outputQueue, t.outputQueue[1:])
36		t.outputQueue = t.outputQueue[:len(t.outputQueue)-1]	36		t.outputQueue = t.outputQueue[:len(t.outputQueue)-1]
37		t.mu.Unlock()	37		t.mu.Unlock()
38	-	if err := t.queueDataJob(dataSendJob(q)); err != nil {	38	+	if err := t.queueDataJob(q); err != nil {
39		t.removeQueuedPending([]queuedSend{q}, err)	39		t.removeQueuedPending([]queuedSend{q}, err)
40		t.outboundMu.Unlock()	40		t.outboundMu.Unlock()
41		return	41		return

M internal/adapters/dgram/transport.go +27 -2


@@ -216,6 +216,15 @@ type Transport struct {
216	216		writeMu sync.Mutex
217	217		// beforeDataPace is a test synchronization hook. It is read under mu.
218	218		beforeDataPace func()
	219	+	// afterPacketProcessed is a test synchronization hook called after an
	220	+	// authenticated packet has been fully processed. It is read under mu.
	221	+	afterPacketProcessed func()
	222	+	// afterACKScheduled is a test synchronization hook called after the ACK
	223	+	// scheduler handles a wake. It is read under mu.
	224	+	afterACKScheduled func()
	225	+	// afterACKQueued is a test synchronization hook called after an ACK is
	226	+	// queued. It is read under mu.
	227	+	afterACKQueued func()
219	228		// outboundMu orders first transmissions and paced batches. Retransmits use
220	229		// writeMu directly because they do not establish new sequence order.
221	230		outboundMu sync.Mutex

@@ -774,14 +783,26 @@ func (t *Transport) readLoop(pc net.PacketConn) {
774	783		}
775	784		if err != nil {
776	785		t.notifyMalformedFragment()
	786	+	t.notifyPacketProcessed()
777	787		continue
778	788		}
779	789		if !complete {
	790	+	t.notifyPacketProcessed()
780	791		continue
781	792		}
782	793		t.recordCompleteRecord()
783	794		t.handleRecord(payload)
784	795		t.checkSilence()
	796	+	t.notifyPacketProcessed()
	797	+	}
	798	+	}
	799	+
	800	+	func (t *Transport) notifyPacketProcessed() {
	801	+	t.mu.Lock()
	802	+	afterPacketProcessed := t.afterPacketProcessed
	803	+	t.mu.Unlock()
	804	+	if afterPacketProcessed != nil {
	805	+	afterPacketProcessed()
785	806		}
786	807		}
787	808

@@ -1266,13 +1287,17 @@ func (t *Transport) hopPacketConnOnce(generation uint64) {
1266	1287		t.hopGeneration = generation
1267	1288		t.mu.Unlock()
1268	1289		pc, err := rebind(old)
	1290	+
	1291	+	// Serialize the swap with writes. writePacket takes writeMu before mu, so
	1292	+	// preserve that order here while retiring old.
	1293	+	t.writeMu.Lock()
	1294	+	defer t.writeMu.Unlock()
	1295	+	t.mu.Lock()
1269	1296		if err != nil {
1270		-	t.mu.Lock()
1271	1297		t.rollbackHopLocked(generation)
1272	1298		t.mu.Unlock()
1273	1299		return
1274	1300		}
1275		-	t.mu.Lock()
1276	1301		if t.closed \|\| t.pc != old \|\| t.health.generation != generation \|\| t.linkState != ports.LinkStateProbing {
1277	1302		t.rollbackHopLocked(generation)
1278	1303		t.mu.Unlock()

 	writeMu sync.Mutex
 	// beforeDataPace is a test synchronization hook. It is read under mu.
 	beforeDataPace func()
+	// afterPacketProcessed is a test synchronization hook called after an
+	// authenticated packet has been fully processed. It is read under mu.
+	afterPacketProcessed func()
+	// afterACKScheduled is a test synchronization hook called after the ACK
+	// scheduler handles a wake. It is read under mu.
+	afterACKScheduled func()
+	// afterACKQueued is a test synchronization hook called after an ACK is
+	// queued. It is read under mu.
+	afterACKQueued func()
 	// outboundMu orders first transmissions and paced batches. Retransmits use
 	// writeMu directly because they do not establish new sequence order.
 	outboundMu sync.Mutex
 		}
 		if err != nil {
 			t.notifyMalformedFragment()
+			t.notifyPacketProcessed()
 			continue
 		}
 		if !complete {
+			t.notifyPacketProcessed()
 			continue
 		}
 		t.recordCompleteRecord()
 		t.handleRecord(payload)
 		t.checkSilence()
+		t.notifyPacketProcessed()
+	}
+}
+func (t *Transport) notifyPacketProcessed() {
+	t.mu.Lock()
+	afterPacketProcessed := t.afterPacketProcessed
+	t.mu.Unlock()
+	if afterPacketProcessed != nil {
+		afterPacketProcessed()
 	}
 }
 	t.hopGeneration = generation
 	t.mu.Unlock()
 	pc, err := rebind(old)
+	// Serialize the swap with writes. writePacket takes writeMu before mu, so
+	// preserve that order here while retiring old.
+	t.writeMu.Lock()
+	defer t.writeMu.Unlock()
+	t.mu.Lock()
 	if err != nil {
-		t.mu.Lock()
 		t.rollbackHopLocked(generation)
 		t.mu.Unlock()
 		return
 	}
-	t.mu.Lock()
 	if t.closed || t.pc != old || t.health.generation != generation || t.linkState != ports.LinkStateProbing {
 		t.rollbackHopLocked(generation)
 		t.mu.Unlock()

M internal/adapters/dgram/transport_test.go +5 -1


@@ -2117,8 +2117,12 @@ func TestLinkHealthHeartbeatControlCannotMaskPendingACKStall(t *testing.T) {
2117	2117		t.Fatal(marshalErr)
2118	2118		}
2119	2119		tr.mu.Lock()
2120		-	pc := tr.pc.(*fakePC)
	2120	+	actual := tr.pc
	2121	+	pc, ok := actual.(*fakePC)
2121	2122		tr.mu.Unlock()
	2123	+	if !ok {
	2124	+	t.Fatalf("transport packet conn = %T, want *fakePC", actual)
	2125	+	}
2122	2126		pc.in <- packet{b: codec.Seal(2, counter, raw, nil), addr: testAddr("b")}
2123	2127		select {
2124	2128		case <-authenticated:

 			t.Fatal(marshalErr)
 		}
 		tr.mu.Lock()
-		pc := tr.pc.(*fakePC)
+		actual := tr.pc
+		pc, ok := actual.(*fakePC)
 		tr.mu.Unlock()
+		if !ok {
+			t.Fatalf("transport packet conn = %T, want *fakePC", actual)
+		}
 		pc.in <- packet{b: codec.Seal(2, counter, raw, nil), addr: testAddr("b")}
 		select {
 		case <-authenticated:

M internal/usecase/daemon/output_state_stream.go +3 -3


@@ -22,10 +22,10 @@ type outputStateStream struct {
22	22		maxOutstanding uint64
23	23		}
24	24
25		-	func newOutputStateStream(max ...uint8) *outputStateStream {
	25	+	func newOutputStateStream(windowSize ...uint8) *outputStateStream {
26	26		window := uint8(maxUnackedOutputStates)
27		-	if len(max) > 0 {
28		-	window = normalizeOutputWindow(max[0])
	27	+	if len(windowSize) > 0 {
	28	+	window = normalizeOutputWindow(windowSize[0])
29	29		}
30	30		return &outputStateStream{
31	31		renderer: renderer.New(renderer.Capabilities{}),


@@ -22,10 +22,10 @@ type outputStateStream struct {
22		maxOutstanding uint64	22		maxOutstanding uint64
23		}	23		}
24			24
25	-	func newOutputStateStream(max ...uint8) *outputStateStream {	25	+	func newOutputStateStream(windowSize ...uint8) *outputStateStream {
26		window := uint8(maxUnackedOutputStates)	26		window := uint8(maxUnackedOutputStates)
27	-	if len(max) > 0 {	27	+	if len(windowSize) > 0 {
28	-	window = normalizeOutputWindow(max[0])	28	+	window = normalizeOutputWindow(windowSize[0])
29		}	29		}
30		return &outputStateStream{	30		return &outputStateStream{
31		renderer: renderer.New(renderer.Capabilities{}),	31		renderer: renderer.New(renderer.Capabilities{}),

M scripts/debug-remote-attach_test.sh +3 -3


@@ -2,13 +2,13 @@
2	2		set -euo pipefail
3	3
4	4		repo_root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
5		-	help="$($repo_root/scripts/debug-remote-attach.sh --help 2>&1)"
	5	+	help="$("$repo_root/scripts/debug-remote-attach.sh" --help 2>&1)"
6	6		grep -Fq -- '--udp-health' <<<"$help"
7	7
8	8		work="$(mktemp -d)"
9	9		trap 'rm -rf "$work"' EXIT
10	10		mkdir -p "$work/bin" "$work/state/vev"
11		-	printf 'udp health peer=192.0.2.1 home=%s\n' "$HOME" >"$work/state/vev/vev-test.log"
	11	+	printf 'udp health peer=192.0.2.1 home=%s/local\n' "$HOME" >"$work/state/vev/vev-test.log"
12	12		cat >"$work/bin/ssh" <<'SSH'
13	13		#!/usr/bin/env bash
14	14		printf 'udp transport health peer=198.51.100.2 key=ssh-ed25519\nskmem:(r0,rb1,t0,tb1,f0,w0,o0,bl0,d0)\n'

@@ -22,7 +22,7 @@ test -n "$bundle"
22	22		test -f "$bundle/local-after/udp-transport-health.out"
23	23		test -f "$bundle/remote-after/udp-socket-memory.out"
24	24		grep -Fq "$bundle/remote-after/udp-socket-memory.out" "$work/run.out"
25		-	if grep -R --exclude='*.raw' -E 'alice@mobile\.example\.com\|192\.0\.2\.1\|198\.51\.100\.2\|ssh-ed25519' "$bundle"; then
	25	+	if grep -R --exclude='*.raw' -F -e 'alice@mobile.example.com' -e '192.0.2.1' -e '198.51.100.2' -e 'ssh-ed25519' -e "$HOME/local" "$bundle"; then
26	26		echo 'unredacted diagnostic value found' >&2
27	27		exit 1
28	28		fi


@@ -2,13 +2,13 @@
2		set -euo pipefail	2		set -euo pipefail
3			3
4		repo_root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"	4		repo_root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
5	-	help="$($repo_root/scripts/debug-remote-attach.sh --help 2>&1)"	5	+	help="$("$repo_root/scripts/debug-remote-attach.sh" --help 2>&1)"
6		grep -Fq -- '--udp-health' <<<"$help"	6		grep -Fq -- '--udp-health' <<<"$help"
7			7
8		work="$(mktemp -d)"	8		work="$(mktemp -d)"
9		trap 'rm -rf "$work"' EXIT	9		trap 'rm -rf "$work"' EXIT
10		mkdir -p "$work/bin" "$work/state/vev"	10		mkdir -p "$work/bin" "$work/state/vev"
11	-	printf 'udp health peer=192.0.2.1 home=%s\n' "$HOME" >"$work/state/vev/vev-test.log"	11	+	printf 'udp health peer=192.0.2.1 home=%s/local\n' "$HOME" >"$work/state/vev/vev-test.log"
12		cat >"$work/bin/ssh" <<'SSH'	12		cat >"$work/bin/ssh" <<'SSH'
13		#!/usr/bin/env bash	13		#!/usr/bin/env bash
14		printf 'udp transport health peer=198.51.100.2 key=ssh-ed25519\nskmem:(r0,rb1,t0,tb1,f0,w0,o0,bl0,d0)\n'	14		printf 'udp transport health peer=198.51.100.2 key=ssh-ed25519\nskmem:(r0,rb1,t0,tb1,f0,w0,o0,bl0,d0)\n'

@@ -22,7 +22,7 @@ test -n "$bundle"
22		test -f "$bundle/local-after/udp-transport-health.out"	22		test -f "$bundle/local-after/udp-transport-health.out"
23		test -f "$bundle/remote-after/udp-socket-memory.out"	23		test -f "$bundle/remote-after/udp-socket-memory.out"
24		grep -Fq "$bundle/remote-after/udp-socket-memory.out" "$work/run.out"	24		grep -Fq "$bundle/remote-after/udp-socket-memory.out" "$work/run.out"
25	-	if grep -R --exclude='*.raw' -E 'alice@mobile\.example\.com\|192\.0\.2\.1\|198\.51\.100\.2\|ssh-ed25519' "$bundle"; then	25	+	if grep -R --exclude='*.raw' -F -e 'alice@mobile.example.com' -e '192.0.2.1' -e '198.51.100.2' -e 'ssh-ed25519' -e "$HOME/local" "$bundle"; then
26		echo 'unredacted diagnostic value found' >&2	26		echo 'unredacted diagnostic value found' >&2
27		exit 1	27		exit 1
28		fi	28		fi