Fix warp issues

dtlstransport.go

@@ -326,6 +326,64 @@
 	})
 }
 
+// dtlsPacketInjector feeds embedded DTLS packets into a dtls.Conn from its own
+// goroutine, in arrival order.
+//
+// Conn.InjectInboundPacket sends on an unbuffered channel that only
+// Conn.nextPacket reads, so it blocks until the DTLS stack asks for its next
+// packet. The SPED callback runs on the ICE agent's task loop, the one
+// goroutine handling all inbound STUN, and blocking that stalls the agent so
+// the handshake never completes. Queue instead and let this goroutine block.
+type dtlsPacketInjector struct {
+	packets   chan pendingDTLSPacket
+	done      chan struct{}
+	closeOnce sync.Once
+}
+
+// dtlsInjectorQueueLen is deliberately generous: a DTLS handshake flight is a
+// handful of packets, so reaching this bound means the DTLS stack has stopped
+// consuming altogether and dropping is the right answer.
+const dtlsInjectorQueueLen = 128
+
+func newDTLSPacketInjector(conn *dtls.Conn) *dtlsPacketInjector {
+	injector := &dtlsPacketInjector{
+		packets: make(chan pendingDTLSPacket, dtlsInjectorQueueLen),
+		done:    make(chan struct{}),
+	}
+
+	go func() {
+		for {
+			select {
+			case <-injector.done:
+				return
+			case packet := <-injector.packets:
+				conn.InjectInboundPacket(packet.packet, packet.rAddr)
+			}
+		}
+	}()
+
+	return injector
+}
+
+// enqueue is the SPED callback handed to ICETransport. It never blocks.
+func (i *dtlsPacketInjector) enqueue(packet []byte, rAddr net.Addr) {
+	buffered := make([]byte, len(packet))
+	copy(buffered, packet)
+
+	select {
+	case i.packets <- pendingDTLSPacket{packet: buffered, rAddr: rAddr}:
+	case <-i.done:
+	default:
+		// Queue full: drop rather than block the caller. DTLS retransmits its
+		// flights, so a drop costs a retransmission. Blocking would cost the
+		// whole ICE agent.
+	}
+}
+
+func (i *dtlsPacketInjector) stop() {
+	i.closeOnce.Do(func() { close(i.done) })
+}
+
 func (t *DTLSTransport) start(remoteParameters DTLSParameters, handshake func(*dtls.Conn) error) error {
 	role, certificate, err := t.prepareStart(remoteParameters)
 	if err != nil {
@@ -347,9 +405,9 @@
 
 	// Configure ICE for SPED after we created the DTLS transport.
 	if t.api.settingEngine.enableSped {
-		t.iceTransport.SetDtlsCallback(func(packet []byte, rAddr net.Addr) {
-			dtlsConn.InjectInboundPacket(packet, rAddr)
-		})
+		injector := newDTLSPacketInjector(dtlsConn)
+		defer injector.stop()
+		t.iceTransport.SetDtlsCallback(injector.enqueue)
 	}
 
 	// This awaits the DTLS handshake.
@@ -402,7 +460,7 @@
 		dtls.WithCertificates(certificate),
 		dtls.WithSRTPProtectionProfiles(t.srtpProtectionProfiles()...),
 		dtls.WithExtendedMasterSecret(t.api.settingEngine.dtls.extendedMasterSecret),
 		// TODO: this should be the default, DTLS runs over ICE which *hopefully* checks the source.
 		dtls.WithInsecureSkipVerify(true),
 		dtls.WithLoggerFactory(t.api.settingEngine.LoggerFactory),
 		dtls.WithVerifyPeerCertificate(t.verifyPeerCertificateFunc()),
@@ -415,7 +473,7 @@
 		)
 	}
 
 	// TODO: should this initially be set to one day for SPED?
 	if t.api.settingEngine.dtls.retransmissionInterval > 0 {
 		sharedOpts = append(
 			sharedOpts,

go.mod

@@ -36,6 +36,6 @@ require (
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
 
-replace github.com/pion/dtls/v3 => github.com/livekit/dtls/v3 v3.1.4-warp.1
+replace github.com/pion/dtls/v3 => github.com/livekit/dtls/v3 v3.1.5-warp.2
 
-replace github.com/pion/ice/v4 => github.com/livekit/ice/v4 v4.2.7-warp.1
+replace github.com/pion/ice/v4 => github.com/livekit/ice/v4 v4.4.0-warp.3

go.sum

@@ -25,10 +25,10 @@ github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
-github.com/livekit/dtls/v3 v3.1.4-warp.1 h1:oS+wSZDhgIXDLX8cmSeFhfdX28XZxwxLGNOmTP8B+yk=
-github.com/livekit/dtls/v3 v3.1.4-warp.1/go.mod h1:cr/qotLISUw/9C1m83ZPNZtj9WnXkYLpfCptPqbkInc=
-github.com/livekit/ice/v4 v4.2.7-warp.1 h1:6ieMfxp6ADkM10SR0Fyu2LYe+TJQADxY//1BJAphn4o=
-github.com/livekit/ice/v4 v4.2.7-warp.1/go.mod h1:9SNPaq0c7El/ki8leJzyCkK10zsskprR3zTNbO3monY=
+github.com/livekit/dtls/v3 v3.1.5-warp.2 h1:ypVipsYIbVKYWUDmrNDKbLZ9NbxhJdMN3xCekKhXNDs=
+github.com/livekit/dtls/v3 v3.1.5-warp.2/go.mod h1:gz1K4jg6c+fq86oQMH4pilpCEOEPwmEr2jY+VcF/mkU=
+github.com/livekit/ice/v4 v4.4.0-warp.3 h1:iSAtEiLeCi/ZKEKOsRrDlBg/BRcyUKpV/NuAXX+dS3o=
+github.com/livekit/ice/v4 v4.4.0-warp.3/go.mod h1:obAyD+J+Hzs7QA7Y8YXHp5uIn6gb7z87pKedXZkrcFU=
 github.com/nxadm/tail v1.4.4/go.mod h1:kenIhsEOeOJmVchQTgglprH7qJGnHDVpk1VPCcaMI8A=
 github.com/nxadm/tail v1.4.8 h1:nPr65rt6Y5JFSKQO7qToXr7pePgD6Gwiw05lkbyAQTE=
 github.com/nxadm/tail v1.4.8/go.mod h1:+ncqLTQzXmGhMZNUePPaPqPvBxHAIsmXswZKocGu+AU=

icetransport.go

@@ -19,6 +19,12 @@ import (
 	"github.com/pion/webrtc/v4/internal/util"
 )
 
+// pendingDTLSPacket is an embedded DTLS packet held until DTLSTransport is ready.
+type pendingDTLSPacket struct {
+	packet []byte
+	rAddr  net.Addr
+}
+
 // ICETransport allows an application access to information about the ICE
 // transport over which packets are sent and received.
 type ICETransport struct {
@@ -42,6 +48,10 @@ type ICETransport struct {
 
 	dtlsCallback func(packet []byte, rAddr net.Addr)
 
+	// Inbound DTLS packets the agent delivered before DTLSTransport configured
+	// its callback (SPED only)
+	pendingDTLSPackets []pendingDTLSPacket
+
 	log logging.LeveledLogger
 }
 
@@ -123,7 +133,16 @@ func (t *ICETransport) StartContext(
 	if agent == nil {
 		return fmt.Errorf("%w: unable to start ICETransport", errICEAgentNotExist)
 	}
-	agent.SetDtlsCallback(t.dtlsCallback)
+
+	// The agent only piggybacks DTLS once a callback is set, and DTLSTransport
+	// cannot set the real one until the mux exists (created below). If ICE
+	// connects in that gap the handshake falls back to plain DTLS and loses the
+	// round trip SPED saves. Install a shim now and buffer until it arrives.
+	if t.gatherer.api.settingEngine.enableSped && t.dtlsCallback == nil {
+		agent.SetDtlsCallback(t.forwardDTLSPacket)
+	} else {
+		agent.SetDtlsCallback(t.dtlsCallback)
+	}
 
 	if err := agent.OnConnectionStateChange(func(iceState ice.ConnectionState) {
 		state := newICETransportStateFromICE(iceState)
@@ -179,6 +198,10 @@ func (t *ICETransport) StartContext(
 	if err != nil {
 		t.lock.Lock()
 
+		// The derived context is unused past this point
+		ctxCancel()
+		t.ctxCancel = nil
+
 		return err
 	}
 
@@ -222,12 +245,53 @@ func (t *ICETransport) StartContext(
 
 func (t *ICETransport) SetDtlsCallback(cb func(packet []byte, rAddr net.Addr)) {
 	t.lock.Lock()
-	defer t.lock.Unlock()
-	if agent := t.gatherer.getAgent(); agent != nil {
-		agent.SetDtlsCallback(cb)
-	} else {
-		t.dtlsCallback = cb
+	agent := t.gatherer.getAgent()
+
+	if agent == nil || !t.gatherer.api.settingEngine.enableSped {
+		if agent != nil {
+			agent.SetDtlsCallback(cb)
+		} else {
+			t.dtlsCallback = cb
+		}
+		t.lock.Unlock()
+
+		return
 	}
+
+	// SPED: StartContext put a shim on the agent that forwards to this field, so
+	// record the callback and hand it whatever arrived before it existed.
+	t.dtlsCallback = cb
+	pending := t.pendingDTLSPackets
+	t.pendingDTLSPackets = nil
+	t.lock.Unlock()
+
+	if cb == nil {
+		// Teardown: take the shim off the agent as well.
+		agent.SetDtlsCallback(nil)
+
+		return
+	}
+
+	for _, p := range pending {
+		cb(p.packet, p.rAddr)
+	}
+}
+
+// forwardDTLSPacket hands an embedded DTLS packet to DTLSTransport, buffering it
+// while the DTLS transport is still being built.
+func (t *ICETransport) forwardDTLSPacket(packet []byte, rAddr net.Addr) {
+	t.lock.Lock()
+	if cb := t.dtlsCallback; cb != nil {
+		t.lock.Unlock()
+		cb(packet, rAddr)
+
+		return
+	}
+
+	buffered := make([]byte, len(packet))
+	copy(buffered, packet)
+	t.pendingDTLSPackets = append(t.pendingDTLSPackets, pendingDTLSPacket{packet: buffered, rAddr: rAddr})
+	t.lock.Unlock()
 }
 
 // restart is not exposed currently because ORTC has users create a whole new ICETransport

peerconnection_test.go

@@ -5,7 +5,6 @@ package webrtc
 
 import (
 	"runtime"
-	"strings"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -969,44 +968,3 @@ func TestICETrickleCapabilityString(t *testing.T) {
 		assert.Equal(t, tt.expected, tt.value.String())
 	}
 }
-
-func TestWarp(t *testing.T) {
-	s := SettingEngine{}
-	s.EnableSped(true)
-	api := NewAPI(WithSettingEngine(s))
-
-	offer, err := api.NewPeerConnection(Configuration{})
-	assert.NoError(t, err)
-	answer, err := api.NewPeerConnection(Configuration{})
-	assert.NoError(t, err)
-
-	peerConnectionsConnected := untilConnectionState(PeerConnectionStateConnected, offer, answer)
-	assert.NoError(t, signalPair(offer, answer))
-	peerConnectionsConnected.Wait()
-
-	closePairNow(t, offer, answer)
-}
-
-func TestWarpClient(t *testing.T) {
-	s := SettingEngine{}
-	s.EnableSped(true)
-	api := NewAPI(WithSettingEngine(s))
-
-	offer, err := api.NewPeerConnection(Configuration{})
-	assert.NoError(t, err)
-	answer, err := api.NewPeerConnection(Configuration{})
-	assert.NoError(t, err)
-
-	peerConnectionsConnected := untilConnectionState(PeerConnectionStateConnected, offer, answer)
-	assert.NoError(t, signalPairWithModification(
-		offer, answer,
-		func(sessionDescription string) string {
-			return strings.ReplaceAll(
-				sessionDescription,
-				"setup:actpass",
-				"setup:active")
-		}))
-	peerConnectionsConnected.Wait()
-
-	closePairNow(t, offer, answer)
-}

sped_warp_test.go

@@ -0,0 +1,54 @@
+// SPDX-FileCopyrightText: 2026 The Pion community <https://pion.ly>
+// SPDX-License-Identifier: MIT
+
+//go:build !js
+
+package webrtc
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestWarp(t *testing.T) {
+	s := SettingEngine{}
+	s.EnableSped(true)
+	api := NewAPI(WithSettingEngine(s))
+
+	offer, err := api.NewPeerConnection(Configuration{})
+	assert.NoError(t, err)
+	answer, err := api.NewPeerConnection(Configuration{})
+	assert.NoError(t, err)
+
+	peerConnectionsConnected := untilConnectionState(PeerConnectionStateConnected, offer, answer)
+	assert.NoError(t, signalPair(offer, answer))
+	peerConnectionsConnected.Wait()
+
+	closePairNow(t, offer, answer)
+}
+
+func TestWarpClient(t *testing.T) {
+	s := SettingEngine{}
+	s.EnableSped(true)
+	api := NewAPI(WithSettingEngine(s))
+
+	offer, err := api.NewPeerConnection(Configuration{})
+	assert.NoError(t, err)
+	answer, err := api.NewPeerConnection(Configuration{})
+	assert.NoError(t, err)
+
+	peerConnectionsConnected := untilConnectionState(PeerConnectionStateConnected, offer, answer)
+	assert.NoError(t, signalPairWithModification(
+		offer, answer,
+		func(sessionDescription string) string {
+			return strings.ReplaceAll(
+				sessionDescription,
+				"setup:actpass",
+				"setup:active")
+		}))
+	peerConnectionsConnected.Wait()
+
+	closePairNow(t, offer, answer)
+}