fix(dgram): exercise real UDP flood failure modes
- commit
- fcb2135314b3fc74314af320280b5373c1c17f74
- author
- brice · · 2026-07-10 00:01 UTC
- parent
- 8c4b9b1
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
1 file · +185 -118
M internal/adapters/dgram/network_sim_test.go +185 -118
| @@ -8,6 +8,7 @@ import ( | |||
| 8 | 8 | "testing" | |
| 9 | 9 | "time" | |
| 10 | 10 | ||
| 11 | + | "github.com/bnema/vev/internal/ports" | |
| 11 | 12 | pdgram "github.com/bnema/vev/pkg/dgram" | |
| 12 | 13 | ) | |
| 13 | 14 | ||
| @@ -28,7 +29,6 @@ type packetMeta struct { | |||
| 28 | 29 | Bytes int | |
| 29 | 30 | At time.Time | |
| 30 | 31 | QueueBytes int | |
| 31 | - | Retransmit bool | |
| 32 | 32 | } | |
| 33 | 33 | ||
| 34 | 34 | // packetPolicy makes loss and constrained links explicit in transport tests. | |
| @@ -196,132 +196,199 @@ func (p *fakePC) SetWriteDeadline(time.Time) error { return nil } | |||
| 196 | 196 | ||
| 197 | 197 | func key() []byte { return bytes.Repeat([]byte{1}, pdgram.KeySize) } | |
| 198 | 198 | ||
| 199 | - | type floodMetrics struct { | |
| 200 | - | packetContact int | |
| 201 | - | completedRecord int | |
| 202 | - | ackProgress int | |
| 203 | - | queueBytes int | |
| 204 | - | retransmits int | |
| 199 | + | // TestTransportFloodClassification drives authenticated datagrams through real | |
| 200 | + | // transports. The metrics are transport state, not simulator bookkeeping. | |
| 201 | + | func TestTransportFloodClassification(t *testing.T) { | |
| 202 | + | t.Run("one real fragment per large record loss contacts without completing", func(t *testing.T) { | |
| 203 | + | clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC)) | |
| 204 | + | aPC, bPC := newPair() | |
| 205 | + | inspect, err := pdgram.NewCodec(key()) | |
| 206 | + | if err != nil { | |
| 207 | + | t.Fatal(err) | |
| 208 | + | } | |
| 209 | + | dropped := map[uint64]bool{} | |
| 210 | + | aPC.drop = func(pkt []byte, _ net.Addr) bool { | |
| 211 | + | _, raw, err := inspect.Open(pkt, 1, nil, nil) | |
| 212 | + | if err != nil { | |
| 213 | + | return false | |
| 214 | + | } | |
| 215 | + | frag, err := pdgram.UnmarshalFragment(raw) | |
| 216 | + | if err != nil || frag.Index != frag.Count-1 || dropped[frag.Seq] { | |
| 217 | + | return false | |
| 218 | + | } | |
| 219 | + | dropped[frag.Seq] = true | |
| 220 | + | return true | |
| 221 | + | } | |
| 222 | + | a, b := newFloodTransports(t, aPC, bPC, clk, 128) | |
| 223 | + | defer closeFloodTransports(a, b) | |
| 224 | + | start := clk.Now() | |
| 225 | + | clk.advance(time.Nanosecond) | |
| 226 | + | for range 2 { | |
| 227 | + | if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: make([]byte, 200)}); err != nil { | |
| 228 | + | t.Fatal(err) | |
| 229 | + | } | |
| 230 | + | } | |
| 231 | + | floodEventually(t, func() bool { return floodState(b).packet && len(dropped) == 2 }) | |
| 232 | + | state := floodState(b) | |
| 233 | + | if !state.packet || state.record || floodState(a).ack { | |
| 234 | + | t.Fatalf("contact=%v complete=%v ack=%v, want contact only", state.packet, state.record, floodState(a).ack) | |
| 235 | + | } | |
| 236 | + | if !floodState(b).packet || !clk.Now().After(start) { | |
| 237 | + | t.Fatal("real authenticated fragments did not update contact") | |
| 238 | + | } | |
| 239 | + | }) | |
| 240 | + | ||
| 241 | + | t.Run("bounded byte queue advances and releases on fake clock", func(t *testing.T) { | |
| 242 | + | clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC)) | |
| 243 | + | limit := 2 * 128 | |
| 244 | + | link := newSimulatedLink(clk, packetPolicy{Delay: time.Second, MaxQueueBytes: limit, BytesPerSecond: 128}) | |
| 245 | + | aPC, bPC := newPairWithLink(link) | |
| 246 | + | a, b := newFloodTransports(t, aPC, bPC, clk, 128) | |
| 247 | + | defer closeFloodTransports(a, b) | |
| 248 | + | a.mu.Lock() | |
| 249 | + | a.rto = time.Hour // Keep the fake-clock queue case about bandwidth, not retries. | |
| 250 | + | a.mu.Unlock() | |
| 251 | + | if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: make([]byte, 200)}); err != nil { | |
| 252 | + | t.Fatal(err) | |
| 253 | + | } | |
| 254 | + | queued := link.queuedBytes() | |
| 255 | + | if queued <= 0 || queued > limit { | |
| 256 | + | t.Fatalf("queued bytes=%d, want 1..%d", queued, limit) | |
| 257 | + | } | |
| 258 | + | clk.advance(10 * time.Second) | |
| 259 | + | link.flush(bPC) | |
| 260 | + | floodEventually(t, func() bool { return floodState(b).packet }) | |
| 261 | + | if got := link.queuedBytes(); got != 0 { | |
| 262 | + | t.Fatalf("queued bytes after fake-clock release=%d, want 0", got) | |
| 263 | + | } | |
| 264 | + | if floodState(b).record || floodState(a).ack { | |
| 265 | + | t.Fatal("dropped queued fragment produced false record or ACK progress") | |
| 266 | + | } | |
| 267 | + | }) | |
| 268 | + | ||
| 269 | + | t.Run("real cumulative ACK loss schedules duplicate retransmits", func(t *testing.T) { | |
| 270 | + | clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC)) | |
| 271 | + | aPC, bPC := newPair() | |
| 272 | + | inspect, err := pdgram.NewCodec(key()) | |
| 273 | + | if err != nil { | |
| 274 | + | t.Fatal(err) | |
| 275 | + | } | |
| 276 | + | bPC.drop = func(pkt []byte, _ net.Addr) bool { | |
| 277 | + | _, raw, err := inspect.Open(pkt, 2, nil, nil) | |
| 278 | + | if err != nil { | |
| 279 | + | return false | |
| 280 | + | } | |
| 281 | + | frag, err := pdgram.UnmarshalFragment(raw) | |
| 282 | + | return err == nil && frag.Count == 1 && len(frag.Data) == 9 && frag.Data[0] == recAck | |
| 283 | + | } | |
| 284 | + | seen := map[uint64]int{} | |
| 285 | + | var seenMu sync.Mutex | |
| 286 | + | aPC.drop = func(pkt []byte, _ net.Addr) bool { | |
| 287 | + | _, 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 | + | } | |
| 299 | + | return false | |
| 300 | + | } | |
| 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 { | |
| 305 | + | t.Fatal(err) | |
| 306 | + | } | |
| 307 | + | floodEventually(t, func() bool { return floodState(b).record }) | |
| 308 | + | waitForManualTimers(t, clk, 4) | |
| 309 | + | 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) | |
| 318 | + | } | |
| 319 | + | }) | |
| 205 | 320 | } | |
| 206 | 321 | ||
| 207 | - | // TestTransportFloodClassification documents the three distinct symptoms that | |
| 208 | - | // used to be conflated as a healthy UDP link: authenticated datagrams arriving, | |
| 209 | - | // complete records arriving, and ACKs retiring sender work. | |
| 210 | - | func TestTransportFloodClassification(t *testing.T) { | |
| 211 | - | const records = 32 | |
| 212 | - | payload := make([]byte, 24*pdgram.DefaultMTU) | |
| 213 | - | fragments, err := pdgram.FragmentPayload(1, payload, pdgram.DefaultMTU) | |
| 322 | + | type floodTransportState struct { | |
| 323 | + | packet, record, ack bool | |
| 324 | + | retransmits uint64 | |
| 325 | + | } | |
| 326 | + | ||
| 327 | + | func floodState(t *Transport) floodTransportState { | |
| 328 | + | t.mu.Lock() | |
| 329 | + | defer t.mu.Unlock() | |
| 330 | + | now := t.clock.Now() | |
| 331 | + | return floodTransportState{ | |
| 332 | + | packet: t.lastAuthenticatedPacket.Equal(now), | |
| 333 | + | record: t.lastCompleteRecord.Equal(now), | |
| 334 | + | ack: t.lastACKProgress.Equal(now), | |
| 335 | + | retransmits: t.retransmits, | |
| 336 | + | } | |
| 337 | + | } | |
| 338 | + | ||
| 339 | + | func newFloodTransports(t *testing.T, aPC, bPC *fakePC, clk *manualClock, mtu int) (*Transport, *Transport) { | |
| 340 | + | t.Helper() | |
| 341 | + | opts := Options{Clock: clk, MTU: mtu, ResendAfter: 10 * time.Millisecond, Heartbeat: time.Hour} | |
| 342 | + | a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, opts) | |
| 214 | 343 | if err != nil { | |
| 215 | 344 | t.Fatal(err) | |
| 216 | 345 | } | |
| 217 | - | if len(fragments) < 24 { | |
| 218 | - | t.Fatalf("fragments=%d, want at least 24", len(fragments)) | |
| 346 | + | b, err := NewTransportWithOptions(bPC, aPC.addr, key(), 2, 1, opts) | |
| 347 | + | if err != nil { | |
| 348 | + | _ = a.Close() | |
| 349 | + | t.Fatal(err) | |
| 219 | 350 | } | |
| 220 | - | packets := make([][]byte, len(fragments)) | |
| 221 | - | for i, fragment := range fragments { | |
| 222 | - | packets[i], err = pdgram.MarshalFragment(fragment) | |
| 223 | - | if err != nil { | |
| 224 | - | t.Fatal(err) | |
| 351 | + | return a, b | |
| 352 | + | } | |
| 353 | + | ||
| 354 | + | func closeFloodTransports(a, b *Transport) { | |
| 355 | + | _ = a.Close() | |
| 356 | + | _ = b.Close() | |
| 357 | + | } | |
| 358 | + | ||
| 359 | + | func floodEventually(t *testing.T, condition func() bool) { | |
| 360 | + | t.Helper() | |
| 361 | + | deadline := time.Now().Add(time.Second) | |
| 362 | + | for time.Now().Before(deadline) { | |
| 363 | + | if condition() { | |
| 364 | + | return | |
| 225 | 365 | } | |
| 366 | + | time.Sleep(time.Millisecond) | |
| 226 | 367 | } | |
| 368 | + | t.Fatal("condition was not met") | |
| 369 | + | } | |
| 227 | 370 | ||
| 228 | - | tests := []struct { | |
| 229 | - | name string | |
| 230 | - | run func(*manualClock) floodMetrics | |
| 231 | - | check func(*testing.T, floodMetrics) | |
| 232 | - | }{ | |
| 233 | - | { | |
| 234 | - | name: "one fragment per record loss has packet contact without record or ACK progress", | |
| 235 | - | run: func(clk *manualClock) floodMetrics { | |
| 236 | - | attempt := 0 | |
| 237 | - | link := newSimulatedLink(clk, packetPolicy{Drop: func(packetMeta) bool { | |
| 238 | - | drop := attempt%len(fragments) == len(fragments)-1 | |
| 239 | - | attempt++ | |
| 240 | - | return drop | |
| 241 | - | }}) | |
| 242 | - | a, b := newPairWithLink(link) | |
| 243 | - | m := floodMetrics{} | |
| 244 | - | for range records { | |
| 245 | - | delivered := 0 | |
| 246 | - | for _, fragment := range packets { | |
| 247 | - | _, _ = a.WriteTo(fragment, b.addr) | |
| 248 | - | select { | |
| 249 | - | case <-b.in: | |
| 250 | - | delivered++ | |
| 251 | - | m.packetContact++ | |
| 252 | - | default: | |
| 253 | - | } | |
| 254 | - | } | |
| 255 | - | if delivered == len(fragments) { | |
| 256 | - | m.completedRecord++ | |
| 257 | - | m.ackProgress++ | |
| 258 | - | } | |
| 259 | - | } | |
| 260 | - | return m | |
| 261 | - | }, | |
| 262 | - | check: func(t *testing.T, m floodMetrics) { | |
| 263 | - | t.Helper() | |
| 264 | - | if m.packetContact == 0 || m.completedRecord != 0 || m.ackProgress != 0 { | |
| 265 | - | t.Fatalf("metrics=%+v, want packet contact only", m) | |
| 266 | - | } | |
| 267 | - | }, | |
| 268 | - | }, | |
| 269 | - | { | |
| 270 | - | name: "bounded bandwidth queue has no false progress and stays bounded", | |
| 271 | - | run: func(clk *manualClock) floodMetrics { | |
| 272 | - | limit := 2 * pdgram.DefaultMTU | |
| 273 | - | link := newSimulatedLink(clk, packetPolicy{Delay: time.Second, MaxQueueBytes: limit, BytesPerSecond: pdgram.DefaultMTU}) | |
| 274 | - | a, b := newPairWithLink(link) | |
| 275 | - | for range records { | |
| 276 | - | for _, fragment := range packets { | |
| 277 | - | _, _ = a.WriteTo(fragment, b.addr) | |
| 278 | - | } | |
| 279 | - | } | |
| 280 | - | return floodMetrics{queueBytes: link.queuedBytes()} | |
| 281 | - | }, | |
| 282 | - | check: func(t *testing.T, m floodMetrics) { | |
| 283 | - | t.Helper() | |
| 284 | - | if m.packetContact != 0 || m.completedRecord != 0 || m.ackProgress != 0 || m.queueBytes <= 0 || m.queueBytes > 2*pdgram.DefaultMTU { | |
| 285 | - | t.Fatalf("metrics=%+v, want bounded queued bytes without progress", m) | |
| 286 | - | } | |
| 287 | - | }, | |
| 288 | - | }, | |
| 289 | - | { | |
| 290 | - | name: "retransmit storm has record progress but no ACK progress", | |
| 291 | - | run: func(clk *manualClock) floodMetrics { | |
| 292 | - | link := newSimulatedLink(clk, packetPolicy{}) | |
| 293 | - | a, b := newPairWithLink(link) | |
| 294 | - | m := floodMetrics{} | |
| 295 | - | for range records { | |
| 296 | - | for attempt := 0; attempt < 3; attempt++ { | |
| 297 | - | for _, fragment := range packets { | |
| 298 | - | _, _ = a.WriteTo(fragment, b.addr) | |
| 299 | - | <-b.in | |
| 300 | - | m.packetContact++ | |
| 301 | - | } | |
| 302 | - | if attempt == 0 { | |
| 303 | - | m.completedRecord++ | |
| 304 | - | } else { | |
| 305 | - | m.retransmits++ | |
| 306 | - | } | |
| 307 | - | } | |
| 308 | - | } | |
| 309 | - | // ACK packets are intentionally lost in this scenario. | |
| 310 | - | return m | |
| 311 | - | }, | |
| 312 | - | check: func(t *testing.T, m floodMetrics) { | |
| 313 | - | t.Helper() | |
| 314 | - | if m.packetContact == 0 || m.completedRecord != records || m.ackProgress != 0 || m.retransmits == 0 { | |
| 315 | - | t.Fatalf("metrics=%+v, want record contact with retransmit-only pressure", m) | |
| 316 | - | } | |
| 317 | - | }, | |
| 318 | - | }, | |
| 371 | + | func TestMalformedAuthenticatedFragmentDoesNotCountAsContact(t *testing.T) { | |
| 372 | + | clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC)) | |
| 373 | + | aPC, bPC := newPair() | |
| 374 | + | a, b := newFloodTransports(t, aPC, bPC, clk, 128) | |
| 375 | + | defer closeFloodTransports(a, b) | |
| 376 | + | b.mu.Lock() | |
| 377 | + | startPacket, startRecord := b.lastAuthenticatedPacket, b.lastCompleteRecord | |
| 378 | + | b.mu.Unlock() | |
| 379 | + | codec, err := pdgram.NewCodec(key()) | |
| 380 | + | if err != nil { | |
| 381 | + | t.Fatal(err) | |
| 319 | 382 | } | |
| 320 | - | for _, tt := range tests { | |
| 321 | - | t.Run(tt.name, func(t *testing.T) { | |
| 322 | - | clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC)) | |
| 323 | - | metrics := tt.run(clk) | |
| 324 | - | tt.check(t, metrics) | |
| 325 | - | }) | |
| 383 | + | clk.advance(time.Second) | |
| 384 | + | if _, err := aPC.WriteTo(codec.Seal(1, 999, []byte("not a fragment"), nil), bPC.addr); err != nil { | |
| 385 | + | t.Fatal(err) | |
| 386 | + | } | |
| 387 | + | time.Sleep(10 * time.Millisecond) | |
| 388 | + | b.mu.Lock() | |
| 389 | + | gotPacket, gotRecord := b.lastAuthenticatedPacket, b.lastCompleteRecord | |
| 390 | + | b.mu.Unlock() | |
| 391 | + | if gotPacket != startPacket || gotRecord != startRecord { | |
| 392 | + | t.Fatal("malformed authenticated plaintext counted as packet contact") | |
| 326 | 393 | } | |
| 327 | 394 | } | |
| @@ -8,6 +8,7 @@ import ( | |||||
| 8 | "testing" | 8 | "testing" | ||
| 9 | "time" | 9 | "time" | ||
| 10 | 10 | ||||
| 11 | + | "github.com/bnema/vev/internal/ports" | |||
| 11 | pdgram "github.com/bnema/vev/pkg/dgram" | 12 | pdgram "github.com/bnema/vev/pkg/dgram" | ||
| 12 | ) | 13 | ) | ||
| 13 | 14 | ||||
| @@ -28,7 +29,6 @@ type packetMeta struct { | |||||
| 28 | Bytes int | 29 | Bytes int | ||
| 29 | At time.Time | 30 | At time.Time | ||
| 30 | QueueBytes int | 31 | QueueBytes int | ||
| 31 | - | Retransmit bool | |||
| 32 | } | 32 | } | ||
| 33 | 33 | ||||
| 34 | // packetPolicy makes loss and constrained links explicit in transport tests. | 34 | // packetPolicy makes loss and constrained links explicit in transport tests. | ||
| @@ -196,132 +196,199 @@ func (p *fakePC) SetWriteDeadline(time.Time) error { return nil } | |||||
| 196 | 196 | ||||
| 197 | func key() []byte { return bytes.Repeat([]byte{1}, pdgram.KeySize) } | 197 | func key() []byte { return bytes.Repeat([]byte{1}, pdgram.KeySize) } | ||
| 198 | 198 | ||||
| 199 | - | type floodMetrics struct { | 199 | + | // TestTransportFloodClassification drives authenticated datagrams through real |
| 200 | - | packetContact int | 200 | + | // transports. The metrics are transport state, not simulator bookkeeping. |
| 201 | - | completedRecord int | 201 | + | func TestTransportFloodClassification(t *testing.T) { |
| 202 | - | ackProgress int | 202 | + | t.Run("one real fragment per large record loss contacts without completing", func(t *testing.T) { |
| 203 | - | queueBytes int | 203 | + | clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC)) |
| 204 | - | retransmits int | 204 | + | aPC, bPC := newPair() |
| 205 | + | inspect, err := pdgram.NewCodec(key()) | |||
| 206 | + | if err != nil { | |||
| 207 | + | t.Fatal(err) | |||
| 208 | + | } | |||
| 209 | + | dropped := map[uint64]bool{} | |||
| 210 | + | aPC.drop = func(pkt []byte, _ net.Addr) bool { | |||
| 211 | + | _, raw, err := inspect.Open(pkt, 1, nil, nil) | |||
| 212 | + | if err != nil { | |||
| 213 | + | return false | |||
| 214 | + | } | |||
| 215 | + | frag, err := pdgram.UnmarshalFragment(raw) | |||
| 216 | + | if err != nil || frag.Index != frag.Count-1 || dropped[frag.Seq] { | |||
| 217 | + | return false | |||
| 218 | + | } | |||
| 219 | + | dropped[frag.Seq] = true | |||
| 220 | + | return true | |||
| 221 | + | } | |||
| 222 | + | a, b := newFloodTransports(t, aPC, bPC, clk, 128) | |||
| 223 | + | defer closeFloodTransports(a, b) | |||
| 224 | + | start := clk.Now() | |||
| 225 | + | clk.advance(time.Nanosecond) | |||
| 226 | + | for range 2 { | |||
| 227 | + | if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: make([]byte, 200)}); err != nil { | |||
| 228 | + | t.Fatal(err) | |||
| 229 | + | } | |||
| 230 | + | } | |||
| 231 | + | floodEventually(t, func() bool { return floodState(b).packet && len(dropped) == 2 }) | |||
| 232 | + | state := floodState(b) | |||
| 233 | + | if !state.packet || state.record || floodState(a).ack { | |||
| 234 | + | t.Fatalf("contact=%v complete=%v ack=%v, want contact only", state.packet, state.record, floodState(a).ack) | |||
| 235 | + | } | |||
| 236 | + | if !floodState(b).packet || !clk.Now().After(start) { | |||
| 237 | + | t.Fatal("real authenticated fragments did not update contact") | |||
| 238 | + | } | |||
| 239 | + | }) | |||
| 240 | + | ||||
| 241 | + | t.Run("bounded byte queue advances and releases on fake clock", func(t *testing.T) { | |||
| 242 | + | clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC)) | |||
| 243 | + | limit := 2 * 128 | |||
| 244 | + | link := newSimulatedLink(clk, packetPolicy{Delay: time.Second, MaxQueueBytes: limit, BytesPerSecond: 128}) | |||
| 245 | + | aPC, bPC := newPairWithLink(link) | |||
| 246 | + | a, b := newFloodTransports(t, aPC, bPC, clk, 128) | |||
| 247 | + | defer closeFloodTransports(a, b) | |||
| 248 | + | a.mu.Lock() | |||
| 249 | + | a.rto = time.Hour // Keep the fake-clock queue case about bandwidth, not retries. | |||
| 250 | + | a.mu.Unlock() | |||
| 251 | + | if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: make([]byte, 200)}); err != nil { | |||
| 252 | + | t.Fatal(err) | |||
| 253 | + | } | |||
| 254 | + | queued := link.queuedBytes() | |||
| 255 | + | if queued <= 0 || queued > limit { | |||
| 256 | + | t.Fatalf("queued bytes=%d, want 1..%d", queued, limit) | |||
| 257 | + | } | |||
| 258 | + | clk.advance(10 * time.Second) | |||
| 259 | + | link.flush(bPC) | |||
| 260 | + | floodEventually(t, func() bool { return floodState(b).packet }) | |||
| 261 | + | if got := link.queuedBytes(); got != 0 { | |||
| 262 | + | t.Fatalf("queued bytes after fake-clock release=%d, want 0", got) | |||
| 263 | + | } | |||
| 264 | + | if floodState(b).record || floodState(a).ack { | |||
| 265 | + | t.Fatal("dropped queued fragment produced false record or ACK progress") | |||
| 266 | + | } | |||
| 267 | + | }) | |||
| 268 | + | ||||
| 269 | + | t.Run("real cumulative ACK loss schedules duplicate retransmits", func(t *testing.T) { | |||
| 270 | + | clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC)) | |||
| 271 | + | aPC, bPC := newPair() | |||
| 272 | + | inspect, err := pdgram.NewCodec(key()) | |||
| 273 | + | if err != nil { | |||
| 274 | + | t.Fatal(err) | |||
| 275 | + | } | |||
| 276 | + | bPC.drop = func(pkt []byte, _ net.Addr) bool { | |||
| 277 | + | _, raw, err := inspect.Open(pkt, 2, nil, nil) | |||
| 278 | + | if err != nil { | |||
| 279 | + | return false | |||
| 280 | + | } | |||
| 281 | + | frag, err := pdgram.UnmarshalFragment(raw) | |||
| 282 | + | return err == nil && frag.Count == 1 && len(frag.Data) == 9 && frag.Data[0] == recAck | |||
| 283 | + | } | |||
| 284 | + | seen := map[uint64]int{} | |||
| 285 | + | var seenMu sync.Mutex | |||
| 286 | + | aPC.drop = func(pkt []byte, _ net.Addr) bool { | |||
| 287 | + | _, 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 | + | } | |||
| 299 | + | return false | |||
| 300 | + | } | |||
| 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 { | |||
| 305 | + | t.Fatal(err) | |||
| 306 | + | } | |||
| 307 | + | floodEventually(t, func() bool { return floodState(b).record }) | |||
| 308 | + | waitForManualTimers(t, clk, 4) | |||
| 309 | + | 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) | |||
| 318 | + | } | |||
| 319 | + | }) | |||
| 205 | } | 320 | } | ||
| 206 | 321 | ||||
| 207 | - | // TestTransportFloodClassification documents the three distinct symptoms that | 322 | + | type floodTransportState struct { |
| 208 | - | // used to be conflated as a healthy UDP link: authenticated datagrams arriving, | 323 | + | packet, record, ack bool |
| 209 | - | // complete records arriving, and ACKs retiring sender work. | 324 | + | retransmits uint64 |
| 210 | - | func TestTransportFloodClassification(t *testing.T) { | 325 | + | } |
| 211 | - | const records = 32 | 326 | + | |
| 212 | - | payload := make([]byte, 24*pdgram.DefaultMTU) | 327 | + | func floodState(t *Transport) floodTransportState { |
| 213 | - | fragments, err := pdgram.FragmentPayload(1, payload, pdgram.DefaultMTU) | 328 | + | t.mu.Lock() |
| 329 | + | defer t.mu.Unlock() | |||
| 330 | + | now := t.clock.Now() | |||
| 331 | + | return floodTransportState{ | |||
| 332 | + | packet: t.lastAuthenticatedPacket.Equal(now), | |||
| 333 | + | record: t.lastCompleteRecord.Equal(now), | |||
| 334 | + | ack: t.lastACKProgress.Equal(now), | |||
| 335 | + | retransmits: t.retransmits, | |||
| 336 | + | } | |||
| 337 | + | } | |||
| 338 | + | ||||
| 339 | + | func newFloodTransports(t *testing.T, aPC, bPC *fakePC, clk *manualClock, mtu int) (*Transport, *Transport) { | |||
| 340 | + | t.Helper() | |||
| 341 | + | opts := Options{Clock: clk, MTU: mtu, ResendAfter: 10 * time.Millisecond, Heartbeat: time.Hour} | |||
| 342 | + | a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, opts) | |||
| 214 | if err != nil { | 343 | if err != nil { | ||
| 215 | t.Fatal(err) | 344 | t.Fatal(err) | ||
| 216 | } | 345 | } | ||
| 217 | - | if len(fragments) < 24 { | 346 | + | b, err := NewTransportWithOptions(bPC, aPC.addr, key(), 2, 1, opts) |
| 218 | - | t.Fatalf("fragments=%d, want at least 24", len(fragments)) | 347 | + | if err != nil { |
| 348 | + | _ = a.Close() | |||
| 349 | + | t.Fatal(err) | |||
| 219 | } | 350 | } | ||
| 220 | - | packets := make([][]byte, len(fragments)) | 351 | + | return a, b |
| 221 | - | for i, fragment := range fragments { | 352 | + | } |
| 222 | - | packets[i], err = pdgram.MarshalFragment(fragment) | 353 | + | |
| 223 | - | if err != nil { | 354 | + | func closeFloodTransports(a, b *Transport) { |
| 224 | - | t.Fatal(err) | 355 | + | _ = a.Close() |
| 356 | + | _ = b.Close() | |||
| 357 | + | } | |||
| 358 | + | ||||
| 359 | + | func floodEventually(t *testing.T, condition func() bool) { | |||
| 360 | + | t.Helper() | |||
| 361 | + | deadline := time.Now().Add(time.Second) | |||
| 362 | + | for time.Now().Before(deadline) { | |||
| 363 | + | if condition() { | |||
| 364 | + | return | |||
| 225 | } | 365 | } | ||
| 366 | + | time.Sleep(time.Millisecond) | |||
| 226 | } | 367 | } | ||
| 368 | + | t.Fatal("condition was not met") | |||
| 369 | + | } | |||
| 227 | 370 | ||||
| 228 | - | tests := []struct { | 371 | + | func TestMalformedAuthenticatedFragmentDoesNotCountAsContact(t *testing.T) { |
| 229 | - | name string | 372 | + | clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC)) |
| 230 | - | run func(*manualClock) floodMetrics | 373 | + | aPC, bPC := newPair() |
| 231 | - | check func(*testing.T, floodMetrics) | 374 | + | a, b := newFloodTransports(t, aPC, bPC, clk, 128) |
| 232 | - | }{ | 375 | + | defer closeFloodTransports(a, b) |
| 233 | - | { | 376 | + | b.mu.Lock() |
| 234 | - | name: "one fragment per record loss has packet contact without record or ACK progress", | 377 | + | startPacket, startRecord := b.lastAuthenticatedPacket, b.lastCompleteRecord |
| 235 | - | run: func(clk *manualClock) floodMetrics { | 378 | + | b.mu.Unlock() |
| 236 | - | attempt := 0 | 379 | + | codec, err := pdgram.NewCodec(key()) |
| 237 | - | link := newSimulatedLink(clk, packetPolicy{Drop: func(packetMeta) bool { | 380 | + | if err != nil { |
| 238 | - | drop := attempt%len(fragments) == len(fragments)-1 | 381 | + | t.Fatal(err) |
| 239 | - | attempt++ | |||
| 240 | - | return drop | |||
| 241 | - | }}) | |||
| 242 | - | a, b := newPairWithLink(link) | |||
| 243 | - | m := floodMetrics{} | |||
| 244 | - | for range records { | |||
| 245 | - | delivered := 0 | |||
| 246 | - | for _, fragment := range packets { | |||
| 247 | - | _, _ = a.WriteTo(fragment, b.addr) | |||
| 248 | - | select { | |||
| 249 | - | case <-b.in: | |||
| 250 | - | delivered++ | |||
| 251 | - | m.packetContact++ | |||
| 252 | - | default: | |||
| 253 | - | } | |||
| 254 | - | } | |||
| 255 | - | if delivered == len(fragments) { | |||
| 256 | - | m.completedRecord++ | |||
| 257 | - | m.ackProgress++ | |||
| 258 | - | } | |||
| 259 | - | } | |||
| 260 | - | return m | |||
| 261 | - | }, | |||
| 262 | - | check: func(t *testing.T, m floodMetrics) { | |||
| 263 | - | t.Helper() | |||
| 264 | - | if m.packetContact == 0 || m.completedRecord != 0 || m.ackProgress != 0 { | |||
| 265 | - | t.Fatalf("metrics=%+v, want packet contact only", m) | |||
| 266 | - | } | |||
| 267 | - | }, | |||
| 268 | - | }, | |||
| 269 | - | { | |||
| 270 | - | name: "bounded bandwidth queue has no false progress and stays bounded", | |||
| 271 | - | run: func(clk *manualClock) floodMetrics { | |||
| 272 | - | limit := 2 * pdgram.DefaultMTU | |||
| 273 | - | link := newSimulatedLink(clk, packetPolicy{Delay: time.Second, MaxQueueBytes: limit, BytesPerSecond: pdgram.DefaultMTU}) | |||
| 274 | - | a, b := newPairWithLink(link) | |||
| 275 | - | for range records { | |||
| 276 | - | for _, fragment := range packets { | |||
| 277 | - | _, _ = a.WriteTo(fragment, b.addr) | |||
| 278 | - | } | |||
| 279 | - | } | |||
| 280 | - | return floodMetrics{queueBytes: link.queuedBytes()} | |||
| 281 | - | }, | |||
| 282 | - | check: func(t *testing.T, m floodMetrics) { | |||
| 283 | - | t.Helper() | |||
| 284 | - | if m.packetContact != 0 || m.completedRecord != 0 || m.ackProgress != 0 || m.queueBytes <= 0 || m.queueBytes > 2*pdgram.DefaultMTU { | |||
| 285 | - | t.Fatalf("metrics=%+v, want bounded queued bytes without progress", m) | |||
| 286 | - | } | |||
| 287 | - | }, | |||
| 288 | - | }, | |||
| 289 | - | { | |||
| 290 | - | name: "retransmit storm has record progress but no ACK progress", | |||
| 291 | - | run: func(clk *manualClock) floodMetrics { | |||
| 292 | - | link := newSimulatedLink(clk, packetPolicy{}) | |||
| 293 | - | a, b := newPairWithLink(link) | |||
| 294 | - | m := floodMetrics{} | |||
| 295 | - | for range records { | |||
| 296 | - | for attempt := 0; attempt < 3; attempt++ { | |||
| 297 | - | for _, fragment := range packets { | |||
| 298 | - | _, _ = a.WriteTo(fragment, b.addr) | |||
| 299 | - | <-b.in | |||
| 300 | - | m.packetContact++ | |||
| 301 | - | } | |||
| 302 | - | if attempt == 0 { | |||
| 303 | - | m.completedRecord++ | |||
| 304 | - | } else { | |||
| 305 | - | m.retransmits++ | |||
| 306 | - | } | |||
| 307 | - | } | |||
| 308 | - | } | |||
| 309 | - | // ACK packets are intentionally lost in this scenario. | |||
| 310 | - | return m | |||
| 311 | - | }, | |||
| 312 | - | check: func(t *testing.T, m floodMetrics) { | |||
| 313 | - | t.Helper() | |||
| 314 | - | if m.packetContact == 0 || m.completedRecord != records || m.ackProgress != 0 || m.retransmits == 0 { | |||
| 315 | - | t.Fatalf("metrics=%+v, want record contact with retransmit-only pressure", m) | |||
| 316 | - | } | |||
| 317 | - | }, | |||
| 318 | - | }, | |||
| 319 | } | 382 | } | ||
| 320 | - | for _, tt := range tests { | 383 | + | clk.advance(time.Second) |
| 321 | - | t.Run(tt.name, func(t *testing.T) { | 384 | + | if _, err := aPC.WriteTo(codec.Seal(1, 999, []byte("not a fragment"), nil), bPC.addr); err != nil { |
| 322 | - | clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC)) | 385 | + | t.Fatal(err) |
| 323 | - | metrics := tt.run(clk) | 386 | + | } |
| 324 | - | tt.check(t, metrics) | 387 | + | time.Sleep(10 * time.Millisecond) |
| 325 | - | }) | 388 | + | b.mu.Lock() |
| 389 | + | gotPacket, gotRecord := b.lastAuthenticatedPacket, b.lastCompleteRecord | |||
| 390 | + | b.mu.Unlock() | |||
| 391 | + | if gotPacket != startPacket || gotRecord != startRecord { | |||
| 392 | + | t.Fatal("malformed authenticated plaintext counted as packet contact") | |||
| 326 | } | 393 | } | ||
| 327 | } | 394 | } | ||