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

@@ -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--
 		}

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)

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)
 			}

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)
+}

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

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()

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
 			}