Skip to content

Commit 0ea9b1d

Browse files
AKomplished-bugAKomplished-bug
committed
fix(rtc): guard remote track event handlers against racing disconnect
track_published, track_subscribed, track_unsubscribed and track_subscription_failed looked up the remote participant (and publication) with an unguarded dict access, raising KeyError when the participant had already been removed by a racing participant_disconnected or unpublish. This mirrors the defensive lookup already applied to track_unpublished in livekit#743, skipping the emit and logging at debug when the entry is gone instead of letting _listen_task log a KeyError.
1 parent 0acc039 commit 0ea9b1d

1 file changed

Lines changed: 82 additions & 33 deletions

File tree

‎livekit-rtc/livekit/rtc/room.py‎

Lines changed: 82 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -817,10 +817,19 @@ def _on_room_event(self, event: proto_room.RoomEvent) -> None:
817817
lpublication._first_subscription.set_result(None)
818818
self.emit("local_track_subscribed", lpublication.track)
819819
elif which == "track_published":
820-
rparticipant = self._remote_participants[event.track_published.participant_identity]
821-
rpublication = RemoteTrackPublication(event.track_published.publication)
822-
rparticipant._track_publications[rpublication.sid] = rpublication
823-
self.emit("track_published", rpublication, rparticipant)
820+
# The participant may already have been removed by a racing
821+
# disconnect or a duplicate event, so the lookup is done defensively
822+
# and the emit is skipped when the participant is gone, mirroring the
823+
# track_unpublished handler, instead of raising a KeyError that
824+
# _listen_task logs as an error.
825+
identity = event.track_published.participant_identity
826+
rparticipant = self._remote_participants.get(identity)
827+
if rparticipant is not None:
828+
rpublication = RemoteTrackPublication(event.track_published.publication)
829+
rparticipant._track_publications[rpublication.sid] = rpublication
830+
self.emit("track_published", rpublication, rparticipant)
831+
else:
832+
logger.debug("track_published for untracked participant %s", identity)
824833
elif which == "track_unpublished":
825834
# The participant or publication may already have been removed by a
826835
# racing disconnect or a duplicate event, so both lookups are done
@@ -839,41 +848,81 @@ def _on_room_event(self, event: proto_room.RoomEvent) -> None:
839848
else:
840849
logger.debug("track_unpublished for untracked participant %s", identity)
841850
elif which == "track_subscribed":
851+
# The participant or publication may already have been removed by a
852+
# racing disconnect or unpublish, so both lookups are done
853+
# defensively and the emit is skipped when either entry is gone,
854+
# mirroring the track_unpublished handler, instead of raising a
855+
# KeyError that _listen_task logs as an error.
842856
owned_track_info = event.track_subscribed.track
843857
track_info = owned_track_info.info
844-
rparticipant = self._remote_participants[event.track_subscribed.participant_identity]
845-
rpublication = rparticipant.track_publications[track_info.sid]
846-
rpublication._subscribed = True
847-
if track_info.kind == TrackKind.KIND_VIDEO:
848-
remote_video_track = RemoteVideoTrack(owned_track_info)
849-
remote_video_track._set_room(self)
850-
rpublication._track = remote_video_track
851-
self.emit("track_subscribed", remote_video_track, rpublication, rparticipant)
852-
elif track_info.kind == TrackKind.KIND_AUDIO:
853-
remote_audio_track = RemoteAudioTrack(owned_track_info)
854-
remote_audio_track._set_room(self)
855-
rpublication._track = remote_audio_track
856-
self.emit("track_subscribed", remote_audio_track, rpublication, rparticipant)
858+
identity = event.track_subscribed.participant_identity
859+
rparticipant = self._remote_participants.get(identity)
860+
rpublication = (
861+
rparticipant.track_publications.get(track_info.sid)
862+
if rparticipant is not None
863+
else None
864+
)
865+
if rpublication is not None:
866+
rpublication._subscribed = True
867+
if track_info.kind == TrackKind.KIND_VIDEO:
868+
remote_video_track = RemoteVideoTrack(owned_track_info)
869+
remote_video_track._set_room(self)
870+
rpublication._track = remote_video_track
871+
self.emit("track_subscribed", remote_video_track, rpublication, rparticipant)
872+
elif track_info.kind == TrackKind.KIND_AUDIO:
873+
remote_audio_track = RemoteAudioTrack(owned_track_info)
874+
remote_audio_track._set_room(self)
875+
rpublication._track = remote_audio_track
876+
self.emit("track_subscribed", remote_audio_track, rpublication, rparticipant)
877+
else:
878+
logger.debug(
879+
"track_subscribed for untracked participant %s or publication %s",
880+
identity,
881+
track_info.sid,
882+
)
857883
elif which == "track_unsubscribed":
884+
# The participant or publication may already have been removed by a
885+
# racing disconnect or unpublish, so both lookups are done
886+
# defensively and the emit is skipped when either entry is gone,
887+
# mirroring the track_unpublished handler, instead of raising a
888+
# KeyError that _listen_task logs as an error.
858889
identity = event.track_unsubscribed.participant_identity
859-
rparticipant = self._remote_participants[identity]
860-
rpublication = rparticipant.track_publications[event.track_unsubscribed.track_sid]
861-
rtrack = rpublication.track
862-
if rtrack is not None:
863-
rtrack._set_room(None)
864-
rpublication._track = None
865-
rpublication._subscribed = False
866-
self.emit("track_unsubscribed", rtrack, rpublication, rparticipant)
890+
rparticipant = self._remote_participants.get(identity)
891+
rpublication = (
892+
rparticipant.track_publications.get(event.track_unsubscribed.track_sid)
893+
if rparticipant is not None
894+
else None
895+
)
896+
if rpublication is not None:
897+
rtrack = rpublication.track
898+
if rtrack is not None:
899+
rtrack._set_room(None)
900+
rpublication._track = None
901+
rpublication._subscribed = False
902+
self.emit("track_unsubscribed", rtrack, rpublication, rparticipant)
903+
else:
904+
logger.debug(
905+
"track_unsubscribed for untracked participant %s or publication %s",
906+
identity,
907+
event.track_unsubscribed.track_sid,
908+
)
867909
elif which == "track_subscription_failed":
910+
# The participant may already have been removed by a racing
911+
# disconnect, so the lookup is done defensively and the emit is
912+
# skipped when the participant is gone, instead of raising a KeyError
913+
# that _listen_task logs as an error.
868914
identity = event.track_subscription_failed.participant_identity
869-
rparticipant = self._remote_participants[identity]
870-
error = event.track_subscription_failed.error
871-
self.emit(
872-
"track_subscription_failed",
873-
rparticipant,
874-
event.track_subscription_failed.track_sid,
875-
error,
876-
)
915+
rparticipant = self._remote_participants.get(identity)
916+
if rparticipant is not None:
917+
error = event.track_subscription_failed.error
918+
self.emit(
919+
"track_subscription_failed",
920+
rparticipant,
921+
event.track_subscription_failed.track_sid,
922+
error,
923+
)
924+
else:
925+
logger.debug("track_subscription_failed for untracked participant %s", identity)
877926
elif which == "track_muted":
878927
identity = event.track_muted.participant_identity
879928
# TODO: pass participant identity

0 commit comments

Comments
 (0)