detached a386d4d

fix(dgram): complete UDP flood diagnostics

commit
a386d4dc2eb6cb37d6833f171acbec61a3df23b2
author
brice · · 2026-07-10 05:52 UTC
parent
7bc9409
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
2 files · +150 -70
Unified
UnifiedSplit
M internal/adapters/dgram/network_sim_test.go +135 -63
@@ -137,8 +137,9 @@ type fakePC struct {
137137 func newPair() (*fakePC, *fakePC) { return newPairWithLink(nil) }
138138
139139 func newPairWithLink(link *simulatedLink) (*fakePC, *fakePC) {
140- a := &fakePC{addr: testAddr("a"), in: make(chan packet, 100), read: make(chan struct{}, 100), peers: map[string]*fakePC{}, link: link}
141- b := &fakePC{addr: testAddr("b"), in: make(chan packet, 100), read: make(chan struct{}, 100), peers: map[string]*fakePC{}, link: link}
140+ const queuePackets = floodRecordCount * floodRecordMTUs * 2
141+ a := &fakePC{addr: testAddr("a"), in: make(chan packet, queuePackets), read: make(chan struct{}, queuePackets), peers: map[string]*fakePC{}, link: link}
142+ b := &fakePC{addr: testAddr("b"), in: make(chan packet, queuePackets), read: make(chan struct{}, queuePackets), peers: map[string]*fakePC{}, link: link}
142143 a.peers["b"] = b
143144 b.peers["a"] = a
144145 return a, b
@@ -206,6 +207,11 @@ func key() []byte { return bytes.Repeat([]byte{1}, pdgram.KeySize) }
206207
207208 const fixtureWaitTimeout = time.Second
208209
210+const (
211+ floodRecordCount = 32
212+ floodRecordMTUs = 24
213+)
214+
209215 // awaitSignal gives fixture barriers a diagnostic deadline. The deadline is a
210216 // hang guard only; tests advance behavior with the manual clock and barriers.
211217 func awaitSignal(t *testing.T, ch <-chan struct{}, context string) {
@@ -252,26 +258,30 @@ func TestTransportFloodClassification(t *testing.T) {
252258 dropped[frag.Seq] = true
253259 return true
254260 }
255- a, b := newFloodTransports(t, aPC, bPC, clk, 128)
261+ const mtu = 128
262+ a, b := newFloodTransports(t, aPC, bPC, clk, mtu)
256263 defer closeFloodTransports(a, b)
257- contacted := make(chan struct{}, 2)
264+ contacted := make(chan struct{}, 1)
258265 b.mu.Lock()
259- b.afterAuthenticatedPacket = func() { contacted <- struct{}{} }
266+ b.afterAuthenticatedPacket = func() {
267+ select {
268+ case contacted <- struct{}{}:
269+ default:
270+ }
271+ }
260272 b.mu.Unlock()
261273 start := clk.Now()
262274 clk.advance(time.Nanosecond)
263- for range 2 {
264- if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: make([]byte, 200)}); err != nil {
265- t.Fatal(err)
266- }
267- }
275+ sendFloodOutputs(t, a, floodRecordCount, mtu)
268276 awaitSignal(t, contacted, "first authenticated fragment")
269- awaitSignal(t, contacted, "second authenticated fragment")
270- state := floodState(b)
271- if !state.packet || state.record || floodState(a).ack {
272- t.Fatalf("contact=%v complete=%v ack=%v, want contact only", state.packet, state.record, floodState(a).ack)
277+ state := floodState(a, b, nil)
278+ if len(dropped) != floodRecordCount {
279+ t.Fatalf("flood state = %+v, dropped records=%d, want exactly one fragment from each of %d records", state, len(dropped), floodRecordCount)
280+ }
281+ if state.packetAge != 0 || state.recordAge != time.Nanosecond || state.ackAge != time.Nanosecond {
282+ t.Fatalf("flood state = %+v, want packet contact only after %d output records of %d MTUs", state, floodRecordCount, floodRecordMTUs)
273283 }
274- if !floodState(b).packet || !clk.Now().After(start) {
284+ if !clk.Now().After(start) {
275285 t.Fatal("real authenticated fragments did not update contact")
276286 }
277287 })
@@ -281,7 +291,8 @@ func TestTransportFloodClassification(t *testing.T) {
281291 limit := 2 * 128
282292 link := newSimulatedLink(clk, packetPolicy{Delay: time.Second, MaxQueueBytes: limit, BytesPerSecond: 128})
283293 aPC, bPC := newPairWithLink(link)
284- a, b := newFloodTransports(t, aPC, bPC, clk, 128)
294+ const mtu = 128
295+ a, b := newFloodTransports(t, aPC, bPC, clk, mtu)
285296 defer closeFloodTransports(a, b)
286297 contacted := make(chan struct{}, 1)
287298 b.mu.Lock()
@@ -290,21 +301,20 @@ func TestTransportFloodClassification(t *testing.T) {
290301 a.mu.Lock()
291302 a.rto = time.Hour // Keep the fake-clock queue case about bandwidth, not retries.
292303 a.mu.Unlock()
293- if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: make([]byte, 200)}); err != nil {
294- t.Fatal(err)
295- }
296- queued := link.queuedBytes()
297- if queued <= 0 || queued > limit {
298- t.Fatalf("queued bytes=%d, want 1..%d", queued, limit)
304+ sendFloodOutputs(t, a, floodRecordCount, mtu)
305+ state := floodState(a, b, link)
306+ if state.queuedBytes <= 0 || state.queuedBytes > limit {
307+ t.Fatalf("flood state = %+v, want queued bytes in 1..%d", state, limit)
299308 }
300309 clk.advance(10 * time.Second)
301310 link.flush(bPC)
302311 awaitSignal(t, contacted, "queued authenticated fragment")
303- if got := link.queuedBytes(); got != 0 {
304- t.Fatalf("queued bytes after fake-clock release=%d, want 0", got)
312+ state = floodState(a, b, link)
313+ if state.queuedBytes != 0 {
314+ t.Fatalf("flood state = %+v, want empty queue after fake-clock release", state)
305315 }
306- if floodState(b).record || floodState(a).ack {
307- t.Fatal("dropped queued fragment produced false record or ACK progress")
316+ if state.packetAge != 0 || state.recordAge != 10*time.Second || state.ackAge != 10*time.Second {
317+ t.Fatalf("flood state = %+v, want delayed packet contact without record or ACK progress", state)
308318 }
309319 })
310320
@@ -316,7 +326,7 @@ func TestTransportFloodClassification(t *testing.T) {
316326 t.Fatal(err)
317327 }
318328
319- acksDropped := make(chan struct{}, 2)
329+ acksDropped := make(chan struct{}, floodRecordCount)
320330 bPC.drop = func(pkt []byte, _ net.Addr) bool {
321331 _, raw, err := inspect.Open(pkt, 2, nil, nil)
322332 if err != nil {
@@ -326,12 +336,16 @@ func TestTransportFloodClassification(t *testing.T) {
326336 if err != nil || frag.Count != 1 || len(frag.Data) != 9 || frag.Data[0] != recAck {
327337 return false
328338 }
329- acksDropped <- struct{}{}
339+ select {
340+ case acksDropped <- struct{}{}:
341+ default:
342+ }
330343 return true
331344 }
332345
333346 retransmitBlocked := make(chan struct{}, 1)
334347 releaseRetransmit := make(chan struct{})
348+ lastRetransmitted := make(chan struct{}, 1)
335349 probeSent := make(chan struct{}, 1)
336350 freshAtPace := make(chan struct{})
337351 allowFreshPace := make(chan struct{})
@@ -363,8 +377,14 @@ func TestTransportFloodClassification(t *testing.T) {
363377 secondWrite := writes[seq] == 2
364378 writesMu.Unlock()
365379 if secondWrite {
366- retransmitBlocked <- struct{}{}
380+ select {
381+ case retransmitBlocked <- struct{}{}:
382+ default:
383+ }
367384 awaitSignal(t, releaseRetransmit, "retransmit release")
385+ if seq == floodRecordCount {
386+ lastRetransmitted <- struct{}{}
387+ }
368388 }
369389 return false
370390 }
@@ -384,28 +404,17 @@ func TestTransportFloodClassification(t *testing.T) {
384404 awaitSignal(t, clk.timerCreated, "transport timer creation")
385405 }
386406
387- // Two fresh, fragmented output records become overdue while their real
407+ // Fresh, fragmented output records become overdue while their real
388408 // cumulative ACKs are lost. Disable only test pacing so the fixture can
389409 // isolate resend-loop serialization rather than pacing delay.
390- a.dataPaceMu.Lock()
391- a.dataPaceRemaining = 100
392- a.dataPaceNext = clk.Now().Add(time.Hour)
393- a.dataPaceMu.Unlock()
394- for range 2 {
395- a.mu.Lock()
396- a.outputNext = time.Time{}
397- a.mu.Unlock()
398- if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: make([]byte, 500)}); err != nil {
399- t.Fatal(err)
400- }
401- }
410+ sendFloodOutputs(t, a, floodRecordCount, 128)
402411 awaitSignal(t, clk.timerCreated, "ACK coalescing timer creation")
403412 clk.advance(maxACKDelay)
404413 awaitSignal(t, acksDropped, "dropped cumulative ACK")
405414
406415 awaitSignal(t, retransmitBlocked, "retransmit pacing barrier")
407- if got := floodState(a).retransmits; got < 2 {
408- t.Fatalf("retransmits=%d, want both overdue large records selected", got)
416+ if state := floodState(a, b, nil); state.retransmits < floodRecordCount {
417+ t.Fatalf("flood state = %+v, want all %d overdue records selected", state, floodRecordCount)
409418 }
410419
411420 // Start a third large output at this same fake-clock instant. The pacing
@@ -419,7 +428,7 @@ func TestTransportFloodClassification(t *testing.T) {
419428 }
420429 a.mu.Unlock()
421430 go func() {
422- freshReturned <- a.Send(ports.Frame{Type: ports.MsgOutput, Payload: make([]byte, 500)})
431+ freshReturned <- a.Send(ports.Frame{Type: ports.MsgOutput, Payload: floodPayload(128)})
423432 }()
424433 awaitSignal(t, freshAtPace, "fresh output pacing barrier")
425434 a.mu.Lock()
@@ -438,33 +447,53 @@ func TestTransportFloodClassification(t *testing.T) {
438447 if err := awaitResult(t, freshReturned, "fresh output completion"); err != nil {
439448 t.Fatal(err)
440449 }
441- eventually(t, time.Second, func() bool {
442- writesMu.Lock()
443- defer writesMu.Unlock()
444- return writes[1] >= 2 && writes[2] >= 2
445- })
450+ awaitSignal(t, lastRetransmitted, "last overdue record retransmission")
446451 writesMu.Lock()
447452 defer writesMu.Unlock()
448- if writes[1] < 2 || writes[2] < 2 || writes[3] == 0 {
449- t.Fatalf("writes=%v, want fresh large traffic and retransmissions for both overdue records", writes)
453+ if writes[1] < 2 || writes[floodRecordCount] < 2 || writes[floodRecordCount+1] == 0 {
454+ t.Fatalf("flood state = %+v, writes=%v, want fresh large traffic and retransmissions across %d overdue records", floodState(a, b, nil), writes, floodRecordCount)
450455 }
451456 })
452457 }
453458
454459 type floodTransportState struct {
455- packet, record, ack bool
456- retransmits uint64
460+ packetAge, recordAge, ackAge time.Duration
461+ queuedBytes int
462+ retransmits uint64
463+}
464+
465+func floodState(sender, receiver *Transport, link *simulatedLink) floodTransportState {
466+ now := sender.clock.Now()
467+ receiver.mu.Lock()
468+ packetAge := now.Sub(receiver.lastAuthenticatedPacket)
469+ recordAge := now.Sub(receiver.lastCompleteRecord)
470+ receiver.mu.Unlock()
471+ sender.mu.Lock()
472+ ackAge := now.Sub(sender.lastACKProgress)
473+ retransmits := sender.retransmits
474+ sender.mu.Unlock()
475+ queuedBytes := 0
476+ if link != nil {
477+ queuedBytes = link.queuedBytes()
478+ }
479+ return floodTransportState{packetAge: packetAge, recordAge: recordAge, ackAge: ackAge, queuedBytes: queuedBytes, retransmits: retransmits}
457480 }
458481
459-func floodState(t *Transport) floodTransportState {
460- t.mu.Lock()
461- defer t.mu.Unlock()
462- now := t.clock.Now()
463- return floodTransportState{
464- packet: t.lastAuthenticatedPacket.Equal(now),
465- record: t.lastCompleteRecord.Equal(now),
466- ack: t.lastACKProgress.Equal(now),
467- retransmits: t.retransmits,
482+func floodPayload(mtu int) []byte { return make([]byte, floodRecordMTUs*mtu) }
483+
484+func sendFloodOutputs(t *testing.T, sender *Transport, count, mtu int) {
485+ t.Helper()
486+ sender.dataPaceMu.Lock()
487+ sender.dataPaceRemaining = count * floodRecordMTUs * 4
488+ sender.dataPaceNext = sender.clock.Now().Add(time.Hour)
489+ sender.dataPaceMu.Unlock()
490+ for range count {
491+ sender.mu.Lock()
492+ sender.outputNext = time.Time{}
493+ sender.mu.Unlock()
494+ if err := sender.Send(ports.Frame{Type: ports.MsgOutput, Payload: floodPayload(mtu)}); err != nil {
495+ t.Fatal(err)
496+ }
468497 }
469498 }
470499
@@ -516,3 +545,46 @@ func TestMalformedAuthenticatedFragmentDoesNotCountAsContact(t *testing.T) {
516545 t.Fatal("malformed authenticated plaintext counted as packet contact")
517546 }
518547 }
548+
549+func TestInconsistentAuthenticatedFragmentDoesNotCountAsContact(t *testing.T) {
550+ clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
551+ aPC, bPC := newPair()
552+ a, b := newFloodTransports(t, aPC, bPC, clk, 128)
553+ defer closeFloodTransports(a, b)
554+ codec, err := pdgram.NewCodec(key())
555+ if err != nil {
556+ t.Fatal(err)
557+ }
558+ accepted := make(chan struct{}, 1)
559+ rejected := make(chan struct{}, 1)
560+ b.mu.Lock()
561+ b.afterAuthenticatedPacket = func() { accepted <- struct{}{} }
562+ b.afterMalformedFragment = func() { rejected <- struct{}{} }
563+ b.mu.Unlock()
564+
565+ writeFragment := func(counter uint64, frag pdgram.Fragment) {
566+ t.Helper()
567+ raw, err := pdgram.MarshalFragment(frag)
568+ if err != nil {
569+ t.Fatal(err)
570+ }
571+ if _, err := aPC.WriteTo(codec.Seal(1, counter, raw, nil), bPC.addr); err != nil {
572+ t.Fatal(err)
573+ }
574+ }
575+ writeFragment(999, pdgram.Fragment{Seq: 42, Index: 0, Count: 2, Data: []byte("first")})
576+ awaitSignal(t, accepted, "accepted incomplete fragment")
577+ b.mu.Lock()
578+ lastContact := b.lastAuthenticatedPacket
579+ b.mu.Unlock()
580+
581+ clk.advance(time.Second)
582+ writeFragment(1000, pdgram.Fragment{Seq: 42, Index: 1, Count: 3, Data: []byte("inconsistent")})
583+ awaitSignal(t, rejected, "inconsistent fragment rejection")
584+ b.mu.Lock()
585+ gotContact := b.lastAuthenticatedPacket
586+ b.mu.Unlock()
587+ if gotContact != lastContact {
588+ t.Fatalf("rejected inconsistent fragment refreshed contact from %v to %v", lastContact, gotContact)
589+ }
590+}
@@ -137,8 +137,9 @@ type fakePC struct {
137 func newPair() (*fakePC, *fakePC) { return newPairWithLink(nil) }137 func newPair() (*fakePC, *fakePC) { return newPairWithLink(nil) }
138 138
139 func newPairWithLink(link *simulatedLink) (*fakePC, *fakePC) {139 func newPairWithLink(link *simulatedLink) (*fakePC, *fakePC) {
140- a := &fakePC{addr: testAddr("a"), in: make(chan packet, 100), read: make(chan struct{}, 100), peers: map[string]*fakePC{}, link: link}140+ const queuePackets = floodRecordCount * floodRecordMTUs * 2
141- b := &fakePC{addr: testAddr("b"), in: make(chan packet, 100), read: make(chan struct{}, 100), peers: map[string]*fakePC{}, link: link}141+ a := &fakePC{addr: testAddr("a"), in: make(chan packet, queuePackets), read: make(chan struct{}, queuePackets), peers: map[string]*fakePC{}, link: link}
142+ b := &fakePC{addr: testAddr("b"), in: make(chan packet, queuePackets), read: make(chan struct{}, queuePackets), peers: map[string]*fakePC{}, link: link}
142 a.peers["b"] = b143 a.peers["b"] = b
143 b.peers["a"] = a144 b.peers["a"] = a
144 return a, b145 return a, b
@@ -206,6 +207,11 @@ func key() []byte { return bytes.Repeat([]byte{1}, pdgram.KeySize) }
206 207
207 const fixtureWaitTimeout = time.Second208 const fixtureWaitTimeout = time.Second
208 209
210+const (
211+ floodRecordCount = 32
212+ floodRecordMTUs = 24
213+)
214+
209 // awaitSignal gives fixture barriers a diagnostic deadline. The deadline is a215 // awaitSignal gives fixture barriers a diagnostic deadline. The deadline is a
210 // hang guard only; tests advance behavior with the manual clock and barriers.216 // hang guard only; tests advance behavior with the manual clock and barriers.
211 func awaitSignal(t *testing.T, ch <-chan struct{}, context string) {217 func awaitSignal(t *testing.T, ch <-chan struct{}, context string) {
@@ -252,26 +258,30 @@ func TestTransportFloodClassification(t *testing.T) {
252 dropped[frag.Seq] = true258 dropped[frag.Seq] = true
253 return true259 return true
254 }260 }
255- a, b := newFloodTransports(t, aPC, bPC, clk, 128)261+ const mtu = 128
262+ a, b := newFloodTransports(t, aPC, bPC, clk, mtu)
256 defer closeFloodTransports(a, b)263 defer closeFloodTransports(a, b)
257- contacted := make(chan struct{}, 2)264+ contacted := make(chan struct{}, 1)
258 b.mu.Lock()265 b.mu.Lock()
259- b.afterAuthenticatedPacket = func() { contacted <- struct{}{} }266+ b.afterAuthenticatedPacket = func() {
267+ select {
268+ case contacted <- struct{}{}:
269+ default:
270+ }
271+ }
260 b.mu.Unlock()272 b.mu.Unlock()
261 start := clk.Now()273 start := clk.Now()
262 clk.advance(time.Nanosecond)274 clk.advance(time.Nanosecond)
263- for range 2 {275+ sendFloodOutputs(t, a, floodRecordCount, mtu)
264- if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: make([]byte, 200)}); err != nil {
265- t.Fatal(err)
266- }
267- }
268 awaitSignal(t, contacted, "first authenticated fragment")276 awaitSignal(t, contacted, "first authenticated fragment")
269- awaitSignal(t, contacted, "second authenticated fragment")277+ state := floodState(a, b, nil)
270- state := floodState(b)278+ if len(dropped) != floodRecordCount {
271- if !state.packet || state.record || floodState(a).ack {279+ t.Fatalf("flood state = %+v, dropped records=%d, want exactly one fragment from each of %d records", state, len(dropped), floodRecordCount)
272- t.Fatalf("contact=%v complete=%v ack=%v, want contact only", state.packet, state.record, floodState(a).ack)280+ }
281+ if state.packetAge != 0 || state.recordAge != time.Nanosecond || state.ackAge != time.Nanosecond {
282+ t.Fatalf("flood state = %+v, want packet contact only after %d output records of %d MTUs", state, floodRecordCount, floodRecordMTUs)
273 }283 }
274- if !floodState(b).packet || !clk.Now().After(start) {284+ if !clk.Now().After(start) {
275 t.Fatal("real authenticated fragments did not update contact")285 t.Fatal("real authenticated fragments did not update contact")
276 }286 }
277 })287 })
@@ -281,7 +291,8 @@ func TestTransportFloodClassification(t *testing.T) {
281 limit := 2 * 128291 limit := 2 * 128
282 link := newSimulatedLink(clk, packetPolicy{Delay: time.Second, MaxQueueBytes: limit, BytesPerSecond: 128})292 link := newSimulatedLink(clk, packetPolicy{Delay: time.Second, MaxQueueBytes: limit, BytesPerSecond: 128})
283 aPC, bPC := newPairWithLink(link)293 aPC, bPC := newPairWithLink(link)
284- a, b := newFloodTransports(t, aPC, bPC, clk, 128)294+ const mtu = 128
295+ a, b := newFloodTransports(t, aPC, bPC, clk, mtu)
285 defer closeFloodTransports(a, b)296 defer closeFloodTransports(a, b)
286 contacted := make(chan struct{}, 1)297 contacted := make(chan struct{}, 1)
287 b.mu.Lock()298 b.mu.Lock()
@@ -290,21 +301,20 @@ func TestTransportFloodClassification(t *testing.T) {
290 a.mu.Lock()301 a.mu.Lock()
291 a.rto = time.Hour // Keep the fake-clock queue case about bandwidth, not retries.302 a.rto = time.Hour // Keep the fake-clock queue case about bandwidth, not retries.
292 a.mu.Unlock()303 a.mu.Unlock()
293- if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: make([]byte, 200)}); err != nil {304+ sendFloodOutputs(t, a, floodRecordCount, mtu)
294- t.Fatal(err)305+ state := floodState(a, b, link)
295- }306+ if state.queuedBytes <= 0 || state.queuedBytes > limit {
296- queued := link.queuedBytes()307+ t.Fatalf("flood state = %+v, want queued bytes in 1..%d", state, limit)
297- if queued <= 0 || queued > limit {
298- t.Fatalf("queued bytes=%d, want 1..%d", queued, limit)
299 }308 }
300 clk.advance(10 * time.Second)309 clk.advance(10 * time.Second)
301 link.flush(bPC)310 link.flush(bPC)
302 awaitSignal(t, contacted, "queued authenticated fragment")311 awaitSignal(t, contacted, "queued authenticated fragment")
303- if got := link.queuedBytes(); got != 0 {312+ state = floodState(a, b, link)
304- t.Fatalf("queued bytes after fake-clock release=%d, want 0", got)313+ if state.queuedBytes != 0 {
314+ t.Fatalf("flood state = %+v, want empty queue after fake-clock release", state)
305 }315 }
306- if floodState(b).record || floodState(a).ack {316+ if state.packetAge != 0 || state.recordAge != 10*time.Second || state.ackAge != 10*time.Second {
307- t.Fatal("dropped queued fragment produced false record or ACK progress")317+ t.Fatalf("flood state = %+v, want delayed packet contact without record or ACK progress", state)
308 }318 }
309 })319 })
310 320
@@ -316,7 +326,7 @@ func TestTransportFloodClassification(t *testing.T) {
316 t.Fatal(err)326 t.Fatal(err)
317 }327 }
318 328
319- acksDropped := make(chan struct{}, 2)329+ acksDropped := make(chan struct{}, floodRecordCount)
320 bPC.drop = func(pkt []byte, _ net.Addr) bool {330 bPC.drop = func(pkt []byte, _ net.Addr) bool {
321 _, raw, err := inspect.Open(pkt, 2, nil, nil)331 _, raw, err := inspect.Open(pkt, 2, nil, nil)
322 if err != nil {332 if err != nil {
@@ -326,12 +336,16 @@ func TestTransportFloodClassification(t *testing.T) {
326 if err != nil || frag.Count != 1 || len(frag.Data) != 9 || frag.Data[0] != recAck {336 if err != nil || frag.Count != 1 || len(frag.Data) != 9 || frag.Data[0] != recAck {
327 return false337 return false
328 }338 }
329- acksDropped <- struct{}{}339+ select {
340+ case acksDropped <- struct{}{}:
341+ default:
342+ }
330 return true343 return true
331 }344 }
332 345
333 retransmitBlocked := make(chan struct{}, 1)346 retransmitBlocked := make(chan struct{}, 1)
334 releaseRetransmit := make(chan struct{})347 releaseRetransmit := make(chan struct{})
348+ lastRetransmitted := make(chan struct{}, 1)
335 probeSent := make(chan struct{}, 1)349 probeSent := make(chan struct{}, 1)
336 freshAtPace := make(chan struct{})350 freshAtPace := make(chan struct{})
337 allowFreshPace := make(chan struct{})351 allowFreshPace := make(chan struct{})
@@ -363,8 +377,14 @@ func TestTransportFloodClassification(t *testing.T) {
363 secondWrite := writes[seq] == 2377 secondWrite := writes[seq] == 2
364 writesMu.Unlock()378 writesMu.Unlock()
365 if secondWrite {379 if secondWrite {
366- retransmitBlocked <- struct{}{}380+ select {
381+ case retransmitBlocked <- struct{}{}:
382+ default:
383+ }
367 awaitSignal(t, releaseRetransmit, "retransmit release")384 awaitSignal(t, releaseRetransmit, "retransmit release")
385+ if seq == floodRecordCount {
386+ lastRetransmitted <- struct{}{}
387+ }
368 }388 }
369 return false389 return false
370 }390 }
@@ -384,28 +404,17 @@ func TestTransportFloodClassification(t *testing.T) {
384 awaitSignal(t, clk.timerCreated, "transport timer creation")404 awaitSignal(t, clk.timerCreated, "transport timer creation")
385 }405 }
386 406
387- // Two fresh, fragmented output records become overdue while their real407+ // Fresh, fragmented output records become overdue while their real
388 // cumulative ACKs are lost. Disable only test pacing so the fixture can408 // cumulative ACKs are lost. Disable only test pacing so the fixture can
389 // isolate resend-loop serialization rather than pacing delay.409 // isolate resend-loop serialization rather than pacing delay.
390- a.dataPaceMu.Lock()410+ sendFloodOutputs(t, a, floodRecordCount, 128)
391- a.dataPaceRemaining = 100
392- a.dataPaceNext = clk.Now().Add(time.Hour)
393- a.dataPaceMu.Unlock()
394- for range 2 {
395- a.mu.Lock()
396- a.outputNext = time.Time{}
397- a.mu.Unlock()
398- if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: make([]byte, 500)}); err != nil {
399- t.Fatal(err)
400- }
401- }
402 awaitSignal(t, clk.timerCreated, "ACK coalescing timer creation")411 awaitSignal(t, clk.timerCreated, "ACK coalescing timer creation")
403 clk.advance(maxACKDelay)412 clk.advance(maxACKDelay)
404 awaitSignal(t, acksDropped, "dropped cumulative ACK")413 awaitSignal(t, acksDropped, "dropped cumulative ACK")
405 414
406 awaitSignal(t, retransmitBlocked, "retransmit pacing barrier")415 awaitSignal(t, retransmitBlocked, "retransmit pacing barrier")
407- if got := floodState(a).retransmits; got < 2 {416+ if state := floodState(a, b, nil); state.retransmits < floodRecordCount {
408- t.Fatalf("retransmits=%d, want both overdue large records selected", got)417+ t.Fatalf("flood state = %+v, want all %d overdue records selected", state, floodRecordCount)
409 }418 }
410 419
411 // Start a third large output at this same fake-clock instant. The pacing420 // Start a third large output at this same fake-clock instant. The pacing
@@ -419,7 +428,7 @@ func TestTransportFloodClassification(t *testing.T) {
419 }428 }
420 a.mu.Unlock()429 a.mu.Unlock()
421 go func() {430 go func() {
422- freshReturned <- a.Send(ports.Frame{Type: ports.MsgOutput, Payload: make([]byte, 500)})431+ freshReturned <- a.Send(ports.Frame{Type: ports.MsgOutput, Payload: floodPayload(128)})
423 }()432 }()
424 awaitSignal(t, freshAtPace, "fresh output pacing barrier")433 awaitSignal(t, freshAtPace, "fresh output pacing barrier")
425 a.mu.Lock()434 a.mu.Lock()
@@ -438,33 +447,53 @@ func TestTransportFloodClassification(t *testing.T) {
438 if err := awaitResult(t, freshReturned, "fresh output completion"); err != nil {447 if err := awaitResult(t, freshReturned, "fresh output completion"); err != nil {
439 t.Fatal(err)448 t.Fatal(err)
440 }449 }
441- eventually(t, time.Second, func() bool {450+ awaitSignal(t, lastRetransmitted, "last overdue record retransmission")
442- writesMu.Lock()
443- defer writesMu.Unlock()
444- return writes[1] >= 2 && writes[2] >= 2
445- })
446 writesMu.Lock()451 writesMu.Lock()
447 defer writesMu.Unlock()452 defer writesMu.Unlock()
448- if writes[1] < 2 || writes[2] < 2 || writes[3] == 0 {453+ if writes[1] < 2 || writes[floodRecordCount] < 2 || writes[floodRecordCount+1] == 0 {
449- t.Fatalf("writes=%v, want fresh large traffic and retransmissions for both overdue records", writes)454+ t.Fatalf("flood state = %+v, writes=%v, want fresh large traffic and retransmissions across %d overdue records", floodState(a, b, nil), writes, floodRecordCount)
450 }455 }
451 })456 })
452 }457 }
453 458
454 type floodTransportState struct {459 type floodTransportState struct {
455- packet, record, ack bool460+ packetAge, recordAge, ackAge time.Duration
456- retransmits uint64461+ queuedBytes int
462+ retransmits uint64
463+}
464+
465+func floodState(sender, receiver *Transport, link *simulatedLink) floodTransportState {
466+ now := sender.clock.Now()
467+ receiver.mu.Lock()
468+ packetAge := now.Sub(receiver.lastAuthenticatedPacket)
469+ recordAge := now.Sub(receiver.lastCompleteRecord)
470+ receiver.mu.Unlock()
471+ sender.mu.Lock()
472+ ackAge := now.Sub(sender.lastACKProgress)
473+ retransmits := sender.retransmits
474+ sender.mu.Unlock()
475+ queuedBytes := 0
476+ if link != nil {
477+ queuedBytes = link.queuedBytes()
478+ }
479+ return floodTransportState{packetAge: packetAge, recordAge: recordAge, ackAge: ackAge, queuedBytes: queuedBytes, retransmits: retransmits}
457 }480 }
458 481
459-func floodState(t *Transport) floodTransportState {482+func floodPayload(mtu int) []byte { return make([]byte, floodRecordMTUs*mtu) }
460- t.mu.Lock()483+
461- defer t.mu.Unlock()484+func sendFloodOutputs(t *testing.T, sender *Transport, count, mtu int) {
462- now := t.clock.Now()485+ t.Helper()
463- return floodTransportState{486+ sender.dataPaceMu.Lock()
464- packet: t.lastAuthenticatedPacket.Equal(now),487+ sender.dataPaceRemaining = count * floodRecordMTUs * 4
465- record: t.lastCompleteRecord.Equal(now),488+ sender.dataPaceNext = sender.clock.Now().Add(time.Hour)
466- ack: t.lastACKProgress.Equal(now),489+ sender.dataPaceMu.Unlock()
467- retransmits: t.retransmits,490+ for range count {
491+ sender.mu.Lock()
492+ sender.outputNext = time.Time{}
493+ sender.mu.Unlock()
494+ if err := sender.Send(ports.Frame{Type: ports.MsgOutput, Payload: floodPayload(mtu)}); err != nil {
495+ t.Fatal(err)
496+ }
468 }497 }
469 }498 }
470 499
@@ -516,3 +545,46 @@ func TestMalformedAuthenticatedFragmentDoesNotCountAsContact(t *testing.T) {
516 t.Fatal("malformed authenticated plaintext counted as packet contact")545 t.Fatal("malformed authenticated plaintext counted as packet contact")
517 }546 }
518 }547 }
548+
549+func TestInconsistentAuthenticatedFragmentDoesNotCountAsContact(t *testing.T) {
550+ clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
551+ aPC, bPC := newPair()
552+ a, b := newFloodTransports(t, aPC, bPC, clk, 128)
553+ defer closeFloodTransports(a, b)
554+ codec, err := pdgram.NewCodec(key())
555+ if err != nil {
556+ t.Fatal(err)
557+ }
558+ accepted := make(chan struct{}, 1)
559+ rejected := make(chan struct{}, 1)
560+ b.mu.Lock()
561+ b.afterAuthenticatedPacket = func() { accepted <- struct{}{} }
562+ b.afterMalformedFragment = func() { rejected <- struct{}{} }
563+ b.mu.Unlock()
564+
565+ writeFragment := func(counter uint64, frag pdgram.Fragment) {
566+ t.Helper()
567+ raw, err := pdgram.MarshalFragment(frag)
568+ if err != nil {
569+ t.Fatal(err)
570+ }
571+ if _, err := aPC.WriteTo(codec.Seal(1, counter, raw, nil), bPC.addr); err != nil {
572+ t.Fatal(err)
573+ }
574+ }
575+ writeFragment(999, pdgram.Fragment{Seq: 42, Index: 0, Count: 2, Data: []byte("first")})
576+ awaitSignal(t, accepted, "accepted incomplete fragment")
577+ b.mu.Lock()
578+ lastContact := b.lastAuthenticatedPacket
579+ b.mu.Unlock()
580+
581+ clk.advance(time.Second)
582+ writeFragment(1000, pdgram.Fragment{Seq: 42, Index: 1, Count: 3, Data: []byte("inconsistent")})
583+ awaitSignal(t, rejected, "inconsistent fragment rejection")
584+ b.mu.Lock()
585+ gotContact := b.lastAuthenticatedPacket
586+ b.mu.Unlock()
587+ if gotContact != lastContact {
588+ t.Fatalf("rejected inconsistent fragment refreshed contact from %v to %v", lastContact, gotContact)
589+ }
590+}
M internal/adapters/dgram/transport.go +15 -7
@@ -790,19 +790,18 @@ func (t *Transport) readLoop(pc net.PacketConn) {
790790 frag, err := pdgram.UnmarshalFragment(pt)
791791 if err != nil {
792792 t.recvMu.Unlock()
793- t.mu.Lock()
794- afterMalformed := t.afterMalformedFragment
795- t.mu.Unlock()
796- if afterMalformed != nil {
797- afterMalformed()
798- }
793+ t.notifyMalformedFragment()
799794 continue
800795 }
801796 payload, complete, err := t.reasm.Add(frag)
802797 inflight := t.reasm.Inflight()
803798 t.recvMu.Unlock()
799+ if err != nil {
800+ t.notifyMalformedFragment()
801+ continue
802+ }
804803 t.recordAuthenticatedFragment(inflight)
805- if err != nil || !complete {
804+ if !complete {
806805 continue
807806 }
808807 t.recordCompleteRecord()
@@ -811,6 +810,15 @@ func (t *Transport) readLoop(pc net.PacketConn) {
811810 }
812811 }
813812
813+func (t *Transport) notifyMalformedFragment() {
814+ t.mu.Lock()
815+ afterMalformed := t.afterMalformedFragment
816+ t.mu.Unlock()
817+ if afterMalformed != nil {
818+ afterMalformed()
819+ }
820+}
821+
814822 func (t *Transport) recordAuthenticatedFragment(inflight int) {
815823 t.mu.Lock()
816824 t.lastAuthenticatedPacket = t.clock.Now()
@@ -790,19 +790,18 @@ func (t *Transport) readLoop(pc net.PacketConn) {
790 frag, err := pdgram.UnmarshalFragment(pt)790 frag, err := pdgram.UnmarshalFragment(pt)
791 if err != nil {791 if err != nil {
792 t.recvMu.Unlock()792 t.recvMu.Unlock()
793- t.mu.Lock()793+ t.notifyMalformedFragment()
794- afterMalformed := t.afterMalformedFragment
795- t.mu.Unlock()
796- if afterMalformed != nil {
797- afterMalformed()
798- }
799 continue794 continue
800 }795 }
801 payload, complete, err := t.reasm.Add(frag)796 payload, complete, err := t.reasm.Add(frag)
802 inflight := t.reasm.Inflight()797 inflight := t.reasm.Inflight()
803 t.recvMu.Unlock()798 t.recvMu.Unlock()
799+ if err != nil {
800+ t.notifyMalformedFragment()
801+ continue
802+ }
804 t.recordAuthenticatedFragment(inflight)803 t.recordAuthenticatedFragment(inflight)
805- if err != nil || !complete {804+ if !complete {
806 continue805 continue
807 }806 }
808 t.recordCompleteRecord()807 t.recordCompleteRecord()
@@ -811,6 +810,15 @@ func (t *Transport) readLoop(pc net.PacketConn) {
811 }810 }
812 }811 }
813 812
813+func (t *Transport) notifyMalformedFragment() {
814+ t.mu.Lock()
815+ afterMalformed := t.afterMalformedFragment
816+ t.mu.Unlock()
817+ if afterMalformed != nil {
818+ afterMalformed()
819+ }
820+}
821+
814 func (t *Transport) recordAuthenticatedFragment(inflight int) {822 func (t *Transport) recordAuthenticatedFragment(inflight int) {
815 t.mu.Lock()823 t.mu.Lock()
816 t.lastAuthenticatedPacket = t.clock.Now()824 t.lastAuthenticatedPacket = t.clock.Now()

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