fix(dgram): isolate ACK deadline scheduling

commit: be2c9fb47bd09f0c219958e64a9b27f6e5a8f842
author: brice · 3mo ago · 2026-07-10 05:53 UTC
parent: a386d4d
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +65 -8

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


@@ -67,6 +67,24 @@ func (t *Transport) takeACK() (uint64, bool) {
67	67		}
68	68
69	69		func (t *Transport) controlLoop() {
	70	+	go t.ackScheduleLoop()
	71	+	for {
	72	+	select {
	73	+	case record := <-t.control:
	74	+	err := t.sendControl(record.kind, record.id)
	75	+	if record.result != nil {
	76	+	select {
	77	+	case record.result <- err:
	78	+	default:
	79	+	}
	80	+	}
	81	+	case <-t.done:
	82	+	return
	83	+	}
	84	+	}
	85	+	}
	86	+
	87	+	func (t *Transport) ackScheduleLoop() {
70	88		var ackTimer ports.Timer
71	89		var ackTimerC <-chan time.Time
72	90		defer func() {

@@ -118,14 +136,6 @@ func (t *Transport) controlLoop() {
118	136		}
119	137
120	138		select {
121		-	case record := <-t.control:
122		-	err := t.sendControl(record.kind, record.id)
123		-	if record.result != nil {
124		-	select {
125		-	case record.result <- err:
126		-	default:
127		-	}
128		-	}
129	139		case <-t.ackWake:
130	140		scheduleACK()
131	141		case <-ackTimerC:

 }
 func (t *Transport) controlLoop() {
+	go t.ackScheduleLoop()
+	for {
+		select {
+		case record := <-t.control:
+			err := t.sendControl(record.kind, record.id)
+			if record.result != nil {
+				select {
+				case record.result <- err:
+				default:
+				}
+			}
+		case <-t.done:
+			return
+		}
+	}
+}
+func (t *Transport) ackScheduleLoop() {
 	var ackTimer ports.Timer
 	var ackTimerC <-chan time.Time
 	defer func() {
 		}
 		select {
-		case record := <-t.control:
-			err := t.sendControl(record.kind, record.id)
-			if record.result != nil {
-				select {
-				case record.result <- err:
-				default:
-				}
-			}
 		case <-t.ackWake:
 			scheduleACK()
 		case <-ackTimerC:

M internal/adapters/dgram/control_plane_test.go +47 -0


@@ -161,6 +161,53 @@ func TestCumulativeACKCoalescesForBoundedDelay(t *testing.T) {
161	161		}
162	162		}
163	163
	164	+	func TestACKDeadlineDispatchesWhileAdvisoryWriteIsBlocked(t *testing.T) {
	165	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
	166	+	aPC, bPC := newPair()
	167	+	a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, Options{
	168	+	Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour,
	169	+	})
	170	+	if err != nil {
	171	+	t.Fatal(err)
	172	+	}
	173	+	defer func() { _ = a.Close() }()
	174	+	for range 3 {
	175	+	awaitSignal(t, clk.timerCreated, "transport timer creation")
	176	+	}
	177	+
	178	+	probeStarted := make(chan struct{}, 1)
	179	+	ackStarted := make(chan struct{}, 1)
	180	+	releaseProbe := make(chan struct{})
	181	+	defer close(releaseProbe)
	182	+	aPC.drop = func(pkt []byte, _ net.Addr) bool {
	183	+	_, raw, err := a.codec.Open(pkt, a.sendDir, nil, nil)
	184	+	if err != nil {
	185	+	return true
	186	+	}
	187	+	frag, err := pdgram.UnmarshalFragment(raw)
	188	+	if err != nil \|\| frag.Count != 1 \|\| len(frag.Data) != 9 {
	189	+	return true
	190	+	}
	191	+	switch frag.Data[0] {
	192	+	case recProbe:
	193	+	probeStarted <- struct{}{}
	194	+	<-releaseProbe
	195	+	case recAck:
	196	+	ackStarted <- struct{}{}
	197	+	}
	198	+	return true
	199	+	}
	200	+
	201	+	if !a.queueControl(recProbe, 1) {
	202	+	t.Fatal("probe control queue unexpectedly full")
	203	+	}
	204	+	awaitSignal(t, probeStarted, "blocked advisory probe write")
	205	+	a.queueACK(1)
	206	+	awaitSignal(t, clk.timerCreated, "ACK coalescing timer creation")
	207	+	clk.advance(maxACKDelay)
	208	+	awaitSignal(t, ackStarted, "ACK write while advisory probe remains blocked")
	209	+	}
	210	+
164	211		func TestProbeDoesNotBlockOnDataWriter(t *testing.T) {
165	212		aPC, bPC := newPair()
166	213		a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, Options{


@@ -161,6 +161,53 @@ func TestCumulativeACKCoalescesForBoundedDelay(t *testing.T) {
161	}	161		}
162	}	162		}
163		163
		164	+	func TestACKDeadlineDispatchesWhileAdvisoryWriteIsBlocked(t *testing.T) {
		165	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
		166	+	aPC, bPC := newPair()
		167	+	a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, Options{
		168	+	Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour,
		169	+	})
		170	+	if err != nil {
		171	+	t.Fatal(err)
		172	+	}
		173	+	defer func() { _ = a.Close() }()
		174	+	for range 3 {
		175	+	awaitSignal(t, clk.timerCreated, "transport timer creation")
		176	+	}
		177	+
		178	+	probeStarted := make(chan struct{}, 1)
		179	+	ackStarted := make(chan struct{}, 1)
		180	+	releaseProbe := make(chan struct{})
		181	+	defer close(releaseProbe)
		182	+	aPC.drop = func(pkt []byte, _ net.Addr) bool {
		183	+	_, raw, err := a.codec.Open(pkt, a.sendDir, nil, nil)
		184	+	if err != nil {
		185	+	return true
		186	+	}
		187	+	frag, err := pdgram.UnmarshalFragment(raw)
		188	+	if err != nil \|\| frag.Count != 1 \|\| len(frag.Data) != 9 {
		189	+	return true
		190	+	}
		191	+	switch frag.Data[0] {
		192	+	case recProbe:
		193	+	probeStarted <- struct{}{}
		194	+	<-releaseProbe
		195	+	case recAck:
		196	+	ackStarted <- struct{}{}
		197	+	}
		198	+	return true
		199	+	}
		200	+
		201	+	if !a.queueControl(recProbe, 1) {
		202	+	t.Fatal("probe control queue unexpectedly full")
		203	+	}
		204	+	awaitSignal(t, probeStarted, "blocked advisory probe write")
		205	+	a.queueACK(1)
		206	+	awaitSignal(t, clk.timerCreated, "ACK coalescing timer creation")
		207	+	clk.advance(maxACKDelay)
		208	+	awaitSignal(t, ackStarted, "ACK write while advisory probe remains blocked")
		209	+	}
		210	+
164	func TestProbeDoesNotBlockOnDataWriter(t *testing.T) {	211		func TestProbeDoesNotBlockOnDataWriter(t *testing.T) {
165	aPC, bPC := newPair()	212		aPC, bPC := newPair()
166	a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, Options{	213		a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, Options{