fix(dgram): synchronize control writes with rebind

commit: 1956fce7836fab2dd83634f43db202dade23db36
author: brice · 3mo ago · 2026-07-15 05:12 UTC
parent: 8661704
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

4 files · +161 -3

A internal/adapters/dgram/ack_rebind_test.go +143 -0

@@ -0,0 +1,143 @@
	1	+	package dgram
	2	+
	3	+	import (
	4	+	"net"
	5	+	"sync"
	6	+	"testing"
	7	+	"time"
	8	+
	9	+	"github.com/bnema/vev/internal/ports"
	10	+	)
	11	+
	12	+	// gatedControlPC stops the first control write after sendControl has selected
	13	+	// its socket and installed its deadline. Close is separately gated so the test
	14	+	// can observe whether a hop attempts to retire that socket too early.
	15	+	type gatedControlPC struct {
	16	+	*fakePC
	17	+	entered chan struct{}
	18	+	releaseWrite chan struct{}
	19	+	closeStarted chan struct{}
	20	+	releaseClose chan struct{}
	21	+	gateWrite sync.Once
	22	+	gateClose sync.Once
	23	+	}
	24	+
	25	+	func (p *gatedControlPC) SetWriteDeadline(time.Time) error {
	26	+	p.gateWrite.Do(func() {
	27	+	close(p.entered)
	28	+	<-p.releaseWrite
	29	+	})
	30	+	return nil
	31	+	}
	32	+
	33	+	func (p *gatedControlPC) Close() error {
	34	+	p.gateClose.Do(func() {
	35	+	close(p.closeStarted)
	36	+	<-p.releaseClose
	37	+	})
	38	+	return p.fakePC.Close()
	39	+	}
	40	+
	41	+	func TestACKWriteCompletesBeforeRebindRetiresPacketConn(t *testing.T) {
	42	+	old, bPC := newPair()
	43	+	gated := &gatedControlPC{
	44	+	fakePC: old,
	45	+	entered: make(chan struct{}),
	46	+	releaseWrite: make(chan struct{}),
	47	+	closeStarted: make(chan struct{}),
	48	+	releaseClose: make(chan struct{}),
	49	+	}
	50	+	var releaseWrite, releaseClose sync.Once
	51	+	release := func() {
	52	+	releaseWrite.Do(func() { close(gated.releaseWrite) })
	53	+	releaseClose.Do(func() { close(gated.releaseClose) })
	54	+	}
	55	+
	56	+	var fresh *fakePC
	57	+	a, err := NewTransportWithOptions(gated, bPC.addr, key(), 1, 2, Options{
	58	+	ResendAfter: time.Hour,
	59	+	Heartbeat: time.Hour,
	60	+	RebindPacketConn: func(net.PacketConn) (net.PacketConn, error) {
	61	+	fresh = &fakePC{
	62	+	addr: testAddr("a2"),
	63	+	in: make(chan packet, 10),
	64	+	read: make(chan struct{}, 10),
	65	+	peers: map[string]*fakePC{"b": bPC},
	66	+	}
	67	+	bPC.peers["a2"] = fresh
	68	+	return fresh, nil
	69	+	},
	70	+	})
	71	+	if err != nil {
	72	+	t.Fatal(err)
	73	+	}
	74	+	b, err := NewTransportWithOptions(bPC, old.addr, key(), 2, 1, Options{
	75	+	ResendAfter: time.Hour,
	76	+	Heartbeat: time.Hour,
	77	+	})
	78	+	if err != nil {
	79	+	release()
	80	+	_ = a.Close()
	81	+	t.Fatal(err)
	82	+	}
	83	+	defer func() {
	84	+	release()
	85	+	_ = a.Close()
	86	+	_ = b.Close()
	87	+	}()
	88	+
	89	+	// A decoded ACK clears this pending record, proving delivery to the peer,
	90	+	// rather than merely observing a write on the old PacketConn.
	91	+	b.mu.Lock()
	92	+	b.pending[7] = &pending{}
	93	+	processed := make(chan struct{}, 1)
	94	+	b.afterPacketProcessed = func() { processed <- struct{}{} }
	95	+	b.mu.Unlock()
	96	+
	97	+	a.mu.Lock()
	98	+	a.linkState = ports.LinkStateProbing
	99	+	a.health.generation = 1
	100	+	a.mu.Unlock()
	101	+	controlDone := make(chan error, 1)
	102	+	go func() { controlDone <- a.sendControl(recAck, 7) }()
	103	+	awaitSignal(t, gated.entered, "ACK control write deadline")
	104	+
	105	+	hopped := make(chan struct{})
	106	+	go func() {
	107	+	a.hopPacketConnOnce(1)
	108	+	close(hopped)
	109	+	}()
	110	+
	111	+	// A hop must wait for the control write, even though this ACK does not wait
	112	+	// for data's writeMu. Before the write is released it must not retire old.
	113	+	select {
	114	+	case <-gated.closeStarted:
	115	+	t.Fatal("rebind began closing packet conn during control write")
	116	+	case <-hopped:
	117	+	t.Fatal("rebind completed during control write")
	118	+	case <-time.After(25 * time.Millisecond):
	119	+	}
	120	+
	121	+	releaseWrite.Do(func() { close(gated.releaseWrite) })
	122	+	if err := awaitResult(t, controlDone, "ACK control write"); err != nil {
	123	+	t.Fatalf("ACK control write: %v", err)
	124	+	}
	125	+	awaitSignal(t, processed, "decoded ACK delivery")
	126	+	b.mu.Lock()
	127	+	_, pending := b.pending[7]
	128	+	b.mu.Unlock()
	129	+	if pending {
	130	+	t.Fatal("decoded ACK did not clear peer pending record")
	131	+	}
	132	+
	133	+	awaitSignal(t, gated.closeStarted, "rebind retires old packet conn")
	134	+	releaseClose.Do(func() { close(gated.releaseClose) })
	135	+	select {
	136	+	case <-hopped:
	137	+	case <-time.After(time.Second):
	138	+	t.Fatal("rebind did not complete after control write")
	139	+	}
	140	+	if fresh == nil {
	141	+	t.Fatal("rebind did not install fresh packet conn")
	142	+	}
	143	+	}

@@ -0,0 +1,143 @@
	1	+	package dgram
	2	+
	3	+	import (
	4	+	"net"
	5	+	"sync"
	6	+	"testing"
	7	+	"time"
	8	+
	9	+	"github.com/bnema/vev/internal/ports"
	10	+	)
	11	+
	12	+	// gatedControlPC stops the first control write after sendControl has selected
	13	+	// its socket and installed its deadline. Close is separately gated so the test
	14	+	// can observe whether a hop attempts to retire that socket too early.
	15	+	type gatedControlPC struct {
	16	+	*fakePC
	17	+	entered chan struct{}
	18	+	releaseWrite chan struct{}
	19	+	closeStarted chan struct{}
	20	+	releaseClose chan struct{}
	21	+	gateWrite sync.Once
	22	+	gateClose sync.Once
	23	+	}
	24	+
	25	+	func (p *gatedControlPC) SetWriteDeadline(time.Time) error {
	26	+	p.gateWrite.Do(func() {
	27	+	close(p.entered)
	28	+	<-p.releaseWrite
	29	+	})
	30	+	return nil
	31	+	}
	32	+
	33	+	func (p *gatedControlPC) Close() error {
	34	+	p.gateClose.Do(func() {
	35	+	close(p.closeStarted)
	36	+	<-p.releaseClose
	37	+	})
	38	+	return p.fakePC.Close()
	39	+	}
	40	+
	41	+	func TestACKWriteCompletesBeforeRebindRetiresPacketConn(t *testing.T) {
	42	+	old, bPC := newPair()
	43	+	gated := &gatedControlPC{
	44	+	fakePC: old,
	45	+	entered: make(chan struct{}),
	46	+	releaseWrite: make(chan struct{}),
	47	+	closeStarted: make(chan struct{}),
	48	+	releaseClose: make(chan struct{}),
	49	+	}
	50	+	var releaseWrite, releaseClose sync.Once
	51	+	release := func() {
	52	+	releaseWrite.Do(func() { close(gated.releaseWrite) })
	53	+	releaseClose.Do(func() { close(gated.releaseClose) })
	54	+	}
	55	+
	56	+	var fresh *fakePC
	57	+	a, err := NewTransportWithOptions(gated, bPC.addr, key(), 1, 2, Options{
	58	+	ResendAfter: time.Hour,
	59	+	Heartbeat: time.Hour,
	60	+	RebindPacketConn: func(net.PacketConn) (net.PacketConn, error) {
	61	+	fresh = &fakePC{
	62	+	addr: testAddr("a2"),
	63	+	in: make(chan packet, 10),
	64	+	read: make(chan struct{}, 10),
	65	+	peers: map[string]*fakePC{"b": bPC},
	66	+	}
	67	+	bPC.peers["a2"] = fresh
	68	+	return fresh, nil
	69	+	},
	70	+	})
	71	+	if err != nil {
	72	+	t.Fatal(err)
	73	+	}
	74	+	b, err := NewTransportWithOptions(bPC, old.addr, key(), 2, 1, Options{
	75	+	ResendAfter: time.Hour,
	76	+	Heartbeat: time.Hour,
	77	+	})
	78	+	if err != nil {
	79	+	release()
	80	+	_ = a.Close()
	81	+	t.Fatal(err)
	82	+	}
	83	+	defer func() {
	84	+	release()
	85	+	_ = a.Close()
	86	+	_ = b.Close()
	87	+	}()
	88	+
	89	+	// A decoded ACK clears this pending record, proving delivery to the peer,
	90	+	// rather than merely observing a write on the old PacketConn.
	91	+	b.mu.Lock()
	92	+	b.pending[7] = &pending{}
	93	+	processed := make(chan struct{}, 1)
	94	+	b.afterPacketProcessed = func() { processed <- struct{}{} }
	95	+	b.mu.Unlock()
	96	+
	97	+	a.mu.Lock()
	98	+	a.linkState = ports.LinkStateProbing
	99	+	a.health.generation = 1
	100	+	a.mu.Unlock()
	101	+	controlDone := make(chan error, 1)
	102	+	go func() { controlDone <- a.sendControl(recAck, 7) }()
	103	+	awaitSignal(t, gated.entered, "ACK control write deadline")
	104	+
	105	+	hopped := make(chan struct{})
	106	+	go func() {
	107	+	a.hopPacketConnOnce(1)
	108	+	close(hopped)
	109	+	}()
	110	+
	111	+	// A hop must wait for the control write, even though this ACK does not wait
	112	+	// for data's writeMu. Before the write is released it must not retire old.
	113	+	select {
	114	+	case <-gated.closeStarted:
	115	+	t.Fatal("rebind began closing packet conn during control write")
	116	+	case <-hopped:
	117	+	t.Fatal("rebind completed during control write")
	118	+	case <-time.After(25 * time.Millisecond):
	119	+	}
	120	+
	121	+	releaseWrite.Do(func() { close(gated.releaseWrite) })
	122	+	if err := awaitResult(t, controlDone, "ACK control write"); err != nil {
	123	+	t.Fatalf("ACK control write: %v", err)
	124	+	}
	125	+	awaitSignal(t, processed, "decoded ACK delivery")
	126	+	b.mu.Lock()
	127	+	_, pending := b.pending[7]
	128	+	b.mu.Unlock()
	129	+	if pending {
	130	+	t.Fatal("decoded ACK did not clear peer pending record")
	131	+	}
	132	+
	133	+	awaitSignal(t, gated.closeStarted, "rebind retires old packet conn")
	134	+	releaseClose.Do(func() { close(gated.releaseClose) })
	135	+	select {
	136	+	case <-hopped:
	137	+	case <-time.After(time.Second):
	138	+	t.Fatal("rebind did not complete after control write")
	139	+	}
	140	+	if fresh == nil {
	141	+	t.Fatal("rebind did not install fresh packet conn")
	142	+	}
	143	+	}

M internal/adapters/dgram/control.go +6 -0


@@ -176,6 +176,12 @@ func (t *Transport) ackSendLoop() {
176	176		}
177	177
178	178		func (t *Transport) sendControl(kind byte, id uint64) error {
	179	+	// Do not take writeMu: control must remain available while data writes are
	180	+	// blocked. The read lock instead keeps the selected connection alive until
	181	+	// every control fragment has been written.
	182	+	t.controlConnMu.RLock()
	183	+	defer t.controlConnMu.RUnlock()
	184	+
179	185		var payload [9]byte
180	186		payload[0] = kind
181	187		binary.BigEndian.PutUint64(payload[1:], id)


@@ -176,6 +176,12 @@ func (t *Transport) ackSendLoop() {
176	}	176		}
177		177
178	func (t *Transport) sendControl(kind byte, id uint64) error {	178		func (t *Transport) sendControl(kind byte, id uint64) error {
		179	+	// Do not take writeMu: control must remain available while data writes are
		180	+	// blocked. The read lock instead keeps the selected connection alive until
		181	+	// every control fragment has been written.
		182	+	t.controlConnMu.RLock()
		183	+	defer t.controlConnMu.RUnlock()
		184	+
179	var payload [9]byte	185		var payload [9]byte
180	payload[0] = kind	186		payload[0] = kind
181	binary.BigEndian.PutUint64(payload[1:], id)	187		binary.BigEndian.PutUint64(payload[1:], id)

M internal/adapters/dgram/transport.go +6 -1


@@ -224,7 +224,12 @@ type Transport struct {
224	224		in chan ports.Frame
225	225		done chan struct{}
226	226
227		-	writeMu sync.Mutex
	227	+	// Connection replacement takes writeMu, controlConnMu, then mu. Data writes
	228	+	// take only writeMu; control writes take controlConnMu's read lock, then mu.
	229	+	// This keeps control independent of data writes while preventing a rebind
	230	+	// from retiring its PacketConn during a control write.
	231	+	writeMu sync.Mutex
	232	+	controlConnMu sync.RWMutex
228	233		// beforeDataPace is a test synchronization hook. It is read under mu.
229	234		beforeDataPace func()
230	235		// afterDataPaceWait observes a data pacer's next manual-clock deadline.

 	in     chan ports.Frame
 	done   chan struct{}
-	writeMu sync.Mutex
+	// Connection replacement takes writeMu, controlConnMu, then mu. Data writes
+	// take only writeMu; control writes take controlConnMu's read lock, then mu.
+	// This keeps control independent of data writes while preventing a rebind
+	// from retiring its PacketConn during a control write.
+	writeMu       sync.Mutex
+	controlConnMu sync.RWMutex
 	// beforeDataPace is a test synchronization hook. It is read under mu.
 	beforeDataPace func()
 	// afterDataPaceWait observes a data pacer's next manual-clock deadline.

M internal/adapters/dgram/transport_health.go +6 -2


@@ -60,10 +60,14 @@ func (t *Transport) hopPacketConnOnce(generation uint64) {
60	60		t.mu.Unlock()
61	61		pc, err := rebind(old)
62	62
63		-	// Serialize the swap with writes. writePacket takes writeMu before mu, so
64		-	// preserve that order here while retiring old.
	63	+	// Lock order is writeMu -> controlConnMu -> mu. Data writes take only
	64	+	// writeMu, while control writes hold controlConnMu's read lock without
	65	+	// writeMu; therefore a hop waits for active control writes but does not make
	66	+	// them wait for a blocked data write.
65	67		t.writeMu.Lock()
66	68		defer t.writeMu.Unlock()
	69	+	t.controlConnMu.Lock()
	70	+	defer t.controlConnMu.Unlock()
67	71		t.mu.Lock()
68	72		if err != nil {
69	73		t.rollbackHopLocked(generation)

 	t.mu.Unlock()
 	pc, err := rebind(old)
-	// Serialize the swap with writes. writePacket takes writeMu before mu, so
+	// Lock order is writeMu -> controlConnMu -> mu. Data writes take only
-	// preserve that order here while retiring old.
+	// writeMu, while control writes hold controlConnMu's read lock without
+	// writeMu; therefore a hop waits for active control writes but does not make
+	// them wait for a blocked data write.
 	t.writeMu.Lock()
 	defer t.writeMu.Unlock()
+	t.controlConnMu.Lock()
+	defer t.controlConnMu.Unlock()
 	t.mu.Lock()
 	if err != nil {
 		t.rollbackHopLocked(generation)