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