detached eac5808

test(dgram): cover UDP control-plane starvation

commit
eac5808d11f8413c106eadf0a974800bf822fdb0
author
brice · · 2026-07-10 00:27 UTC
parent
2479821
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
1 file · +122 -0
Unified
UnifiedSplit
A internal/adapters/dgram/control_plane_test.go +122 -0
@@ -0,0 +1,122 @@
1+package dgram
2+
3+import (
4+ "net"
5+ "testing"
6+ "time"
7+
8+ "github.com/bnema/vev/internal/ports"
9+ pdgram "github.com/bnema/vev/pkg/dgram"
10+)
11+
12+func TestReadLoopDoesNotWriteACKSynchronously(t *testing.T) {
13+ clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
14+ aPC, bPC := newPair()
15+ a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, Options{
16+ Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour,
17+ })
18+ if err != nil {
19+ t.Fatal(err)
20+ }
21+ b, err := NewTransportWithOptions(bPC, aPC.addr, key(), 2, 1, Options{
22+ Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour,
23+ })
24+ if err != nil {
25+ _ = a.Close()
26+ t.Fatal(err)
27+ }
28+ releaseACK := make(chan struct{})
29+ defer close(releaseACK)
30+ defer closeFloodTransports(a, b)
31+
32+ ackStarted := make(chan struct{}, 1)
33+ aPC.drop = func(_ []byte, _ net.Addr) bool {
34+ ackStarted <- struct{}{}
35+ <-releaseACK
36+ return true
37+ }
38+ contacts := make(chan struct{}, 2)
39+ a.mu.Lock()
40+ a.afterAuthenticatedPacket = func() { contacts <- struct{}{} }
41+ // Keep diagnostics local to this test so each snapshot has an explicit
42+ // causative event instead of depending on observer scheduling.
43+ a.diagnosticCh = make(chan Diagnostic, 1)
44+ a.mu.Unlock()
45+
46+ if err := b.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("first")}); err != nil {
47+ t.Fatal(err)
48+ }
49+ awaitSignal(t, contacts, "first authenticated record")
50+ awaitSignal(t, ackStarted, "blocked ACK")
51+
52+ clk.advance(time.Second)
53+ if err := b.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("second")}); err != nil {
54+ t.Fatal(err)
55+ }
56+ awaitSignal(t, contacts, "second authenticated record while ACK write is blocked")
57+ a.emitDiagnostic()
58+ diagnostic := awaitResult(t, a.diagnosticCh, "second-record diagnostic")
59+ if diagnostic.SinceCompleteRecord != 0 {
60+ t.Fatalf("second record age=%v, want 0 before ACK release", diagnostic.SinceCompleteRecord)
61+ }
62+}
63+
64+func TestHeartbeatContinuesDuringRetransmitStorm(t *testing.T) {
65+ clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
66+ aPC, bPC := newPair()
67+ const heartbeat = 11 * time.Millisecond
68+ a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, Options{
69+ Clock: clk, MTU: 128, ResendAfter: 10 * time.Millisecond, Heartbeat: heartbeat,
70+ })
71+ if err != nil {
72+ t.Fatal(err)
73+ }
74+ defer func() { _ = a.Close() }()
75+ for range 2 {
76+ awaitSignal(t, clk.timerCreated, "resend and heartbeat timer creation")
77+ }
78+
79+ retransmitStarted := make(chan struct{}, 1)
80+ releaseRetransmit := make(chan struct{})
81+ defer close(releaseRetransmit)
82+ heartbeats := make(chan struct{}, 2)
83+ aPC.drop = func(pkt []byte, _ net.Addr) bool {
84+ _, raw, err := a.codec.Open(pkt, a.sendDir, nil, nil)
85+ if err != nil {
86+ t.Errorf("open outbound packet: %v", err)
87+ return true
88+ }
89+ frag, err := pdgram.UnmarshalFragment(raw)
90+ if err != nil {
91+ t.Errorf("unmarshal outbound fragment: %v", err)
92+ return true
93+ }
94+ if frag.Count == 1 && len(frag.Data) == 9 && frag.Data[0] == recProbe {
95+ heartbeats <- struct{}{}
96+ return true
97+ }
98+ select {
99+ case retransmitStarted <- struct{}{}:
100+ <-releaseRetransmit
101+ default:
102+ }
103+ return true
104+ }
105+
106+ now := clk.Now()
107+ a.mu.Lock()
108+ for seq := uint64(1); seq <= 64; seq++ {
109+ a.pending[seq] = &pending{
110+ frame: ports.Frame{Type: ports.MsgInput, Payload: make([]byte, 24*a.mtu)},
111+ enqueued: now.Add(-time.Second), first: now.Add(-time.Second), last: now.Add(-time.Second),
112+ }
113+ }
114+ a.mu.Unlock()
115+
116+ clk.advance(10 * time.Millisecond)
117+ awaitSignal(t, retransmitStarted, "first retransmit in 64-record storm")
118+ for interval := 1; interval <= 2; interval++ {
119+ clk.advance(heartbeat)
120+ awaitSignal(t, heartbeats, "heartbeat during retransmit storm")
121+ }
122+}
@@ -0,0 +1,122 @@
1+package dgram
2+
3+import (
4+ "net"
5+ "testing"
6+ "time"
7+
8+ "github.com/bnema/vev/internal/ports"
9+ pdgram "github.com/bnema/vev/pkg/dgram"
10+)
11+
12+func TestReadLoopDoesNotWriteACKSynchronously(t *testing.T) {
13+ clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
14+ aPC, bPC := newPair()
15+ a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, Options{
16+ Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour,
17+ })
18+ if err != nil {
19+ t.Fatal(err)
20+ }
21+ b, err := NewTransportWithOptions(bPC, aPC.addr, key(), 2, 1, Options{
22+ Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour,
23+ })
24+ if err != nil {
25+ _ = a.Close()
26+ t.Fatal(err)
27+ }
28+ releaseACK := make(chan struct{})
29+ defer close(releaseACK)
30+ defer closeFloodTransports(a, b)
31+
32+ ackStarted := make(chan struct{}, 1)
33+ aPC.drop = func(_ []byte, _ net.Addr) bool {
34+ ackStarted <- struct{}{}
35+ <-releaseACK
36+ return true
37+ }
38+ contacts := make(chan struct{}, 2)
39+ a.mu.Lock()
40+ a.afterAuthenticatedPacket = func() { contacts <- struct{}{} }
41+ // Keep diagnostics local to this test so each snapshot has an explicit
42+ // causative event instead of depending on observer scheduling.
43+ a.diagnosticCh = make(chan Diagnostic, 1)
44+ a.mu.Unlock()
45+
46+ if err := b.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("first")}); err != nil {
47+ t.Fatal(err)
48+ }
49+ awaitSignal(t, contacts, "first authenticated record")
50+ awaitSignal(t, ackStarted, "blocked ACK")
51+
52+ clk.advance(time.Second)
53+ if err := b.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("second")}); err != nil {
54+ t.Fatal(err)
55+ }
56+ awaitSignal(t, contacts, "second authenticated record while ACK write is blocked")
57+ a.emitDiagnostic()
58+ diagnostic := awaitResult(t, a.diagnosticCh, "second-record diagnostic")
59+ if diagnostic.SinceCompleteRecord != 0 {
60+ t.Fatalf("second record age=%v, want 0 before ACK release", diagnostic.SinceCompleteRecord)
61+ }
62+}
63+
64+func TestHeartbeatContinuesDuringRetransmitStorm(t *testing.T) {
65+ clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
66+ aPC, bPC := newPair()
67+ const heartbeat = 11 * time.Millisecond
68+ a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, Options{
69+ Clock: clk, MTU: 128, ResendAfter: 10 * time.Millisecond, Heartbeat: heartbeat,
70+ })
71+ if err != nil {
72+ t.Fatal(err)
73+ }
74+ defer func() { _ = a.Close() }()
75+ for range 2 {
76+ awaitSignal(t, clk.timerCreated, "resend and heartbeat timer creation")
77+ }
78+
79+ retransmitStarted := make(chan struct{}, 1)
80+ releaseRetransmit := make(chan struct{})
81+ defer close(releaseRetransmit)
82+ heartbeats := make(chan struct{}, 2)
83+ aPC.drop = func(pkt []byte, _ net.Addr) bool {
84+ _, raw, err := a.codec.Open(pkt, a.sendDir, nil, nil)
85+ if err != nil {
86+ t.Errorf("open outbound packet: %v", err)
87+ return true
88+ }
89+ frag, err := pdgram.UnmarshalFragment(raw)
90+ if err != nil {
91+ t.Errorf("unmarshal outbound fragment: %v", err)
92+ return true
93+ }
94+ if frag.Count == 1 && len(frag.Data) == 9 && frag.Data[0] == recProbe {
95+ heartbeats <- struct{}{}
96+ return true
97+ }
98+ select {
99+ case retransmitStarted <- struct{}{}:
100+ <-releaseRetransmit
101+ default:
102+ }
103+ return true
104+ }
105+
106+ now := clk.Now()
107+ a.mu.Lock()
108+ for seq := uint64(1); seq <= 64; seq++ {
109+ a.pending[seq] = &pending{
110+ frame: ports.Frame{Type: ports.MsgInput, Payload: make([]byte, 24*a.mtu)},
111+ enqueued: now.Add(-time.Second), first: now.Add(-time.Second), last: now.Add(-time.Second),
112+ }
113+ }
114+ a.mu.Unlock()
115+
116+ clk.advance(10 * time.Millisecond)
117+ awaitSignal(t, retransmitStarted, "first retransmit in 64-record storm")
118+ for interval := 1; interval <= 2; interval++ {
119+ clk.advance(heartbeat)
120+ awaitSignal(t, heartbeats, "heartbeat during retransmit storm")
121+ }
122+}

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 12

Tags 11

↑↓ moveenter openesc close