From 6bd5504bffe7956febc2cee66b76a731e3ae798c Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Tue, 6 Dec 2022 14:46:59 +0530 Subject: [PATCH] Add option to issue full reconnect on a publication error. (#1214) * Add option to issue full reconnect on a publication error. Leaving the publication error timeout at 30 seconds as there are some publications taking long. Also, there are cases where the peer connection fails after 30 seconds. The peer connection failure happens after publication error is detected. But, 30 seconds is a good amount of time for publication to establish. * prevent recursive lock --- pkg/config/config.go | 3 + pkg/config/config_test.go | 11 +-- pkg/rtc/participant.go | 73 ++++++++++++-------- pkg/rtc/supervisor/participant_supervisor.go | 24 +++++++ pkg/rtc/types/interfaces.go | 9 ++- pkg/service/roommanager.go | 6 ++ 6 files changed, 90 insertions(+), 36 deletions(-) diff --git a/pkg/config/config.go b/pkg/config/config.go index 80e4e508d..da7349155 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -85,6 +85,9 @@ type RTCConfig struct { // for testing, disable UDP ForceTCP bool `yaml:"force_tcp,omitempty"` + + // force a reconnect on a publication error + ReconnectOnPublicationError *bool `yaml:"reconnect_on_publication_error,omitempty"` } type TURNServer struct { diff --git a/pkg/config/config_test.go b/pkg/config/config_test.go index dd1114354..99e1ce9ce 100644 --- a/pkg/config/config_test.go +++ b/pkg/config/config_test.go @@ -42,10 +42,11 @@ func TestGeneratedFlags(t *testing.T) { app.Flags = append(app.Flags, generatedFlags...) set := flag.NewFlagSet("test", 0) - set.Bool("rtc.use_ice_lite", true, "") // bool - set.String("redis.address", "localhost:6379", "") // string - set.Uint("prometheus_port", 9999, "") // uint32 - set.Bool("rtc.allow_tcp_fallback", true, "") // pointer + set.Bool("rtc.use_ice_lite", true, "") // bool + set.String("redis.address", "localhost:6379", "") // string + set.Uint("prometheus_port", 9999, "") // uint32 + set.Bool("rtc.allow_tcp_fallback", true, "") // pointer + set.Bool("rtc.reconnect_on_publication_error", true, "") // pointer c := cli.NewContext(app, set, nil) conf, err := NewConfig("", true, c, nil) @@ -56,4 +57,6 @@ func TestGeneratedFlags(t *testing.T) { require.Equal(t, uint32(9999), conf.PrometheusPort) require.NotNil(t, conf.RTC.AllowTCPFallback) require.True(t, *conf.RTC.AllowTCPFallback) + require.NotNil(t, conf.RTC.ReconnectOnPublicationError) + require.True(t, *conf.RTC.ReconnectOnPublicationError) } diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 8756e959f..256514ca0 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -64,30 +64,31 @@ type SubscribeRequest struct { } type ParticipantParams struct { - Identity livekit.ParticipantIdentity - Name livekit.ParticipantName - SID livekit.ParticipantID - Config *WebRTCConfig - Sink routing.MessageSink - AudioConfig config.AudioConfig - VideoConfig config.VideoConfig - ProtocolVersion types.ProtocolVersion - Telemetry telemetry.TelemetryService - PLIThrottleConfig config.PLIThrottleConfig - CongestionControlConfig config.CongestionControlConfig - EnabledCodecs []*livekit.Codec - Logger logger.Logger - SimTracks map[uint32]SimulcastTrackInfo - Grants *auth.ClaimGrants - InitialVersion uint32 - ClientConf *livekit.ClientConfiguration - ClientInfo ClientInfo - Region string - Migration bool - AdaptiveStream bool - AllowTCPFallback bool - TURNSEnabled bool - GetParticipantInfo func(pID livekit.ParticipantID) *livekit.ParticipantInfo + Identity livekit.ParticipantIdentity + Name livekit.ParticipantName + SID livekit.ParticipantID + Config *WebRTCConfig + Sink routing.MessageSink + AudioConfig config.AudioConfig + VideoConfig config.VideoConfig + ProtocolVersion types.ProtocolVersion + Telemetry telemetry.TelemetryService + PLIThrottleConfig config.PLIThrottleConfig + CongestionControlConfig config.CongestionControlConfig + EnabledCodecs []*livekit.Codec + Logger logger.Logger + SimTracks map[uint32]SimulcastTrackInfo + Grants *auth.ClaimGrants + InitialVersion uint32 + ClientConf *livekit.ClientConfiguration + ClientInfo ClientInfo + Region string + Migration bool + AdaptiveStream bool + AllowTCPFallback bool + TURNSEnabled bool + GetParticipantInfo func(pID livekit.ParticipantID) *livekit.ParticipantInfo + ReconnectOnPublicationError bool } type ParticipantImpl struct { @@ -198,6 +199,8 @@ func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) { p.grants = params.Grants p.SetResponseSink(params.Sink) + p.supervisor.OnPublicationError(p.onPublicationError) + var err error // keep last participants and when updates were sent if p.updateCache, err = lru.New(128); err != nil { @@ -1484,7 +1487,7 @@ func (p *ParticipantImpl) addPendingTrackLocked(req *livekit.AddTrackRequest) *l } else { p.pendingTracks[req.Cid].trackInfos = append(p.pendingTracks[req.Cid].trackInfos, ti) } - p.params.Logger.Debugw("pending track queued", "trackID", ti.Sid, "track", ti.String(), "request", req.String()) + p.params.Logger.Infow("pending track queued", "trackID", ti.Sid, "track", ti.String(), "request", req.String()) return nil } @@ -1492,7 +1495,7 @@ func (p *ParticipantImpl) addPendingTrackLocked(req *livekit.AddTrackRequest) *l p.supervisor.SetPublicationMute(livekit.TrackID(ti.Sid), ti.Muted) p.pendingTracks[req.Cid] = &pendingTrackInfo{trackInfos: []*livekit.TrackInfo{ti}} - p.params.Logger.Debugw("pending track added", "trackID", ti.Sid, "track", ti.String(), "request", req.String()) + p.params.Logger.Infow("pending track added", "trackID", ti.Sid, "track", ti.String(), "request", req.String()) return ti } @@ -1947,13 +1950,12 @@ func (p *ParticipantImpl) GetCachedDownTrack(trackID livekit.TrackID) (*webrtc.R return nil, sfu.DownTrackState{} } -func (p *ParticipantImpl) onAnyTransportNegotiationFailed() { - p.params.Logger.Infow("negotiation failed, starting full reconnect") +func (p *ParticipantImpl) issueFullReconnect(reason types.ParticipantCloseReason) { _ = p.writeMessage(&livekit.SignalResponse{ Message: &livekit.SignalResponse_Leave{ Leave: &livekit.LeaveRequest{ CanReconnect: true, - Reason: types.ParticipantCloseReasonNegotiateFailed.ToDisconnectReason(), + Reason: reason.ToDisconnectReason(), }, }, }) @@ -1963,6 +1965,19 @@ func (p *ParticipantImpl) onAnyTransportNegotiationFailed() { p.supervisor.Stop() } +func (p *ParticipantImpl) onPublicationError(trackID livekit.TrackID) { + p.params.Logger.Infow("publication failed", "trackID", trackID) + if p.params.ReconnectOnPublicationError { + p.params.Logger.Infow("starting full reconnect") + p.issueFullReconnect(types.ParticipantCloseReasonPublicationError) + } +} + +func (p *ParticipantImpl) onAnyTransportNegotiationFailed() { + p.params.Logger.Infow("negotiation failed, starting full reconnect") + p.issueFullReconnect(types.ParticipantCloseReasonNegotiateFailed) +} + func (p *ParticipantImpl) EnqueueSubscribeTrack(trackID livekit.TrackID, isRelayed bool, f func(sub types.LocalParticipant) error) bool { // do not queue subscription is participant is already closed/disconnected if p.isClosed.Load() || p.State() == livekit.ParticipantInfo_DISCONNECTED { diff --git a/pkg/rtc/supervisor/participant_supervisor.go b/pkg/rtc/supervisor/participant_supervisor.go index d9e55c921..b8e847236 100644 --- a/pkg/rtc/supervisor/participant_supervisor.go +++ b/pkg/rtc/supervisor/participant_supervisor.go @@ -27,6 +27,8 @@ type ParticipantSupervisor struct { subscriptions map[livekit.TrackID]types.OperationMonitor isStopped atomic.Bool + + onPublicationError func(trackID livekit.TrackID) } func NewParticipantSupervisor(params ParticipantSupervisorParams) *ParticipantSupervisor { @@ -45,6 +47,20 @@ func (p *ParticipantSupervisor) Stop() { p.isStopped.Store(true) } +func (p *ParticipantSupervisor) OnPublicationError(f func(trackID livekit.TrackID)) { + p.lock.Lock() + defer p.lock.Unlock() + + p.onPublicationError = f +} + +func (p *ParticipantSupervisor) getOnPublicationError() func(trackID livekit.TrackID) { + p.lock.RLock() + defer p.lock.RUnlock() + + return p.onPublicationError +} + func (p *ParticipantSupervisor) SetPublisherPeerConnectionConnected(isConnected bool) { p.lock.Lock() p.isPublisherConnected = isConnected @@ -141,11 +157,13 @@ func (p *ParticipantSupervisor) checkState() { } func (p *ParticipantSupervisor) checkPublications() { + var erroredPublications []livekit.TrackID var removablePublications []livekit.TrackID p.lock.RLock() for trackID, pm := range p.publications { if err := pm.Check(); err != nil { p.params.Logger.Errorw("supervisor error on publication", err, "trackID", trackID) + erroredPublications = append(erroredPublications, trackID) } else { if pm.IsIdle() { removablePublications = append(removablePublications, trackID) @@ -159,6 +177,12 @@ func (p *ParticipantSupervisor) checkPublications() { delete(p.publications, trackID) } p.lock.Unlock() + + if onPublicationError := p.getOnPublicationError(); onPublicationError != nil { + for _, trackID := range erroredPublications { + onPublicationError(trackID) + } + } } func (p *ParticipantSupervisor) checkSubscriptions() { diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 21538a7e1..6efa176f2 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -84,6 +84,7 @@ const ( ParticipantCloseReasonNegotiateFailed ParticipantCloseReasonMigrationRequested ParticipantCloseReasonOvercommitted + ParticipantCloseReasonPublicationError ) func (p ParticipantCloseReason) String() string { @@ -122,10 +123,12 @@ func (p ParticipantCloseReason) String() string { return "SIMULATE_SERVER_LEAVE" case ParticipantCloseReasonNegotiateFailed: return "NEGOTIATE_FAILED" - case ParticipantCloseReasonOvercommitted: - return "OVERCOMMITTED" case ParticipantCloseReasonMigrationRequested: return "MIGRATION_REQUESTED" + case ParticipantCloseReasonOvercommitted: + return "OVERCOMMITTED" + case ParticipantCloseReasonPublicationError: + return "PUBLICATION_ERROR" default: return fmt.Sprintf("%d", int(p)) } @@ -156,7 +159,7 @@ func (p ParticipantCloseReason) ToDisconnectReason() livekit.DisconnectReason { return livekit.DisconnectReason_SERVER_SHUTDOWN case ParticipantCloseReasonOvercommitted: return livekit.DisconnectReason_SERVER_SHUTDOWN - case ParticipantCloseReasonNegotiateFailed: + case ParticipantCloseReasonNegotiateFailed, ParticipantCloseReasonPublicationError: return livekit.DisconnectReason_STATE_MISMATCH default: // the other types will map to unknown reason diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index 5ba4a1824..daa8cbfdd 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -280,6 +280,11 @@ func (r *RoomManager) StartSession( if r.config.RTC.AllowTCPFallback != nil { allowFallback = *r.config.RTC.AllowTCPFallback } + // default do not force full reconnect on a publication error + reconnectOnPublicationError := false + if r.config.RTC.ReconnectOnPublicationError != nil { + reconnectOnPublicationError = *r.config.RTC.ReconnectOnPublicationError + } participant, err = rtc.NewParticipant(rtc.ParticipantParams{ Identity: pi.Identity, Name: pi.Name, @@ -307,6 +312,7 @@ func (r *RoomManager) StartSession( } return nil }, + ReconnectOnPublicationError: reconnectOnPublicationError, }) if err != nil { return err