From 11ae7fdbb63077ea5f9b1873bff06f7bb418ecf1 Mon Sep 17 00:00:00 2001 From: cnderrauber Date: Wed, 8 Mar 2023 15:16:40 +0800 Subject: [PATCH] Don't switch candidate if signal closed when pc failed (#1498) * Don't switch candidate if signal closed when pc failed * change comment * test case --- pkg/rtc/room.go | 61 +++++++++++++------ pkg/rtc/room_test.go | 14 ++--- pkg/rtc/transport.go | 10 ++- pkg/rtc/transportmanager.go | 17 ++++-- pkg/rtc/types/interfaces.go | 1 + .../typesfakes/fake_local_participant.go | 39 ++++++++++++ pkg/service/roommanager.go | 7 ++- 7 files changed, 111 insertions(+), 38 deletions(-) diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index e44f568f0..f319010e4 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -55,9 +55,10 @@ type Room struct { trackManager *RoomTrackManager // map of identity -> Participant - participants map[livekit.ParticipantIdentity]types.LocalParticipant - participantOpts map[livekit.ParticipantIdentity]*ParticipantOptions - bufferFactory *buffer.FactoryOfBufferFactory + participants map[livekit.ParticipantIdentity]types.LocalParticipant + participantOpts map[livekit.ParticipantIdentity]*ParticipantOptions + participantRequestSources map[livekit.ParticipantIdentity]routing.MessageSource + bufferFactory *buffer.FactoryOfBufferFactory // batch update participant info for non-publishers batchedUpdates map[livekit.ParticipantIdentity]*livekit.ParticipantInfo @@ -89,20 +90,21 @@ func NewRoom( egressLauncher EgressLauncher, ) *Room { r := &Room{ - protoRoom: proto.Clone(room).(*livekit.Room), - internal: internal, - Logger: LoggerWithRoom(logger.GetLogger(), livekit.RoomName(room.Name), livekit.RoomID(room.Sid)), - config: config, - audioConfig: audioConfig, - telemetry: telemetry, - egressLauncher: egressLauncher, - trackManager: NewRoomTrackManager(), - serverInfo: serverInfo, - participants: make(map[livekit.ParticipantIdentity]types.LocalParticipant), - participantOpts: make(map[livekit.ParticipantIdentity]*ParticipantOptions), - bufferFactory: buffer.NewFactoryOfBufferFactory(config.Receiver.PacketBufferSize), - batchedUpdates: make(map[livekit.ParticipantIdentity]*livekit.ParticipantInfo), - closed: make(chan struct{}), + protoRoom: proto.Clone(room).(*livekit.Room), + internal: internal, + Logger: LoggerWithRoom(logger.GetLogger(), livekit.RoomName(room.Name), livekit.RoomID(room.Sid)), + config: config, + audioConfig: audioConfig, + telemetry: telemetry, + egressLauncher: egressLauncher, + trackManager: NewRoomTrackManager(), + serverInfo: serverInfo, + participants: make(map[livekit.ParticipantIdentity]types.LocalParticipant), + participantOpts: make(map[livekit.ParticipantIdentity]*ParticipantOptions), + participantRequestSources: make(map[livekit.ParticipantIdentity]routing.MessageSource), + bufferFactory: buffer.NewFactoryOfBufferFactory(config.Receiver.PacketBufferSize), + batchedUpdates: make(map[livekit.ParticipantIdentity]*livekit.ParticipantInfo), + closed: make(chan struct{}), } if r.protoRoom.EmptyTimeout == 0 { r.protoRoom.EmptyTimeout = DefaultEmptyTimeout @@ -225,7 +227,7 @@ func (r *Room) Release() { r.holds.Dec() } -func (r *Room) Join(participant types.LocalParticipant, opts *ParticipantOptions, iceServers []*livekit.ICEServer) error { +func (r *Room) Join(participant types.LocalParticipant, requestSource routing.MessageSource, opts *ParticipantOptions, iceServers []*livekit.ICEServer) error { r.lock.Lock() defer r.lock.Unlock() @@ -339,6 +341,7 @@ func (r *Room) Join(participant types.LocalParticipant, opts *ParticipantOptions r.participants[participant.Identity()] = participant r.participantOpts[participant.Identity()] = opts + r.participantRequestSources[participant.Identity()] = requestSource if r.onParticipantChanged != nil { r.onParticipantChanged(participant) @@ -376,11 +379,30 @@ func (r *Room) Join(participant types.LocalParticipant, opts *ParticipantOptions return nil } -func (r *Room) ResumeParticipant(p types.LocalParticipant, responseSink routing.MessageSink, iceServers []*livekit.ICEServer, reason livekit.ReconnectReason) error { +func (r *Room) ReplaceParticipantRequestSource(identity livekit.ParticipantIdentity, reqSource routing.MessageSource) { + r.lock.Lock() + if rs, ok := r.participantRequestSources[identity]; ok { + rs.Close() + } + r.participantRequestSources[identity] = reqSource + + r.lock.Unlock() +} + +func (r *Room) GetParticipantRequestSource(identity livekit.ParticipantIdentity) routing.MessageSource { + r.lock.RLock() + defer r.lock.RUnlock() + return r.participantRequestSources[identity] +} + +func (r *Room) ResumeParticipant(p types.LocalParticipant, requestSource routing.MessageSource, responseSink routing.MessageSink, iceServers []*livekit.ICEServer, reason livekit.ReconnectReason) error { + r.ReplaceParticipantRequestSource(p.Identity(), requestSource) // close previous sink, and link to new one p.CloseSignalConnection() p.SetResponseSink(responseSink) + p.SetSignalSourceValid(true) + if err := p.SendReconnectResponse(&livekit.ReconnectResponse{ IceServers: iceServers, ClientConfiguration: p.GetClientConfiguration(), @@ -413,6 +435,7 @@ func (r *Room) RemoveParticipant(identity livekit.ParticipantIdentity, pID livek delete(r.participants, identity) delete(r.participantOpts, identity) + delete(r.participantRequestSources, identity) if !p.Hidden() { r.protoRoom.NumParticipants-- } diff --git a/pkg/rtc/room_test.go b/pkg/rtc/room_test.go index 392508a55..2452eb6cc 100644 --- a/pkg/rtc/room_test.go +++ b/pkg/rtc/room_test.go @@ -72,7 +72,7 @@ func TestRoomJoin(t *testing.T) { rm := newRoomWithParticipants(t, testRoomOpts{num: numParticipants}) pNew := newMockParticipant("new", types.CurrentProtocol, false, false) - _ = rm.Join(pNew, nil, iceServersForRoom) + _ = rm.Join(pNew, nil, nil, iceServersForRoom) // expect new participant to get a JoinReply res := pNew.SendJoinResponseArgsForCall(0) @@ -87,7 +87,7 @@ func TestRoomJoin(t *testing.T) { rm := newRoomWithParticipants(t, testRoomOpts{num: numExisting}) p := newMockParticipant("new", types.CurrentProtocol, false, false) - err := rm.Join(p, &ParticipantOptions{AutoSubscribe: true}, iceServersForRoom) + err := rm.Join(p, nil, &ParticipantOptions{AutoSubscribe: true}, iceServersForRoom) require.NoError(t, err) stateChangeCB := p.OnStateChangeArgsForCall(0) @@ -141,7 +141,7 @@ func TestRoomJoin(t *testing.T) { rm.protoRoom.MaxParticipants = 1 p := newMockParticipant("second", types.ProtocolVersion(0), false, false) - err := rm.Join(p, nil, iceServersForRoom) + err := rm.Join(p, nil, nil, iceServersForRoom) require.Equal(t, ErrMaxParticipantsExceeded, err) }) } @@ -355,7 +355,7 @@ func TestRoomClosure(t *testing.T) { require.Len(t, rm.GetParticipants(), 0) require.True(t, isClosed) - require.Equal(t, ErrRoomClosed, rm.Join(p, nil, iceServersForRoom)) + require.Equal(t, ErrRoomClosed, rm.Join(p, nil, nil, iceServersForRoom)) }) t.Run("room does not close before empty timeout", func(t *testing.T) { @@ -635,7 +635,7 @@ func TestHiddenParticipants(t *testing.T) { defer rm.Close() pNew := newMockParticipant("new", types.CurrentProtocol, false, false) - rm.Join(pNew, nil, iceServersForRoom) + rm.Join(pNew, nil, nil, iceServersForRoom) // expect new participant to get a JoinReply res := pNew.SendJoinResponseArgsForCall(0) @@ -650,7 +650,7 @@ func TestHiddenParticipants(t *testing.T) { rm := newRoomWithParticipants(t, testRoomOpts{num: 2}) hidden := newMockParticipant("hidden", types.CurrentProtocol, true, false) - err := rm.Join(hidden, &ParticipantOptions{AutoSubscribe: true}, iceServersForRoom) + err := rm.Join(hidden, nil, &ParticipantOptions{AutoSubscribe: true}, iceServersForRoom) require.NoError(t, err) stateChangeCB := hidden.OnStateChangeArgsForCall(0) @@ -705,7 +705,7 @@ func newRoomWithParticipants(t *testing.T, opts testRoomOpts) *Room { for i := 0; i < opts.num+opts.numHidden; i++ { identity := livekit.ParticipantIdentity(fmt.Sprintf("p%d", i)) participant := newMockParticipant(identity, opts.protocol, i >= opts.num, true) - err := rm.Join(participant, &ParticipantOptions{AutoSubscribe: true}, iceServersForRoom) + err := rm.Join(participant, nil, &ParticipantOptions{AutoSubscribe: true}, iceServersForRoom) require.NoError(t, err) participant.StateReturns(livekit.ParticipantInfo_ACTIVE) participant.IsReadyReturns(true) diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index d8cf2daee..cbc70ba2c 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -46,6 +46,7 @@ const ( minTcpICEConnectTimeout = 5 * time.Second maxTcpICEConnectTimeout = 12 * time.Second // js-sdk has a default 15s timeout for first connection, let server detect failure earlier before that + minConnectTimeoutAfterICE = 10 * time.Second maxConnectTimeoutAfterICE = 20 * time.Second // max duration for waiting pc to connect after ICE is connected shortConnectionThreshold = 90 * time.Second @@ -451,8 +452,7 @@ func (t *PCTransport) setICEConnectedAt(at time.Time) { // set failure timer for dtls handshake iceDuration := at.Sub(t.iceStartedAt) - // let ice has chance to become disconnected if user close/refresh client app before transport full connected. - connTimeoutAfterICE := iceDisconnectedTimeout + time.Second + iceDuration + connTimeoutAfterICE := minConnectTimeoutAfterICE if connTimeoutAfterICE < 3*iceDuration { connTimeoutAfterICE = 3 * iceDuration } @@ -462,10 +462,8 @@ func (t *PCTransport) setICEConnectedAt(at time.Time) { t.params.Logger.Debugw("setting connection timer after ice connected", "timeout", connTimeoutAfterICE, "iceDuration", iceDuration) t.connectAfterICETimer = time.AfterFunc(connTimeoutAfterICE, func() { state := t.pc.ConnectionState() - iceState := t.pc.ICEConnectionState() - // if ice connected, pc is still checking or connected but not fully established after timeout, then fire connection fail - if iceState == webrtc.ICEConnectionStateConnected && state != webrtc.PeerConnectionStateClosed && - state != webrtc.PeerConnectionStateFailed && !t.isFullyEstablished() { + // if pc is still checking or connected but not fully established after timeout, then fire connection fail + if state != webrtc.PeerConnectionStateClosed && state != webrtc.PeerConnectionStateFailed && !t.isFullyEstablished() { t.params.Logger.Infow("connect timeout after ICE connected", "timeout", connTimeoutAfterICE, "iceDuration", iceDuration) t.handleConnectionFailed(false) } diff --git a/pkg/rtc/transportmanager.go b/pkg/rtc/transportmanager.go index 569bd6b8c..d4d347ba3 100644 --- a/pkg/rtc/transportmanager.go +++ b/pkg/rtc/transportmanager.go @@ -62,6 +62,7 @@ type TransportManager struct { isTransportReconfigured bool lastFailure time.Time lastSignalAt time.Time + signalSourceValid atomic.Bool pendingOfferPublisher *webrtc.SessionDescription pendingDataChannelsPublisher []*livekit.DataChannelInfo @@ -177,6 +178,8 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro } } + t.signalSourceValid.Store(true) + return t, nil } @@ -536,11 +539,12 @@ func (t *TransportManager) handleConnectionFailed(isShortLived bool) { } lastSignalSince := time.Since(t.lastSignalAt) - if lastSignalSince > iceFailedTimeout { - // the failed might cause by network interrupt because we have not seen any signal in the time window too + signalValid := t.signalSourceValid.Load() + if lastSignalSince > iceFailedTimeout || !signalValid { + // the failed might cause by network interrupt because signal closed or we have not seen any signal in the time window, // so don't switch to next candidate type - t.params.Logger.Infow("ignoring prefer candidate check by ICE failure because no signal received in the ice failed window", - "lastSignalSince", lastSignalSince) + t.params.Logger.Infow("ignoring prefer candidate check by ICE failure because signal connection interrupted", + "lastSignalSince", lastSignalSince, "signalValid", signalValid) t.failureCount = 0 t.lastFailure = time.Time{} t.lock.Unlock() @@ -742,3 +746,8 @@ func (t *TransportManager) LastSeenSignalAt() time.Time { func (t *TransportManager) canUseICETCP() bool { return t.params.TCPFallbackRTTThreshold == 0 || int(t.signalingRTT) < t.params.TCPFallbackRTTThreshold } + +func (t *TransportManager) SetSignalSourceValid(valid bool) { + t.signalSourceValid.Store(valid) + t.params.Logger.Debugw("signal source valid", "valid", valid) +} diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 6c160b6a8..6bfc95edf 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -246,6 +246,7 @@ type LocalParticipant interface { SetResponseSink(sink routing.MessageSink) CloseSignalConnection() UpdateLastSeenSignal() + SetSignalSourceValid(valid bool) // permissions ClaimGrants() *auth.ClaimGrants diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index 1d2f4bb19..67ce745dd 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -653,6 +653,11 @@ type FakeLocalParticipant struct { setResponseSinkArgsForCall []struct { arg1 routing.MessageSink } + SetSignalSourceValidStub func(bool) + setSignalSourceValidMutex sync.RWMutex + setSignalSourceValidArgsForCall []struct { + arg1 bool + } SetTrackMutedStub func(livekit.TrackID, bool, bool) setTrackMutedMutex sync.RWMutex setTrackMutedArgsForCall []struct { @@ -4297,6 +4302,38 @@ func (fake *FakeLocalParticipant) SetResponseSinkArgsForCall(i int) routing.Mess return argsForCall.arg1 } +func (fake *FakeLocalParticipant) SetSignalSourceValid(arg1 bool) { + fake.setSignalSourceValidMutex.Lock() + fake.setSignalSourceValidArgsForCall = append(fake.setSignalSourceValidArgsForCall, struct { + arg1 bool + }{arg1}) + stub := fake.SetSignalSourceValidStub + fake.recordInvocation("SetSignalSourceValid", []interface{}{arg1}) + fake.setSignalSourceValidMutex.Unlock() + if stub != nil { + fake.SetSignalSourceValidStub(arg1) + } +} + +func (fake *FakeLocalParticipant) SetSignalSourceValidCallCount() int { + fake.setSignalSourceValidMutex.RLock() + defer fake.setSignalSourceValidMutex.RUnlock() + return len(fake.setSignalSourceValidArgsForCall) +} + +func (fake *FakeLocalParticipant) SetSignalSourceValidCalls(stub func(bool)) { + fake.setSignalSourceValidMutex.Lock() + defer fake.setSignalSourceValidMutex.Unlock() + fake.SetSignalSourceValidStub = stub +} + +func (fake *FakeLocalParticipant) SetSignalSourceValidArgsForCall(i int) bool { + fake.setSignalSourceValidMutex.RLock() + defer fake.setSignalSourceValidMutex.RUnlock() + argsForCall := fake.setSignalSourceValidArgsForCall[i] + return argsForCall.arg1 +} + func (fake *FakeLocalParticipant) SetTrackMuted(arg1 livekit.TrackID, arg2 bool, arg3 bool) { fake.setTrackMutedMutex.Lock() fake.setTrackMutedArgsForCall = append(fake.setTrackMutedArgsForCall, struct { @@ -5320,6 +5357,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} { defer fake.setPermissionMutex.RUnlock() fake.setResponseSinkMutex.RLock() defer fake.setResponseSinkMutex.RUnlock() + fake.setSignalSourceValidMutex.RLock() + defer fake.setSignalSourceValidMutex.RUnlock() fake.setTrackMutedMutex.RLock() defer fake.setTrackMutedMutex.RUnlock() fake.startMutex.RLock() diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index 3f7db4b1a..28254ea62 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -243,7 +243,7 @@ func (r *RoomManager) StartSession( if iceConfig == nil { iceConfig = &livekit.ICEConfig{} } - if err = room.ResumeParticipant(participant, responseSink, + if err = room.ResumeParticipant(participant, requestSource, responseSink, r.iceServersForRoom(protoRoom, iceConfig.PreferenceSubscriber == livekit.ICECandidateType_ICT_TLS), pi.ReconnectReason); err != nil { logger.Warnw("could not resume participant", err, "participant", pi.Identity) @@ -343,7 +343,7 @@ func (r *RoomManager) StartSession( opts := rtc.ParticipantOptions{ AutoSubscribe: pi.AutoSubscribe, } - if err = room.Join(participant, &opts, r.iceServersForRoom(protoRoom, iceConfig.PreferenceSubscriber == livekit.ICECandidateType_ICT_TLS)); err != nil { + if err = room.Join(participant, requestSource, &opts, r.iceServersForRoom(protoRoom, iceConfig.PreferenceSubscriber == livekit.ICECandidateType_ICT_TLS)); err != nil { pLogger.Errorw("could not join room", err) _ = participant.Close(true, types.ParticipantCloseReasonJoinFailed) return err @@ -508,6 +508,9 @@ func (r *RoomManager) rtcSessionWorker(room *rtc.Room, participant types.LocalPa // In single node mode, the request source is directly tied to the signal message channel // this means ICE restart isn't possible in single node mode if obj == nil { + if room.GetParticipantRequestSource(participant.Identity()) == requestSource { + participant.SetSignalSourceValid(false) + } return }