Use ICE candidate buffering on the server

It is unknown yet if Pion handles remote candidate additions with "no
remote description" gracefully.

TODO: check and remove later

pkg/network/webrtc/ice.go

@@ -0,0 +1,34 @@
+package webrtc
+
+import (
+	"sync"
+
+	"github.com/pion/webrtc/v4"
+)
+
+type IceCandidateBuffer struct {
+	mu  sync.Mutex
+	buf []webrtc.ICECandidateInit
+}
+
+func (b *IceCandidateBuffer) push(c webrtc.ICECandidateInit) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	b.buf = append(b.buf, c)
+}
+
+// FlushAll atomically swaps the internal buffer with a new, empty one
+// and returns the old buffer's contents for processing.
+func (b *IceCandidateBuffer) FlushAll() []webrtc.ICECandidateInit {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	oldBuffer := b.buf
+	b.buf = nil // Or b.buf = make([]Candidate, 0, initialCapacity)
+	return oldBuffer
+}
+
+func (b *IceCandidateBuffer) Clear() {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	b.buf = nil
+}

pkg/network/webrtc/webrtc.go

@@ -20,13 +20,17 @@ type Peer struct {
 	a *webrtc.TrackLocalStaticSample
 	v *webrtc.TrackLocalStaticSample
 	d *webrtc.DataChannel
+
+	icb *IceCandidateBuffer
 }
 
 var samplePool sync.Pool
 
 type Decoder func(data string, obj any) error
 
-func New(log *logger.Logger, api *ApiFactory) *Peer { return &Peer{api: api, log: log} }
+func New(log *logger.Logger, api *ApiFactory) *Peer {
+	return &Peer{api: api, log: log, icb: &IceCandidateBuffer{}}
+}
 
 func (p *Peer) NewConnection(vCodec, aCodec string, onICECandidate func(ice any)) (err error) {
 	if p.conn != nil && p.conn.ConnectionState() == webrtc.PeerConnectionStateConnected {
@@ -66,6 +70,13 @@ func (p *Peer) NewConnection(vCodec, aCodec string, onICECandidate func(ice any)
 	})
 	p.conn.OnICECandidate(p.handleICECandidate(onICECandidate))
 
+	p.conn.OnSignalingStateChange(func(ss webrtc.SignalingState) {
+		p.log.Debug().Msgf("[rtc] [sig] %s", ss.Get())
+		if ss == webrtc.SignalingStateStable {
+			p.flushPendingCandidates()
+		}
+	})
+
 	// plug in the [video] track (out)
 	if p.v, err = p.AddTrack("video", "video", vCodec); err != nil {
 		return err
@@ -168,6 +179,7 @@ func (p *Peer) SetRemoteSDP(sdp string, decoder Decoder) error {
 		p.log.Error().Err(err).Msg("Set remote description from peer failed")
 		return err
 	}
+	p.flushPendingCandidates()
 	p.log.Debug().Msg("Set Remote Description")
 	return nil
 }
@@ -234,18 +246,17 @@ func (p *Peer) handleICEState(onConnect func()) func(webrtc.ICEConnectionState)
 }
 
 func (p *Peer) AddCandidate(candidate string, decoder Decoder) error {
-	// !to add test when the connection is closed but it is still
-	// receiving ice candidates
-
 	var iceCandidate webrtc.ICECandidateInit
 	if err := decoder(candidate, &iceCandidate); err != nil {
 		return err
 	}
-	if err := p.conn.AddICECandidate(iceCandidate); err != nil {
-		return err
-	}
-	p.log.Debug().Str("candidate", iceCandidate.Candidate).Msg("Ice")
-	return nil
+	p.log.Debug().
+		Str("x", "[ice]").
+		Str("remote", iceCandidate.Candidate).
+		Msg("[rtc]")
+	buffered := p.conn.RemoteDescription() == nil ||
+		p.conn.SignalingState() != webrtc.SignalingStateStable
+	return p.addCandidate(iceCandidate, buffered)
 }
 
 func (p *Peer) AddChannel(label string, conf *webrtc.DataChannelInit, onMessage func([]byte)) (*webrtc.DataChannel, error) {
@@ -282,7 +293,41 @@ func (p *Peer) Disconnect() {
 		// ignore this due to DTLS fatal: conn is closed
 		_ = p.conn.Close()
 	}
+	p.icb.Clear()
 	p.log.Debug().Msg("WebRTC stop")
 }
 
+func (p *Peer) addCandidate(candidate webrtc.ICECandidateInit, wait bool) error {
+	if wait {
+		p.icb.push(candidate)
+		return nil
+	}
+	if err := p.conn.AddICECandidate(candidate); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (p *Peer) flushPendingCandidates() {
+	prev := p.icb.FlushAll()
+	if len(prev) == 0 {
+		return
+	}
+
+	p.log.Debug().
+		Str("x", "[ice]").
+		Msg(fmt.Sprintf("[rtc] buf (%d) flush", len(prev)))
+
+	for _, candidate := range prev {
+		if err := p.addCandidate(candidate, false); err != nil {
+			p.log.Error().
+				Str("x", "[ice]").
+				Str("remote", candidate.Candidate).
+				Err(err).
+				Msg("[rtc]")
+		}
+	}
+	clear(prev)
+}
+
 func (p *Peer) logx(err error) { p.log.Error().Err(err) }

web/js/network/webrtc.js

@@ -186,6 +186,13 @@ export const webrtc = {
             log.debug("[rtc] negotiation");
         };
         pc.ontrack = (event) => stream.addTrack(event.track);
+        pc.onsignalingstatechange = () => {
+            log.debug(`[rtc] [sig] state: ${pc.signalingState}`);
+
+            if (pc.signalingState === "stable") {
+                ice.flush(pc);
+            }
+        };
 
         connectionTime = performance.now();
         if (initiator) {
@@ -225,7 +232,9 @@ export const webrtc = {
     },
     candidate: (/** @type {RTCIceCandidateInit | string} */ candidate) => {
         log.debug(`[rtc] [ice] remote`, candidate);
-        if (pc) ice.add(pc, candidate, !pc.remoteDescription);
+        const buffered =
+            !pc.remoteDescription || pc.signalingState !== "stable";
+        if (pc) ice.add(pc, candidate, buffered);
     },
     send: (chan, data) => {
         const ch = channels.get(chan);