quic/conn.go - net - Git at Google

 // Copyright 2023 The Go Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style
 // license that can be found in the LICENSE file.

 //go:build go1.21

 package quic

 import (
 	"context"
 	"crypto/tls"
 	"errors"
 	"fmt"
 	"log/slog"
 	"net/netip"
 	"time"
 )

 // A Conn is a QUIC connection.
 //
 // Multiple goroutines may invoke methods on a Conn simultaneously.
 type Conn struct {
 	side      connSide
 	endpoint  *Endpoint
 	config    *Config
 	testHooks connTestHooks
 	peerAddr  netip.AddrPort
 	localAddr netip.AddrPort

 	msgc  chan any
 	donec chan struct{} // closed when conn loop exits

 	w           packetWriter
 	acks        [numberSpaceCount]ackState // indexed by number space
 	lifetime    lifetimeState
 	idle        idleState
 	connIDState connIDState
 	loss        lossState
 	streams     streamsState
 	path        pathState

 	// Packet protection keys, CRYPTO streams, and TLS state.
 	keysInitial   fixedKeyPair
 	keysHandshake fixedKeyPair
 	keysAppData   updatingKeyPair
 	crypto        [numberSpaceCount]cryptoStream
 	tls           *tls.QUICConn

 	// retryToken is the token provided by the peer in a Retry packet.
 	retryToken []byte

 	// handshakeConfirmed is set when the handshake is confirmed.
 	// For server connections, it tracks sending HANDSHAKE_DONE.
 	handshakeConfirmed sentVal

 	peerAckDelayExponent int8 // -1 when unknown

 	// Tests only: Send a PING in a specific number space.
 	testSendPingSpace numberSpace
 	testSendPing      sentVal

 	log *slog.Logger
 }

 // connTestHooks override conn behavior in tests.
 type connTestHooks interface {
 	// init is called after a conn is created.
 	init()

 	// nextMessage is called to request the next event from msgc.
 	// Used to give tests control of the connection event loop.
 	nextMessage(msgc chan any, nextTimeout time.Time) (now time.Time, message any)

 	// handleTLSEvent is called with each TLS event.
 	handleTLSEvent(tls.QUICEvent)

 	// newConnID is called to generate a new connection ID.
 	// Permits tests to generate consistent connection IDs rather than random ones.
 	newConnID(seq int64) ([]byte, error)

 	// waitUntil blocks until the until func returns true or the context is done.
 	// Used to synchronize asynchronous blocking operations in tests.
 	waitUntil(ctx context.Context, until func() bool) error

 	// timeNow returns the current time.
 	timeNow() time.Time
 }

 // newServerConnIDs is connection IDs associated with a new server connection.
 type newServerConnIDs struct {
 	srcConnID         []byte // source from client's current Initial
 	dstConnID         []byte // destination from client's current Initial
 	originalDstConnID []byte // destination from client's first Initial
 	retrySrcConnID    []byte // source from server's Retry
 }

 func newConn(now time.Time, side connSide, cids newServerConnIDs, peerHostname string, peerAddr netip.AddrPort, config *Config, e *Endpoint) (conn *Conn, _ error) {
 	c := &Conn{
 		side:                 side,
 		endpoint:             e,
 		config:               config,
 		peerAddr:             unmapAddrPort(peerAddr),
 		msgc:                 make(chan any, 1),
 		donec:                make(chan struct{}),
 		peerAckDelayExponent: -1,
 	}
 	defer func() {
 		// If we hit an error in newConn, close donec so tests don't get stuck waiting for it.
 		// This is only relevant if we've got a bug, but it makes tracking that bug down
 		// much easier.
 		if conn == nil {
 			close(c.donec)
 		}
 	}()

 	// A one-element buffer allows us to wake a Conn's event loop as a
 	// non-blocking operation.
 	c.msgc = make(chan any, 1)

 	if e.testHooks != nil {
 		e.testHooks.newConn(c)
 	}

 	// initialConnID is the connection ID used to generate Initial packet protection keys.
 	var initialConnID []byte
 	if c.side == clientSide {
 		if err := c.connIDState.initClient(c); err != nil {
 			return nil, err
 		}
 		initialConnID, _ = c.connIDState.dstConnID()
 	} else {
 		initialConnID = cids.originalDstConnID
 		if cids.retrySrcConnID != nil {
 			initialConnID = cids.retrySrcConnID
 		}
 		if err := c.connIDState.initServer(c, cids); err != nil {
 			return nil, err
 		}
 	}

 	// TODO: PMTU discovery.
 	c.logConnectionStarted(cids.originalDstConnID, peerAddr)
 	c.keysAppData.init()
 	c.loss.init(c.side, smallestMaxDatagramSize, now)
 	c.streamsInit()
 	c.lifetimeInit()
 	c.restartIdleTimer(now)

 	if err := c.startTLS(now, initialConnID, peerHostname, transportParameters{
 		initialSrcConnID:               c.connIDState.srcConnID(),
 		originalDstConnID:              cids.originalDstConnID,
 		retrySrcConnID:                 cids.retrySrcConnID,
 		ackDelayExponent:               ackDelayExponent,
 		maxUDPPayloadSize:              maxUDPPayloadSize,
 		maxAckDelay:                    maxAckDelay,
 		disableActiveMigration:         true,
 		initialMaxData:                 config.maxConnReadBufferSize(),
 		initialMaxStreamDataBidiLocal:  config.maxStreamReadBufferSize(),
 		initialMaxStreamDataBidiRemote: config.maxStreamReadBufferSize(),
 		initialMaxStreamDataUni:        config.maxStreamReadBufferSize(),
 		initialMaxStreamsBidi:          c.streams.remoteLimit[bidiStream].max,
 		initialMaxStreamsUni:           c.streams.remoteLimit[uniStream].max,
 		activeConnIDLimit:              activeConnIDLimit,
 	}); err != nil {
 		return nil, err
 	}

 	if c.testHooks != nil {
 		c.testHooks.init()
 	}
 	go c.loop(now)
 	return c, nil
 }

 func (c *Conn) String() string {
 	return fmt.Sprintf("quic.Conn(%v,->%v)", c.side, c.peerAddr)
 }

 // confirmHandshake is called when the handshake is confirmed.
 // https://www.rfc-editor.org/rfc/rfc9001#section-4.1.2
 func (c *Conn) confirmHandshake(now time.Time) {
 	// If handshakeConfirmed is unset, the handshake is not confirmed.
 	// If it is unsent, the handshake is confirmed and we need to send a HANDSHAKE_DONE.
 	// If it is sent, we have sent a HANDSHAKE_DONE.
 	// If it is received, the handshake is confirmed and we do not need to send anything.
 	if c.handshakeConfirmed.isSet() {
 		return // already confirmed
 	}
 	if c.side == serverSide {
 		// When the server confirms the handshake, it sends a HANDSHAKE_DONE.
 		c.handshakeConfirmed.setUnsent()
 		c.endpoint.serverConnEstablished(c)
 	} else {
 		// The client never sends a HANDSHAKE_DONE, so we set handshakeConfirmed
 		// to the received state, indicating that the handshake is confirmed and we
 		// don't need to send anything.
 		c.handshakeConfirmed.setReceived()
 	}
 	c.restartIdleTimer(now)
 	c.loss.confirmHandshake()
 	// "An endpoint MUST discard its Handshake keys when the TLS handshake is confirmed"
 	// https://www.rfc-editor.org/rfc/rfc9001#section-4.9.2-1
 	c.discardKeys(now, handshakeSpace)
 }

 // discardKeys discards unused packet protection keys.
 // https://www.rfc-editor.org/rfc/rfc9001#section-4.9
 func (c *Conn) discardKeys(now time.Time, space numberSpace) {
 	switch space {
 	case initialSpace:
 		c.keysInitial.discard()
 	case handshakeSpace:
 		c.keysHandshake.discard()
 	}
 	c.loss.discardKeys(now, c.log, space)
 }

 // receiveTransportParameters applies transport parameters sent by the peer.
 func (c *Conn) receiveTransportParameters(p transportParameters) error {
 	isRetry := c.retryToken != nil
 	if err := c.connIDState.validateTransportParameters(c, isRetry, p); err != nil {
 		return err
 	}
 	c.streams.outflow.setMaxData(p.initialMaxData)
 	c.streams.localLimit[bidiStream].setMax(p.initialMaxStreamsBidi)
 	c.streams.localLimit[uniStream].setMax(p.initialMaxStreamsUni)
 	c.streams.peerInitialMaxStreamDataBidiLocal = p.initialMaxStreamDataBidiLocal
 	c.streams.peerInitialMaxStreamDataRemote[bidiStream] = p.initialMaxStreamDataBidiRemote
 	c.streams.peerInitialMaxStreamDataRemote[uniStream] = p.initialMaxStreamDataUni
 	c.receivePeerMaxIdleTimeout(p.maxIdleTimeout)
 	c.peerAckDelayExponent = p.ackDelayExponent
 	c.loss.setMaxAckDelay(p.maxAckDelay)
 	if err := c.connIDState.setPeerActiveConnIDLimit(c, p.activeConnIDLimit); err != nil {
 		return err
 	}
 	if p.preferredAddrConnID != nil {
 		var (
 			seq           int64 = 1 // sequence number of this conn id is 1
 			retirePriorTo int64 = 0 // retire nothing
 			resetToken    [16]byte
 		)
 		copy(resetToken[:], p.preferredAddrResetToken)
 		if err := c.connIDState.handleNewConnID(c, seq, retirePriorTo, p.preferredAddrConnID, resetToken); err != nil {
 			return err
 		}
 	}
 	// TODO: stateless_reset_token
 	// TODO: max_udp_payload_size
 	// TODO: disable_active_migration
 	// TODO: preferred_address
 	return nil
 }

 type (
 	timerEvent struct{}
 	wakeEvent  struct{}
 )

 var errIdleTimeout = errors.New("idle timeout")

 // loop is the connection main loop.
 //
 // Except where otherwise noted, all connection state is owned by the loop goroutine.
 //
 // The loop processes messages from c.msgc and timer events.
 // Other goroutines may examine or modify conn state by sending the loop funcs to execute.
 func (c *Conn) loop(now time.Time) {
 	defer c.cleanup()

 	// The connection timer sends a message to the connection loop on expiry.
 	// We need to give it an expiry when creating it, so set the initial timeout to
 	// an arbitrary large value. The timer will be reset before this expires (and it
 	// isn't a problem if it does anyway). Skip creating the timer in tests which
 	// take control of the connection message loop.
 	var timer *time.Timer
 	var lastTimeout time.Time
 	hooks := c.testHooks
 	if hooks == nil {
 		timer = time.AfterFunc(1*time.Hour, func() {
 			c.sendMsg(timerEvent{})
 		})
 		defer timer.Stop()
 	}

 	for c.lifetime.state != connStateDone {
 		sendTimeout := c.maybeSend(now) // try sending

 		// Note that we only need to consider the ack timer for the App Data space,
 		// since the Initial and Handshake spaces always ack immediately.
 		nextTimeout := sendTimeout
 		nextTimeout = firstTime(nextTimeout, c.idle.nextTimeout)
 		if c.isAlive() {
 			nextTimeout = firstTime(nextTimeout, c.loss.timer)
 			nextTimeout = firstTime(nextTimeout, c.acks[appDataSpace].nextAck)
 		} else {
 			nextTimeout = firstTime(nextTimeout, c.lifetime.drainEndTime)
 		}

 		var m any
 		if hooks != nil {
 			// Tests only: Wait for the test to tell us to continue.
 			now, m = hooks.nextMessage(c.msgc, nextTimeout)
 		} else if !nextTimeout.IsZero() && nextTimeout.Before(now) {
 			// A connection timer has expired.
 			now = time.Now()
 			m = timerEvent{}
 		} else {
 			// Reschedule the connection timer if necessary
 			// and wait for the next event.
 			if !nextTimeout.Equal(lastTimeout) && !nextTimeout.IsZero() {
 				// Resetting a timer created with time.AfterFunc guarantees
 				// that the timer will run again. We might generate a spurious
 				// timer event under some circumstances, but that's okay.
 				timer.Reset(nextTimeout.Sub(now))
 				lastTimeout = nextTimeout
 			}
 			m = <-c.msgc
 			now = time.Now()
 		}
 		switch m := m.(type) {
 		case *datagram:
 			if !c.handleDatagram(now, m) {
 				if c.logEnabled(QLogLevelPacket) {
 					c.logPacketDropped(m)
 				}
 			}
 			m.recycle()
 		case timerEvent:
 			// A connection timer has expired.
 			if c.idleAdvance(now) {
 				// The connection idle timer has expired.
 				c.abortImmediately(now, errIdleTimeout)
 				return
 			}
 			c.loss.advance(now, c.handleAckOrLoss)
 			if c.lifetimeAdvance(now) {
 				// The connection has completed the draining period,
 				// and may be shut down.
 				return
 			}
 		case wakeEvent:
 			// We're being woken up to try sending some frames.
 		case func(time.Time, *Conn):
 			// Send a func to msgc to run it on the main Conn goroutine
 			m(now, c)
 		default:
 			panic(fmt.Sprintf("quic: unrecognized conn message %T", m))
 		}
 	}
 }

 func (c *Conn) cleanup() {
 	c.logConnectionClosed()
 	c.endpoint.connDrained(c)
 	c.tls.Close()
 	close(c.donec)
 }

 // sendMsg sends a message to the conn's loop.
 // It does not wait for the message to be processed.
 // The conn may close before processing the message, in which case it is lost.
 func (c *Conn) sendMsg(m any) {
 	select {
 	case c.msgc <- m:
 	case <-c.donec:
 	}
 }

 // wake wakes up the conn's loop.
 func (c *Conn) wake() {
 	select {
 	case c.msgc <- wakeEvent{}:
 	default:
 	}
 }

 // runOnLoop executes a function within the conn's loop goroutine.
 func (c *Conn) runOnLoop(ctx context.Context, f func(now time.Time, c *Conn)) error {
 	donec := make(chan struct{})
 	msg := func(now time.Time, c *Conn) {
 		defer close(donec)
 		f(now, c)
 	}
 	if c.testHooks != nil {
 		// In tests, we can't rely on being able to send a message immediately:
 		// c.msgc might be full, and testConnHooks.nextMessage might be waiting
 		// for us to block before it processes the next message.
 		// To avoid a deadlock, we send the message in waitUntil.
 		// If msgc is empty, the message is buffered.
 		// If msgc is full, we block and let nextMessage process the queue.
 		msgc := c.msgc
 		c.testHooks.waitUntil(ctx, func() bool {
 			for {
 				select {
 				case msgc <- msg:
 					msgc = nil // send msg only once
 				case <-donec:
 					return true
 				case <-c.donec:
 					return true
 				default:
 					return false
 				}
 			}
 		})
 	} else {
 		c.sendMsg(msg)
 	}
 	select {
 	case <-donec:
 	case <-c.donec:
 		return errors.New("quic: connection closed")
 	}
 	return nil
 }

 func (c *Conn) waitOnDone(ctx context.Context, ch <-chan struct{}) error {
 	if c.testHooks != nil {
 		return c.testHooks.waitUntil(ctx, func() bool {
 			select {
 			case <-ch:
 				return true
 			default:
 			}
 			return false
 		})
 	}
 	// Check the channel before the context.
 	// We always prefer to return results when available,
 	// even when provided with an already-canceled context.
 	select {
 	case <-ch:
 		return nil
 	default:
 	}
 	select {
 	case <-ch:
 	case <-ctx.Done():
 		return ctx.Err()
 	}
 	return nil
 }

 // firstTime returns the earliest non-zero time, or zero if both times are zero.
 func firstTime(a, b time.Time) time.Time {
 	switch {
 	case a.IsZero():
 		return b
 	case b.IsZero():
 		return a
 	case a.Before(b):
 		return a
 	default:
 		return b
 	}
 }
	// Copyright 2023 The Go Authors. All rights reserved.
	// Use of this source code is governed by a BSD-style
	// license that can be found in the LICENSE file.

	//go:build go1.21

	package quic

	import (
	"context"
	"crypto/tls"
	"errors"
	"fmt"
	"log/slog"
	"net/netip"
	"time"
	)

	// A Conn is a QUIC connection.
	//
	// Multiple goroutines may invoke methods on a Conn simultaneously.
	type Conn struct {
	side connSide
	endpoint *Endpoint
	config *Config
	testHooks connTestHooks
	peerAddr netip.AddrPort
	localAddr netip.AddrPort

	msgc chan any
	donec chan struct{} // closed when conn loop exits

	w packetWriter
	acks [numberSpaceCount]ackState // indexed by number space
	lifetime lifetimeState
	idle idleState
	connIDState connIDState
	loss lossState
	streams streamsState
	path pathState

	// Packet protection keys, CRYPTO streams, and TLS state.
	keysInitial fixedKeyPair
	keysHandshake fixedKeyPair
	keysAppData updatingKeyPair
	crypto [numberSpaceCount]cryptoStream
	tls *tls.QUICConn

	// retryToken is the token provided by the peer in a Retry packet.
	retryToken []byte

	// handshakeConfirmed is set when the handshake is confirmed.
	// For server connections, it tracks sending HANDSHAKE_DONE.
	handshakeConfirmed sentVal

	peerAckDelayExponent int8 // -1 when unknown

	// Tests only: Send a PING in a specific number space.
	testSendPingSpace numberSpace
	testSendPing sentVal

	log *slog.Logger
	}

	// connTestHooks override conn behavior in tests.
	type connTestHooks interface {
	// init is called after a conn is created.
	init()

	// nextMessage is called to request the next event from msgc.
	// Used to give tests control of the connection event loop.
	nextMessage(msgc chan any, nextTimeout time.Time) (now time.Time, message any)

	// handleTLSEvent is called with each TLS event.
	handleTLSEvent(tls.QUICEvent)

	// newConnID is called to generate a new connection ID.
	// Permits tests to generate consistent connection IDs rather than random ones.
	newConnID(seq int64) ([]byte, error)

	// waitUntil blocks until the until func returns true or the context is done.
	// Used to synchronize asynchronous blocking operations in tests.
	waitUntil(ctx context.Context, until func() bool) error

	// timeNow returns the current time.
	timeNow() time.Time
	}

	// newServerConnIDs is connection IDs associated with a new server connection.
	type newServerConnIDs struct {
	srcConnID []byte // source from client's current Initial
	dstConnID []byte // destination from client's current Initial
	originalDstConnID []byte // destination from client's first Initial
	retrySrcConnID []byte // source from server's Retry
	}

	func newConn(now time.Time, side connSide, cids newServerConnIDs, peerHostname string, peerAddr netip.AddrPort, config Config, e Endpoint) (conn *Conn, _ error) {
	c := &Conn{
	side: side,
	endpoint: e,
	config: config,
	peerAddr: unmapAddrPort(peerAddr),
	msgc: make(chan any, 1),
	donec: make(chan struct{}),
	peerAckDelayExponent: -1,
	}
	defer func() {
	// If we hit an error in newConn, close donec so tests don't get stuck waiting for it.
	// This is only relevant if we've got a bug, but it makes tracking that bug down
	// much easier.
	if conn == nil {
	close(c.donec)
	}
	}()

	// A one-element buffer allows us to wake a Conn's event loop as a
	// non-blocking operation.
	c.msgc = make(chan any, 1)

	if e.testHooks != nil {
	e.testHooks.newConn(c)
	}

	// initialConnID is the connection ID used to generate Initial packet protection keys.
	var initialConnID []byte
	if c.side == clientSide {
	if err := c.connIDState.initClient(c); err != nil {
	return nil, err
	}
	initialConnID, _ = c.connIDState.dstConnID()
	} else {
	initialConnID = cids.originalDstConnID
	if cids.retrySrcConnID != nil {
	initialConnID = cids.retrySrcConnID
	}
	if err := c.connIDState.initServer(c, cids); err != nil {
	return nil, err
	}
	}

	// TODO: PMTU discovery.
	c.logConnectionStarted(cids.originalDstConnID, peerAddr)
	c.keysAppData.init()
	c.loss.init(c.side, smallestMaxDatagramSize, now)
	c.streamsInit()
	c.lifetimeInit()
	c.restartIdleTimer(now)

	if err := c.startTLS(now, initialConnID, peerHostname, transportParameters{
	initialSrcConnID: c.connIDState.srcConnID(),
	originalDstConnID: cids.originalDstConnID,
	retrySrcConnID: cids.retrySrcConnID,
	ackDelayExponent: ackDelayExponent,
	maxUDPPayloadSize: maxUDPPayloadSize,
	maxAckDelay: maxAckDelay,
	disableActiveMigration: true,
	initialMaxData: config.maxConnReadBufferSize(),
	initialMaxStreamDataBidiLocal: config.maxStreamReadBufferSize(),
	initialMaxStreamDataBidiRemote: config.maxStreamReadBufferSize(),
	initialMaxStreamDataUni: config.maxStreamReadBufferSize(),
	initialMaxStreamsBidi: c.streams.remoteLimit[bidiStream].max,
	initialMaxStreamsUni: c.streams.remoteLimit[uniStream].max,
	activeConnIDLimit: activeConnIDLimit,
	}); err != nil {
	return nil, err
	}

	if c.testHooks != nil {
	c.testHooks.init()
	}
	go c.loop(now)
	return c, nil
	}

	func (c *Conn) String() string {
	return fmt.Sprintf("quic.Conn(%v,->%v)", c.side, c.peerAddr)
	}

	// confirmHandshake is called when the handshake is confirmed.
	// https://www.rfc-editor.org/rfc/rfc9001#section-4.1.2
	func (c *Conn) confirmHandshake(now time.Time) {
	// If handshakeConfirmed is unset, the handshake is not confirmed.
	// If it is unsent, the handshake is confirmed and we need to send a HANDSHAKE_DONE.
	// If it is sent, we have sent a HANDSHAKE_DONE.
	// If it is received, the handshake is confirmed and we do not need to send anything.
	if c.handshakeConfirmed.isSet() {
	return // already confirmed
	}
	if c.side == serverSide {
	// When the server confirms the handshake, it sends a HANDSHAKE_DONE.
	c.handshakeConfirmed.setUnsent()
	c.endpoint.serverConnEstablished(c)
	} else {
	// The client never sends a HANDSHAKE_DONE, so we set handshakeConfirmed
	// to the received state, indicating that the handshake is confirmed and we
	// don't need to send anything.
	c.handshakeConfirmed.setReceived()
	}
	c.restartIdleTimer(now)
	c.loss.confirmHandshake()
	// "An endpoint MUST discard its Handshake keys when the TLS handshake is confirmed"
	// https://www.rfc-editor.org/rfc/rfc9001#section-4.9.2-1
	c.discardKeys(now, handshakeSpace)
	}

	// discardKeys discards unused packet protection keys.
	// https://www.rfc-editor.org/rfc/rfc9001#section-4.9
	func (c *Conn) discardKeys(now time.Time, space numberSpace) {
	switch space {
	case initialSpace:
	c.keysInitial.discard()
	case handshakeSpace:
	c.keysHandshake.discard()
	}
	c.loss.discardKeys(now, c.log, space)
	}

	// receiveTransportParameters applies transport parameters sent by the peer.
	func (c *Conn) receiveTransportParameters(p transportParameters) error {
	isRetry := c.retryToken != nil
	if err := c.connIDState.validateTransportParameters(c, isRetry, p); err != nil {
	return err
	}
	c.streams.outflow.setMaxData(p.initialMaxData)
	c.streams.localLimit[bidiStream].setMax(p.initialMaxStreamsBidi)
	c.streams.localLimit[uniStream].setMax(p.initialMaxStreamsUni)
	c.streams.peerInitialMaxStreamDataBidiLocal = p.initialMaxStreamDataBidiLocal
	c.streams.peerInitialMaxStreamDataRemote[bidiStream] = p.initialMaxStreamDataBidiRemote
	c.streams.peerInitialMaxStreamDataRemote[uniStream] = p.initialMaxStreamDataUni
	c.receivePeerMaxIdleTimeout(p.maxIdleTimeout)
	c.peerAckDelayExponent = p.ackDelayExponent
	c.loss.setMaxAckDelay(p.maxAckDelay)
	if err := c.connIDState.setPeerActiveConnIDLimit(c, p.activeConnIDLimit); err != nil {
	return err
	}
	if p.preferredAddrConnID != nil {
	var (
	seq int64 = 1 // sequence number of this conn id is 1
	retirePriorTo int64 = 0 // retire nothing
	resetToken [16]byte
	)
	copy(resetToken[:], p.preferredAddrResetToken)
	if err := c.connIDState.handleNewConnID(c, seq, retirePriorTo, p.preferredAddrConnID, resetToken); err != nil {
	return err
	}
	}
	// TODO: stateless_reset_token
	// TODO: max_udp_payload_size
	// TODO: disable_active_migration
	// TODO: preferred_address
	return nil
	}

	type (
	timerEvent struct{}
	wakeEvent struct{}
	)

	var errIdleTimeout = errors.New("idle timeout")

	// loop is the connection main loop.
	//
	// Except where otherwise noted, all connection state is owned by the loop goroutine.
	//
	// The loop processes messages from c.msgc and timer events.
	// Other goroutines may examine or modify conn state by sending the loop funcs to execute.
	func (c *Conn) loop(now time.Time) {
	defer c.cleanup()

	// The connection timer sends a message to the connection loop on expiry.
	// We need to give it an expiry when creating it, so set the initial timeout to
	// an arbitrary large value. The timer will be reset before this expires (and it
	// isn't a problem if it does anyway). Skip creating the timer in tests which
	// take control of the connection message loop.
	var timer *time.Timer
	var lastTimeout time.Time
	hooks := c.testHooks
	if hooks == nil {
	timer = time.AfterFunc(1*time.Hour, func() {
	c.sendMsg(timerEvent{})
	})
	defer timer.Stop()
	}

	for c.lifetime.state != connStateDone {
	sendTimeout := c.maybeSend(now) // try sending

	// Note that we only need to consider the ack timer for the App Data space,
	// since the Initial and Handshake spaces always ack immediately.
	nextTimeout := sendTimeout
	nextTimeout = firstTime(nextTimeout, c.idle.nextTimeout)
	if c.isAlive() {
	nextTimeout = firstTime(nextTimeout, c.loss.timer)
	nextTimeout = firstTime(nextTimeout, c.acks[appDataSpace].nextAck)
	} else {
	nextTimeout = firstTime(nextTimeout, c.lifetime.drainEndTime)
	}

	var m any
	if hooks != nil {
	// Tests only: Wait for the test to tell us to continue.
	now, m = hooks.nextMessage(c.msgc, nextTimeout)
	} else if !nextTimeout.IsZero() && nextTimeout.Before(now) {
	// A connection timer has expired.
	now = time.Now()
	m = timerEvent{}
	} else {
	// Reschedule the connection timer if necessary
	// and wait for the next event.
	if !nextTimeout.Equal(lastTimeout) && !nextTimeout.IsZero() {
	// Resetting a timer created with time.AfterFunc guarantees
	// that the timer will run again. We might generate a spurious
	// timer event under some circumstances, but that's okay.
	timer.Reset(nextTimeout.Sub(now))
	lastTimeout = nextTimeout
	}
	m = <-c.msgc
	now = time.Now()
	}
	switch m := m.(type) {
	case *datagram:
	if !c.handleDatagram(now, m) {
	if c.logEnabled(QLogLevelPacket) {
	c.logPacketDropped(m)
	}
	}
	m.recycle()
	case timerEvent:
	// A connection timer has expired.
	if c.idleAdvance(now) {
	// The connection idle timer has expired.
	c.abortImmediately(now, errIdleTimeout)
	return
	}
	c.loss.advance(now, c.handleAckOrLoss)
	if c.lifetimeAdvance(now) {
	// The connection has completed the draining period,
	// and may be shut down.
	return
	}
	case wakeEvent:
	// We're being woken up to try sending some frames.
	case func(time.Time, *Conn):
	// Send a func to msgc to run it on the main Conn goroutine
	m(now, c)
	default:
	panic(fmt.Sprintf("quic: unrecognized conn message %T", m))
	}
	}
	}

	func (c *Conn) cleanup() {
	c.logConnectionClosed()
	c.endpoint.connDrained(c)
	c.tls.Close()
	close(c.donec)
	}

	// sendMsg sends a message to the conn's loop.
	// It does not wait for the message to be processed.
	// The conn may close before processing the message, in which case it is lost.
	func (c *Conn) sendMsg(m any) {
	select {
	case c.msgc <- m:
	case <-c.donec:
	}
	}

	// wake wakes up the conn's loop.
	func (c *Conn) wake() {
	select {
	case c.msgc <- wakeEvent{}:
	default:
	}
	}

	// runOnLoop executes a function within the conn's loop goroutine.
	func (c Conn) runOnLoop(ctx context.Context, f func(now time.Time, c Conn)) error {
	donec := make(chan struct{})
	msg := func(now time.Time, c *Conn) {
	defer close(donec)
	f(now, c)
	}
	if c.testHooks != nil {
	// In tests, we can't rely on being able to send a message immediately:
	// c.msgc might be full, and testConnHooks.nextMessage might be waiting
	// for us to block before it processes the next message.
	// To avoid a deadlock, we send the message in waitUntil.
	// If msgc is empty, the message is buffered.
	// If msgc is full, we block and let nextMessage process the queue.
	msgc := c.msgc
	c.testHooks.waitUntil(ctx, func() bool {
	for {
	select {
	case msgc <- msg:
	msgc = nil // send msg only once
	case <-donec:
	return true
	case <-c.donec:
	return true
	default:
	return false
	}
	}
	})
	} else {
	c.sendMsg(msg)
	}
	select {
	case <-donec:
	case <-c.donec:
	return errors.New("quic: connection closed")
	}
	return nil
	}

	func (c *Conn) waitOnDone(ctx context.Context, ch <-chan struct{}) error {
	if c.testHooks != nil {
	return c.testHooks.waitUntil(ctx, func() bool {
	select {
	case <-ch:
	return true
	default:
	}
	return false
	})
	}
	// Check the channel before the context.
	// We always prefer to return results when available,
	// even when provided with an already-canceled context.
	select {
	case <-ch:
	return nil
	default:
	}
	select {
	case <-ch:
	case <-ctx.Done():
	return ctx.Err()
	}
	return nil
	}

	// firstTime returns the earliest non-zero time, or zero if both times are zero.
	func firstTime(a, b time.Time) time.Time {
	switch {
	case a.IsZero():
	return b
	case b.IsZero():
	return a
	case a.Before(b):
	return a
	default:
	return b
	}
	}