From 2e70bed995745aa9fac48565bff2f92456e4cca8 Mon Sep 17 00:00:00 2001 From: Joshua Lee <24319042+bclswl0827@users.noreply.github.com> Date: Wed, 29 Jul 2026 15:15:46 +0800 Subject: [PATCH 1/4] ice: add extensible relay candidate providers Allow external relay transports to register candidate providers without adding protocol-specific logic to pion-ice. Expose relay local preference configuration and add tests for custom protocols, multiple providers, and invalid candidate cleanup. --- agent.go | 2 + agent_options.go | 26 +++++ candidate_base.go | 20 ++-- candidate_relay.go | 16 ++- gather.go | 100 +++++++++++----- relay_candidate.go | 26 +++++ relay_candidate_provider_test.go | 195 +++++++++++++++++++++++++++++++ 7 files changed, 344 insertions(+), 41 deletions(-) create mode 100644 relay_candidate.go create mode 100644 relay_candidate_provider_test.go diff --git a/agent.go b/agent.go index d9e67644..eadbf2a2 100644 --- a/agent.go +++ b/agent.go @@ -182,6 +182,8 @@ type Agent struct { lastRenominationTime time.Time turnClientFactory func(*turn.ClientConfig) (turnClient, error) + + relayCandidateProviders []RelayCandidateProvider } // NewAgent creates a new Agent. diff --git a/agent_options.go b/agent_options.go index dd124655..5f99695a 100644 --- a/agent_options.go +++ b/agent_options.go @@ -949,6 +949,32 @@ func WithCandidateTypes(candidateTypes []CandidateType) AgentOption { } } +// WithRelayCandidateProvider registers a provider for non-TURN relay +// candidates. The provider is used in addition to the relay candidates +// gathered from WithUrls and is ignored when relay candidates are disabled. +func WithRelayCandidateProvider(provider RelayCandidateProvider) AgentOption { + return WithRelayCandidateProviders(provider) +} + +// WithRelayCandidateProviders registers providers for non-TURN relay +// candidates. Providers are called independently during candidate gathering, +// so an application can add multiple relay transports without modifying ICE. +func WithRelayCandidateProviders(providers ...RelayCandidateProvider) AgentOption { + return func(a *Agent) error { + if a.constructed { + return ErrAgentOptionNotUpdatable + } + + for _, provider := range providers { + if provider != nil { + a.relayCandidateProviders = append(a.relayCandidateProviders, provider) + } + } + + return nil + } +} + // WithAutomaticRenomination enables automatic renomination of candidate pairs // when better pairs become available after initial connection establishment. // This feature requires renomination to be enabled and both agents to support it. diff --git a/candidate_base.go b/candidate_base.go index b4f48df9..69ee836f 100644 --- a/candidate_base.go +++ b/candidate_base.go @@ -976,17 +976,15 @@ func UnmarshalCandidate(raw string) (Candidate, error) { //nolint:cyclop return candidate, nil case "relay": candidate, err := NewCandidateRelay(&CandidateRelayConfig{ - "", - protocol, - address, - port, - uint16(component), //nolint:gosec // G115 no overflow we read 5 digits - uint32(priority), //nolint:gosec // G115 no overflow we read 5 digits - foundation, - raddr, - rport, - "", - nil, + CandidateID: "", + Network: protocol, + Address: address, + Port: port, + Component: uint16(component), //nolint:gosec // G115 no overflow we read 5 digits + Priority: uint32(priority), //nolint:gosec // G115 no overflow we read 5 digits + Foundation: foundation, + RelAddr: raddr, + RelPort: rport, }) if err != nil { return nil, err diff --git a/candidate_relay.go b/candidate_relay.go index 47d40c43..934410da 100644 --- a/candidate_relay.go +++ b/candidate_relay.go @@ -38,7 +38,11 @@ type CandidateRelayConfig struct { RelAddr string RelPort int RelayProtocol string - OnClose func() error + // RelayLocalPreference controls the local preference component of the + // relay candidate priority. It is supplied by the relay implementation so + // custom relay transports do not need to be added to ICE's protocol switch. + RelayLocalPreference uint16 + OnClose func() error } // NewCandidateRelay creates a new relay candidate. @@ -59,6 +63,14 @@ func NewCandidateRelay(config *CandidateRelayConfig) (*CandidateRelay, error) { return nil, err } + localPreference := config.RelayLocalPreference + // Candidates created without relay metadata retain the default preference. + // Built-in TURN candidates and custom providers with a protocol name supply + // their preference explicitly. + if config.RelayProtocol == "" && localPreference == 0 { + localPreference = defaultLocalPreference + } + candidate := &CandidateRelay{ candidateBase: candidateBase{ id: candidateID, @@ -73,7 +85,7 @@ func NewCandidateRelay(config *CandidateRelayConfig) (*CandidateRelay, error) { Address: config.RelAddr, Port: config.RelPort, }, - relayLocalPreference: relayProtocolPreference(config.RelayProtocol), + relayLocalPreference: localPreference, }, relayProtocol: config.RelayProtocol, onClose: config.OnClose, diff --git a/gather.go b/gather.go index b357e066..c5e0c72d 100644 --- a/gather.go +++ b/gather.go @@ -276,11 +276,20 @@ func (a *Agent) gatherCandidatesInternal(ctx context.Context) { case CandidateTypeServerReflexive: a.gatherServerReflexiveCandidates(ctx, &wg) case CandidateTypeRelay: - wg.Add(1) + // TURN and custom relay providers are independent gathering paths. + // Run them concurrently so a slow or unavailable TURN server does not + // delay candidates from another relay implementation. + wg.Add(1 + len(a.relayCandidateProviders)) go func() { + defer wg.Done() a.gatherCandidatesRelay(ctx, a.urls) - wg.Done() }() + for _, provider := range a.relayCandidateProviders { + go func() { + defer wg.Done() + a.gatherCandidatesFromProvider(ctx, provider) + }() + } case CandidateTypePeerReflexive, CandidateTypeUnspecified: } } @@ -289,6 +298,38 @@ func (a *Agent) gatherCandidatesInternal(ctx context.Context) { wg.Wait() } +func (a *Agent) gatherCandidatesFromProvider(ctx context.Context, provider RelayCandidateProvider) { + candidates, err := provider.GatherCandidates(ctx, a.localUfrag, a.localPwd) + if err != nil { + a.log.Warnf("Failed to gather custom relay candidates: %v", err) + + return + } + + for _, item := range candidates { + if item.Conn == nil { + a.log.Warn("Ignoring custom relay candidate with nil packet connection") + + continue + } + if item.Config.Component == 0 { + item.Config.Component = ComponentRTP + } + candidate, err := NewCandidateRelay(&item.Config) + if err != nil { + _ = item.Conn.Close() + a.log.Warnf("Failed to create custom relay candidate: %v", err) + + continue + } + if err := a.addCandidate(ctx, candidate, item.Conn); err != nil { + _ = candidate.close() + _ = item.Conn.Close() + a.log.Warnf("Failed to add custom relay candidate: %v", err) + } + } +} + func (a *Agent) gatherServerReflexiveCandidates(ctx context.Context, wg *sync.WaitGroup) { replaceSrflx := a.addressRewriteMapper != nil && a.addressRewriteMapper.shouldReplace(CandidateTypeServerReflexive) if !replaceSrflx { @@ -1228,14 +1269,15 @@ func (a *Agent) gatherCandidatesRelay(ctx context.Context, urls []*stun.URI) { // Relay allocations currently produce UDP relay endpoints regardless of // whether the TURN control connection uses UDP/TCP/TLS/DTLS. a.addRelayCandidates(ctx, relayEndpoint{ - network: udp, - address: rAddr.IP, - port: rAddr.Port, - relAddr: relAddr, - relPort: relPort, - iface: findIfaceForIP(ifaces, net.ParseIP(relAddr)), - protocol: relayProtocol, - conn: relayConn, + network: udp, + address: rAddr.IP, + port: rAddr.Port, + relAddr: relAddr, + relPort: relPort, + iface: findIfaceForIP(ifaces, net.ParseIP(relAddr)), + protocol: relayProtocol, + localPreference: relayProtocolPreference(relayProtocol), + conn: relayConn, onClose: func() error { client.Close() @@ -1250,16 +1292,17 @@ func (a *Agent) gatherCandidatesRelay(ctx context.Context, urls []*stun.URI) { } type relayEndpoint struct { - network string - address net.IP - port int - relAddr string - relPort int - protocol string - iface string - conn net.PacketConn - onClose func() error - closeConn func() + network string + address net.IP + port int + relAddr string + relPort int + protocol string + localPreference uint16 + iface string + conn net.PacketConn + onClose func() error + closeConn func() } func (a *Agent) resolveRelayAddresses(ep relayEndpoint) ([]net.IP, bool) { @@ -1346,14 +1389,15 @@ func findIfaceForIP(ifaces []ifaceAddr, ip net.IP) string { func (a *Agent) createRelayCandidate(ctx context.Context, ep relayEndpoint, ip net.IP, onClose func() error) error { relayConfig := CandidateRelayConfig{ - Network: ep.network, - Component: ComponentRTP, - Address: ip.String(), - Port: ep.port, - RelAddr: ep.relAddr, - RelPort: ep.relPort, - RelayProtocol: ep.protocol, - OnClose: onClose, + Network: ep.network, + Component: ComponentRTP, + Address: ip.String(), + Port: ep.port, + RelAddr: ep.relAddr, + RelPort: ep.relPort, + RelayProtocol: ep.protocol, + RelayLocalPreference: ep.localPreference, + OnClose: onClose, } candidate, err := NewCandidateRelay(&relayConfig) if err != nil { diff --git a/relay_candidate.go b/relay_candidate.go new file mode 100644 index 00000000..b5062d37 --- /dev/null +++ b/relay_candidate.go @@ -0,0 +1,26 @@ +// SPDX-FileCopyrightText: 2026 The Pion community +// SPDX-License-Identifier: MIT + +package ice + +import ( + "context" + "net" +) + +// RelayCandidate is a locally allocated relay candidate and the packet +// connection used to reach the relay. The connection must preserve packet +// boundaries and return the remote candidate address from ReadFrom. +type RelayCandidate struct { + Config CandidateRelayConfig + Conn net.PacketConn +} + +// RelayCandidateProvider allocates non-TURN relay candidates. The provider is +// called during candidate gathering when relay candidates are enabled. A +// provider may use the ICE credentials to bind the allocation to this agent, +// but it must not assume that the credentials are the relay authentication +// credentials. +type RelayCandidateProvider interface { + GatherCandidates(context.Context, string, string) ([]RelayCandidate, error) +} diff --git a/relay_candidate_provider_test.go b/relay_candidate_provider_test.go new file mode 100644 index 00000000..0e123d1b --- /dev/null +++ b/relay_candidate_provider_test.go @@ -0,0 +1,195 @@ +// SPDX-FileCopyrightText: 2026 The Pion community +// SPDX-License-Identifier: MIT + +package ice + +import ( + "context" + "io" + "net" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +type relayProviderTestPacketConn struct { + addr net.Addr + closeCount atomic.Int32 +} + +func (c *relayProviderTestPacketConn) ReadFrom([]byte) (int, net.Addr, error) { + return 0, c.addr, io.EOF +} + +func (c *relayProviderTestPacketConn) WriteTo(payload []byte, _ net.Addr) (int, error) { + return len(payload), nil +} + +func (c *relayProviderTestPacketConn) Close() error { + c.closeCount.Add(1) + + return nil +} + +func (c *relayProviderTestPacketConn) LocalAddr() net.Addr { + return c.addr +} + +func (c *relayProviderTestPacketConn) SetDeadline(_ time.Time) error { + return nil +} + +func (c *relayProviderTestPacketConn) SetReadDeadline(_ time.Time) error { + return nil +} + +func (c *relayProviderTestPacketConn) SetWriteDeadline(_ time.Time) error { + return nil +} + +type relayProviderTestProvider struct { + called atomic.Int32 + packetConn *relayProviderTestPacketConn + relayProtocol string + localPreference uint16 + address string + port int +} + +func (p *relayProviderTestProvider) GatherCandidates(context.Context, string, string) ([]RelayCandidate, error) { + p.called.Add(1) + + return []RelayCandidate{{ + Config: CandidateRelayConfig{ + Network: NetworkTypeUDP4.String(), + Address: p.address, + Port: p.port, + Component: ComponentRTP, + RelayProtocol: p.relayProtocol, + RelayLocalPreference: p.localPreference, + }, + Conn: p.packetConn, + }}, nil +} + +func TestNewCandidateRelayAcceptsExternalProtocolAndPreference(t *testing.T) { + candidate, err := NewCandidateRelay(&CandidateRelayConfig{ + Network: NetworkTypeUDP4.String(), + Address: "192.0.2.10", + Port: 5000, + Component: ComponentRTP, + RelayProtocol: "quic", + RelayLocalPreference: 37, + }) + require.NoError(t, err) + + require.Equal(t, "quic", candidate.RelayProtocol()) + require.Equal(t, uint16(37), candidate.LocalPreference()) + require.Equal(t, uint32(37*256+255), candidate.Priority()) +} + +func TestAgentGathersFromMultipleRelayProviders(t *testing.T) { + first := &relayProviderTestProvider{ + packetConn: &relayProviderTestPacketConn{addr: &net.UDPAddr{IP: net.IPv4(192, 0, 2, 1), Port: 6001}}, + relayProtocol: "websocket", + localPreference: 21, + address: "192.0.2.1", + port: 6001, + } + second := &relayProviderTestProvider{ + packetConn: &relayProviderTestPacketConn{addr: &net.UDPAddr{IP: net.IPv4(192, 0, 2, 2), Port: 6002}}, + relayProtocol: "quic", + localPreference: 42, + address: "192.0.2.2", + port: 6002, + } + + agent, err := NewAgentWithOptions( + WithCandidateTypes([]CandidateType{CandidateTypeRelay}), + WithNetworkTypes([]NetworkType{NetworkTypeUDP4}), + WithRelayCandidateProviders(first, second), + WithMulticastDNSMode(MulticastDNSModeDisabled), + ) + require.NoError(t, err) + defer func() { require.NoError(t, agent.Close()) }() + require.Equal(t, []CandidateType{CandidateTypeRelay}, agent.candidateTypes) + require.Len(t, agent.relayCandidateProviders, 2) + + gathered := make(chan struct{}) + require.NoError(t, agent.OnCandidate(func(candidate Candidate) { + if candidate == nil { + close(gathered) + } + })) + require.NoError(t, agent.GatherCandidates()) + require.Eventually(t, func() bool { + select { + case <-gathered: + return true + default: + return false + } + }, time.Second, time.Millisecond) + + require.Equal(t, int32(1), first.called.Load()) + require.Equal(t, int32(1), second.called.Load()) + + candidates, err := agent.GetLocalCandidates() + require.NoError(t, err) + require.Len(t, candidates, 2) + + byProtocol := make(map[string]*CandidateRelay, len(candidates)) + for _, candidate := range candidates { + relay, ok := candidate.(*CandidateRelay) + require.True(t, ok) + byProtocol[relay.RelayProtocol()] = relay + } + + require.Equal(t, uint16(21), byProtocol["websocket"].LocalPreference()) + require.Equal(t, uint16(42), byProtocol["quic"].LocalPreference()) +} + +func TestAgentClosesInvalidExternalRelayCandidate(t *testing.T) { + packetConn := &relayProviderTestPacketConn{ + addr: &net.UDPAddr{IP: net.IPv4(192, 0, 2, 3), Port: 6003}, + } + provider := &relayProviderTestProvider{ + packetConn: packetConn, + relayProtocol: "custom", + address: "not-an-ip-address", + port: 6003, + } + + agent, err := NewAgentWithOptions( + WithCandidateTypes([]CandidateType{CandidateTypeRelay}), + WithNetworkTypes([]NetworkType{NetworkTypeUDP4}), + WithRelayCandidateProvider(provider), + WithMulticastDNSMode(MulticastDNSModeDisabled), + ) + require.NoError(t, err) + defer func() { require.NoError(t, agent.Close()) }() + + gathered := make(chan struct{}) + require.NoError(t, agent.OnCandidate(func(candidate Candidate) { + if candidate == nil { + close(gathered) + } + })) + require.NoError(t, agent.GatherCandidates()) + require.Eventually(t, func() bool { + select { + case <-gathered: + return true + default: + return false + } + }, time.Second, time.Millisecond) + + require.Equal(t, int32(1), provider.called.Load()) + require.Equal(t, int32(1), packetConn.closeCount.Load()) + candidates, err := agent.GetLocalCandidates() + require.NoError(t, err) + require.Empty(t, candidates) +} From cade5b9e7106b69beec02e5aa6b0894a7761966e Mon Sep 17 00:00:00 2001 From: Joshua Lee <24319042+bclswl0827@users.noreply.github.com> Date: Mon, 3 Aug 2026 16:50:00 +0000 Subject: [PATCH 2/4] Add manual local candidate registration Expose Agent.AddLocalCandidate so applications can create candidates with their own packet connections and register them before gathering. Remove provider-based relay gathering while preserving existing relay protocol preference handling. --- agent.go | 15 ++- agent_options.go | 26 ----- candidate_base.go | 20 ++-- candidate_relay.go | 16 +-- gather.go | 100 +++++----------- local_candidate_test.go | 135 +++++++++++++++++++++ relay_candidate.go | 26 ----- relay_candidate_provider_test.go | 195 ------------------------------- 8 files changed, 189 insertions(+), 344 deletions(-) create mode 100644 local_candidate_test.go delete mode 100644 relay_candidate.go delete mode 100644 relay_candidate_provider_test.go diff --git a/agent.go b/agent.go index eadbf2a2..b976edb6 100644 --- a/agent.go +++ b/agent.go @@ -182,8 +182,6 @@ type Agent struct { lastRenominationTime time.Time turnClientFactory func(*turn.ClientConfig) (turnClient, error) - - relayCandidateProviders []RelayCandidateProvider } // NewAgent creates a new Agent. @@ -1005,6 +1003,19 @@ func (a *Agent) AddRemoteCandidate(cand Candidate) error { return nil } +// AddLocalCandidate adds a local candidate and the packet connection used to +// send and receive packets for it. +func (a *Agent) AddLocalCandidate(cand Candidate, candidateConn net.PacketConn) error { + if cand == nil { + return nil + } + if candidateConn == nil { + return fmt.Errorf("candidate packet connection is nil") + } + + return a.addCandidate(a.loop, cand, candidateConn) +} + func (a *Agent) resolveAndAddMulticastCandidate(cand *CandidateHost) { if a.mDNSConn == nil { return diff --git a/agent_options.go b/agent_options.go index 5f99695a..dd124655 100644 --- a/agent_options.go +++ b/agent_options.go @@ -949,32 +949,6 @@ func WithCandidateTypes(candidateTypes []CandidateType) AgentOption { } } -// WithRelayCandidateProvider registers a provider for non-TURN relay -// candidates. The provider is used in addition to the relay candidates -// gathered from WithUrls and is ignored when relay candidates are disabled. -func WithRelayCandidateProvider(provider RelayCandidateProvider) AgentOption { - return WithRelayCandidateProviders(provider) -} - -// WithRelayCandidateProviders registers providers for non-TURN relay -// candidates. Providers are called independently during candidate gathering, -// so an application can add multiple relay transports without modifying ICE. -func WithRelayCandidateProviders(providers ...RelayCandidateProvider) AgentOption { - return func(a *Agent) error { - if a.constructed { - return ErrAgentOptionNotUpdatable - } - - for _, provider := range providers { - if provider != nil { - a.relayCandidateProviders = append(a.relayCandidateProviders, provider) - } - } - - return nil - } -} - // WithAutomaticRenomination enables automatic renomination of candidate pairs // when better pairs become available after initial connection establishment. // This feature requires renomination to be enabled and both agents to support it. diff --git a/candidate_base.go b/candidate_base.go index 69ee836f..b4f48df9 100644 --- a/candidate_base.go +++ b/candidate_base.go @@ -976,15 +976,17 @@ func UnmarshalCandidate(raw string) (Candidate, error) { //nolint:cyclop return candidate, nil case "relay": candidate, err := NewCandidateRelay(&CandidateRelayConfig{ - CandidateID: "", - Network: protocol, - Address: address, - Port: port, - Component: uint16(component), //nolint:gosec // G115 no overflow we read 5 digits - Priority: uint32(priority), //nolint:gosec // G115 no overflow we read 5 digits - Foundation: foundation, - RelAddr: raddr, - RelPort: rport, + "", + protocol, + address, + port, + uint16(component), //nolint:gosec // G115 no overflow we read 5 digits + uint32(priority), //nolint:gosec // G115 no overflow we read 5 digits + foundation, + raddr, + rport, + "", + nil, }) if err != nil { return nil, err diff --git a/candidate_relay.go b/candidate_relay.go index 934410da..47d40c43 100644 --- a/candidate_relay.go +++ b/candidate_relay.go @@ -38,11 +38,7 @@ type CandidateRelayConfig struct { RelAddr string RelPort int RelayProtocol string - // RelayLocalPreference controls the local preference component of the - // relay candidate priority. It is supplied by the relay implementation so - // custom relay transports do not need to be added to ICE's protocol switch. - RelayLocalPreference uint16 - OnClose func() error + OnClose func() error } // NewCandidateRelay creates a new relay candidate. @@ -63,14 +59,6 @@ func NewCandidateRelay(config *CandidateRelayConfig) (*CandidateRelay, error) { return nil, err } - localPreference := config.RelayLocalPreference - // Candidates created without relay metadata retain the default preference. - // Built-in TURN candidates and custom providers with a protocol name supply - // their preference explicitly. - if config.RelayProtocol == "" && localPreference == 0 { - localPreference = defaultLocalPreference - } - candidate := &CandidateRelay{ candidateBase: candidateBase{ id: candidateID, @@ -85,7 +73,7 @@ func NewCandidateRelay(config *CandidateRelayConfig) (*CandidateRelay, error) { Address: config.RelAddr, Port: config.RelPort, }, - relayLocalPreference: localPreference, + relayLocalPreference: relayProtocolPreference(config.RelayProtocol), }, relayProtocol: config.RelayProtocol, onClose: config.OnClose, diff --git a/gather.go b/gather.go index c5e0c72d..b357e066 100644 --- a/gather.go +++ b/gather.go @@ -276,20 +276,11 @@ func (a *Agent) gatherCandidatesInternal(ctx context.Context) { case CandidateTypeServerReflexive: a.gatherServerReflexiveCandidates(ctx, &wg) case CandidateTypeRelay: - // TURN and custom relay providers are independent gathering paths. - // Run them concurrently so a slow or unavailable TURN server does not - // delay candidates from another relay implementation. - wg.Add(1 + len(a.relayCandidateProviders)) + wg.Add(1) go func() { - defer wg.Done() a.gatherCandidatesRelay(ctx, a.urls) + wg.Done() }() - for _, provider := range a.relayCandidateProviders { - go func() { - defer wg.Done() - a.gatherCandidatesFromProvider(ctx, provider) - }() - } case CandidateTypePeerReflexive, CandidateTypeUnspecified: } } @@ -298,38 +289,6 @@ func (a *Agent) gatherCandidatesInternal(ctx context.Context) { wg.Wait() } -func (a *Agent) gatherCandidatesFromProvider(ctx context.Context, provider RelayCandidateProvider) { - candidates, err := provider.GatherCandidates(ctx, a.localUfrag, a.localPwd) - if err != nil { - a.log.Warnf("Failed to gather custom relay candidates: %v", err) - - return - } - - for _, item := range candidates { - if item.Conn == nil { - a.log.Warn("Ignoring custom relay candidate with nil packet connection") - - continue - } - if item.Config.Component == 0 { - item.Config.Component = ComponentRTP - } - candidate, err := NewCandidateRelay(&item.Config) - if err != nil { - _ = item.Conn.Close() - a.log.Warnf("Failed to create custom relay candidate: %v", err) - - continue - } - if err := a.addCandidate(ctx, candidate, item.Conn); err != nil { - _ = candidate.close() - _ = item.Conn.Close() - a.log.Warnf("Failed to add custom relay candidate: %v", err) - } - } -} - func (a *Agent) gatherServerReflexiveCandidates(ctx context.Context, wg *sync.WaitGroup) { replaceSrflx := a.addressRewriteMapper != nil && a.addressRewriteMapper.shouldReplace(CandidateTypeServerReflexive) if !replaceSrflx { @@ -1269,15 +1228,14 @@ func (a *Agent) gatherCandidatesRelay(ctx context.Context, urls []*stun.URI) { // Relay allocations currently produce UDP relay endpoints regardless of // whether the TURN control connection uses UDP/TCP/TLS/DTLS. a.addRelayCandidates(ctx, relayEndpoint{ - network: udp, - address: rAddr.IP, - port: rAddr.Port, - relAddr: relAddr, - relPort: relPort, - iface: findIfaceForIP(ifaces, net.ParseIP(relAddr)), - protocol: relayProtocol, - localPreference: relayProtocolPreference(relayProtocol), - conn: relayConn, + network: udp, + address: rAddr.IP, + port: rAddr.Port, + relAddr: relAddr, + relPort: relPort, + iface: findIfaceForIP(ifaces, net.ParseIP(relAddr)), + protocol: relayProtocol, + conn: relayConn, onClose: func() error { client.Close() @@ -1292,17 +1250,16 @@ func (a *Agent) gatherCandidatesRelay(ctx context.Context, urls []*stun.URI) { } type relayEndpoint struct { - network string - address net.IP - port int - relAddr string - relPort int - protocol string - localPreference uint16 - iface string - conn net.PacketConn - onClose func() error - closeConn func() + network string + address net.IP + port int + relAddr string + relPort int + protocol string + iface string + conn net.PacketConn + onClose func() error + closeConn func() } func (a *Agent) resolveRelayAddresses(ep relayEndpoint) ([]net.IP, bool) { @@ -1389,15 +1346,14 @@ func findIfaceForIP(ifaces []ifaceAddr, ip net.IP) string { func (a *Agent) createRelayCandidate(ctx context.Context, ep relayEndpoint, ip net.IP, onClose func() error) error { relayConfig := CandidateRelayConfig{ - Network: ep.network, - Component: ComponentRTP, - Address: ip.String(), - Port: ep.port, - RelAddr: ep.relAddr, - RelPort: ep.relPort, - RelayProtocol: ep.protocol, - RelayLocalPreference: ep.localPreference, - OnClose: onClose, + Network: ep.network, + Component: ComponentRTP, + Address: ip.String(), + Port: ep.port, + RelAddr: ep.relAddr, + RelPort: ep.relPort, + RelayProtocol: ep.protocol, + OnClose: onClose, } candidate, err := NewCandidateRelay(&relayConfig) if err != nil { diff --git a/local_candidate_test.go b/local_candidate_test.go new file mode 100644 index 00000000..3507a06e --- /dev/null +++ b/local_candidate_test.go @@ -0,0 +1,135 @@ +// SPDX-FileCopyrightText: 2026 The Pion community +// SPDX-License-Identifier: MIT + +package ice + +import ( + "io" + "net" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +type localCandidatePacketConn struct { + addr net.Addr + closeCount atomic.Int32 +} + +func (c *localCandidatePacketConn) ReadFrom([]byte) (int, net.Addr, error) { + return 0, c.addr, io.EOF +} + +func (c *localCandidatePacketConn) WriteTo(payload []byte, _ net.Addr) (int, error) { + return len(payload), nil +} + +func (c *localCandidatePacketConn) Close() error { + c.closeCount.Add(1) + + return nil +} + +func (c *localCandidatePacketConn) LocalAddr() net.Addr { return c.addr } +func (c *localCandidatePacketConn) SetDeadline(time.Time) error { return nil } +func (c *localCandidatePacketConn) SetReadDeadline(time.Time) error { return nil } +func (c *localCandidatePacketConn) SetWriteDeadline(time.Time) error { return nil } + +func TestAddLocalCandidateRegistersExternalRelay(t *testing.T) { + agent, err := NewAgentWithOptions( + WithCandidateTypes([]CandidateType{}), + WithMulticastDNSMode(MulticastDNSModeDisabled), + ) + require.NoError(t, err) + defer func() { require.NoError(t, agent.Close()) }() + + gatheringComplete := make(chan struct{}) + candidates := make(chan Candidate, 1) + require.NoError(t, agent.OnCandidate(func(candidate Candidate) { + if candidate == nil { + close(gatheringComplete) + + return + } + candidates <- candidate + })) + + candidate, err := NewCandidateRelay(&CandidateRelayConfig{ + Network: NetworkTypeUDP4.String(), + Address: "192.0.2.10", + Port: 5000, + Component: ComponentRTP, + RelayProtocol: "custom", + }) + require.NoError(t, err) + packetConn := &localCandidatePacketConn{ + addr: &net.UDPAddr{IP: net.IPv4(192, 0, 2, 10), Port: 5000}, + } + + require.NoError(t, agent.AddLocalCandidate(candidate, packetConn)) + localCandidates, err := agent.GetLocalCandidates() + require.NoError(t, err) + require.Len(t, localCandidates, 1) + require.Equal(t, "custom", localCandidates[0].(*CandidateRelay).RelayProtocol()) + + select { + case got := <-candidates: + require.Equal(t, candidate, got) + case <-time.After(time.Second): + t.Fatal("timed out waiting for local candidate callback") + } + + require.NoError(t, agent.GatherCandidates()) + select { + case <-gatheringComplete: + case <-time.After(time.Second): + t.Fatal("timed out waiting for gathering completion") + } +} + +func TestAddLocalCandidateRejectsNilPacketConn(t *testing.T) { + agent, err := NewAgentWithOptions(WithMulticastDNSMode(MulticastDNSModeDisabled)) + require.NoError(t, err) + defer func() { require.NoError(t, agent.Close()) }() + + candidate, err := NewCandidateRelay(&CandidateRelayConfig{ + Network: NetworkTypeUDP4.String(), + Address: "192.0.2.11", + Port: 5001, + Component: ComponentRTP, + }) + require.NoError(t, err) + require.Error(t, agent.AddLocalCandidate(candidate, nil)) +} + +func TestAddLocalCandidateClosesDuplicateConnection(t *testing.T) { + agent, err := NewAgentWithOptions(WithMulticastDNSMode(MulticastDNSModeDisabled)) + require.NoError(t, err) + defer func() { require.NoError(t, agent.Close()) }() + + require.NoError(t, agent.OnCandidate(func(Candidate) {})) + config := CandidateRelayConfig{ + Network: NetworkTypeUDP4.String(), + Address: "192.0.2.12", + Port: 5002, + Component: ComponentRTP, + } + first, err := NewCandidateRelay(&config) + require.NoError(t, err) + second, err := NewCandidateRelay(&config) + require.NoError(t, err) + firstConn := &localCandidatePacketConn{addr: &net.UDPAddr{IP: net.IPv4(192, 0, 2, 12), Port: 5002}} + secondConn := &localCandidatePacketConn{addr: &net.UDPAddr{IP: net.IPv4(192, 0, 2, 12), Port: 5002}} + + require.NoError(t, agent.AddLocalCandidate(first, firstConn)) + require.NoError(t, agent.AddLocalCandidate(second, secondConn)) + require.Eventually(t, func() bool { + return secondConn.closeCount.Load() == 1 + }, time.Second, time.Millisecond) + + localCandidates, err := agent.GetLocalCandidates() + require.NoError(t, err) + require.Len(t, localCandidates, 1) +} diff --git a/relay_candidate.go b/relay_candidate.go deleted file mode 100644 index b5062d37..00000000 --- a/relay_candidate.go +++ /dev/null @@ -1,26 +0,0 @@ -// SPDX-FileCopyrightText: 2026 The Pion community -// SPDX-License-Identifier: MIT - -package ice - -import ( - "context" - "net" -) - -// RelayCandidate is a locally allocated relay candidate and the packet -// connection used to reach the relay. The connection must preserve packet -// boundaries and return the remote candidate address from ReadFrom. -type RelayCandidate struct { - Config CandidateRelayConfig - Conn net.PacketConn -} - -// RelayCandidateProvider allocates non-TURN relay candidates. The provider is -// called during candidate gathering when relay candidates are enabled. A -// provider may use the ICE credentials to bind the allocation to this agent, -// but it must not assume that the credentials are the relay authentication -// credentials. -type RelayCandidateProvider interface { - GatherCandidates(context.Context, string, string) ([]RelayCandidate, error) -} diff --git a/relay_candidate_provider_test.go b/relay_candidate_provider_test.go deleted file mode 100644 index 0e123d1b..00000000 --- a/relay_candidate_provider_test.go +++ /dev/null @@ -1,195 +0,0 @@ -// SPDX-FileCopyrightText: 2026 The Pion community -// SPDX-License-Identifier: MIT - -package ice - -import ( - "context" - "io" - "net" - "sync/atomic" - "testing" - "time" - - "github.com/stretchr/testify/require" -) - -type relayProviderTestPacketConn struct { - addr net.Addr - closeCount atomic.Int32 -} - -func (c *relayProviderTestPacketConn) ReadFrom([]byte) (int, net.Addr, error) { - return 0, c.addr, io.EOF -} - -func (c *relayProviderTestPacketConn) WriteTo(payload []byte, _ net.Addr) (int, error) { - return len(payload), nil -} - -func (c *relayProviderTestPacketConn) Close() error { - c.closeCount.Add(1) - - return nil -} - -func (c *relayProviderTestPacketConn) LocalAddr() net.Addr { - return c.addr -} - -func (c *relayProviderTestPacketConn) SetDeadline(_ time.Time) error { - return nil -} - -func (c *relayProviderTestPacketConn) SetReadDeadline(_ time.Time) error { - return nil -} - -func (c *relayProviderTestPacketConn) SetWriteDeadline(_ time.Time) error { - return nil -} - -type relayProviderTestProvider struct { - called atomic.Int32 - packetConn *relayProviderTestPacketConn - relayProtocol string - localPreference uint16 - address string - port int -} - -func (p *relayProviderTestProvider) GatherCandidates(context.Context, string, string) ([]RelayCandidate, error) { - p.called.Add(1) - - return []RelayCandidate{{ - Config: CandidateRelayConfig{ - Network: NetworkTypeUDP4.String(), - Address: p.address, - Port: p.port, - Component: ComponentRTP, - RelayProtocol: p.relayProtocol, - RelayLocalPreference: p.localPreference, - }, - Conn: p.packetConn, - }}, nil -} - -func TestNewCandidateRelayAcceptsExternalProtocolAndPreference(t *testing.T) { - candidate, err := NewCandidateRelay(&CandidateRelayConfig{ - Network: NetworkTypeUDP4.String(), - Address: "192.0.2.10", - Port: 5000, - Component: ComponentRTP, - RelayProtocol: "quic", - RelayLocalPreference: 37, - }) - require.NoError(t, err) - - require.Equal(t, "quic", candidate.RelayProtocol()) - require.Equal(t, uint16(37), candidate.LocalPreference()) - require.Equal(t, uint32(37*256+255), candidate.Priority()) -} - -func TestAgentGathersFromMultipleRelayProviders(t *testing.T) { - first := &relayProviderTestProvider{ - packetConn: &relayProviderTestPacketConn{addr: &net.UDPAddr{IP: net.IPv4(192, 0, 2, 1), Port: 6001}}, - relayProtocol: "websocket", - localPreference: 21, - address: "192.0.2.1", - port: 6001, - } - second := &relayProviderTestProvider{ - packetConn: &relayProviderTestPacketConn{addr: &net.UDPAddr{IP: net.IPv4(192, 0, 2, 2), Port: 6002}}, - relayProtocol: "quic", - localPreference: 42, - address: "192.0.2.2", - port: 6002, - } - - agent, err := NewAgentWithOptions( - WithCandidateTypes([]CandidateType{CandidateTypeRelay}), - WithNetworkTypes([]NetworkType{NetworkTypeUDP4}), - WithRelayCandidateProviders(first, second), - WithMulticastDNSMode(MulticastDNSModeDisabled), - ) - require.NoError(t, err) - defer func() { require.NoError(t, agent.Close()) }() - require.Equal(t, []CandidateType{CandidateTypeRelay}, agent.candidateTypes) - require.Len(t, agent.relayCandidateProviders, 2) - - gathered := make(chan struct{}) - require.NoError(t, agent.OnCandidate(func(candidate Candidate) { - if candidate == nil { - close(gathered) - } - })) - require.NoError(t, agent.GatherCandidates()) - require.Eventually(t, func() bool { - select { - case <-gathered: - return true - default: - return false - } - }, time.Second, time.Millisecond) - - require.Equal(t, int32(1), first.called.Load()) - require.Equal(t, int32(1), second.called.Load()) - - candidates, err := agent.GetLocalCandidates() - require.NoError(t, err) - require.Len(t, candidates, 2) - - byProtocol := make(map[string]*CandidateRelay, len(candidates)) - for _, candidate := range candidates { - relay, ok := candidate.(*CandidateRelay) - require.True(t, ok) - byProtocol[relay.RelayProtocol()] = relay - } - - require.Equal(t, uint16(21), byProtocol["websocket"].LocalPreference()) - require.Equal(t, uint16(42), byProtocol["quic"].LocalPreference()) -} - -func TestAgentClosesInvalidExternalRelayCandidate(t *testing.T) { - packetConn := &relayProviderTestPacketConn{ - addr: &net.UDPAddr{IP: net.IPv4(192, 0, 2, 3), Port: 6003}, - } - provider := &relayProviderTestProvider{ - packetConn: packetConn, - relayProtocol: "custom", - address: "not-an-ip-address", - port: 6003, - } - - agent, err := NewAgentWithOptions( - WithCandidateTypes([]CandidateType{CandidateTypeRelay}), - WithNetworkTypes([]NetworkType{NetworkTypeUDP4}), - WithRelayCandidateProvider(provider), - WithMulticastDNSMode(MulticastDNSModeDisabled), - ) - require.NoError(t, err) - defer func() { require.NoError(t, agent.Close()) }() - - gathered := make(chan struct{}) - require.NoError(t, agent.OnCandidate(func(candidate Candidate) { - if candidate == nil { - close(gathered) - } - })) - require.NoError(t, agent.GatherCandidates()) - require.Eventually(t, func() bool { - select { - case <-gathered: - return true - default: - return false - } - }, time.Second, time.Millisecond) - - require.Equal(t, int32(1), provider.called.Load()) - require.Equal(t, int32(1), packetConn.closeCount.Load()) - candidates, err := agent.GetLocalCandidates() - require.NoError(t, err) - require.Empty(t, candidates) -} From d7617c60e1385837c2edc967a0f9cc73e2387295 Mon Sep 17 00:00:00 2001 From: Joshua Lee <24319042+bclswl0827@users.noreply.github.com> Date: Tue, 4 Aug 2026 16:33:20 +0000 Subject: [PATCH 3/4] Fix lint violations in local candidate support Use a static error for nil packet connections, check candidate type assertions, and replace forbidden test fatal calls with testify assertions. --- agent.go | 2 +- errors.go | 4 ++++ local_candidate_test.go | 8 +++++--- 3 files changed, 10 insertions(+), 4 deletions(-) diff --git a/agent.go b/agent.go index cf98bd6a..c607a23c 100644 --- a/agent.go +++ b/agent.go @@ -1010,7 +1010,7 @@ func (a *Agent) AddLocalCandidate(cand Candidate, candidateConn net.PacketConn) return nil } if candidateConn == nil { - return fmt.Errorf("candidate packet connection is nil") + return ErrCandidatePacketConnNil } return a.addCandidate(a.loop, cand, candidateConn) diff --git a/errors.go b/errors.go index 2a051267..e68289dc 100644 --- a/errors.go +++ b/errors.go @@ -64,6 +64,10 @@ var ( // ErrNoOnCandidateHandler indicates agent was started without OnCandidate. ErrNoOnCandidateHandler = errors.New("no OnCandidate provided") + // ErrCandidatePacketConnNil indicates a local candidate was added without + // a packet connection. + ErrCandidatePacketConnNil = errors.New("candidate packet connection is nil") + // ErrMultipleGatherAttempted indicates GatherCandidates has been called multiple times. ErrMultipleGatherAttempted = errors.New("attempting to gather candidates during gathering state") diff --git a/local_candidate_test.go b/local_candidate_test.go index 3507a06e..b704db20 100644 --- a/local_candidate_test.go +++ b/local_candidate_test.go @@ -72,20 +72,22 @@ func TestAddLocalCandidateRegistersExternalRelay(t *testing.T) { localCandidates, err := agent.GetLocalCandidates() require.NoError(t, err) require.Len(t, localCandidates, 1) - require.Equal(t, "custom", localCandidates[0].(*CandidateRelay).RelayProtocol()) + relayCandidate, ok := localCandidates[0].(*CandidateRelay) + require.True(t, ok) + require.Equal(t, "custom", relayCandidate.RelayProtocol()) select { case got := <-candidates: require.Equal(t, candidate, got) case <-time.After(time.Second): - t.Fatal("timed out waiting for local candidate callback") + require.FailNow(t, "timed out waiting for local candidate callback") } require.NoError(t, agent.GatherCandidates()) select { case <-gatheringComplete: case <-time.After(time.Second): - t.Fatal("timed out waiting for gathering completion") + require.FailNow(t, "timed out waiting for gathering completion") } } From 9586aeab89fe86efeb05f6dd6b9f72aa70e354ae Mon Sep 17 00:00:00 2001 From: Joshua Lee <24319042+bclswl0827@users.noreply.github.com> Date: Sat, 8 Aug 2026 07:54:20 +0000 Subject: [PATCH 4/4] Handle duplicate local candidates safely Return ErrDuplicateCandidate when AddLocalCandidate receives a candidate that is already registered. Preserve ownership of the caller-provided candidate and packet connection instead of closing them. Keep the existing cleanup behavior for candidates gathered internally. Move the local candidate tests into agent_test.go and cover both duplicate-handling paths. --- agent.go | 20 ++++- agent_test.go | 165 +++++++++++++++++++++++++++++++++++++++- errors.go | 3 + gather.go | 12 +-- local_candidate_test.go | 137 --------------------------------- 5 files changed, 189 insertions(+), 148 deletions(-) delete mode 100644 local_candidate_test.go diff --git a/agent.go b/agent.go index 64d5a824..8fc495b0 100644 --- a/agent.go +++ b/agent.go @@ -1037,7 +1037,7 @@ func (a *Agent) AddLocalCandidate(cand Candidate, candidateConn net.PacketConn) return ErrCandidatePacketConnNil } - return a.addCandidate(a.loop, cand, candidateConn) + return a.addCandidate(a.loop, cand, candidateConn, true) } func (a *Agent) resolveAndAddMulticastCandidate(cand *CandidateHost) { @@ -1364,15 +1364,21 @@ func (a *Agent) shouldAcceptRemoteCandidate(cand Candidate) bool { return true } -func (a *Agent) addCandidate(ctx context.Context, cand Candidate, candidateConn net.PacketConn) error { +func (a *Agent) addCandidate(ctx context.Context, cand Candidate, candidateConn net.PacketConn, errorOnDuplicate bool) error { if err := ctx.Err(); err != nil { return err } - return a.loop.Run(ctx, func(context.Context) { + duplicate := false + err := a.loop.Run(ctx, func(context.Context) { set := a.localCandidates[cand.NetworkType()] for _, candidate := range set { if candidate.Equal(cand) { + if errorOnDuplicate { + duplicate = true + return + } + a.log.Debugf("Ignore duplicate candidate: %s", cand) if err := cand.close(); err != nil { a.log.Warnf("Failed to close duplicate candidate: %v", err) @@ -1403,6 +1409,14 @@ func (a *Agent) addCandidate(ctx context.Context, cand Candidate, candidateConn a.candidateNotifier.EnqueueCandidate(cand) } }) + if err != nil { + return err + } + if duplicate { + return ErrDuplicateCandidate + } + + return nil } func (a *Agent) setCandidateExtensions(cand Candidate) { diff --git a/agent_test.go b/agent_test.go index 803711c4..560541b4 100644 --- a/agent_test.go +++ b/agent_test.go @@ -51,6 +51,30 @@ type blockingWritePacketConn struct { closeOnce sync.Once } +type localCandidatePacketConn struct { + addr net.Addr + closeCount atomic.Int32 +} + +func (c *localCandidatePacketConn) ReadFrom([]byte) (int, net.Addr, error) { + return 0, c.addr, io.EOF +} + +func (c *localCandidatePacketConn) WriteTo(payload []byte, _ net.Addr) (int, error) { + return len(payload), nil +} + +func (c *localCandidatePacketConn) Close() error { + c.closeCount.Add(1) + + return nil +} + +func (c *localCandidatePacketConn) LocalAddr() net.Addr { return c.addr } +func (c *localCandidatePacketConn) SetDeadline(time.Time) error { return nil } +func (c *localCandidatePacketConn) SetReadDeadline(time.Time) error { return nil } +func (c *localCandidatePacketConn) SetWriteDeadline(time.Time) error { return nil } + func newBlockingWritePacketConn() *blockingWritePacketConn { return &blockingWritePacketConn{ writeStarted: make(chan struct{}), @@ -2711,6 +2735,143 @@ func TestAddRemoteCandidateHonorsRemoteIPFilter(t *testing.T) { }, time.Second, 10*time.Millisecond) } +func TestAddLocalCandidateRegistersExternalRelay(t *testing.T) { + agent, err := NewAgentWithOptions( + WithCandidateTypes([]CandidateType{}), + WithMulticastDNSMode(MulticastDNSModeDisabled), + ) + require.NoError(t, err) + defer func() { require.NoError(t, agent.Close()) }() + + gatheringComplete := make(chan struct{}) + candidates := make(chan Candidate, 1) + require.NoError(t, agent.OnCandidate(func(candidate Candidate) { + if candidate == nil { + close(gatheringComplete) + + return + } + candidates <- candidate + })) + + candidate, err := NewCandidateRelay(&CandidateRelayConfig{ + Network: NetworkTypeUDP4.String(), + Address: "192.0.2.10", + Port: 5000, + Component: ComponentRTP, + RelayProtocol: "custom", + }) + require.NoError(t, err) + packetConn := &localCandidatePacketConn{ + addr: &net.UDPAddr{IP: net.IPv4(192, 0, 2, 10), Port: 5000}, + } + + require.NoError(t, agent.AddLocalCandidate(candidate, packetConn)) + localCandidates, err := agent.GetLocalCandidates() + require.NoError(t, err) + require.Len(t, localCandidates, 1) + relayCandidate, ok := localCandidates[0].(*CandidateRelay) + require.True(t, ok) + require.Equal(t, "custom", relayCandidate.RelayProtocol()) + + select { + case got := <-candidates: + require.Equal(t, candidate, got) + case <-time.After(time.Second): + require.FailNow(t, "timed out waiting for local candidate callback") + } + + require.NoError(t, agent.GatherCandidates()) + select { + case <-gatheringComplete: + case <-time.After(time.Second): + require.FailNow(t, "timed out waiting for gathering completion") + } +} + +func TestAddLocalCandidateRejectsNilPacketConn(t *testing.T) { + agent, err := NewAgentWithOptions(WithMulticastDNSMode(MulticastDNSModeDisabled)) + require.NoError(t, err) + defer func() { require.NoError(t, agent.Close()) }() + + candidate, err := NewCandidateRelay(&CandidateRelayConfig{ + Network: NetworkTypeUDP4.String(), + Address: "192.0.2.11", + Port: 5001, + Component: ComponentRTP, + }) + require.NoError(t, err) + require.ErrorIs(t, agent.AddLocalCandidate(candidate, nil), ErrCandidatePacketConnNil) +} + +func TestAddLocalCandidateRejectsDuplicateWithoutClosing(t *testing.T) { + agent, err := NewAgentWithOptions(WithMulticastDNSMode(MulticastDNSModeDisabled)) + require.NoError(t, err) + defer func() { require.NoError(t, agent.Close()) }() + + require.NoError(t, agent.OnCandidate(func(Candidate) {})) + var candidateCloseCount atomic.Int32 + candidate, err := NewCandidateRelay(&CandidateRelayConfig{ + Network: NetworkTypeUDP4.String(), + Address: "192.0.2.12", + Port: 5002, + Component: ComponentRTP, + OnClose: func() error { + candidateCloseCount.Add(1) + + return nil + }, + }) + require.NoError(t, err) + packetConn := &localCandidatePacketConn{ + addr: &net.UDPAddr{IP: net.IPv4(192, 0, 2, 12), Port: 5002}, + } + + require.NoError(t, agent.AddLocalCandidate(candidate, packetConn)) + require.ErrorIs(t, agent.AddLocalCandidate(candidate, packetConn), ErrDuplicateCandidate) + require.Zero(t, candidateCloseCount.Load()) + require.Zero(t, packetConn.closeCount.Load()) + + localCandidates, err := agent.GetLocalCandidates() + require.NoError(t, err) + require.Equal(t, []Candidate{candidate}, localCandidates) +} + +func TestAddCandidateClosesDuplicate(t *testing.T) { + agent, err := NewAgentWithOptions(WithMulticastDNSMode(MulticastDNSModeDisabled)) + require.NoError(t, err) + defer func() { require.NoError(t, agent.Close()) }() + + require.NoError(t, agent.OnCandidate(func(Candidate) {})) + config := CandidateRelayConfig{ + Network: NetworkTypeUDP4.String(), + Address: "192.0.2.13", + Port: 5003, + Component: ComponentRTP, + } + first, err := NewCandidateRelay(&config) + require.NoError(t, err) + var duplicateCloseCount atomic.Int32 + config.OnClose = func() error { + duplicateCloseCount.Add(1) + + return nil + } + duplicate, err := NewCandidateRelay(&config) + require.NoError(t, err) + firstConn := &localCandidatePacketConn{ + addr: &net.UDPAddr{IP: net.IPv4(192, 0, 2, 13), Port: 5003}, + } + duplicateConn := &localCandidatePacketConn{ + addr: &net.UDPAddr{IP: net.IPv4(192, 0, 2, 13), Port: 5003}, + } + + require.NoError(t, agent.addCandidate(context.Background(), first, firstConn, false)) + require.NoError(t, agent.addCandidate(context.Background(), duplicate, duplicateConn, false)) + require.Equal(t, int32(1), duplicateCloseCount.Load()) + require.Equal(t, int32(1), duplicateConn.closeCount.Load()) +} + func TestGetLocalCandidates(t *testing.T) { var config AgentConfig @@ -2736,7 +2897,7 @@ func TestGetLocalCandidates(t *testing.T) { expectedCandidates = append(expectedCandidates, cand) - err = agent.addCandidate(context.Background(), cand, dummyConn) + err = agent.addCandidate(context.Background(), cand, dummyConn, false) require.NoError(t, err) } @@ -3376,7 +3537,7 @@ func TestSetCandidatesUfrag(t *testing.T) { cand, errCand := NewCandidateHost(&cfg) require.NoError(t, errCand) - err = agent.addCandidate(context.Background(), cand, dummyConn) + err = agent.addCandidate(context.Background(), cand, dummyConn, false) require.NoError(t, err) } diff --git a/errors.go b/errors.go index e68289dc..9fd874f6 100644 --- a/errors.go +++ b/errors.go @@ -68,6 +68,9 @@ var ( // a packet connection. ErrCandidatePacketConnNil = errors.New("candidate packet connection is nil") + // ErrDuplicateCandidate indicates a local candidate has already been added. + ErrDuplicateCandidate = errors.New("candidate already added") + // ErrMultipleGatherAttempted indicates GatherCandidates has been called multiple times. ErrMultipleGatherAttempted = errors.New("attempting to gather candidates during gathering state") diff --git a/gather.go b/gather.go index b357e066..745881fd 100644 --- a/gather.go +++ b/gather.go @@ -484,7 +484,7 @@ func (a *Agent) gatherCandidatesLocal(ctx context.Context, networkTypes []Networ continue } - if err := a.addCandidate(ctx, candidateHost, connAndPort.conn); err != nil { + if err := a.addCandidate(ctx, candidateHost, connAndPort.conn, false); err != nil { if closeErr := candidateHost.close(); closeErr != nil { a.log.Warnf("Failed to close candidate: %v", closeErr) } @@ -593,7 +593,7 @@ func (a *Agent) gatherCandidatesLocalUDPMux(ctx context.Context) error { //nolin continue } - if err := a.addCandidate(ctx, c, conn); err != nil { + if err := a.addCandidate(ctx, c, conn, false); err != nil { if closeErr := c.close(); closeErr != nil { a.log.Warnf("Failed to close candidate: %v", closeErr) } @@ -708,7 +708,7 @@ func (a *Agent) gatherCandidatesSrflxMapped(ctx context.Context, networkTypes [] continue } - if err := a.addCandidate(ctx, c, currentConn); err != nil { + if err := a.addCandidate(ctx, c, currentConn, false); err != nil { if closeErr := c.close(); closeErr != nil { a.log.Warnf("Failed to close candidate: %v", closeErr) } @@ -797,7 +797,7 @@ func (a *Agent) gatherCandidatesSrflxUDPMux(ctx context.Context, urls []*stun.UR return } - if err := a.addCandidate(ctx, c, conn); err != nil { + if err := a.addCandidate(ctx, c, conn, false); err != nil { if closeErr := c.close(); closeErr != nil { a.log.Warnf("Failed to close candidate: %v", closeErr) } @@ -921,7 +921,7 @@ func (a *Agent) gatherCandidatesSrflx(ctx context.Context, urls []*stun.URI, net return } - if err := a.addCandidate(ctx, c, conn); err != nil { + if err := a.addCandidate(ctx, c, conn, false); err != nil { if closeErr := c.close(); closeErr != nil { a.log.Warnf("Failed to close candidate: %v", closeErr) } @@ -1362,7 +1362,7 @@ func (a *Agent) createRelayCandidate(ctx context.Context, ep relayEndpoint, ip n return err } - if err := a.addCandidate(ctx, candidate, ep.conn); err != nil { + if err := a.addCandidate(ctx, candidate, ep.conn, false); err != nil { if closeErr := candidate.close(); closeErr != nil { a.log.Warnf("Failed to close candidate: %v", closeErr) } diff --git a/local_candidate_test.go b/local_candidate_test.go deleted file mode 100644 index b704db20..00000000 --- a/local_candidate_test.go +++ /dev/null @@ -1,137 +0,0 @@ -// SPDX-FileCopyrightText: 2026 The Pion community -// SPDX-License-Identifier: MIT - -package ice - -import ( - "io" - "net" - "sync/atomic" - "testing" - "time" - - "github.com/stretchr/testify/require" -) - -type localCandidatePacketConn struct { - addr net.Addr - closeCount atomic.Int32 -} - -func (c *localCandidatePacketConn) ReadFrom([]byte) (int, net.Addr, error) { - return 0, c.addr, io.EOF -} - -func (c *localCandidatePacketConn) WriteTo(payload []byte, _ net.Addr) (int, error) { - return len(payload), nil -} - -func (c *localCandidatePacketConn) Close() error { - c.closeCount.Add(1) - - return nil -} - -func (c *localCandidatePacketConn) LocalAddr() net.Addr { return c.addr } -func (c *localCandidatePacketConn) SetDeadline(time.Time) error { return nil } -func (c *localCandidatePacketConn) SetReadDeadline(time.Time) error { return nil } -func (c *localCandidatePacketConn) SetWriteDeadline(time.Time) error { return nil } - -func TestAddLocalCandidateRegistersExternalRelay(t *testing.T) { - agent, err := NewAgentWithOptions( - WithCandidateTypes([]CandidateType{}), - WithMulticastDNSMode(MulticastDNSModeDisabled), - ) - require.NoError(t, err) - defer func() { require.NoError(t, agent.Close()) }() - - gatheringComplete := make(chan struct{}) - candidates := make(chan Candidate, 1) - require.NoError(t, agent.OnCandidate(func(candidate Candidate) { - if candidate == nil { - close(gatheringComplete) - - return - } - candidates <- candidate - })) - - candidate, err := NewCandidateRelay(&CandidateRelayConfig{ - Network: NetworkTypeUDP4.String(), - Address: "192.0.2.10", - Port: 5000, - Component: ComponentRTP, - RelayProtocol: "custom", - }) - require.NoError(t, err) - packetConn := &localCandidatePacketConn{ - addr: &net.UDPAddr{IP: net.IPv4(192, 0, 2, 10), Port: 5000}, - } - - require.NoError(t, agent.AddLocalCandidate(candidate, packetConn)) - localCandidates, err := agent.GetLocalCandidates() - require.NoError(t, err) - require.Len(t, localCandidates, 1) - relayCandidate, ok := localCandidates[0].(*CandidateRelay) - require.True(t, ok) - require.Equal(t, "custom", relayCandidate.RelayProtocol()) - - select { - case got := <-candidates: - require.Equal(t, candidate, got) - case <-time.After(time.Second): - require.FailNow(t, "timed out waiting for local candidate callback") - } - - require.NoError(t, agent.GatherCandidates()) - select { - case <-gatheringComplete: - case <-time.After(time.Second): - require.FailNow(t, "timed out waiting for gathering completion") - } -} - -func TestAddLocalCandidateRejectsNilPacketConn(t *testing.T) { - agent, err := NewAgentWithOptions(WithMulticastDNSMode(MulticastDNSModeDisabled)) - require.NoError(t, err) - defer func() { require.NoError(t, agent.Close()) }() - - candidate, err := NewCandidateRelay(&CandidateRelayConfig{ - Network: NetworkTypeUDP4.String(), - Address: "192.0.2.11", - Port: 5001, - Component: ComponentRTP, - }) - require.NoError(t, err) - require.Error(t, agent.AddLocalCandidate(candidate, nil)) -} - -func TestAddLocalCandidateClosesDuplicateConnection(t *testing.T) { - agent, err := NewAgentWithOptions(WithMulticastDNSMode(MulticastDNSModeDisabled)) - require.NoError(t, err) - defer func() { require.NoError(t, agent.Close()) }() - - require.NoError(t, agent.OnCandidate(func(Candidate) {})) - config := CandidateRelayConfig{ - Network: NetworkTypeUDP4.String(), - Address: "192.0.2.12", - Port: 5002, - Component: ComponentRTP, - } - first, err := NewCandidateRelay(&config) - require.NoError(t, err) - second, err := NewCandidateRelay(&config) - require.NoError(t, err) - firstConn := &localCandidatePacketConn{addr: &net.UDPAddr{IP: net.IPv4(192, 0, 2, 12), Port: 5002}} - secondConn := &localCandidatePacketConn{addr: &net.UDPAddr{IP: net.IPv4(192, 0, 2, 12), Port: 5002}} - - require.NoError(t, agent.AddLocalCandidate(first, firstConn)) - require.NoError(t, agent.AddLocalCandidate(second, secondConn)) - require.Eventually(t, func() bool { - return secondConn.closeCount.Load() == 1 - }, time.Second, time.Millisecond) - - localCandidates, err := agent.GetLocalCandidates() - require.NoError(t, err) - require.Len(t, localCandidates, 1) -}