fix(quicnettest): emit the impairment event before scheduling the packet
A zero-latency packet could be delivered and answered before ingest emitted its event, so the reply's event was observed first.
- commit
- fe90d45828e97e53992daf093bcfca0bf8a8b812
- author
- brice · · 2026-10-08 08:07 UTC
- parent
- dce3fad
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email - refs
- fix/ci-race-and-proxy-event-order
1 file · +11 -8
M internal/adapters/quicnettest/proxy.go +11 -8
| @@ -468,6 +468,17 @@ func (p *Proxy) ingest(toServer bool, raw []byte, dst *net.UDPAddr) { | |||
| 468 | 468 | counters.reordered.Add(1) | |
| 469 | 469 | } | |
| 470 | 470 | due := now.Add(link.BaseLatency + decision.ExtraDelay) | |
| 471 | + | // Emit before scheduling: once enqueued, a zero-latency copy can be | |
| 472 | + | // delivered and answered before this call returns, and the reply's event | |
| 473 | + | // must not be observed ahead of the packet that caused it. | |
| 474 | + | p.emit(Event{ | |
| 475 | + | Direction: directionOf(toServer), | |
| 476 | + | Sequence: seq, | |
| 477 | + | Bytes: len(raw), | |
| 478 | + | At: now, | |
| 479 | + | Due: due, | |
| 480 | + | Decision: decision, | |
| 481 | + | }) | |
| 471 | 482 | for copyIndex := 0; copyIndex <= decision.Duplicates; copyIndex++ { | |
| 472 | 483 | copyDue := due.Add(time.Duration(copyIndex) * link.DuplicateDelay) | |
| 473 | 484 | if !p.enqueue(packet{ | |
| @@ -480,14 +491,6 @@ func (p *Proxy) ingest(toServer bool, raw []byte, dst *net.UDPAddr) { | |||
| 480 | 491 | counters.overflowDrops.Add(1) | |
| 481 | 492 | } | |
| 482 | 493 | } | |
| 483 | - | p.emit(Event{ | |
| 484 | - | Direction: directionOf(toServer), | |
| 485 | - | Sequence: seq, | |
| 486 | - | Bytes: len(raw), | |
| 487 | - | At: now, | |
| 488 | - | Due: due, | |
| 489 | - | Decision: decision, | |
| 490 | - | }) | |
| 491 | 494 | } | |
| 492 | 495 | ||
| 493 | 496 | func (p *Proxy) enqueue(pkt packet) bool { | |
| @@ -468,6 +468,17 @@ func (p *Proxy) ingest(toServer bool, raw []byte, dst *net.UDPAddr) { | |||||
| 468 | counters.reordered.Add(1) | 468 | counters.reordered.Add(1) | ||
| 469 | } | 469 | } | ||
| 470 | due := now.Add(link.BaseLatency + decision.ExtraDelay) | 470 | due := now.Add(link.BaseLatency + decision.ExtraDelay) | ||
| 471 | + | // Emit before scheduling: once enqueued, a zero-latency copy can be | |||
| 472 | + | // delivered and answered before this call returns, and the reply's event | |||
| 473 | + | // must not be observed ahead of the packet that caused it. | |||
| 474 | + | p.emit(Event{ | |||
| 475 | + | Direction: directionOf(toServer), | |||
| 476 | + | Sequence: seq, | |||
| 477 | + | Bytes: len(raw), | |||
| 478 | + | At: now, | |||
| 479 | + | Due: due, | |||
| 480 | + | Decision: decision, | |||
| 481 | + | }) | |||
| 471 | for copyIndex := 0; copyIndex <= decision.Duplicates; copyIndex++ { | 482 | for copyIndex := 0; copyIndex <= decision.Duplicates; copyIndex++ { | ||
| 472 | copyDue := due.Add(time.Duration(copyIndex) * link.DuplicateDelay) | 483 | copyDue := due.Add(time.Duration(copyIndex) * link.DuplicateDelay) | ||
| 473 | if !p.enqueue(packet{ | 484 | if !p.enqueue(packet{ | ||
| @@ -480,14 +491,6 @@ func (p *Proxy) ingest(toServer bool, raw []byte, dst *net.UDPAddr) { | |||||
| 480 | counters.overflowDrops.Add(1) | 491 | counters.overflowDrops.Add(1) | ||
| 481 | } | 492 | } | ||
| 482 | } | 493 | } | ||
| 483 | - | p.emit(Event{ | |||
| 484 | - | Direction: directionOf(toServer), | |||
| 485 | - | Sequence: seq, | |||
| 486 | - | Bytes: len(raw), | |||
| 487 | - | At: now, | |||
| 488 | - | Due: due, | |||
| 489 | - | Decision: decision, | |||
| 490 | - | }) | |||
| 491 | } | 494 | } | ||
| 492 | 495 | ||||
| 493 | func (p *Proxy) enqueue(pkt packet) bool { | 496 | func (p *Proxy) enqueue(pkt packet) bool { | ||