Repository: pion/ice
Pull request #680 (2024-04-16 18:40:29)

[Cherrypick] Match libwebrtc protocol priority
Cherrypick from master (since we will keep using v2 for some time)

#### Description
Match libwebrtc's TURN protocol priority (TCP vs UDP vs TLS)

#### Reference issue
Fixes reproducibility issues as seen in the field



Patch:
diff --git a/agent.go b/agent.go
index 5e4fcb9..82ee75b 100644
--- a/agent.go
+++ b/agent.go
@@ -7,6 +7,7 @@ package ice
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"math"
 	"net"
@@ -126,12 +127,16 @@ type Agent struct {
 	taskLoopDone chan struct{}
 	err          atomicx.Error
 
+	// Callback that allows user to implement custom behavior
+	// for STUN Binding Requests
+	userBindingRequestHandler func(m *stun.Message, local, remote Candidate, pair *CandidatePair) bool
+
 	gatherCandidateCancel func()
 	gatherCandidateDone   chan struct{}
 
-	chanCandidate     chan Candidate
-	chanCandidatePair chan *CandidatePair
-	chanState         chan ConnectionState
+	connectionStateNotifier       *handlerNotifier
+	candidateNotifier             *handlerNotifier
+	selectedCandidatePairNotifier *handlerNotifier
 
 	loggerFactory logging.LoggerFactory
 	log           logging.LeveledLogger
@@ -228,9 +233,6 @@ func (a *Agent) taskLoop() {
 
 		after()
 
-		close(a.chanState)
-		close(a.chanCandidate)
-		close(a.chanCandidatePair)
 		close(a.taskLoopDone)
 	}()
 
@@ -278,33 +280,30 @@ func NewAgent(config *AgentConfig) (*Agent, error) { //nolint:gocognit
 	startedCtx, startedFn := context.WithCancel(context.Background())
 
 	a := &Agent{
-		chanTask:          make(chan task),
-		chanState:         make(chan ConnectionState),
-		chanCandidate:     make(chan Candidate),
-		chanCandidatePair: make(chan *CandidatePair),
-		tieBreaker:        globalMathRandomGenerator.Uint64(),
-		lite:              config.Lite,
-		gatheringState:    GatheringStateNew,
-		connectionState:   ConnectionStateNew,
-		localCandidates:   make(map[NetworkType][]Candidate),
-		remoteCandidates:  make(map[NetworkType][]Candidate),
-		urls:              config.Urls,
-		networkTypes:      config.NetworkTypes,
-		onConnected:       make(chan struct{}),
-		buf:               packetio.NewBuffer(),
-		done:              make(chan struct{}),
-		taskLoopDone:      make(chan struct{}),
-		startedCh:         startedCtx.Done(),
-		startedFn:         startedFn,
-		portMin:           config.PortMin,
-		portMax:           config.PortMax,
-		loggerFactory:     loggerFactory,
-		log:               log,
-		net:               config.Net,
-		proxyDialer:       config.ProxyDialer,
-		tcpMux:            config.TCPMux,
-		udpMux:            config.UDPMux,
-		udpMuxSrflx:       config.UDPMuxSrflx,
+		chanTask:         make(chan task),
+		tieBreaker:       globalMathRandomGenerator.Uint64(),
+		lite:             config.Lite,
+		gatheringState:   GatheringStateNew,
+		connectionState:  ConnectionStateNew,
+		localCandidates:  make(map[NetworkType][]Candidate),
+		remoteCandidates: make(map[NetworkType][]Candidate),
+		urls:             config.Urls,
+		networkTypes:     config.NetworkTypes,
+		onConnected:      make(chan struct{}),
+		buf:              packetio.NewBuffer(),
+		done:             make(chan struct{}),
+		taskLoopDone:     make(chan struct{}),
+		startedCh:        startedCtx.Done(),
+		startedFn:        startedFn,
+		portMin:          config.PortMin,
+		portMax:          config.PortMax,
+		loggerFactory:    loggerFactory,
+		log:              log,
+		net:              config.Net,
+		proxyDialer:      config.ProxyDialer,
+		tcpMux:           config.TCPMux,
+		udpMux:           config.UDPMux,
+		udpMuxSrflx:      config.UDPMuxSrflx,
 
 		mDNSMode: mDNSMode,
 		mDNSName: mDNSName,
@@ -322,7 +321,12 @@ func NewAgent(config *AgentConfig) (*Agent, error) { //nolint:gocognit
 		includeLoopback: config.IncludeLoopback,
 
 		disableActiveTCP: config.DisableActiveTCP,
+
+		userBindingRequestHandler: config.BindingRequestHandler,
 	}
+	a.connectionStateNotifier = &handlerNotifier{connectionStateFunc: a.onConnectionStateChange, done: make(chan struct{})}
+	a.candidateNotifier = &handlerNotifier{candidateFunc: a.onCandidate, done: make(chan struct{})}
+	a.selectedCandidatePairNotifier = &handlerNotifier{candidatePairFunc: a.onSelectedCandidatePairChange, done: make(chan struct{})}
 
 	if a.net == nil {
 		a.net, err = stdnet.NewNet()
@@ -366,13 +370,6 @@ func NewAgent(config *AgentConfig) (*Agent, error) { //nolint:gocognit
 
 	go a.taskLoop()
 
-	// CandidatePair and ConnectionState are usually changed at once.
-	// Blocking one by the other one causes deadlock.
-	// Hence, we call handlers from independent Goroutines.
-	go a.candidatePairRoutine()
-	go a.connectionStateRoutine()
-	go a.candidateRoutine()
-
 	// Restart is also used to initialize the agent for the first time
 	if err := a.Restart(config.LocalUfrag, config.LocalPwd); err != nil {
 		a.closeMulticastConn()
@@ -510,12 +507,7 @@ func (a *Agent) updateConnectionState(newState ConnectionState) {
 
 		a.log.Infof("Setting new connection state: %s", newState)
 		a.connectionState = newState
-
-		// Call handler after finishing current task since we may be holding the agent lock
-		// and the handler may also require it
-		a.afterRun(func(ctx context.Context) {
-			a.chanState <- newState
-		})
+		a.connectionStateNotifier.EnqueueConnectionState(newState)
 	}
 }
 
@@ -534,12 +526,7 @@ func (a *Agent) setSelectedPair(p *CandidatePair) {
 	a.updateConnectionState(ConnectionStateConnected)
 
 	// Notify when the selected pair changes
-	a.afterRun(func(ctx context.Context) {
-		select {
-		case a.chanCandidatePair <- p:
-		case <-ctx.Done():
-		}
-	})
+	a.selectedCandidatePairNotifier.EnqueueSelectedCandidatePair(p)
 
 	// Signal connected
 	a.onConnectedOnce.Do(func() { close(a.onConnected) })
@@ -775,7 +762,7 @@ func (a *Agent) addRemotePassiveTCPCandidate(remoteCandidate Candidate) {
 
 		localCandidate.start(a, conn, a.startedCh)
 		a.localCandidates[localCandidate.NetworkType()] = append(a.localCandidates[localCandidate.NetworkType()], localCandidate)
-		a.chanCandidate <- localCandidate
+		a.candidateNotifier.EnqueueCandidate(localCandidate)
 
 		a.addPair(localCandidate, remoteCandidate)
 	}
@@ -845,7 +832,7 @@ func (a *Agent) addCandidate(ctx context.Context, c Candidate, candidateConn net
 
 		a.requestConnectivityCheck()
 
-		a.chanCandidate <- c
+		a.candidateNotifier.EnqueueCandidate(c)
 	})
 }
 
@@ -929,7 +916,21 @@ func (a *Agent) removeUfragFromMux() {
 
 // Close cleans up the Agent
 func (a *Agent) Close() error {
+	return a.close(false)
+}
+
+// GracefulClose cleans up the Agent and waits for any goroutines it started
+// to complete. This is only safe to call outside of Agent callbacks or if in a callback,
+// in its own goroutine.
+func (a *Agent) GracefulClose() error {
+	return a.close(true)
+}
+
+func (a *Agent) close(graceful bool) error {
 	if err := a.ok(); err != nil {
+		if errors.Is(err, ErrClosed) {
+			return nil
+		}
 		return err
 	}
 
@@ -944,7 +945,14 @@ func (a *Agent) Close() error {
 	a.removeUfragFromMux()
 
 	close(a.done)
+	// the loop is safe to wait on no matter what
 	<-a.taskLoopDone
+
+	// but we are in less control of the notifiers, so we will
+	// pass through `graceful`.
+	a.connectionStateNotifier.Close(graceful)
+	a.candidateNotifier.Close(graceful)
+	a.selectedCandidatePairNotifier.Close(graceful)
 	return nil
 }
 
@@ -1281,7 +1289,7 @@ func (a *Agent) setGatheringState(newState GatheringState) error {
 	done := make(chan struct{})
 	if err := a.run(a.context(), func(ctx context.Context, agent *Agent) {
 		if a.gatheringState != newState && newState == GatheringStateComplete {
-			a.chanCandidate <- nil
+			a.candidateNotifier.EnqueueCandidate(nil)
 		}
 
 		a.gatheringState = newState
diff --git a/agent_config.go b/agent_config.go
index 9629fd5..7f7b67a 100644
--- a/agent_config.go
+++ b/agent_config.go
@@ -188,6 +188,13 @@ type AgentConfig struct {
 	// DisableActiveTCP can be used to disable Active TCP candidates. Otherwise when TCP is enabled
 	// Active TCP candidates will be created when a new passive TCP remote candidate is added.
 	DisableActiveTCP bool
+
+	// BindingRequestHandler allows applications to perform logic on incoming STUN Binding Requests
+	// This was implemented to allow users to
+	// * Log incoming Binding Requests for debugging
+	// * Implement draft-thatcher-ice-renomination
+	// * Implement custom CandidatePair switching logic
+	BindingRequestHandler func(m *stun.Message, local, remote Candidate, pair *CandidatePair) bool
 }
 
 // initWithDefaults populates an agent and falls back to defaults if fields are unset
diff --git a/agent_handlers.go b/agent_handlers.go
index c5a5ec0..0c02277 100644
--- a/agent_handlers.go
+++ b/agent_handlers.go
@@ -3,6 +3,8 @@
 
 package ice
 
+import "sync"
+
 // OnConnectionStateChange sets a handler that is fired when the connection state changes
 func (a *Agent) OnConnectionStateChange(f func(ConnectionState)) error {
 	a.onConnectionStateChangeHdlr.Store(f)
@@ -41,20 +43,141 @@ func (a *Agent) onConnectionStateChange(s ConnectionState) {
 	}
 }
 
-func (a *Agent) candidatePairRoutine() {
-	for p := range a.chanCandidatePair {
-		a.onSelectedCandidatePairChange(p)
+type handlerNotifier struct {
+	sync.Mutex
+	running   bool
+	notifiers sync.WaitGroup
+
+	connectionStates    []ConnectionState
+	connectionStateFunc func(ConnectionState)
+
+	candidates    []Candidate
+	candidateFunc func(Candidate)
+
+	selectedCandidatePairs []*CandidatePair
+	candidatePairFunc      func(*CandidatePair)
+
+	// State for closing
+	done chan struct{}
+}
+
+func (h *handlerNotifier) Close(graceful bool) {
+	if graceful {
+		// if we were closed ungracefully before, we now
+		// want ot wait.
+		defer h.notifiers.Wait()
+	}
+
+	h.Lock()
+
+	select {
+	case <-h.done:
+		h.Unlock()
+		return
+	default:
+	}
+	close(h.done)
+	h.Unlock()
+}
+
+func (h *handlerNotifier) EnqueueConnectionState(s ConnectionState) {
+	h.Lock()
+	defer h.Unlock()
+
+	select {
+	case <-h.done:
+		return
+	default:
+	}
+
+	notify := func() {
+		defer h.notifiers.Done()
+		for {
+			h.Lock()
+			if len(h.connectionStates) == 0 {
+				h.running = false
+				h.Unlock()
+				return
+			}
+			notification := h.connectionStates[0]
+			h.connectionStates = h.connectionStates[1:]
+			h.Unlock()
+			h.connectionStateFunc(notification)
+		}
+	}
+
+	h.connectionStates = append(h.connectionStates, s)
+	if !h.running {
+		h.running = true
+		h.notifiers.Add(1)
+		go notify()
 	}
 }
 
-func (a *Agent) connectionStateRoutine() {
-	for s := range a.chanState {
-		go a.onConnectionStateChange(s)
+func (h *handlerNotifier) EnqueueCandidate(c Candidate) {
+	h.Lock()
+	defer h.Unlock()
+
+	select {
+	case <-h.done:
+		return
+	default:
+	}
+
+	notify := func() {
+		defer h.notifiers.Done()
+		for {
+			h.Lock()
+			if len(h.candidates) == 0 {
+				h.running = false
+				h.Unlock()
+				return
+			}
+			notification := h.candidates[0]
+			h.candidates = h.candidates[1:]
+			h.Unlock()
+			h.candidateFunc(notification)
+		}
+	}
+
+	h.candidates = append(h.candidates, c)
+	if !h.running {
+		h.running = true
+		h.notifiers.Add(1)
+		go notify()
 	}
 }
 
-func (a *Agent) candidateRoutine() {
-	for c := range a.chanCandidate {
-		a.onCandidate(c)
+func (h *handlerNotifier) EnqueueSelectedCandidatePair(p *CandidatePair) {
+	h.Lock()
+	defer h.Unlock()
+
+	select {
+	case <-h.done:
+		return
+	default:
+	}
+
+	notify := func() {
+		defer h.notifiers.Done()
+		for {
+			h.Lock()
+			if len(h.selectedCandidatePairs) == 0 {
+				h.running = false
+				h.Unlock()
+				return
+			}
+			notification := h.selectedCandidatePairs[0]
+			h.selectedCandidatePairs = h.selectedCandidatePairs[1:]
+			h.Unlock()
+			h.candidatePairFunc(notification)
+		}
+	}
+
+	h.selectedCandidatePairs = append(h.selectedCandidatePairs, p)
+	if !h.running {
+		h.running = true
+		h.notifiers.Add(1)
+		go notify()
 	}
 }
diff --git a/candidate_base.go b/candidate_base.go
index 7f97be8..f725985 100644
--- a/candidate_base.go
+++ b/candidate_base.go
@@ -31,8 +31,8 @@ type candidateBase struct {
 
 	resolvedAddr net.Addr
 
-	lastSent     atomic.Int64
-	lastReceived atomic.Int64
+	lastSent     atomic.Value
+	lastReceived atomic.Value
 	conn         net.PacketConn
 
 	currAgent *Agent
@@ -400,27 +400,27 @@ func (c *candidateBase) String() string {
 // LastReceived returns a time.Time indicating the last time
 // this candidate was received
 func (c *candidateBase) LastReceived() time.Time {
-	if lastReceived := c.lastReceived.Load(); lastReceived != 0 {
-		return time.Unix(0, lastReceived)
+	if lastReceived, ok := c.lastReceived.Load().(time.Time); ok {
+		return lastReceived
 	}
 	return time.Time{}
 }
 
 func (c *candidateBase) setLastReceived(t time.Time) {
-	c.lastReceived.Store(t.UnixNano())
+	c.lastReceived.Store(t)
 }
 
 // LastSent returns a time.Time indicating the last time
 // this candidate was sent
 func (c *candidateBase) LastSent() time.Time {
-	if lastSent := c.lastSent.Load(); lastSent != 0 {
-		return time.Unix(0, lastSent)
+	if lastSent, ok := c.lastSent.Load().(time.Time); ok {
+		return lastSent
 	}
 	return time.Time{}
 }
 
 func (c *candidateBase) setLastSent(t time.Time) {
-	c.lastSent.Store(t.UnixNano())
+	c.lastSent.Store(t)
 }
 
 func (c *candidateBase) seen(outbound bool) {
@@ -447,6 +447,13 @@ func (c *candidateBase) copy() (Candidate, error) {
 	return UnmarshalCandidate(c.Marshal())
 }
 
+func removeZoneIDFromAddress(addr string) string {
+	if i := strings.Index(addr, "%"); i != -1 {
+		return addr[:i]
+	}
+	return addr
+}
+
 // Marshal returns the string representation of the ICECandidate
 func (c *candidateBase) Marshal() string {
 	val := c.Foundation()
@@ -459,7 +466,7 @@ func (c *candidateBase) Marshal() string {
 		c.Component(),
 		c.NetworkType().NetworkShort(),
 		c.Priority(),
-		c.Address(),
+		removeZoneIDFromAddress(c.Address()),
 		c.Port(),
 		c.Type())
 
@@ -509,7 +516,7 @@ func UnmarshalCandidate(raw string) (Candidate, error) {
 	priority := uint32(priorityRaw)
 
 	// Address
-	address := split[4]
+	address := removeZoneIDFromAddress(split[4])
 
 	// Port
 	rawPort, err := strconv.ParseUint(split[5], 10, 16)
diff --git a/candidate_relay.go b/candidate_relay.go
index 449d077..fa5297b 100644
--- a/candidate_relay.go
+++ b/candidate_relay.go
@@ -70,6 +70,23 @@ func NewCandidateRelay(config *CandidateRelayConfig) (*CandidateRelay, error) {
 	}, nil
 }
 
+// LocalPreference returns the local preference for this candidate
+func (c *CandidateRelay) LocalPreference() uint16 {
+	// These preference values come from libwebrtc
+	// https://github.com/mozilla/libwebrtc/blob/1389c76d9c79839a2ca069df1db48aa3f2e6a1ac/p2p/base/turn_port.cc#L61
+	var relayPreference uint16
+	switch c.relayProtocol {
+	case relayProtocolTLS, relayProtocolDTLS:
+		relayPreference = 2
+	case tcp:
+		relayPreference = 1
+	default:
+		relayPreference = 0
+	}
+
+	return c.candidateBase.LocalPreference() + relayPreference
+}
+
 // RelayProtocol returns the protocol used between the endpoint and the relay server.
 func (c *CandidateRelay) RelayProtocol() string {
 	return c.relayProtocol
diff --git a/gather.go b/gather.go
index f93e6c1..024247d 100644
--- a/gather.go
+++ b/gather.go
@@ -616,7 +616,7 @@ func (a *Agent) gatherCandidatesRelay(ctx context.Context, urls []*stun.URI) { /
 
 				relAddr = conn.LocalAddr().(*net.UDPAddr).IP.String() //nolint:forcetypeassert
 				relPort = conn.LocalAddr().(*net.UDPAddr).Port        //nolint:forcetypeassert
-				relayProtocol = "dtls"
+				relayProtocol = relayProtocolDTLS
 				locConn = &fakenet.PacketConn{Conn: conn}
 			case url.Proto == stun.ProtoTypeTCP && url.Scheme == stun.SchemeTypeTURNS:
 				tcpAddr, resolvErr := a.net.ResolveTCPAddr(NetworkTypeTCP4.String(), turnServerAddr)
@@ -646,7 +646,7 @@ func (a *Agent) gatherCandidatesRelay(ctx context.Context, urls []*stun.URI) { /
 
 				relAddr = conn.LocalAddr().(*net.TCPAddr).IP.String() //nolint:forcetypeassert
 				relPort = conn.LocalAddr().(*net.TCPAddr).Port        //nolint:forcetypeassert
-				relayProtocol = "tls"
+				relayProtocol = relayProtocolTLS
 				locConn = turn.NewSTUNConn(conn)
 			default:
 				a.log.Warnf("Unable to handle URL in gatherCandidatesRelay %v", url)
diff --git a/go.mod b/go.mod
index a53211d..7f181e7 100644
--- a/go.mod
+++ b/go.mod
@@ -10,7 +10,7 @@ require (
 	github.com/pion/mdns v0.0.12
 	github.com/pion/randutil v0.1.0
 	github.com/pion/stun v0.6.1
-	github.com/pion/transport/v2 v2.2.2
+	github.com/pion/transport/v2 v2.2.10
 	github.com/pion/turn/v2 v2.1.3
 	github.com/stretchr/testify v1.8.4
 	golang.org/x/net v0.20.0
diff --git a/go.sum b/go.sum
index b5dfe68..55fea23 100644
--- a/go.sum
+++ b/go.sum
@@ -19,8 +19,8 @@ github.com/pion/randutil v0.1.0/go.mod h1:XcJrSMMbbMRhASFVOlj/5hQial/Y8oH/HVo7TB
 github.com/pion/stun v0.6.1 h1:8lp6YejULeHBF8NmV8e2787BogQhduZugh5PdhDyyN4=
 github.com/pion/stun v0.6.1/go.mod h1:/hO7APkX4hZKu/D0f2lHzNyvdkTGtIy3NDmLR7kSz/8=
 github.com/pion/transport/v2 v2.2.1/go.mod h1:cXXWavvCnFF6McHTft3DWS9iic2Mftcz1Aq29pGcU5g=
-github.com/pion/transport/v2 v2.2.2 h1:yv+EKSU2dpmInuCebQ1rsBFCYL7p+aV90xIlshSBO+A=
-github.com/pion/transport/v2 v2.2.2/go.mod h1:OJg3ojoBJopjEeECq2yJdXH9YVrUJ1uQ++NjXLOUorc=
+github.com/pion/transport/v2 v2.2.10 h1:ucLBLE8nuxiHfvkFKnkDQRYWYfp8ejf4YBOPfaQpw6Q=
+github.com/pion/transport/v2 v2.2.10/go.mod h1:sq1kSLWs+cHW9E+2fJP95QudkzbK7wscs8yYgQToO5E=
 github.com/pion/transport/v3 v3.0.1 h1:gDTlPJwROfSfz6QfSi0ZmeCSkFcnWWiiR9ES0ouANiM=
 github.com/pion/transport/v3 v3.0.1/go.mod h1:UY7kiITrlMv7/IKgd5eTUcaahZx5oUN3l9SzK5f5xE0=
 github.com/pion/turn/v2 v2.1.3 h1:pYxTVWG2gpC97opdRc5IGsQ1lJ9O/IlNhkzj7MMrGAA=
@@ -35,11 +35,12 @@ github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO
 github.com/stretchr/testify v1.8.3/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
 github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=
 github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
+github.com/wlynxg/anet v0.0.3 h1:PvR53psxFXstc12jelG6f1Lv4MWqE0tI76/hHGjh9rg=
+github.com/wlynxg/anet v0.0.3/go.mod h1:eay5PRQr7fIVAMbTbchTnO9gG65Hg/uYGdc7mguHxoA=
 github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.8.0/go.mod h1:mRqEX+O9/h5TFCrQhkgjo2yKi0yYA+9ecGkdQoHrywE=
-golang.org/x/crypto v0.11.0/go.mod h1:xgJhtzW8F9jGdVFWZESrid1U1bjeNy4zgy5cRr/CIio=
 golang.org/x/crypto v0.12.0/go.mod h1:NF0Gs7EO5K4qLn+Ylc+fih8BSTeIjAP05siRnAh98yw=
 golang.org/x/crypto v0.18.0 h1:PGVlW0xEltQnzFZ55hkuX5+KLyrMYhHld1YHO4AKcdc=
 golang.org/x/crypto v0.18.0/go.mod h1:R0j02AL6hcrfOiy9T4ZYp/rcWeMxM3L6QYxlOuEG1mg=
@@ -51,7 +52,6 @@ golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug
 golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
 golang.org/x/net v0.9.0/go.mod h1:d48xBJpPfHeWQsugry2m+kC02ZBRGRgulfHnEXEuWns=
 golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg=
-golang.org/x/net v0.13.0/go.mod h1:zEVYFnQC7m/vmpQFELhcD1EWkZlX69l4oqgmer6hfKA=
 golang.org/x/net v0.14.0/go.mod h1:PpSgVXXLK0OxS0F31C1/tv6XNguvCrnXIDrFMspZIUI=
 golang.org/x/net v0.20.0 h1:aCL9BSgETF1k+blQaYUBx9hJ9LOGP3gAVemcZlf1Kpo=
 golang.org/x/net v0.20.0/go.mod h1:z8BVo6PvndSri0LbOE3hAn0apkU+1YvI6E70E9jsnvY=
@@ -67,7 +67,6 @@ golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.7.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.9.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.10.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.11.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.16.0 h1:xWw16ngr6ZMtmxDyKyIgsE93KNKz5HKmMa3b8ALHidU=
 golang.org/x/sys v0.16.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
@@ -76,7 +75,6 @@ golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuX
 golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
 golang.org/x/term v0.7.0/go.mod h1:P32HKFT3hSsZrRxla30E9HqToFYAQPCMs/zFMBUFqPY=
 golang.org/x/term v0.8.0/go.mod h1:xPskH00ivmX89bAKVGSKKtLOWNx2+17Eiy94tnKShWo=
-golang.org/x/term v0.10.0/go.mod h1:lpqdcUyK/oCiQxvxVrppt5ggO2KCZ5QblwqPnfZ6d5o=
 golang.org/x/term v0.11.0/go.mod h1:zC9APTIj3jG3FdV/Ons+XE1riIZXG4aZ4GTHiPZJPIU=
 golang.org/x/term v0.16.0/go.mod h1:yn7UURbUtPyrVJPGPq404EukNFxcm/foM+bV/bfcDsY=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
@@ -84,7 +82,6 @@ golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
 golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
-golang.org/x/text v0.11.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
 golang.org/x/text v0.12.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
 golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
diff --git a/ice.go b/ice.go
index bd55120..73262dd 100644
--- a/ice.go
+++ b/ice.go
@@ -83,3 +83,8 @@ func (t GatheringState) String() string {
 		return ErrUnknownType.Error()
 	}
 }
+
+const (
+	relayProtocolDTLS = "dtls"
+	relayProtocolTLS  = "tls"
+)
diff --git a/selection.go b/selection.go
index f854581..69bde40 100644
--- a/selection.go
+++ b/selection.go
@@ -59,7 +59,7 @@ func (s *controllingSelector) ContactCandidates() {
 	default:
 		p := s.agent.getBestValidCandidatePair()
 		if p != nil && s.isNominatable(p.Local) && s.isNominatable(p.Remote) {
-			s.log.Tracef("Nominatable pair found, nominating (%s, %s)", p.Local.String(), p.Remote.String())
+			s.log.Tracef("Nominatable pair found, nominating (%s, %s)", p.Local, p.Remote)
 			p.nominated = true
 			s.nominatedPair = p
 			s.nominatePair(p)
@@ -87,7 +87,7 @@ func (s *controllingSelector) nominatePair(pair *CandidatePair) {
 		return
 	}
 
-	s.log.Tracef("Ping STUN (nominate candidate pair) from %s to %s", pair.Local.String(), pair.Remote.String())
+	s.log.Tracef("Ping STUN (nominate candidate pair) from %s to %s", pair.Local, pair.Remote)
 	s.agent.sendBindingRequest(msg, pair.Local, pair.Remote)
 }
 
@@ -106,12 +106,17 @@ func (s *controllingSelector) HandleBindingRequest(m *stun.Message, local, remot
 		if bestPair == nil {
 			s.log.Tracef("No best pair available")
 		} else if bestPair.equal(p) && s.isNominatable(p.Local) && s.isNominatable(p.Remote) {
-			s.log.Tracef("The candidate (%s, %s) is the best candidate available, marking it as nominated",
-				p.Local.String(), p.Remote.String())
+			s.log.Tracef("The candidate (%s, %s) is the best candidate available, marking it as nominated", p.Local, p.Remote)
 			s.nominatedPair = p
 			s.nominatePair(p)
 		}
 	}
+
+	if s.agent.userBindingRequestHandler != nil {
+		if shouldSwitch := s.agent.userBindingRequestHandler(m, local, remote, p); shouldSwitch {
+			s.agent.setSelectedPair(p)
+		}
+	}
 }
 
 func (s *controllingSelector) HandleSuccessResponse(m *stun.Message, local, remote Candidate, remoteAddr net.Addr) {
@@ -130,7 +135,7 @@ func (s *controllingSelector) HandleSuccessResponse(m *stun.Message, local, remo
 		return
 	}
 
-	s.log.Tracef("Inbound STUN (SuccessResponse) from %s to %s", remote.String(), local.String())
+	s.log.Tracef("Inbound STUN (SuccessResponse) from %s to %s", remote, local)
 	p := s.agent.findPair(local, remote)
 
 	if p == nil {
@@ -221,7 +226,7 @@ func (s *controlledSelector) HandleSuccessResponse(m *stun.Message, local, remot
 		return
 	}
 
-	s.log.Tracef("Inbound STUN (SuccessResponse) from %s to %s", remote.String(), local.String())
+	s.log.Tracef("Inbound STUN (SuccessResponse) from %s to %s", remote, local)
 
 	p := s.agent.findPair(local, remote)
 	if p == nil {
@@ -243,14 +248,12 @@ func (s *controlledSelector) HandleSuccessResponse(m *stun.Message, local, remot
 }
 
 func (s *controlledSelector) HandleBindingRequest(m *stun.Message, local, remote Candidate) {
-	useCandidate := m.Contains(stun.AttrUseCandidate)
-
 	p := s.agent.findPair(local, remote)
 	if p == nil {
 		p = s.agent.addPair(local, remote)
 	}
 
-	if useCandidate {
+	if m.Contains(stun.AttrUseCandidate) {
 		// https://tools.ietf.org/html/rfc8445#section-7.3.1.5
 
 		if p.state == CandidatePairStateSucceeded {
@@ -258,8 +261,8 @@ func (s *controlledSelector) HandleBindingRequest(m *stun.Message, local, remote
 			// previously sent by this pair produced a successful response and
 			// generated a valid pair (Section 7.2.5.3.2).  The agent sets the
 			// nominated flag value of the valid pair to true.
-			if selectedPair := s.agent.getSelectedPair(); selectedPair == nil ||
-				(selectedPair != p && selectedPair.priority() <= p.priority()) {
+			selectedPair := s.agent.getSelectedPair()
+			if selectedPair == nil || (selectedPair != p && selectedPair.priority() <= p.priority()) {
 				s.agent.setSelectedPair(p)
 			} else if selectedPair != p {
 				s.log.Tracef("Ignore nominate new pair %s, already nominated pair %s", p, selectedPair)
@@ -279,6 +282,12 @@ func (s *controlledSelector) HandleBindingRequest(m *stun.Message, local, remote
 
 	s.agent.sendBindingSuccess(m, local, remote)
 	s.PingCandidate(local, remote)
+
+	if s.agent.userBindingRequestHandler != nil {
+		if shouldSwitch := s.agent.userBindingRequestHandler(m, local, remote, p); shouldSwitch {
+			s.agent.setSelectedPair(p)
+		}
+	}
 }
 
 type liteSelector struct {
diff --git a/udp_mux.go b/udp_mux.go
index ba0efbe..ba2a7e3 100644
--- a/udp_mux.go
+++ b/udp_mux.go
@@ -331,7 +331,7 @@ func (m *UDPMuxDefault) connWorker() {
 		}
 
 		if destinationConn == nil {
-			m.params.Logger.Tracef("Dropping packet from %s, addr: %s", udpAddr.String(), addr.String())
+			m.params.Logger.Tracef("Dropping packet from %s, addr: %s", udpAddr, addr)
 			continue
 		}
 
diff --git a/udp_muxed_conn.go b/udp_muxed_conn.go
index 7802cd1..322d020 100644
--- a/udp_muxed_conn.go
+++ b/udp_muxed_conn.go
@@ -88,7 +88,7 @@ func (c *udpMuxedConn) ReadFrom(b []byte) (n int, rAddr net.Addr, err error) {
 
 		if c.state == udpMuxedConnClosed {
 			c.mu.Unlock()
-			return 0, nil, io.ErrClosedPipe
+			return 0, nil, io.EOF
 		}
 
 		c.state = udpMuxedConnWaiting
@@ -97,7 +97,7 @@ func (c *udpMuxedConn) ReadFrom(b []byte) (n int, rAddr net.Addr, err error) {
 		select {
 		case <-c.notify:
 		case <-c.closedChan:
-			return 0, nil, io.ErrClosedPipe
+			return 0, nil, io.EOF
 		}
 	}
 }
@@ -147,6 +147,8 @@ func (c *udpMuxedConn) Close() error {
 
 			pkt = next
 		}
+		c.bufHead = nil
+		c.bufTail = nil
 
 		c.state = udpMuxedConnClosed
 		close(c.closedChan)


Test patch:
diff --git a/agent_handlers_test.go b/agent_handlers_test.go
new file mode 100644
index 0000000..7354b82
--- /dev/null
+++ b/agent_handlers_test.go
@@ -0,0 +1,75 @@
+// SPDX-FileCopyrightText: 2023 The Pion community <https://pion.ly>
+// SPDX-License-Identifier: MIT
+
+package ice
+
+import (
+	"testing"
+	"time"
+
+	"github.com/pion/transport/v2/test"
+)
+
+func TestConnectionStateNotifier(t *testing.T) {
+	t.Run("TestManyUpdates", func(t *testing.T) {
+		report := test.CheckRoutines(t)
+		defer report()
+		updates := make(chan struct{}, 1)
+		c := &handlerNotifier{
+			connectionStateFunc: func(_ ConnectionState) {
+				updates <- struct{}{}
+			},
+			done: make(chan struct{}),
+		}
+		// Enqueue all updates upfront to ensure that it
+		// doesn't block
+		for i := 0; i < 10000; i++ {
+			c.EnqueueConnectionState(ConnectionStateNew)
+		}
+		done := make(chan struct{})
+		go func() {
+			for i := 0; i < 10000; i++ {
+				<-updates
+			}
+			select {
+			case <-updates:
+				t.Errorf("received more updates than expected")
+			case <-time.After(1 * time.Second):
+			}
+			close(done)
+		}()
+		<-done
+		c.Close(true)
+	})
+	t.Run("TestUpdateOrdering", func(t *testing.T) {
+		report := test.CheckRoutines(t)
+		defer report()
+		updates := make(chan ConnectionState)
+		c := &handlerNotifier{
+			connectionStateFunc: func(cs ConnectionState) {
+				updates <- cs
+			},
+			done: make(chan struct{}),
+		}
+		done := make(chan struct{})
+		go func() {
+			for i := 0; i < 10000; i++ {
+				x := <-updates
+				if x != ConnectionState(i) {
+					t.Errorf("expected %d got %d", x, i)
+				}
+			}
+			select {
+			case <-updates:
+				t.Errorf("received more updates than expected")
+			case <-time.After(1 * time.Second):
+			}
+			close(done)
+		}()
+		for i := 0; i < 10000; i++ {
+			c.EnqueueConnectionState(ConnectionState(i))
+		}
+		<-done
+		c.Close(true)
+	})
+}
diff --git a/agent_test.go b/agent_test.go
index d41dc36..0835408 100644
--- a/agent_test.go
+++ b/agent_test.go
@@ -1799,3 +1799,54 @@ func TestAcceptAggressiveNomination(t *testing.T) {
 		return
 	}
 }
+
+// Close can deadlock but GracefulClose must not
+func TestAgentGracefulCloseDeadlock(t *testing.T) {
+	defer test.CheckRoutinesStrict(t)()
+	defer test.TimeOut(time.Second * 5).Stop()
+
+	config := &AgentConfig{
+		NetworkTypes: supportedNetworkTypes(),
+	}
+	aAgent, err := NewAgent(config)
+	require.NoError(t, err)
+
+	bAgent, err := NewAgent(config)
+	require.NoError(t, err)
+
+	var connected, closeNow, closed sync.WaitGroup
+	connected.Add(2)
+	closeNow.Add(1)
+	closed.Add(2)
+	closeHdlr := func(agent *Agent) {
+		check(agent.OnConnectionStateChange(func(cs ConnectionState) {
+			if cs == ConnectionStateConnected {
+				connected.Done()
+				closeNow.Wait()
+
+				go func() {
+					if err := agent.GracefulClose(); err != nil {
+						require.NoError(t, err)
+					}
+					closed.Done()
+				}()
+			}
+		}))
+	}
+
+	closeHdlr(aAgent)
+	closeHdlr(bAgent)
+
+	t.Log("connecting agents")
+	_, _ = connect(aAgent, bAgent)
+
+	t.Log("waiting for them to confirm connection in callback")
+	connected.Wait()
+
+	t.Log("tell them to close themselves in the same callback and wait")
+	closeNow.Done()
+	closed.Wait()
+
+	require.NoError(t, aAgent.Close())
+	require.NoError(t, bAgent.Close())
+}
diff --git a/candidate_test.go b/candidate_test.go
index 76d1052..64b4e3b 100644
--- a/candidate_test.go
+++ b/candidate_test.go
@@ -184,7 +184,7 @@ func TestCandidateLastSent(t *testing.T) {
 	assert.Equal(t, candidate.LastSent(), time.Time{})
 	now := time.Now()
 	candidate.setLastSent(now)
-	assert.EqualValues(t, 0, now.Sub(candidate.LastSent()))
+	assert.Equal(t, candidate.LastSent(), now)
 }
 
 func TestCandidateLastReceived(t *testing.T) {
@@ -192,7 +192,7 @@ func TestCandidateLastReceived(t *testing.T) {
 	assert.Equal(t, candidate.LastReceived(), time.Time{})
 	now := time.Now()
 	candidate.setLastReceived(now)
-	assert.EqualValues(t, 0, now.Sub(candidate.LastReceived()))
+	assert.Equal(t, candidate.LastReceived(), now)
 }
 
 func TestCandidateFoundation(t *testing.T) {
@@ -391,7 +391,7 @@ func TestCandidateMarshal(t *testing.T) {
 
 		assert.NoError(t, err)
 
-		assert.True(t, test.candidate.Equal(actualCandidate))
+		assert.Truef(t, test.candidate.Equal(actualCandidate), "%s != %s", test.candidate.String(), actualCandidate.String())
 		assert.Equal(t, test.marshaled, actualCandidate.Marshal())
 	}
 }
@@ -435,3 +435,40 @@ func TestCandidateWriteTo(t *testing.T) {
 	_, err = c1.writeTo([]byte("test"), c2)
 	assert.Error(t, err, "writing to closed conn")
 }
+
+func TestMarshalUnmarshalCandidateWithZoneID(t *testing.T) {
+	candidateWithZoneID := &CandidateHost{
+		candidateBase{
+			networkType:        NetworkTypeUDP6,
+			candidateType:      CandidateTypeHost,
+			address:            "fcd9:e3b8:12ce:9fc5:74a5:c6bb:d8b:e08a%Local Connection",
+			port:               53987,
+			priorityOverride:   500,
+			foundationOverride: "750",
+		},
+		"",
+	}
+	candidateStr := "750 0 udp 500 fcd9:e3b8:12ce:9fc5:74a5:c6bb:d8b:e08a 53987 typ host"
+	require.Equal(t, candidateStr, candidateWithZoneID.Marshal())
+
+	candidate := &CandidateHost{
+		candidateBase{
+			networkType:        NetworkTypeUDP6,
+			candidateType:      CandidateTypeHost,
+			address:            "fcd9:e3b8:12ce:9fc5:74a5:c6bb:d8b:e08a",
+			port:               53987,
+			priorityOverride:   500,
+			foundationOverride: "750",
+		},
+		"",
+	}
+	candidateWithZoneIDStr := "750 0 udp 500 fcd9:e3b8:12ce:9fc5:74a5:c6bb:d8b:e08a%eth0 53987 typ host"
+	candidate2, err := UnmarshalCandidate(candidateWithZoneIDStr)
+	require.NoError(t, err)
+	require.Truef(t, candidate.Equal(candidate2), "%s != %s", candidate.String(), candidate2.String())
+
+	candidateWithZoneIDStr2 := "750 0 udp 500 fcd9:e3b8:12ce:9fc5:74a5:c6bb:d8b:e08a%eth0%eth1 53987 typ host"
+	candidate2, err = UnmarshalCandidate(candidateWithZoneIDStr2)
+	require.NoError(t, err)
+	require.Truef(t, candidate.Equal(candidate2), "%s != %s", candidate.String(), candidate2.String())
+}
diff --git a/selection_test.go b/selection_test.go
new file mode 100644
index 0000000..5caea14
--- /dev/null
+++ b/selection_test.go
@@ -0,0 +1,156 @@
+// SPDX-FileCopyrightText: 2023 The Pion community <https://pion.ly>
+// SPDX-License-Identifier: MIT
+
+//go:build !js
+// +build !js
+
+package ice
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"io"
+	"net"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/pion/stun"
+	"github.com/pion/transport/v2/test"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func sendUntilDone(t *testing.T, writingConn, readingConn net.Conn, maxAttempts int) bool {
+	testMessage := []byte("Hello World")
+	testBuffer := make([]byte, len(testMessage))
+
+	readDone, readDoneCancel := context.WithCancel(context.Background())
+	go func() {
+		_, err := readingConn.Read(testBuffer)
+		if errors.Is(err, io.EOF) {
+			return
+		}
+
+		require.NoError(t, err)
+		require.True(t, bytes.Equal(testMessage, testBuffer))
+
+		readDoneCancel()
+	}()
+
+	attempts := 0
+	for {
+		select {
+		case <-time.After(5 * time.Millisecond):
+			if attempts > maxAttempts {
+				return false
+			}
+
+			_, err := writingConn.Write(testMessage)
+			require.NoError(t, err)
+			attempts++
+		case <-readDone.Done():
+			return true
+		}
+	}
+}
+
+func TestBindingRequestHandler(t *testing.T) {
+	defer test.CheckRoutines(t)()
+	defer test.TimeOut(time.Second * 30).Stop()
+
+	var switchToNewCandidatePair, controlledLoggingFired atomic.Value
+	oneHour := time.Hour
+	keepaliveInterval := time.Millisecond * 20
+
+	aNotifier, aConnected := onConnected()
+	bNotifier, bConnected := onConnected()
+	controllingAgent, err := NewAgent(&AgentConfig{
+		NetworkTypes:      []NetworkType{NetworkTypeUDP4, NetworkTypeUDP6},
+		MulticastDNSMode:  MulticastDNSModeDisabled,
+		KeepaliveInterval: &keepaliveInterval,
+		CheckInterval:     &oneHour,
+		BindingRequestHandler: func(_ *stun.Message, _, _ Candidate, _ *CandidatePair) bool {
+			controlledLoggingFired.Store(true)
+			return false
+		},
+	})
+	require.NoError(t, err)
+	require.NoError(t, controllingAgent.OnConnectionStateChange(aNotifier))
+
+	controlledAgent, err := NewAgent(&AgentConfig{
+		NetworkTypes:      []NetworkType{NetworkTypeUDP4},
+		MulticastDNSMode:  MulticastDNSModeDisabled,
+		KeepaliveInterval: &keepaliveInterval,
+		CheckInterval:     &oneHour,
+		BindingRequestHandler: func(_ *stun.Message, _, _ Candidate, _ *CandidatePair) bool {
+			// Don't switch candidate pair until we are ready
+			val, ok := switchToNewCandidatePair.Load().(bool)
+			return ok && val
+		},
+	})
+	require.NoError(t, err)
+	require.NoError(t, controlledAgent.OnConnectionStateChange(bNotifier))
+
+	controlledConn, controllingConn := connect(controlledAgent, controllingAgent)
+	<-aConnected
+	<-bConnected
+
+	// Assert we have connected and can send data
+	require.True(t, sendUntilDone(t, controlledConn, controllingConn, 100))
+
+	// Take the lock on the controlling Agent and unset state
+	assert.NoError(t, controlledAgent.run(controlledAgent.context(), func(_ context.Context, controlledAgent *Agent) {
+		for net, cs := range controlledAgent.remoteCandidates {
+			for _, c := range cs {
+				require.NoError(t, c.close())
+			}
+			delete(controlledAgent.remoteCandidates, net)
+		}
+
+		for _, c := range controlledAgent.localCandidates[NetworkTypeUDP4] {
+			cast, ok := c.(*CandidateHost)
+			require.True(t, ok)
+			cast.remoteCandidateCaches = map[AddrPort]Candidate{}
+		}
+
+		controlledAgent.setSelectedPair(nil)
+		controlledAgent.checklist = make([]*CandidatePair, 0)
+	}))
+
+	// Assert that Selected Candidate pair has only been unset on Controlled side
+	candidatePair, err := controlledAgent.GetSelectedCandidatePair()
+	assert.Nil(t, candidatePair)
+	assert.NoError(t, err)
+
+	candidatePair, err = controllingAgent.GetSelectedCandidatePair()
+	assert.NotNil(t, candidatePair)
+	assert.NoError(t, err)
+
+	// Sending will fail, we no longer have a selected candidate pair
+	require.False(t, sendUntilDone(t, controlledConn, controllingConn, 20))
+
+	// Send STUN Binding requests until a new Selected Candidate Pair has been set by BindingRequestHandler
+	switchToNewCandidatePair.Store(true)
+	for {
+		controllingAgent.requestConnectivityCheck()
+
+		candidatePair, err = controlledAgent.GetSelectedCandidatePair()
+		require.NoError(t, err)
+		if candidatePair != nil {
+			break
+		}
+
+		time.Sleep(time.Millisecond * 5)
+	}
+
+	// We have a new selected candidate pair because of BindingRequestHandler, test that it works
+	require.True(t, sendUntilDone(t, controllingConn, controlledConn, 100))
+
+	fired, ok := controlledLoggingFired.Load().(bool)
+	require.True(t, ok)
+	require.True(t, fired)
+
+	closePipe(t, controllingConn, controlledConn)
+}