Send disconnected participant update for reconnecting user (#1495)

* Send disconnected participant update for reconnecting user

* clean code

pkg/rtc/participant.go

@@ -50,6 +50,12 @@ type downTrackState struct {
 	downTrack   sfu.DownTrackState
 }
 
+type participantUpdateInfo struct {
+	version   uint32
+	state     livekit.ParticipantInfo_State
+	updatedAt time.Time
+}
+
 type ParticipantParams struct {
 	Identity                     livekit.ParticipantIdentity
 	Name                         livekit.ParticipantName
@@ -125,7 +131,7 @@ type ParticipantImpl struct {
 	queuedUpdates []*livekit.ParticipantInfo
 	// cache of recently sent updates, to ensuring ordering by version
 	// guarded by updateLock
-	updateCache *lru.Cache[livekit.ParticipantID, uint32]
+	updateCache *lru.Cache[livekit.ParticipantID, participantUpdateInfo]
 	updateLock  utils.Mutex
 
 	dataChannelStats *telemetry.BytesTrackStats
@@ -192,7 +198,7 @@ func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) {
 
 	var err error
 	// keep last participants and when updates were sent
-	if p.updateCache, err = lru.New[livekit.ParticipantID, uint32](128); err != nil {
+	if p.updateCache, err = lru.New[livekit.ParticipantID, participantUpdateInfo](128); err != nil {
 		return nil, err
 	}
 
@@ -905,7 +911,7 @@ func (p *ParticipantImpl) VerifySubscribeParticipantInfo(pID livekit.Participant
 		// we have not sent a JoinResponse yet. metadata would be covered in JoinResponse
 		return
 	}
-	if v, ok := p.updateCache.Get(pID); ok && v >= version {
+	if info, ok := p.updateCache.Get(pID); ok && info.version >= version {
 		return
 	}
 

pkg/rtc/participant_signal.go

@@ -2,6 +2,7 @@ package rtc
 
 import (
 	"fmt"
+	"time"
 
 	"github.com/pion/webrtc/v3"
 
@@ -33,7 +34,7 @@ func (p *ParticipantImpl) SendJoinResponse(joinResponse *livekit.JoinResponse) e
 	// keep track of participant updates and versions
 	p.updateLock.Lock()
 	for _, op := range joinResponse.OtherParticipants {
-		p.updateCache.Add(livekit.ParticipantID(op.Sid), op.Version)
+		p.updateCache.Add(livekit.ParticipantID(op.Sid), participantUpdateInfo{version: op.Version, state: op.State, updatedAt: time.Now()})
 	}
 	p.updateLock.Unlock()
 
@@ -84,13 +85,13 @@ func (p *ParticipantImpl) SendParticipantUpdate(participantsToUpdate []*livekit.
 		if lastVersion, ok := p.updateCache.Get(pID); ok {
 			// this is a message delivered out of order, a more recent version of the message had already been
 			// sent.
-			if pi.Version < lastVersion {
+			if pi.Version < lastVersion.version {
 				p.params.Logger.Debugw("skipping outdated participant update", "version", pi.Version, "lastVersion", lastVersion)
 				isValid = false
 			}
 		}
 		if isValid {
-			p.updateCache.Add(pID, pi.Version)
+			p.updateCache.Add(pID, participantUpdateInfo{version: pi.Version, state: pi.State, updatedAt: time.Now()})
 			validUpdates = append(validUpdates, pi)
 		}
 	}
@@ -182,10 +183,47 @@ func (p *ParticipantImpl) SendReconnectResponse(reconnectResponse *livekit.Recon
 	if !p.params.ClientInfo.CanHandleReconnectResponse() {
 		return nil
 	}
-	return p.writeMessage(&livekit.SignalResponse{
+	if err := p.writeMessage(&livekit.SignalResponse{
 		Message: &livekit.SignalResponse_Reconnect{
 			Reconnect: reconnectResponse,
 		},
+	}); err != nil {
+		return err
+	}
+
+	if p.params.ProtocolVersion.SupportHandlesDisconnectedUpdate() {
+		return p.sendDisconnectUpdatesForReconnect()
+	}
+
+	return nil
+}
+
+func (p *ParticipantImpl) sendDisconnectUpdatesForReconnect() error {
+	lastSignalAt := p.TransportManager.LastSeenSignalAt()
+	var disconnectedParticipants []*livekit.ParticipantInfo
+	p.updateLock.Lock()
+	keys := p.updateCache.Keys()
+	for i := len(keys) - 1; i >= 0; i-- {
+		if info, ok := p.updateCache.Get(keys[i]); ok {
+			if info.updatedAt.Before(lastSignalAt) {
+				break
+			} else if info.state == livekit.ParticipantInfo_DISCONNECTED {
+				disconnectedParticipants = append(disconnectedParticipants, &livekit.ParticipantInfo{
+					Sid:     string(keys[i]),
+					Version: info.version,
+					State:   livekit.ParticipantInfo_DISCONNECTED,
+				})
+			}
+		}
+	}
+	p.updateLock.Unlock()
+
+	return p.writeMessage(&livekit.SignalResponse{
+		Message: &livekit.SignalResponse_Update{
+			Update: &livekit.ParticipantUpdate{
+				Participants: disconnectedParticipants,
+			},
+		},
 	})
 }
 

pkg/rtc/transportmanager.go

@@ -733,6 +733,12 @@ func (t *TransportManager) SinceLastSignal() time.Duration {
 	return time.Since(t.lastSignalAt)
 }
 
+func (t *TransportManager) LastSeenSignalAt() time.Time {
+	t.lock.RLock()
+	defer t.lock.RUnlock()
+	return t.lastSignalAt
+}
+
 func (t *TransportManager) canUseICETCP() bool {
 	return t.params.TCPFallbackRTTThreshold == 0 || int(t.signalingRTT) < t.params.TCPFallbackRTTThreshold
 }

pkg/rtc/types/protocol_version.go

@@ -2,7 +2,7 @@ package types
 
 type ProtocolVersion int
 
-const CurrentProtocol = 8
+const CurrentProtocol = 9
 
 func (v ProtocolVersion) SupportsPackedStreamId() bool {
 	return v > 0
@@ -53,3 +53,7 @@ func (v ProtocolVersion) SupportsUnpublish() bool {
 func (v ProtocolVersion) SupportFastStart() bool {
 	return v > 7
 }
+
+func (v ProtocolVersion) SupportHandlesDisconnectedUpdate() bool {
+	return v > 8
+}