diff --git a/dtlstransport.go b/dtlstransport.go index 8d9b4f29c73..9a1bb7f2f08 100644 --- a/dtlstransport.go +++ b/dtlstransport.go @@ -326,6 +326,64 @@ func (t *DTLSTransport) StartContext(ctx context.Context, remoteParameters DTLSP }) } +// 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 @@ func (t *DTLSTransport) start(remoteParameters DTLSParameters, handshake func(*d // 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. diff --git a/go.mod b/go.mod index 4ccb67c14c9..6c0d4baaac2 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index fb8ed9b6d33..04c5c380ee8 100644 --- a/go.sum +++ b/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= diff --git a/icetransport.go b/icetransport.go index a8d799ff7be..c026c895044 100644 --- a/icetransport.go +++ b/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 diff --git a/peerconnection_test.go b/peerconnection_test.go index 4877d7cf4a1..09f09eba167 100644 --- a/peerconnection_test.go +++ b/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) -} diff --git a/sped_offer_test.go b/sped_offer_test.go deleted file mode 100644 index 8343832d6c2..00000000000 --- a/sped_offer_test.go +++ /dev/null @@ -1,64 +0,0 @@ -// SPDX-FileCopyrightText: 2026 The Pion community -// SPDX-License-Identifier: MIT - -//go:build !js - -package webrtc - -import ( - "strings" - "testing" - - "github.com/stretchr/testify/assert" -) - -// TestSpedOfferAdvertisesGoogSpedV1 guards the downstream (server-offers) path: -// when SPED is enabled the offer MUST carry `a=ice-options:trickle goog-sped-v1`, -// otherwise a libwebrtc answerer will not enable DTLS-in-STUN and the subscriber -// PC silently falls back to plain DTLS (no fold). The "trickle" token must be -// present too, or a pion peer treats us as non-trickle and de-paces ICE. -func TestSpedOfferAdvertisesGoogSpedV1(t *testing.T) { - s := SettingEngine{} - s.EnableSped(true) - pc, err := NewAPI(WithSettingEngine(s)).NewPeerConnection(Configuration{}) - assert.NoError(t, err) - defer func() { _ = pc.Close() }() - - _, err = pc.AddTransceiverFromKind(RTPCodecTypeAudio, RTPTransceiverInit{ - Direction: RTPTransceiverDirectionSendonly, - }) - assert.NoError(t, err) - - offer, err := pc.CreateOffer(nil) - assert.NoError(t, err) - - var iceOptions string - for _, line := range strings.Split(offer.SDP, "\r\n") { - if strings.HasPrefix(line, "a=ice-options:") { - iceOptions = line - break - } - } - assert.NotEmpty(t, iceOptions, "SPED offer must contain an a=ice-options line") - assert.Contains(t, iceOptions, "goog-sped-v1", - "SPED offer must advertise goog-sped-v1 so a libwebrtc answerer enables the fold") - assert.Contains(t, iceOptions, "trickle", - "ice-options must keep trickle or a pion peer de-paces ICE and the fold breaks") -} - -// TestNoSpedNoGoogSpedV1 confirms the option is opt-in: with SPED disabled the -// offer must not advertise goog-sped-v1. -func TestNoSpedNoGoogSpedV1(t *testing.T) { - pc, err := NewAPI().NewPeerConnection(Configuration{}) - assert.NoError(t, err) - defer func() { _ = pc.Close() }() - - _, err = pc.AddTransceiverFromKind(RTPCodecTypeAudio, RTPTransceiverInit{ - Direction: RTPTransceiverDirectionSendonly, - }) - assert.NoError(t, err) - - offer, err := pc.CreateOffer(nil) - assert.NoError(t, err) - assert.NotContains(t, offer.SDP, "goog-sped-v1") -} diff --git a/sped_warp_test.go b/sped_warp_test.go new file mode 100644 index 00000000000..5ad3e65ba1d --- /dev/null +++ b/sped_warp_test.go @@ -0,0 +1,54 @@ +// SPDX-FileCopyrightText: 2026 The Pion community +// 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) +}