detached 56d79df

test(dgram): characterize retransmit starvation under output load

commit
56d79dfc90f61fc12c381de676a6184773c698a5
author
brice · · 2026-07-10 00:08 UTC
parent
fcb2135
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
2 files · +102 -32
Unified
UnifiedSplit
M internal/adapters/dgram/network_sim_test.go +89 -28
@@ -266,55 +266,116 @@ func TestTransportFloodClassification(t *testing.T) {
266266 }
267267 })
268268
269- t.Run("real cumulative ACK loss schedules duplicate retransmits", func(t *testing.T) {
269+ t.Run("large retransmits synchronously delay heartbeat health work", func(t *testing.T) {
270270 clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
271271 aPC, bPC := newPair()
272272 inspect, err := pdgram.NewCodec(key())
273273 if err != nil {
274274 t.Fatal(err)
275275 }
276+
277+ acksDropped := make(chan struct{}, 2)
276278 bPC.drop = func(pkt []byte, _ net.Addr) bool {
277279 _, raw, err := inspect.Open(pkt, 2, nil, nil)
278280 if err != nil {
279281 return false
280282 }
281283 frag, err := pdgram.UnmarshalFragment(raw)
282- return err == nil && frag.Count == 1 && len(frag.Data) == 9 && frag.Data[0] == recAck
284+ if err != nil || frag.Count != 1 || len(frag.Data) != 9 || frag.Data[0] != recAck {
285+ return false
286+ }
287+ acksDropped <- struct{}{}
288+ return true
283289 }
284- seen := map[uint64]int{}
285- var seenMu sync.Mutex
290+
291+ retransmitBlocked := make(chan struct{}, 1)
292+ releaseRetransmit := make(chan struct{})
293+ probeSent := make(chan struct{}, 1)
294+ writes := make(map[uint64]int)
295+ var writesMu sync.Mutex
286296 aPC.drop = func(pkt []byte, _ net.Addr) bool {
287297 _, raw, err := inspect.Open(pkt, 1, nil, nil)
288- if err == nil {
289- frag, err := pdgram.UnmarshalFragment(raw)
290- if err == nil && frag.Count == 1 {
291- seq, _, _, ok := decodeData(frag.Data)
292- if ok {
293- seenMu.Lock()
294- seen[seq]++
295- seenMu.Unlock()
296- }
297- }
298+ if err != nil {
299+ return false
300+ }
301+ frag, err := pdgram.UnmarshalFragment(raw)
302+ if err != nil {
303+ return false
304+ }
305+ if frag.Count == 1 && len(frag.Data) == 9 && frag.Data[0] == recProbe {
306+ probeSent <- struct{}{}
307+ return false
308+ }
309+ if frag.Index != 0 {
310+ return false
311+ }
312+ seq, _, _, ok := decodeData(frag.Data)
313+ if !ok {
314+ return false
315+ }
316+ writesMu.Lock()
317+ writes[seq]++
318+ secondWrite := writes[seq] == 2
319+ writesMu.Unlock()
320+ if secondWrite {
321+ retransmitBlocked <- struct{}{}
322+ <-releaseRetransmit
298323 }
299324 return false
300325 }
301- a, b := newFloodTransports(t, aPC, bPC, clk, 128)
302- defer closeFloodTransports(a, b)
303- clk.advance(time.Nanosecond)
304- if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("retry")}); err != nil {
326+
327+ opts := Options{Clock: clk, MTU: 128, ResendAfter: 10 * time.Millisecond, Heartbeat: 11 * time.Millisecond}
328+ a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, opts)
329+ if err != nil {
330+ t.Fatal(err)
331+ }
332+ b, err := NewTransportWithOptions(bPC, aPC.addr, key(), 2, 1, opts)
333+ if err != nil {
334+ _ = a.Close()
305335 t.Fatal(err)
306336 }
307- floodEventually(t, func() bool { return floodState(b).record })
308- waitForManualTimers(t, clk, 4)
337+ defer closeFloodTransports(a, b)
338+ for range 4 { // Both transports install resend and heartbeat timers.
339+ <-clk.timerCreated
340+ }
341+
342+ // Two fresh, fragmented output records become overdue while their real
343+ // cumulative ACKs are lost. Disable only test pacing so the fixture can
344+ // isolate resend-loop serialization rather than pacing delay.
345+ a.dataPaceMu.Lock()
346+ a.dataPaceRemaining = 100
347+ a.dataPaceNext = clk.Now().Add(time.Hour)
348+ a.dataPaceMu.Unlock()
349+ for range 2 {
350+ a.mu.Lock()
351+ a.outputNext = time.Time{}
352+ a.mu.Unlock()
353+ if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: make([]byte, 500)}); err != nil {
354+ t.Fatal(err)
355+ }
356+ }
357+ for range 2 {
358+ <-acksDropped
359+ }
360+
309361 clk.advance(10 * time.Millisecond)
310- floodEventually(t, func() bool {
311- seenMu.Lock()
312- defer seenMu.Unlock()
313- return seen[1] >= 2
314- })
315- state := floodState(a)
316- if state.ack || state.retransmits == 0 {
317- t.Fatalf("ack=%v retransmits=%d, want lost ACK and scheduled retry", state.ack, state.retransmits)
362+ <-retransmitBlocked // resendPending is synchronously inside writePacket.
363+ clk.advance(time.Millisecond)
364+ select {
365+ case <-probeSent:
366+ t.Fatal("heartbeat probe ran while a large retransmission held the resend loop")
367+ default:
368+ }
369+ if got := floodState(a).retransmits; got < 2 {
370+ t.Fatalf("retransmits=%d, want both overdue large records selected", got)
371+ }
372+
373+ close(releaseRetransmit)
374+ <-probeSent
375+ writesMu.Lock()
376+ defer writesMu.Unlock()
377+ if writes[1] < 2 || writes[2] < 2 {
378+ t.Fatalf("writes=%v, want fresh large traffic and retransmissions for both records", writes)
318379 }
319380 })
320381 }
@@ -266,55 +266,116 @@ func TestTransportFloodClassification(t *testing.T) {
266 }266 }
267 })267 })
268 268
269- t.Run("real cumulative ACK loss schedules duplicate retransmits", func(t *testing.T) {269+ t.Run("large retransmits synchronously delay heartbeat health work", func(t *testing.T) {
270 clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))270 clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
271 aPC, bPC := newPair()271 aPC, bPC := newPair()
272 inspect, err := pdgram.NewCodec(key())272 inspect, err := pdgram.NewCodec(key())
273 if err != nil {273 if err != nil {
274 t.Fatal(err)274 t.Fatal(err)
275 }275 }
276+
277+ acksDropped := make(chan struct{}, 2)
276 bPC.drop = func(pkt []byte, _ net.Addr) bool {278 bPC.drop = func(pkt []byte, _ net.Addr) bool {
277 _, raw, err := inspect.Open(pkt, 2, nil, nil)279 _, raw, err := inspect.Open(pkt, 2, nil, nil)
278 if err != nil {280 if err != nil {
279 return false281 return false
280 }282 }
281 frag, err := pdgram.UnmarshalFragment(raw)283 frag, err := pdgram.UnmarshalFragment(raw)
282- return err == nil && frag.Count == 1 && len(frag.Data) == 9 && frag.Data[0] == recAck284+ if err != nil || frag.Count != 1 || len(frag.Data) != 9 || frag.Data[0] != recAck {
285+ return false
286+ }
287+ acksDropped <- struct{}{}
288+ return true
283 }289 }
284- seen := map[uint64]int{}290+
285- var seenMu sync.Mutex291+ retransmitBlocked := make(chan struct{}, 1)
292+ releaseRetransmit := make(chan struct{})
293+ probeSent := make(chan struct{}, 1)
294+ writes := make(map[uint64]int)
295+ var writesMu sync.Mutex
286 aPC.drop = func(pkt []byte, _ net.Addr) bool {296 aPC.drop = func(pkt []byte, _ net.Addr) bool {
287 _, raw, err := inspect.Open(pkt, 1, nil, nil)297 _, raw, err := inspect.Open(pkt, 1, nil, nil)
288- if err == nil {298+ if err != nil {
289- frag, err := pdgram.UnmarshalFragment(raw)299+ return false
290- if err == nil && frag.Count == 1 {300+ }
291- seq, _, _, ok := decodeData(frag.Data)301+ frag, err := pdgram.UnmarshalFragment(raw)
292- if ok {302+ if err != nil {
293- seenMu.Lock()303+ return false
294- seen[seq]++304+ }
295- seenMu.Unlock()305+ if frag.Count == 1 && len(frag.Data) == 9 && frag.Data[0] == recProbe {
296- }306+ probeSent <- struct{}{}
297- }307+ return false
308+ }
309+ if frag.Index != 0 {
310+ return false
311+ }
312+ seq, _, _, ok := decodeData(frag.Data)
313+ if !ok {
314+ return false
315+ }
316+ writesMu.Lock()
317+ writes[seq]++
318+ secondWrite := writes[seq] == 2
319+ writesMu.Unlock()
320+ if secondWrite {
321+ retransmitBlocked <- struct{}{}
322+ <-releaseRetransmit
298 }323 }
299 return false324 return false
300 }325 }
301- a, b := newFloodTransports(t, aPC, bPC, clk, 128)326+
302- defer closeFloodTransports(a, b)327+ opts := Options{Clock: clk, MTU: 128, ResendAfter: 10 * time.Millisecond, Heartbeat: 11 * time.Millisecond}
303- clk.advance(time.Nanosecond)328+ a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, opts)
304- if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("retry")}); err != nil {329+ if err != nil {
330+ t.Fatal(err)
331+ }
332+ b, err := NewTransportWithOptions(bPC, aPC.addr, key(), 2, 1, opts)
333+ if err != nil {
334+ _ = a.Close()
305 t.Fatal(err)335 t.Fatal(err)
306 }336 }
307- floodEventually(t, func() bool { return floodState(b).record })337+ defer closeFloodTransports(a, b)
308- waitForManualTimers(t, clk, 4)338+ for range 4 { // Both transports install resend and heartbeat timers.
339+ <-clk.timerCreated
340+ }
341+
342+ // Two fresh, fragmented output records become overdue while their real
343+ // cumulative ACKs are lost. Disable only test pacing so the fixture can
344+ // isolate resend-loop serialization rather than pacing delay.
345+ a.dataPaceMu.Lock()
346+ a.dataPaceRemaining = 100
347+ a.dataPaceNext = clk.Now().Add(time.Hour)
348+ a.dataPaceMu.Unlock()
349+ for range 2 {
350+ a.mu.Lock()
351+ a.outputNext = time.Time{}
352+ a.mu.Unlock()
353+ if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: make([]byte, 500)}); err != nil {
354+ t.Fatal(err)
355+ }
356+ }
357+ for range 2 {
358+ <-acksDropped
359+ }
360+
309 clk.advance(10 * time.Millisecond)361 clk.advance(10 * time.Millisecond)
310- floodEventually(t, func() bool {362+ <-retransmitBlocked // resendPending is synchronously inside writePacket.
311- seenMu.Lock()363+ clk.advance(time.Millisecond)
312- defer seenMu.Unlock()364+ select {
313- return seen[1] >= 2365+ case <-probeSent:
314- })366+ t.Fatal("heartbeat probe ran while a large retransmission held the resend loop")
315- state := floodState(a)367+ default:
316- if state.ack || state.retransmits == 0 {368+ }
317- t.Fatalf("ack=%v retransmits=%d, want lost ACK and scheduled retry", state.ack, state.retransmits)369+ if got := floodState(a).retransmits; got < 2 {
370+ t.Fatalf("retransmits=%d, want both overdue large records selected", got)
371+ }
372+
373+ close(releaseRetransmit)
374+ <-probeSent
375+ writesMu.Lock()
376+ defer writesMu.Unlock()
377+ if writes[1] < 2 || writes[2] < 2 {
378+ t.Fatalf("writes=%v, want fresh large traffic and retransmissions for both records", writes)
318 }379 }
319 })380 })
320 }381 }
M internal/adapters/dgram/transport_test.go +13 -4
@@ -102,13 +102,18 @@ func (c fixedClock) NewTimer(d time.Duration) ports.Timer {
102102 }
103103
104104 type manualClock struct {
105- mu sync.Mutex
106- now time.Time
107- timers map[*manualTimer]struct{}
105+ mu sync.Mutex
106+ now time.Time
107+ timers map[*manualTimer]struct{}
108+ timerCreated chan struct{}
108109 }
109110
110111 func newManualClock(now time.Time) *manualClock {
111- return &manualClock{now: now, timers: make(map[*manualTimer]struct{})}
112+ return &manualClock{
113+ now: now,
114+ timers: make(map[*manualTimer]struct{}),
115+ timerCreated: make(chan struct{}, 64),
116+ }
112117 }
113118
114119 func (c *manualClock) Now() time.Time {
@@ -122,6 +127,10 @@ func (c *manualClock) NewTimer(d time.Duration) ports.Timer {
122127 tm := &manualTimer{clock: c, c: make(chan time.Time, 1), deadline: c.now.Add(d), active: true}
123128 c.timers[tm] = struct{}{}
124129 c.mu.Unlock()
130+ select {
131+ case c.timerCreated <- struct{}{}:
132+ default:
133+ }
125134 return tm
126135 }
127136
@@ -102,13 +102,18 @@ func (c fixedClock) NewTimer(d time.Duration) ports.Timer {
102 }102 }
103 103
104 type manualClock struct {104 type manualClock struct {
105- mu sync.Mutex105+ mu sync.Mutex
106- now time.Time106+ now time.Time
107- timers map[*manualTimer]struct{}107+ timers map[*manualTimer]struct{}
108+ timerCreated chan struct{}
108 }109 }
109 110
110 func newManualClock(now time.Time) *manualClock {111 func newManualClock(now time.Time) *manualClock {
111- return &manualClock{now: now, timers: make(map[*manualTimer]struct{})}112+ return &manualClock{
113+ now: now,
114+ timers: make(map[*manualTimer]struct{}),
115+ timerCreated: make(chan struct{}, 64),
116+ }
112 }117 }
113 118
114 func (c *manualClock) Now() time.Time {119 func (c *manualClock) Now() time.Time {
@@ -122,6 +127,10 @@ func (c *manualClock) NewTimer(d time.Duration) ports.Timer {
122 tm := &manualTimer{clock: c, c: make(chan time.Time, 1), deadline: c.now.Add(d), active: true}127 tm := &manualTimer{clock: c, c: make(chan time.Time, 1), deadline: c.now.Add(d), active: true}
123 c.timers[tm] = struct{}{}128 c.timers[tm] = struct{}{}
124 c.mu.Unlock()129 c.mu.Unlock()
130+ select {
131+ case c.timerCreated <- struct{}{}:
132+ default:
133+ }
125 return tm134 return tm
126 }135 }
127 136

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