From 93f5b3eb615c792b8749b0532ff9855d9fd88418 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Sat, 26 Sep 2026 01:33:38 +0530 Subject: [PATCH] Add a local participant method to get recently disconnected (#4911) participants. While migrating, the old node will have the recently disconnected participants in the migrating out participant's cache and needs to sent out to the participant from the migrating in node via the new WebSocket. This method can be used to exchange that data between the two nodes. Keeping it as `recently` disconnected and leaving the semantics upto how recent for implementation. --- pkg/rtc/participant_internal_test.go | 5 ++ pkg/rtc/participant_signal.go | 25 +++++- pkg/rtc/room.go | 2 +- pkg/rtc/types/interfaces.go | 3 +- .../typesfakes/fake_local_participant.go | 86 +++++++++++++++++-- 5 files changed, 107 insertions(+), 14 deletions(-) diff --git a/pkg/rtc/participant_internal_test.go b/pkg/rtc/participant_internal_test.go index da41e7ff0..7492beaf7 100644 --- a/pkg/rtc/participant_internal_test.go +++ b/pkg/rtc/participant_internal_test.go @@ -917,6 +917,7 @@ func TestMigratingInParticipantWaitsForReconnectResponse(t *testing.T) { require.NoError(t, p.HandleReconnectAndSendResponse( livekit.ReconnectReason_RR_UNKNOWN, &livekit.ReconnectResponse{LastMessageSeq: 21}, + nil, )) require.True(t, p.IsReady()) @@ -942,6 +943,7 @@ func TestMigratingInParticipantWaitsForReconnectResponse(t *testing.T) { require.NoError(t, p.HandleReconnectAndSendResponse( livekit.ReconnectReason_RR_UNKNOWN, &livekit.ReconnectResponse{}, + nil, )) require.Equal(t, 2, sink.WriteMessageCallCount()) @@ -964,6 +966,7 @@ func TestMigratingInParticipantWaitsForReconnectResponse(t *testing.T) { require.NoError(t, p.HandleReconnectAndSendResponse( livekit.ReconnectReason_RR_UNKNOWN, &livekit.ReconnectResponse{}, + nil, )) require.True(t, p.IsReady()) require.Zero(t, sink.WriteMessageCallCount()) @@ -1004,6 +1007,7 @@ func TestResumedParticipantWaitsForReconnectResponse(t *testing.T) { require.NoError(t, p.HandleReconnectAndSendResponse( livekit.ReconnectReason_RR_SIGNAL_DISCONNECTED, &livekit.ReconnectResponse{LastMessageSeq: 7}, + nil, )) require.Equal(t, 2, sink.WriteMessageCallCount()) @@ -1040,6 +1044,7 @@ func TestResumedParticipantWaitsForReconnectResponse(t *testing.T) { require.NoError(t, p.HandleReconnectAndSendResponse( livekit.ReconnectReason_RR_SIGNAL_DISCONNECTED, &livekit.ReconnectResponse{}, + nil, )) require.Zero(t, sink.WriteMessageCallCount()) diff --git a/pkg/rtc/participant_signal.go b/pkg/rtc/participant_signal.go index e32a739be..9202ebb91 100644 --- a/pkg/rtc/participant_signal.go +++ b/pkg/rtc/participant_signal.go @@ -175,7 +175,11 @@ func (p *ParticipantImpl) SendRoomMovedResponse(roomMovedResponse *livekit.RoomM return p.signaller.WriteMessage(p.signalling.SignalRoomMovedResponse(roomMovedResponse)) } -func (p *ParticipantImpl) HandleReconnectAndSendResponse(reconnectReason livekit.ReconnectReason, reconnectResponse *livekit.ReconnectResponse) error { +func (p *ParticipantImpl) HandleReconnectAndSendResponse( + reconnectReason livekit.ReconnectReason, + reconnectResponse *livekit.ReconnectResponse, + recentlyDisconnectedParticipants []*livekit.ParticipantInfo, // when migrating in, these are the participants disconnected received from migrate out node +) error { p.TransportManager.HandleClientReconnect(reconnectReason) if !p.params.ClientInfo.CanHandleReconnectResponse() { @@ -198,6 +202,11 @@ func (p *ParticipantImpl) HandleReconnectAndSendResponse(reconnectReason livekit } if p.params.ProtocolVersion.SupportsDisconnectedUpdate() { + if len(recentlyDisconnectedParticipants) != 0 { + if err := p.signaller.WriteMessage(p.signalling.SignalParticipantUpdate(recentlyDisconnectedParticipants)); err != nil { + return err + } + } return p.sendDisconnectUpdatesForReconnect() } @@ -243,11 +252,19 @@ func (p *ParticipantImpl) flushQueuedUpdates() { } func (p *ParticipantImpl) sendDisconnectUpdatesForReconnect() error { + recentlyDisconnectedParticipants := p.GetRecentlyDisconnectedParticipants() + if len(recentlyDisconnectedParticipants) != 0 { + return p.signaller.WriteMessage(p.signalling.SignalParticipantUpdate(recentlyDisconnectedParticipants)) + } + return nil +} + +func (p *ParticipantImpl) GetRecentlyDisconnectedParticipants() []*livekit.ParticipantInfo { // look back a little more than last signal receive time as WebSocket close // on client side might have flushed messages and client application may have // lost connectivity earlier. lastSignalAt := p.TransportManager.LastSeenSignalAt().Add(-3 * time.Second) - var disconnectedParticipants []*livekit.ParticipantInfo + var recentlyDisconnectedParticipants []*livekit.ParticipantInfo p.updateLock.Lock() keys := p.updateCache.Keys() for i := len(keys) - 1; i >= 0; i-- { @@ -257,7 +274,7 @@ func (p *ParticipantImpl) sendDisconnectUpdatesForReconnect() error { } if info.state == livekit.ParticipantInfo_DISCONNECTED { - disconnectedParticipants = append(disconnectedParticipants, &livekit.ParticipantInfo{ + recentlyDisconnectedParticipants = append(recentlyDisconnectedParticipants, &livekit.ParticipantInfo{ Sid: string(keys[i]), Identity: string(info.identity), Version: info.version, @@ -268,7 +285,7 @@ func (p *ParticipantImpl) sendDisconnectUpdatesForReconnect() error { } p.updateLock.Unlock() - return p.signaller.WriteMessage(p.signalling.SignalParticipantUpdate(disconnectedParticipants)) + return recentlyDisconnectedParticipants } func (p *ParticipantImpl) sendICECandidate(ic *webrtc.ICECandidate, target livekit.SignalTarget) error { diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 3ae984632..f2891e79a 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -586,7 +586,7 @@ func (r *Room) ResumeParticipant( ClientConfiguration: p.GetClientConfiguration(), ServerInfo: r.serverInfo, LastMessageSeq: p.GetLastReliableSequence(false), - }); err != nil { + }, nil); err != nil { return err } diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 87b89ffe1..a8e3331df 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -583,10 +583,11 @@ type LocalParticipant interface { SendConnectionQualityUpdate(update *livekit.ConnectionQualityUpdate) error SendSubscriptionPermissionUpdate(publisherID livekit.ParticipantID, trackID livekit.TrackID, allowed bool) error SendRefreshToken(token string) error - HandleReconnectAndSendResponse(reconnectReason livekit.ReconnectReason, reconnectResponse *livekit.ReconnectResponse) error + HandleReconnectAndSendResponse(reconnectReason livekit.ReconnectReason, reconnectResponse *livekit.ReconnectResponse, recentlyDisconnectedParticipants []*livekit.ParticipantInfo) error IssueFullReconnect(reason ParticipantCloseReason) SendRoomMovedResponse(moved *livekit.RoomMovedResponse) error SendDataTrackSubscriberHandles(handles map[uint32]*livekit.DataTrackSubscriberHandles_PublishedDataTrack) error + GetRecentlyDisconnectedParticipants() []*livekit.ParticipantInfo AddOnClose(key string, callback func(LocalParticipant)) OnClaimsChanged(callback func(LocalParticipant)) diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index b6f349e70..a417a5b31 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -527,6 +527,16 @@ type FakeLocalParticipant struct { result1 string result2 error } + GetRecentlyDisconnectedParticipantsStub func() []*livekit.ParticipantInfo + getRecentlyDisconnectedParticipantsMutex sync.RWMutex + getRecentlyDisconnectedParticipantsArgsForCall []struct { + } + getRecentlyDisconnectedParticipantsReturns struct { + result1 []*livekit.ParticipantInfo + } + getRecentlyDisconnectedParticipantsReturnsOnCall map[int]struct { + result1 []*livekit.ParticipantInfo + } GetReporterStub func() roomobs.ParticipantSessionReporter getReporterMutex sync.RWMutex getReporterArgsForCall []struct { @@ -692,11 +702,12 @@ type FakeLocalParticipant struct { arg1 *sfu.DownTrack arg2 *rtcp.ReceiverReport } - HandleReconnectAndSendResponseStub func(livekit.ReconnectReason, *livekit.ReconnectResponse) error + HandleReconnectAndSendResponseStub func(livekit.ReconnectReason, *livekit.ReconnectResponse, []*livekit.ParticipantInfo) error handleReconnectAndSendResponseMutex sync.RWMutex handleReconnectAndSendResponseArgsForCall []struct { arg1 livekit.ReconnectReason arg2 *livekit.ReconnectResponse + arg3 []*livekit.ParticipantInfo } handleReconnectAndSendResponseReturns struct { result1 error @@ -4264,6 +4275,59 @@ func (fake *FakeLocalParticipant) GetPublisherICESessionUfragReturnsOnCall(i int }{result1, result2} } +func (fake *FakeLocalParticipant) GetRecentlyDisconnectedParticipants() []*livekit.ParticipantInfo { + fake.getRecentlyDisconnectedParticipantsMutex.Lock() + ret, specificReturn := fake.getRecentlyDisconnectedParticipantsReturnsOnCall[len(fake.getRecentlyDisconnectedParticipantsArgsForCall)] + fake.getRecentlyDisconnectedParticipantsArgsForCall = append(fake.getRecentlyDisconnectedParticipantsArgsForCall, struct { + }{}) + stub := fake.GetRecentlyDisconnectedParticipantsStub + fakeReturns := fake.getRecentlyDisconnectedParticipantsReturns + fake.recordInvocation("GetRecentlyDisconnectedParticipants", []interface{}{}) + fake.getRecentlyDisconnectedParticipantsMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalParticipant) GetRecentlyDisconnectedParticipantsCallCount() int { + fake.getRecentlyDisconnectedParticipantsMutex.RLock() + defer fake.getRecentlyDisconnectedParticipantsMutex.RUnlock() + return len(fake.getRecentlyDisconnectedParticipantsArgsForCall) +} + +func (fake *FakeLocalParticipant) GetRecentlyDisconnectedParticipantsCalls(stub func() []*livekit.ParticipantInfo) { + fake.getRecentlyDisconnectedParticipantsMutex.Lock() + defer fake.getRecentlyDisconnectedParticipantsMutex.Unlock() + fake.GetRecentlyDisconnectedParticipantsStub = stub +} + +func (fake *FakeLocalParticipant) GetRecentlyDisconnectedParticipantsReturns(result1 []*livekit.ParticipantInfo) { + fake.getRecentlyDisconnectedParticipantsMutex.Lock() + defer fake.getRecentlyDisconnectedParticipantsMutex.Unlock() + fake.GetRecentlyDisconnectedParticipantsStub = nil + fake.getRecentlyDisconnectedParticipantsReturns = struct { + result1 []*livekit.ParticipantInfo + }{result1} +} + +func (fake *FakeLocalParticipant) GetRecentlyDisconnectedParticipantsReturnsOnCall(i int, result1 []*livekit.ParticipantInfo) { + fake.getRecentlyDisconnectedParticipantsMutex.Lock() + defer fake.getRecentlyDisconnectedParticipantsMutex.Unlock() + fake.GetRecentlyDisconnectedParticipantsStub = nil + if fake.getRecentlyDisconnectedParticipantsReturnsOnCall == nil { + fake.getRecentlyDisconnectedParticipantsReturnsOnCall = make(map[int]struct { + result1 []*livekit.ParticipantInfo + }) + } + fake.getRecentlyDisconnectedParticipantsReturnsOnCall[i] = struct { + result1 []*livekit.ParticipantInfo + }{result1} +} + func (fake *FakeLocalParticipant) GetReporter() roomobs.ParticipantSessionReporter { fake.getReporterMutex.Lock() ret, specificReturn := fake.getReporterReturnsOnCall[len(fake.getReporterArgsForCall)] @@ -5168,19 +5232,25 @@ func (fake *FakeLocalParticipant) HandleReceiverReportArgsForCall(i int) (*sfu.D return argsForCall.arg1, argsForCall.arg2 } -func (fake *FakeLocalParticipant) HandleReconnectAndSendResponse(arg1 livekit.ReconnectReason, arg2 *livekit.ReconnectResponse) error { +func (fake *FakeLocalParticipant) HandleReconnectAndSendResponse(arg1 livekit.ReconnectReason, arg2 *livekit.ReconnectResponse, arg3 []*livekit.ParticipantInfo) error { + var arg3Copy []*livekit.ParticipantInfo + if arg3 != nil { + arg3Copy = make([]*livekit.ParticipantInfo, len(arg3)) + copy(arg3Copy, arg3) + } fake.handleReconnectAndSendResponseMutex.Lock() ret, specificReturn := fake.handleReconnectAndSendResponseReturnsOnCall[len(fake.handleReconnectAndSendResponseArgsForCall)] fake.handleReconnectAndSendResponseArgsForCall = append(fake.handleReconnectAndSendResponseArgsForCall, struct { arg1 livekit.ReconnectReason arg2 *livekit.ReconnectResponse - }{arg1, arg2}) + arg3 []*livekit.ParticipantInfo + }{arg1, arg2, arg3Copy}) stub := fake.HandleReconnectAndSendResponseStub fakeReturns := fake.handleReconnectAndSendResponseReturns - fake.recordInvocation("HandleReconnectAndSendResponse", []interface{}{arg1, arg2}) + fake.recordInvocation("HandleReconnectAndSendResponse", []interface{}{arg1, arg2, arg3Copy}) fake.handleReconnectAndSendResponseMutex.Unlock() if stub != nil { - return stub(arg1, arg2) + return stub(arg1, arg2, arg3) } if specificReturn { return ret.result1 @@ -5194,17 +5264,17 @@ func (fake *FakeLocalParticipant) HandleReconnectAndSendResponseCallCount() int return len(fake.handleReconnectAndSendResponseArgsForCall) } -func (fake *FakeLocalParticipant) HandleReconnectAndSendResponseCalls(stub func(livekit.ReconnectReason, *livekit.ReconnectResponse) error) { +func (fake *FakeLocalParticipant) HandleReconnectAndSendResponseCalls(stub func(livekit.ReconnectReason, *livekit.ReconnectResponse, []*livekit.ParticipantInfo) error) { fake.handleReconnectAndSendResponseMutex.Lock() defer fake.handleReconnectAndSendResponseMutex.Unlock() fake.HandleReconnectAndSendResponseStub = stub } -func (fake *FakeLocalParticipant) HandleReconnectAndSendResponseArgsForCall(i int) (livekit.ReconnectReason, *livekit.ReconnectResponse) { +func (fake *FakeLocalParticipant) HandleReconnectAndSendResponseArgsForCall(i int) (livekit.ReconnectReason, *livekit.ReconnectResponse, []*livekit.ParticipantInfo) { fake.handleReconnectAndSendResponseMutex.RLock() defer fake.handleReconnectAndSendResponseMutex.RUnlock() argsForCall := fake.handleReconnectAndSendResponseArgsForCall[i] - return argsForCall.arg1, argsForCall.arg2 + return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3 } func (fake *FakeLocalParticipant) HandleReconnectAndSendResponseReturns(result1 error) {