Repository: quic-go/quic-go
Pull request #3854 (2023-05-23 11:01:48)

server: send version negotiation and invalid token packets from a single Go routine
closes: https://github.com/quic-go/quic-go/issues/3765



codecov[bot]: ## [Codecov](https://app.codecov.io/gh/quic-go/quic-go/pull/3854?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) Report
> Merging [#3854](https://app.codecov.io/gh/quic-go/quic-go/pull/3854?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) (0c77ff9) into [master](https://app.codecov.io/gh/quic-go/quic-go/commit/f401a73d271b76a353313de227fb19ff53a8bedd?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) (f401a73) will **decrease** coverage by `0.08%`.
> The diff coverage is `68.24%`.

```diff
@@            Coverage Diff             @@
##           master    #3854      +/-   ##
==========================================
- Coverage   83.92%   83.84%   -0.08%     
==========================================
  Files         146      146              
  Lines       14307    14347      +40     
==========================================
+ Hits        12006    12028      +22     
- Misses       1860     1875      +15     
- Partials      441      444       +3     
```


| [Impacted Files](https://app.codecov.io/gh/quic-go/quic-go/pull/3854?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) | Coverage Δ | |
|---|---|---|
| [server.go](https://app.codecov.io/gh/quic-go/quic-go/pull/3854?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-c2VydmVyLmdv) | `76.25% <68.24%> (-1.63%)` | :arrow_down: |

... and [3 files with indirect coverage changes](https://app.codecov.io/gh/quic-go/quic-go/pull/3854/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None)



Patch:
diff --git a/internal/utils/rtt_stats.go b/internal/utils/rtt_stats.go
index 527539e1..2cd9a191 100644
--- a/internal/utils/rtt_stats.go
+++ b/internal/utils/rtt_stats.go
@@ -103,8 +103,12 @@ func (r *RTTStats) SetMaxAckDelay(mad time.Duration) {
 // SetInitialRTT sets the initial RTT.
 // It is used during the 0-RTT handshake when restoring the RTT stats from the session state.
 func (r *RTTStats) SetInitialRTT(t time.Duration) {
+	// On the server side, by the time we get to process the session ticket,
+	// we might already have obtained an RTT measurement.
+	// This can happen if we received the ClientHello in multiple pieces, and one of those pieces was lost.
+	// Discard the restored value. A fresh measurement is always better.
 	if r.hasMeasurement {
-		panic("initial RTT set after first measurement")
+		return
 	}
 	r.smoothedRTT = t
 	r.latestRTT = t
diff --git a/internal/wire/ack_frame.go b/internal/wire/ack_frame.go
index f145c8b4..2f816c78 100644
--- a/internal/wire/ack_frame.go
+++ b/internal/wire/ack_frame.go
@@ -90,13 +90,22 @@ func parseAckFrame(r *bytes.Reader, typ uint64, ackDelayExponent uint8, _ protoc
 		return nil, errInvalidAckRanges
 	}
 
-	// parse (and skip) the ECN section
 	if ecn {
-		for i := 0; i < 3; i++ {
-			if _, err := quicvarint.Read(r); err != nil {
-				return nil, err
-			}
+		ect0, err := quicvarint.Read(r)
+		if err != nil {
+			return nil, err
+		}
+		frame.ECT0 = ect0
+		ect1, err := quicvarint.Read(r)
+		if err != nil {
+			return nil, err
+		}
+		frame.ECT1 = ect1
+		ecnce, err := quicvarint.Read(r)
+		if err != nil {
+			return nil, err
 		}
+		frame.ECNCE = ecnce
 	}
 
 	return frame, nil
diff --git a/packet_packer.go b/packet_packer.go
index 14befd46..180a03e1 100644
--- a/packet_packer.go
+++ b/packet_packer.go
@@ -417,7 +417,7 @@ func (p *packetPacker) PackCoalescedPacket(onlyAck bool, v protocol.VersionNumbe
 			if oneRTTPayload.length > 0 {
 				size += p.shortHeaderPacketLength(connID, oneRTTPacketNumberLen, oneRTTPayload) + protocol.ByteCount(oneRTTSealer.Overhead())
 			}
-		} else if p.perspective == protocol.PerspectiveClient { // 0-RTT
+		} else if p.perspective == protocol.PerspectiveClient && !onlyAck { // 0-RTT packets can't contain ACK frames
 			var err error
 			zeroRTTSealer, err = p.cryptoSetup.Get0RTTSealer()
 			if err != nil && err != handshake.ErrKeysDropped && err != handshake.ErrKeysNotYetAvailable {
diff --git a/server.go b/server.go
index f5b04549..4043c58c 100644
--- a/server.go
+++ b/server.go
@@ -98,10 +98,12 @@ type baseServer struct {
 		protocol.VersionNumber,
 	) quicConn
 
-	serverError error
-	errorChan   chan struct{}
-	closed      bool
-	running     chan struct{} // closed as soon as run() returns
+	serverError             error
+	errorChan               chan struct{}
+	closed                  bool
+	running                 chan struct{} // closed as soon as run() returns
+	versionNegotiationQueue chan *receivedPacket
+	invalidTokenQueue       chan *receivedPacket
 
 	connQueue    chan quicConn
 	connQueueLen int32 // to be used as an atomic
@@ -229,26 +231,29 @@ func newServer(
 		return nil, err
 	}
 	s := &baseServer{
-		conn:             conn,
-		tlsConf:          tlsConf,
-		config:           config,
-		tokenGenerator:   tokenGenerator,
-		connIDGenerator:  connIDGenerator,
-		connHandler:      connHandler,
-		connQueue:        make(chan quicConn),
-		errorChan:        make(chan struct{}),
-		running:          make(chan struct{}),
-		receivedPackets:  make(chan *receivedPacket, protocol.MaxServerUnprocessedPackets),
-		newConn:          newConnection,
-		tracer:           tracer,
-		logger:           utils.DefaultLogger.WithPrefix("server"),
-		acceptEarlyConns: acceptEarly,
-		onClose:          onClose,
+		conn:                    conn,
+		tlsConf:                 tlsConf,
+		config:                  config,
+		tokenGenerator:          tokenGenerator,
+		connIDGenerator:         connIDGenerator,
+		connHandler:             connHandler,
+		connQueue:               make(chan quicConn),
+		errorChan:               make(chan struct{}),
+		running:                 make(chan struct{}),
+		receivedPackets:         make(chan *receivedPacket, protocol.MaxServerUnprocessedPackets),
+		versionNegotiationQueue: make(chan *receivedPacket, 4),
+		invalidTokenQueue:       make(chan *receivedPacket, 4),
+		newConn:                 newConnection,
+		tracer:                  tracer,
+		logger:                  utils.DefaultLogger.WithPrefix("server"),
+		acceptEarlyConns:        acceptEarly,
+		onClose:                 onClose,
 	}
 	if acceptEarly {
 		s.zeroRTTQueues = map[protocol.ConnectionID]*zeroRTTQueue{}
 	}
 	go s.run()
+	go s.runSendQueue()
 	s.logger.Debugf("Listening for %s connections on %s", conn.LocalAddr().Network(), conn.LocalAddr().String())
 	return s, nil
 }
@@ -272,6 +277,19 @@ func (s *baseServer) run() {
 	}
 }
 
+func (s *baseServer) runSendQueue() {
+	for {
+		select {
+		case <-s.running:
+			return
+		case p := <-s.versionNegotiationQueue:
+			s.maybeSendVersionNegotiationPacket(p)
+		case p := <-s.invalidTokenQueue:
+			s.maybeSendInvalidToken(p)
+		}
+	}
+}
+
 // Accept returns connections that already completed the handshake.
 // It is only valid if acceptEarlyConns is false.
 func (s *baseServer) Accept(ctx context.Context) (Connection, error) {
@@ -353,27 +371,28 @@ func (s *baseServer) handlePacketImpl(p *receivedPacket) bool /* is the buffer s
 		panic(fmt.Sprintf("misrouted packet: %#v", p.data))
 	}
 	v, err := wire.ParseVersion(p.data)
+	// drop the packet if we failed to parse the protocol version
+	if err != nil {
+		s.logger.Debugf("Dropping a packet with an unknown version")
+		if s.tracer != nil {
+			s.tracer.DroppedPacket(p.remoteAddr, logging.PacketTypeNotDetermined, p.Size(), logging.PacketDropUnexpectedPacket)
+		}
+		return false
+	}
 	// send a Version Negotiation Packet if the client is speaking a different protocol version
-	if err != nil || !protocol.IsSupportedVersion(s.config.Versions, v) {
-		if err != nil || p.Size() < protocol.MinUnknownVersionPacketSize {
-			s.logger.Debugf("Dropping a packet with an unknown version that is too small (%d bytes)", p.Size())
-			if s.tracer != nil {
-				s.tracer.DroppedPacket(p.remoteAddr, logging.PacketTypeNotDetermined, p.Size(), logging.PacketDropUnexpectedPacket)
-			}
+	if !protocol.IsSupportedVersion(s.config.Versions, v) {
+		if s.config.DisableVersionNegotiationPackets {
 			return false
 		}
-		_, src, dest, err := wire.ParseArbitraryLenConnectionIDs(p.data)
-		if err != nil { // should never happen
-			s.logger.Debugf("Dropping a packet with an unknown version for which we failed to parse connection IDs")
+
+		if p.Size() < protocol.MinUnknownVersionPacketSize {
+			s.logger.Debugf("Dropping a packet with an unsupported version number %d that is too small (%d bytes)", v, p.Size())
 			if s.tracer != nil {
 				s.tracer.DroppedPacket(p.remoteAddr, logging.PacketTypeNotDetermined, p.Size(), logging.PacketDropUnexpectedPacket)
 			}
 			return false
 		}
-		if !s.config.DisableVersionNegotiationPackets {
-			go s.sendVersionNegotiationPacket(p.remoteAddr, src, dest, p.info.OOB(), v)
-		}
-		return false
+		return s.enqueueVersionNegotiationPacket(p)
 	}
 
 	if wire.Is0RTTPacket(p.data) {
@@ -557,12 +576,7 @@ func (s *baseServer) handleInitialImpl(p *receivedPacket, hdr *wire.Header) erro
 			// For Retry tokens, we send an INVALID_ERROR if
 			// * the token is too old, or
 			// * the token is invalid, in case of a retry token.
-			go func() {
-				defer p.buffer.Release()
-				if err := s.maybeSendInvalidToken(p, hdr); err != nil {
-					s.logger.Debugf("Error sending INVALID_TOKEN error: %s", err)
-				}
-			}()
+			s.enqueueInvalidToken(p)
 			return nil
 		}
 	}
@@ -730,7 +744,27 @@ func (s *baseServer) sendRetry(remoteAddr net.Addr, hdr *wire.Header, info *pack
 	return err
 }
 
-func (s *baseServer) maybeSendInvalidToken(p *receivedPacket, hdr *wire.Header) error {
+func (s *baseServer) enqueueInvalidToken(p *receivedPacket) {
+	select {
+	case s.invalidTokenQueue <- p:
+	default:
+		// it's fine to drop INVALID_TOKEN packets when we are busy
+		p.buffer.Release()
+	}
+}
+
+func (s *baseServer) maybeSendInvalidToken(p *receivedPacket) {
+	defer p.buffer.Release()
+
+	hdr, _, _, err := wire.ParsePacket(p.data)
+	if err != nil {
+		if s.tracer != nil {
+			s.tracer.DroppedPacket(p.remoteAddr, logging.PacketTypeNotDetermined, p.Size(), logging.PacketDropHeaderParseError)
+		}
+		s.logger.Debugf("Error parsing packet: %s", err)
+		return
+	}
+
 	// Only send INVALID_TOKEN if we can unprotect the packet.
 	// This makes sure that we won't send it for packets that were corrupted.
 	sealer, opener := handshake.NewInitialAEAD(hdr.DestConnectionID, protocol.PerspectiveServer, hdr.Version)
@@ -740,21 +774,21 @@ func (s *baseServer) maybeSendInvalidToken(p *receivedPacket, hdr *wire.Header)
 		if s.tracer != nil {
 			s.tracer.DroppedPacket(p.remoteAddr, logging.PacketTypeInitial, p.Size(), logging.PacketDropHeaderParseError)
 		}
-		// don't return the error here. Just drop the packet.
-		return nil
+		return
 	}
 	hdrLen := extHdr.ParsedLen()
 	if _, err := opener.Open(data[hdrLen:hdrLen], data[hdrLen:], extHdr.PacketNumber, data[:hdrLen]); err != nil {
-		// don't return the error here. Just drop the packet.
 		if s.tracer != nil {
 			s.tracer.DroppedPacket(p.remoteAddr, logging.PacketTypeInitial, p.Size(), logging.PacketDropPayloadDecryptError)
 		}
-		return nil
+		return
 	}
 	if s.logger.Debug() {
 		s.logger.Debugf("Client sent an invalid retry token. Sending INVALID_TOKEN to %s.", p.remoteAddr)
 	}
-	return s.sendError(p.remoteAddr, hdr, sealer, qerr.InvalidToken, p.info)
+	if err := s.sendError(p.remoteAddr, hdr, sealer, qerr.InvalidToken, p.info); err != nil {
+		s.logger.Debugf("Error sending INVALID_TOKEN error: %s", err)
+	}
 }
 
 func (s *baseServer) sendConnectionRefused(remoteAddr net.Addr, hdr *wire.Header, info *packetInfo) error {
@@ -807,14 +841,41 @@ func (s *baseServer) sendError(remoteAddr net.Addr, hdr *wire.Header, sealer han
 	return err
 }
 
-func (s *baseServer) sendVersionNegotiationPacket(remote net.Addr, src, dest protocol.ArbitraryLenConnectionID, oob []byte, v protocol.VersionNumber) {
+func (s *baseServer) enqueueVersionNegotiationPacket(p *receivedPacket) (bufferInUse bool) {
+	select {
+	case s.versionNegotiationQueue <- p:
+		return true
+	default:
+		// it's fine to not send version negotiation packets when we are busy
+	}
+	return false
+}
+
+func (s *baseServer) maybeSendVersionNegotiationPacket(p *receivedPacket) {
+	defer p.buffer.Release()
+
+	v, err := wire.ParseVersion(p.data)
+	if err != nil {
+		s.logger.Debugf("failed to parse version for sending version negotiation packet: %s", err)
+		return
+	}
+
+	_, src, dest, err := wire.ParseArbitraryLenConnectionIDs(p.data)
+	if err != nil { // should never happen
+		s.logger.Debugf("Dropping a packet with an unknown version for which we failed to parse connection IDs")
+		if s.tracer != nil {
+			s.tracer.DroppedPacket(p.remoteAddr, logging.PacketTypeNotDetermined, p.Size(), logging.PacketDropUnexpectedPacket)
+		}
+		return
+	}
+
 	s.logger.Debugf("Client offered version %s, sending Version Negotiation", v)
 
 	data := wire.ComposeVersionNegotiation(dest, src, s.config.Versions)
 	if s.tracer != nil {
-		s.tracer.SentVersionNegotiationPacket(remote, src, dest, s.config.Versions)
+		s.tracer.SentVersionNegotiationPacket(p.remoteAddr, src, dest, s.config.Versions)
 	}
-	if _, err := s.conn.WritePacket(data, remote, oob); err != nil {
+	if _, err := s.conn.WritePacket(data, p.remoteAddr, p.info.OOB()); err != nil {
 		s.logger.Debugf("Error sending Version Negotiation: %s", err)
 	}
 }


Test patch:
diff --git a/integrationtests/tools/proxy/proxy.go b/integrationtests/tools/proxy/proxy.go
index 01e6a4a9..4ed2d89d 100644
--- a/integrationtests/tools/proxy/proxy.go
+++ b/integrationtests/tools/proxy/proxy.go
@@ -169,6 +169,12 @@ func NewQuicProxy(local string, opts *Opts) (*QuicProxy, error) {
 	if err != nil {
 		return nil, err
 	}
+	if err := conn.SetReadBuffer(protocol.DesiredReceiveBufferSize); err != nil {
+		return nil, err
+	}
+	if err := conn.SetWriteBuffer(protocol.DesiredSendBufferSize); err != nil {
+		return nil, err
+	}
 	raddr, err := net.ResolveUDPAddr("udp", opts.RemoteAddr)
 	if err != nil {
 		return nil, err
@@ -225,13 +231,19 @@ func (p *QuicProxy) LocalPort() int {
 }
 
 func (p *QuicProxy) newConnection(cliAddr *net.UDPAddr) (*connection, error) {
-	srvudp, err := net.DialUDP("udp", nil, p.serverAddr)
+	conn, err := net.DialUDP("udp", nil, p.serverAddr)
 	if err != nil {
 		return nil, err
 	}
+	if err := conn.SetReadBuffer(protocol.DesiredReceiveBufferSize); err != nil {
+		return nil, err
+	}
+	if err := conn.SetWriteBuffer(protocol.DesiredSendBufferSize); err != nil {
+		return nil, err
+	}
 	return &connection{
 		ClientAddr:      cliAddr,
-		ServerConn:      srvudp,
+		ServerConn:      conn,
 		incomingPackets: make(chan packetEntry, 10),
 		Incoming:        newQueue(),
 		Outgoing:        newQueue(),
diff --git a/internal/utils/rtt_stats_test.go b/internal/utils/rtt_stats_test.go
index 29892c6f..eae31926 100644
--- a/internal/utils/rtt_stats_test.go
+++ b/internal/utils/rtt_stats_test.go
@@ -154,4 +154,14 @@ var _ = Describe("RTT stats", func() {
 		Expect(rttStats.SmoothedRTT()).To(Equal(200 * time.Millisecond))
 		Expect(rttStats.MeanDeviation()).To(Equal(100 * time.Millisecond))
 	})
+
+	It("doesn't restore the RTT if we already have a measurement", func() {
+		const rtt = 10 * time.Millisecond
+		rttStats.UpdateRTT(rtt, 0, time.Now())
+		Expect(rttStats.LatestRTT()).To(Equal(rtt))
+		Expect(rttStats.SmoothedRTT()).To(Equal(rtt))
+		rttStats.SetInitialRTT(time.Minute)
+		Expect(rttStats.LatestRTT()).To(Equal(rtt))
+		Expect(rttStats.SmoothedRTT()).To(Equal(rtt))
+	})
 })
diff --git a/internal/wire/ack_frame_test.go b/internal/wire/ack_frame_test.go
index 49d11374..f80335c0 100644
--- a/internal/wire/ack_frame_test.go
+++ b/internal/wire/ack_frame_test.go
@@ -170,6 +170,9 @@ var _ = Describe("ACK Frame (for IETF QUIC)", func() {
 				Expect(frame.LargestAcked()).To(Equal(protocol.PacketNumber(100)))
 				Expect(frame.LowestAcked()).To(Equal(protocol.PacketNumber(90)))
 				Expect(frame.HasMissingRanges()).To(BeFalse())
+				Expect(frame.ECT0).To(BeEquivalentTo(0x42))
+				Expect(frame.ECT1).To(BeEquivalentTo(0x12345))
+				Expect(frame.ECNCE).To(BeEquivalentTo(0x12345678))
 				Expect(b.Len()).To(BeZero())
 			})
 
diff --git a/packet_packer_test.go b/packet_packer_test.go
index 31caa17a..05dc8689 100644
--- a/packet_packer_test.go
+++ b/packet_packer_test.go
@@ -316,6 +316,12 @@ var _ = Describe("Packet packer", func() {
 				Expect(p.longHdrPackets[0].EncryptionLevel()).To(Equal(protocol.Encryption0RTT))
 				Expect(p.longHdrPackets[0].frames).To(Equal([]*ackhandler.Frame{cf}))
 			})
+
+			It("doesn't add an ACK-only 0-RTT packet", func() { // ACK frames cannot be sent in 0-RTT packets
+				p, err := packer.PackCoalescedPacket(true, protocol.Version1)
+				Expect(err).ToNot(HaveOccurred())
+				Expect(p).To(BeNil())
+			})
 		})
 
 		Context("packing CONNECTION_CLOSE", func() {