Repository: pion/turn

Issue:
Force use of channels
Hi,

Is there a way to force use of Channels instead of Indications where in practice my app seems to be randomly defaulting either on Indications or Channels at connection start-up.

What are the forces in effect influencing the behavior?

Discussion:
@rg0now: You mean on the pion/turn client side? My understanding is that the client by default creates a Channel for every peer address:port you send to (see [here](https://github.com/pion/turn/blob/dac75d84d97eccc56e476b84010e6006ef83460f/internal/client/udp_conn.go#L182)) and currently there is no way to override this. As per the server: if the client requests a channel we create one, I'm not sure how to control this.@berkant: Sorry for the late followup. Just went over a long stream of STUN traffic in WireShark, and it seems there is a fallback for good to indications if a channel bind request to refresh the channel fails with a Nonce stale response from the server. It seems to affect egress first where ingress from the TURN server keeps running over channels for a little while more at which point it also falls back to indications (Data Indications from client's POV).

So, it appears the issue is around turn not retrying to refresh a channel so as to keep it alive upon error.@rg0now: Is this an issue then? The pion/turn client somehow forgets to refresh an allocation? I'm not sure I follow your train of thought. @berkant: Bindings, not allocations. There is actually [timers to refresh both allocs and perms](https://github.com/pion/turn/blob/master/internal/client/udp_conn.go#L68-L85) but not for channel bindings. So yeah, it would mean the end of channel-based comms once the binding state switches to failed for any reason because [UDPConn.bind](https://github.com/pion/turn/blob/55f84c235638bc182421da7e277197f680ba4649/internal/client/udp_conn.go#L418) doesn't seem to handle stale nonces unlike peers such as [CreatePermissions](https://github.com/pion/turn/blob/master/internal/client/udp_conn.go#L379-L383).@rg0now: I had to look up the [RFC](https://datatracker.ietf.org/doc/html/rfc8656#name-channels) to remind myself how channel timeouts go:

> Channel bindings last for 10 minutes unless refreshed; this lifetime was chosen to be longer than the permission lifetime. Channel bindings are refreshed by sending another ChannelBind request rebinding the channel to the peer. Like permissions (but unlike allocations), there is no way to explicitly delete a channel binding; the client must simply wait for it to time out.

So there seem to be two bugs here:
- pion/turn clients [do not start a timer](https://github.com/pion/turn/blob/master/internal/client/udp_conn.go#L68-L85) to refresh channel bindings and correspondingly never refresh channels,
- `bind`, which I guess should be the function to handle channel refresh would call (once implemented), does not handle stale nonces the same way as [`CreatePermission` and `refreshPermissions` do](https://github.com/pion/turn/blob/55f84c235638bc182421da7e277197f680ba4649/internal/client/allocation.go#L126).

I have to admit I'm not that familiar with the client-side code, is the above analysis right? Seems like the [server side is OK](https://github.com/pion/turn/blob/55f84c235638bc182421da7e277197f680ba4649/internal/allocation/allocation.go#L142) though.@berkant: > is the above analysis right?

Yes. Client in fact does refresh channel bindings [at write time](https://github.com/pion/turn/blob/dac75d84d97eccc56e476b84010e6006ef83460f/internal/client/udp_conn.go#L240-L252) as needed, however, as noted earlier, stale nonces aren't handled and are recognized as like any error, causing the binding to be marked as failed. `WriteTo` method is then bound only to [block of code that does indications](https://github.com/pion/turn/blob/dac75d84d97eccc56e476b84010e6006ef83460f/internal/client/udp_conn.go#L191-L231).

Solution patch:
diff --git a/internal/client/binding.go b/internal/client/binding.go
index 1203cce..0720d03 100644
--- a/internal/client/binding.go
+++ b/internal/client/binding.go
@@ -61,6 +61,12 @@ func (b *binding) refreshedAt() time.Time {
 	return b._refreshedAt
 }
 
+func (b *binding) ok() bool {
+	state := b.state()
+
+	return state == bindingStateReady || state == bindingStateRefresh
+}
+
 // Thread-safe binding map.
 type bindingManager struct {
 	chanMap map[uint16]*binding
@@ -159,3 +165,15 @@ func (mgr *bindingManager) size() int {
 
 	return len(mgr.chanMap)
 }
+
+func (mgr *bindingManager) all() []*binding {
+	mgr.mutex.RLock()
+	defer mgr.mutex.RUnlock()
+
+	list := make([]*binding, 0, len(mgr.chanMap))
+	for _, b := range mgr.chanMap {
+		list = append(list, b)
+	}
+
+	return list
+}
diff --git a/internal/client/udp_conn.go b/internal/client/udp_conn.go
index 5108150..7f5e1d8 100644
--- a/internal/client/udp_conn.go
+++ b/internal/client/udp_conn.go
@@ -17,14 +17,17 @@ import (
 )
 
 const (
-	maxReadQueueSize    = 1024
-	permRefreshInterval = 120 * time.Second
-	maxRetryAttempts    = 3
+	maxReadQueueSize       = 1024
+	permRefreshInterval    = 120 * time.Second
+	bindingRefreshInterval = 5 * time.Minute
+	bindingCheckInterval   = 30 * time.Second
+	maxRetryAttempts       = 3
 )
 
 const (
 	timerIDRefreshAlloc int = iota
 	timerIDRefreshPerms
+	timerIDCheckBindings
 )
 
 type inboundData struct {
@@ -35,9 +38,10 @@ type inboundData struct {
 // UDPConn is the implementation of the Conn and PacketConn interfaces for UDP network connections.
 // compatible with net.PacketConn and net.Conn.
 type UDPConn struct {
-	bindingMgr *bindingManager   // Thread-safe
-	readCh     chan *inboundData // Thread-safe
-	closeCh    chan struct{}     // Thread-safe
+	bindingMgr         *bindingManager   // Thread-safe
+	checkBindingsTimer *PeriodicTimer    // Thread-safe
+	readCh             chan *inboundData // Thread-safe
+	closeCh            chan struct{}     // Thread-safe
 	allocation
 }
 
@@ -77,12 +81,25 @@ func NewUDPConn(config *AllocationConfig) *UDPConn {
 		permRefreshInterval,
 	)
 
+	conn.checkBindingsTimer = NewPeriodicTimer(
+		timerIDCheckBindings,
+		func(timerID int) {
+			for _, bound := range conn.bindingMgr.all() {
+				conn.maybeBind(bound)
+			}
+		},
+		bindingCheckInterval,
+	)
+
 	if conn.refreshAllocTimer.Start() {
 		conn.log.Debugf("Started refresh allocation timer")
 	}
 	if conn.refreshPermsTimer.Start() {
 		conn.log.Debugf("Started refresh permission timer")
 	}
+	if conn.checkBindingsTimer.Start() {
+		conn.log.Debugf("Started check bindings timer")
+	}
 
 	return conn
 }
@@ -185,31 +202,11 @@ func (c *UDPConn) WriteTo(payload []byte, addr net.Addr) (int, error) { //nolint
 		bound = c.bindingMgr.create(addr)
 	}
 
-	bindSt := bound.state()
-
 	//nolint:nestif
-	if bindSt == bindingStateIdle || bindSt == bindingStateRequest || bindSt == bindingStateFailed {
-		func() {
-			// Block only callers with the same binding until
-			// the binding transaction has been complete
-			bound.muBind.Lock()
-			defer bound.muBind.Unlock()
-
-			// Binding state may have been changed while waiting. check again.
-			if bound.state() == bindingStateIdle {
-				bound.setState(bindingStateRequest)
-				go func() {
-					err2 := c.bind(bound)
-					if err2 != nil {
-						c.log.Warnf("Failed to bind bind(): %s", err2)
-						bound.setState(bindingStateFailed)
-						// Keep going...
-					} else {
-						bound.setState(bindingStateReady)
-					}
-				}()
-			}
-		}()
+	if !bound.ok() {
+		// Try to establish an initial binding with the server.
+		// Writes still occur via indications meanwhile.
+		c.maybeBind(bound)
 
 		// Send data using SendIndication
 		peerAddr := addr2PeerAddress(addr)
@@ -225,34 +222,10 @@ func (c *UDPConn) WriteTo(payload []byte, addr net.Addr) (int, error) { //nolint
 			return 0, err
 		}
 
-		// Indication has no transaction (fire-and-forget)
-
 		return c.client.WriteTo(msg.Raw, c.serverAddr)
 	}
 
-	// Binding is either ready
-
-	// Check if the binding needs a refresh
-	func() {
-		bound.muBind.Lock()
-		defer bound.muBind.Unlock()
-
-		if bound.state() == bindingStateReady && time.Since(bound.refreshedAt()) > 5*time.Minute {
-			bound.setState(bindingStateRefresh)
-			go func() {
-				if bindErr := c.bind(bound); bindErr != nil {
-					c.log.Warnf("Failed to bind() for refresh: %s", bindErr)
-					bound.setState(bindingStateFailed)
-					// Keep going...
-				} else {
-					bound.setRefreshedAt(time.Now())
-					bound.setState(bindingStateReady)
-				}
-			}()
-		}
-	}()
-
-	// Send via ChannelData
+	// Binding is ready beyond this point, so send over it.
 	_, err = c.sendChannelData(payload, bound.number)
 	if err != nil {
 		return 0, err
@@ -266,6 +239,7 @@ func (c *UDPConn) WriteTo(payload []byte, addr net.Addr) (int, error) { //nolint
 func (c *UDPConn) Close() error {
 	c.refreshAllocTimer.Stop()
 	c.refreshPermsTimer.Stop()
+	c.checkBindingsTimer.Stop()
 
 	select {
 	case <-c.closeCh:
@@ -420,6 +394,44 @@ func (c *UDPConn) FindAddrByChannelNumber(chNum uint16) (net.Addr, bool) {
 	return b.addr, true
 }
 
+func (c *UDPConn) maybeBind(bound *binding) {
+	bind := func() {
+		var err error
+		for i := 0; i < maxRetryAttempts; i++ {
+			if err = c.bind(bound); !errors.Is(err, errTryAgain) {
+				break
+			}
+		}
+		if err != nil {
+			c.log.Warnf("Failed to bind channel %d: %s", bound.number, err)
+			bound.setState(bindingStateFailed)
+
+			return
+		}
+		bound.setRefreshedAt(time.Now())
+		bound.setState(bindingStateReady)
+	}
+
+	// Block only callers with the same binding until
+	// the binding transaction has been complete
+	bound.muBind.Lock()
+	defer bound.muBind.Unlock()
+
+	state := bound.state()
+	switch {
+	case state == bindingStateIdle:
+		bound.setState(bindingStateRequest)
+	case state == bindingStateReady && time.Since(bound.refreshedAt()) > bindingRefreshInterval:
+		bound.setState(bindingStateRefresh)
+	default:
+		return
+	}
+
+	// Establish binding with the server if eligible
+	// with regard to cases right above.
+	go bind()
+}
+
 func (c *UDPConn) bind(bound *binding) error {
 	setters := []stun.Setter{
 		stun.TransactionID,
@@ -446,8 +458,15 @@ func (c *UDPConn) bind(bound *binding) error {
 	}
 
 	res := trRes.Msg
+	if res.Type.Class == stun.ClassErrorResponse {
+		var code stun.ErrorCodeAttribute
+		if err = code.GetFrom(res); err == nil {
+			if code.Code == stun.CodeStaleNonce {
+				c.setNonceFromMsg(res)
 
-	if res.Type != stun.NewType(stun.MethodChannelBind, stun.ClassSuccessResponse) {
+				return errTryAgain
+			}
+		}
 		return fmt.Errorf("unexpected response type %s", res.Type) //nolint // dynamic errors
 	}
 



Test patch:
diff --git a/internal/client/binding_test.go b/internal/client/binding_test.go
index f437626..ba18447 100644
--- a/internal/client/binding_test.go
+++ b/internal/client/binding_test.go
@@ -46,6 +46,13 @@ func TestBindingManager(t *testing.T) {
 			assert.Equal(t, b0, b2, "should match")
 		}
 
+		all := bm.all()
+		for _, b := range all {
+			found, ok := bm.findByNumber(b.number)
+			assert.True(t, ok, "should exist")
+			assert.Equal(t, b, found, "should match")
+		}
+		assert.Equal(t, count, len(all), "should match")
 		assert.Equal(t, count, bm.size(), "should match")
 		assert.Equal(t, count, len(bm.addrMap), "should match")
 
@@ -60,6 +67,7 @@ func TestBindingManager(t *testing.T) {
 
 		assert.Equal(t, 0, bm.size(), "should match")
 		assert.Equal(t, 0, len(bm.addrMap), "should match")
+		assert.Equal(t, 0, len(bm.all()), "should match")
 	})
 
 	t.Run("failure test", func(t *testing.T) {
diff --git a/internal/client/udp_conn_test.go b/internal/client/udp_conn_test.go
index 40c8444..4c11b21 100644
--- a/internal/client/udp_conn_test.go
+++ b/internal/client/udp_conn_test.go
@@ -7,36 +7,143 @@ import (
 	"errors"
 	"net"
 	"testing"
+	"time"
 
+	"github.com/pion/logging"
 	"github.com/pion/stun/v3"
 	"github.com/stretchr/testify/assert"
 )
 
 func TestUDPConn(t *testing.T) {
-	t.Run("bind()", func(t *testing.T) {
-		client := &mockClient{
-			performTransaction: func(*stun.Message, net.Addr, bool) (TransactionResult, error) {
-				return TransactionResult{}, errFake
+	makeConn := func(client *mockClient, bm *bindingManager) UDPConn {
+		return UDPConn{
+			allocation: allocation{
+				client: client,
+				log:    logging.NewDefaultLoggerFactory().NewLogger("test"),
 			},
+			bindingMgr: bm,
+		}
+	}
+
+	staleNonceMsg := func() *stun.Message {
+		return stun.MustBuild(
+			stun.NewType(stun.MethodChannelBind, stun.ClassErrorResponse),
+			stun.CodeStaleNonce,
+			stun.NewNonce("new-nonce-123"),
+		)
+	}
+
+	t.Run("maybeBind()", func(t *testing.T) {
+		tests := []struct {
+			name          string
+			initialState  bindingState
+			interimState  bindingState
+			finalState    bindingState
+			pastInterval  bool
+			shouldSucceed bool
+		}{
+			{"idle -> request -> ready", bindingStateIdle, bindingStateRequest, bindingStateReady, false, true},
+			{"idle -> request -> failed", bindingStateIdle, bindingStateRequest, bindingStateFailed, false, false},
+			{"ready (stale) -> refresh -> ready", bindingStateReady, bindingStateRefresh, bindingStateReady, true, true},
+			{"ready (stale) -> refresh -> failed", bindingStateReady, bindingStateRefresh, bindingStateFailed, true, false},
+
+			// Noop cases:
+			{"ready (noop)", bindingStateReady, bindingStateReady, bindingStateReady, false, true},
+			{"request (noop)", bindingStateRequest, bindingStateRequest, bindingStateRequest, false, true},
+			{"refresh (noop)", bindingStateRefresh, bindingStateRefresh, bindingStateRefresh, false, true},
+			{"failed (noop)", bindingStateFailed, bindingStateFailed, bindingStateFailed, false, true},
 		}
 
-		bm := newBindingManager()
-		b := bm.create(&net.UDPAddr{
-			IP:   net.ParseIP("127.0.0.1"),
-			Port: 1234,
-		})
+		for _, tt := range tests {
+			t.Run(tt.name, func(t *testing.T) {
+				unblock := make(chan struct{})
+
+				bm := newBindingManager()
+				bound := bm.create(&net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 1234})
+				conn := makeConn(&mockClient{
+					performTransaction: func(msg *stun.Message, addr net.Addr, dontWait bool) (TransactionResult, error) {
+						<-unblock
+						if tt.shouldSucceed {
+							return TransactionResult{Msg: new(stun.Message)}, nil
+						}
+
+						return TransactionResult{Msg: staleNonceMsg()}, nil
+					},
+				}, bm)
+
+				bound.setState(tt.initialState)
+				if tt.pastInterval {
+					bound.setRefreshedAt(time.Now().Add(-(bindingRefreshInterval + 1*time.Minute)))
+				}
 
-		conn := UDPConn{
-			allocation: allocation{
-				client: client,
+				conn.maybeBind(bound)
+				assert.Equal(t, tt.interimState, bound.state())
+
+				// Release barrier so inner bind() can move forward.
+				close(unblock)
+
+				assert.Eventually(t, func() bool {
+					return bound.state() == tt.finalState
+				}, 5*time.Second, 10*time.Millisecond)
+			})
+		}
+	})
+
+	t.Run("bind()", func(t *testing.T) {
+		tests := []struct {
+			name                 string
+			transactionFn        func(*stun.Message, net.Addr, bool) (TransactionResult, error)
+			expectErr            error
+			expectBindingDeleted bool
+			expectNonceChanged   bool
+		}{
+			{
+				name: "PerformTransaction returns error",
+				transactionFn: func(*stun.Message, net.Addr, bool) (TransactionResult, error) {
+					return TransactionResult{}, errFake
+				},
+				expectErr:            errFake,
+				expectBindingDeleted: true,
+			},
+			{
+				name: "ErrorResponse with CodeStaleNonce triggers nonce update",
+				transactionFn: func(*stun.Message, net.Addr, bool) (TransactionResult, error) {
+					return TransactionResult{Msg: staleNonceMsg()}, nil
+				},
+				expectErr:          errTryAgain,
+				expectNonceChanged: true,
 			},
-			bindingMgr: bm,
 		}
 
-		err := conn.bind(b)
-		assert.Error(t, err, "should fail")
-		assert.Equal(t, 0, len(bm.chanMap), "should be 0")
-		assert.Equal(t, 0, len(bm.addrMap), "should be 0")
+		for _, tt := range tests {
+			t.Run(tt.name, func(t *testing.T) {
+				bm := newBindingManager()
+				bound := bm.create(&net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 1234})
+				conn := makeConn(&mockClient{performTransaction: tt.transactionFn}, bm)
+
+				nonceT0 := conn.nonce()
+
+				err := conn.bind(bound)
+				if tt.expectErr == nil {
+					assert.NoError(t, err)
+				} else {
+					assert.ErrorIs(t, err, tt.expectErr)
+				}
+
+				if tt.expectBindingDeleted {
+					assert.Empty(t, bm.chanMap)
+					assert.Empty(t, bm.addrMap)
+				}
+
+				nonceT1 := conn.nonce()
+				if tt.expectNonceChanged {
+					assert.NotEqual(t, nonceT0, nonceT1, "should change")
+					assert.NotEmpty(t, nonceT1, "should be non-empty")
+				} else {
+					assert.Equal(t, nonceT0, nonceT1, "should remain unchanged")
+				}
+			})
+		}
 	})
 
 	t.Run("WriteTo()", func(t *testing.T) {