diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 5ec0a5db3..8008a21c3 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -36,6 +36,7 @@ const ( rttUpdateInterval = 5 * time.Second stateActiveCond = 3 // reliableDCOpen,lossyDCOpen,PeerConnectionStateConnected + initNetWorkCost = 100 ) type pendingTrackInfo struct { @@ -125,6 +126,7 @@ type ParticipantImpl struct { onClaimsChanged func(participant types.LocalParticipant) activeCounter atomic.Int32 + networkCost int32 } func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) { @@ -146,6 +148,7 @@ func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) { disallowedSubscriptions: make(map[livekit.TrackID]livekit.ParticipantID), connectedAt: time.Now(), rttUpdatedAt: time.Now(), + networkCost: initNetWorkCost, } p.version.Store(params.InitialVersion) p.migrateState.Store(types.MigrateStateInit) @@ -600,6 +603,9 @@ func (p *ParticipantImpl) SetMigrateState(s types.MigrateState) { p.pendingOffer = nil // in case of migration, subscriber data channel will not fire OnOpen callback p.activeCounter.CAS(0, 2) + + // use lower network cost to make sure migrated's node has higher priority candidate + p.networkCost-- } p.lock.Unlock() if s == types.MigrateStateComplete { diff --git a/pkg/rtc/participant_signal.go b/pkg/rtc/participant_signal.go index 5197758c3..383536f0c 100644 --- a/pkg/rtc/participant_signal.go +++ b/pkg/rtc/participant_signal.go @@ -178,6 +178,7 @@ func (p *ParticipantImpl) sendIceCandidate(c *webrtc.ICECandidate, target liveki // write candidate p.params.Logger.Debugw("sending ice candidates", "candidate", c.String(), "target", target) + ci.Candidate += fmt.Sprintf(" network-cost %d", p.networkCost) trickle := ToProtoTrickle(ci) trickle.Target = target _ = p.writeMessage(&livekit.SignalResponse{