chore: Fix race in collecting ICE Candidates (#76)

* chore: Fix race in collecting ICE Candidates

This logic was flawed previously. ICE Candidates could collect
before a negotiation was triggered, which led to a race where
candidates would be lost. Candidates can no longer be lost,
and we removed some code 😎.

* Add comment describing fix

* Use upstream dependency to fix goroutine leak

* Use upstream dependency to fix goroutine leak

peer/channel.go

@@ -271,16 +271,16 @@ func (c *Channel) closeWithError(err error) error {
 	} else {
 		c.closeError = err
 	}
+	if c.rwc != nil {
+		_ = c.rwc.Close()
+	}
+	_ = c.dc.Close()
+
 	close(c.closed)
 	close(c.sendMore)
 	c.conn.dcDisconnectListeners.Sub(1)
 	c.conn.dcFailedListeners.Sub(1)
 	c.conn.dcClosedWaitGroup.Done()
-
-	if c.rwc != nil {
-		_ = c.rwc.Close()
-	}
-	_ = c.dc.Close()
 	return err
 }
 

peer/conn.go

@@ -73,6 +73,7 @@ func newWithClientOrServer(servers []webrtc.ICEServer, client bool, opts *ConnOp
 		dcFailedChannel:                 make(chan struct{}),
 		localCandidateChannel:           make(chan webrtc.ICECandidateInit),
 		localSessionDescriptionChannel:  make(chan webrtc.SessionDescription),
+		pendingCandidates:               make([]webrtc.ICECandidateInit, 0),
 		remoteSessionDescriptionChannel: make(chan webrtc.SessionDescription),
 	}
 	if client {
@@ -118,7 +119,9 @@ type Conn struct {
 	localCandidateChannel           chan webrtc.ICECandidateInit
 	localSessionDescriptionChannel  chan webrtc.SessionDescription
 	remoteSessionDescriptionChannel chan webrtc.SessionDescription
-	remoteSessionDescriptionMutex   sync.Mutex
+
+	pendingCandidates      []webrtc.ICECandidateInit
+	pendingCandidatesMutex sync.Mutex
 
 	pingChannelID     uint16
 	pingEchoChannelID uint16
@@ -134,6 +137,25 @@ type Conn struct {
 
 func (c *Conn) init() error {
 	c.rtc.OnNegotiationNeeded(c.negotiate)
+	c.rtc.OnICECandidate(func(iceCandidate *webrtc.ICECandidate) {
+		if iceCandidate == nil {
+			return
+		}
+		// ICE Candidates on a remote peer are reset when an offer
+		// is received. We must wait until the offer<->answer has
+		// been negotiated to flush candidates.
+		c.pendingCandidatesMutex.Lock()
+		defer c.pendingCandidatesMutex.Unlock()
+		if c.rtc.RemoteDescription() == nil {
+			c.pendingCandidates = append(c.pendingCandidates, iceCandidate.ToJSON())
+			return
+		}
+		select {
+		case <-c.closed:
+			break
+		case c.localCandidateChannel <- iceCandidate.ToJSON():
+		}
+	})
 	c.rtc.OnDataChannel(func(dc *webrtc.DataChannel) {
 		select {
 		case <-c.closed:
@@ -232,12 +254,6 @@ func (c *Conn) pingEchoChannel() (*Channel, error) {
 
 func (c *Conn) negotiate() {
 	c.opts.Logger.Debug(context.Background(), "negotiating")
-	flushCandidates := c.proxyICECandidates()
-
-	// Locks while the negotiation for a remote session
-	// description is taking place.
-	c.remoteSessionDescriptionMutex.Lock()
-	defer c.remoteSessionDescriptionMutex.Unlock()
 
 	if c.offerrer {
 		offer, err := c.rtc.CreateOffer(&webrtc.OfferOptions{})
@@ -291,44 +307,17 @@ func (c *Conn) negotiate() {
 		}
 	}
 
-	flushCandidates()
-	c.opts.Logger.Debug(context.Background(), "flushed candidates")
-}
-
-func (c *Conn) proxyICECandidates() func() {
-	var (
-		mut     sync.Mutex
-		queue   = []webrtc.ICECandidateInit{}
-		flushed = false
-	)
-	c.rtc.OnICECandidate(func(iceCandidate *webrtc.ICECandidate) {
-		if iceCandidate == nil {
-			return
-		}
-		mut.Lock()
-		defer mut.Unlock()
-		if !flushed {
-			queue = append(queue, iceCandidate.ToJSON())
-			return
-		}
+	c.pendingCandidatesMutex.Lock()
+	defer c.pendingCandidatesMutex.Unlock()
+	for _, pendingCandidate := range c.pendingCandidates {
 		select {
 		case <-c.closed:
 			return
-		case c.localCandidateChannel <- iceCandidate.ToJSON():
+		case c.localCandidateChannel <- pendingCandidate:
 		}
-	})
-	return func() {
-		mut.Lock()
-		defer mut.Unlock()
-		for _, q := range queue {
-			select {
-			case <-c.closed:
-				break
-			case c.localCandidateChannel <- q:
-			}
-		}
-		flushed = true
 	}
+	c.pendingCandidates = make([]webrtc.ICECandidateInit, 0)
+	c.opts.Logger.Debug(context.Background(), "flushed candidates")
 }
 
 // LocalCandidate returns a channel that emits when a local candidate
@@ -339,9 +328,6 @@ func (c *Conn) LocalCandidate() <-chan webrtc.ICECandidateInit {
 
 // AddRemoteCandidate adds a remote candidate to the RTC connection.
 func (c *Conn) AddRemoteCandidate(i webrtc.ICECandidateInit) error {
-	// Prevents candidates from being added before an offer<->answer has occurred.
-	c.remoteSessionDescriptionMutex.Lock()
-	defer c.remoteSessionDescriptionMutex.Unlock()
 	return c.rtc.AddICECandidate(i)
 }