diff --git a/config-sample.yaml b/config-sample.yaml index e4fd969c2..07574c922 100644 --- a/config-sample.yaml +++ b/config-sample.yaml @@ -166,6 +166,10 @@ rtc: # # max number of bytes to buffer for data channel. 0 means unlimited. # # when this limit is breached, data messages will be dropped till the buffered amount drops below this limit. # data_channel_max_buffered_amount: 0 + # # target latency for data track channels. under downlink congestion, frames are dropped + # # to keep the per-subscriber buffer bounded (a fresh frame is preferable to a stale queued one). + # # 0 disables buffering control (unbounded buffering). defaults to 100ms. + # datachannel_data_track_target_latency: 100ms # when enabled, LiveKit will expose prometheus metrics on :6789/metrics # prometheus_port: 6789 diff --git a/pkg/config/config.go b/pkg/config/config.go index 90135c781..73ffaca0e 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -150,6 +150,12 @@ type RTCConfig struct { // Target latency for lossy data channels, used to drop packets to reduce latency. DatachannelLossyTargetLatency time.Duration `yaml:"datachannel_lossy_target_latency,omitempty"` + // Target latency for data track channels, used to drop packets to keep the + // per-subscriber buffer bounded under downlink congestion. Data tracks are a + // low-latency lossy transport, so a stale queued frame is worse than a dropped + // one. Set to 0 to disable buffering control (unbounded buffering). + DatachannelDataTrackTargetLatency time.Duration `yaml:"datachannel_data_track_target_latency,omitempty"` + ForwardStats ForwardStatsConfig `yaml:"forward_stats,omitempty"` // enable rtp stream restart detection for published tracks @@ -454,10 +460,11 @@ var DefaultConfig = Config{ ICEPortRangeEnd: 0, STUNServers: []string{}, }, - PacketBufferSize: 500, - PacketBufferSizeVideo: 500, - PacketBufferSizeAudio: 200, - PLIThrottle: sfu.DefaultPLIThrottleConfig, + PacketBufferSize: 500, + PacketBufferSizeVideo: 500, + PacketBufferSizeAudio: 200, + PLIThrottle: sfu.DefaultPLIThrottleConfig, + DatachannelDataTrackTargetLatency: 100 * time.Millisecond, CongestionControl: CongestionControlConfig{ Enabled: true, AllowPause: false, diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index d6130cfce..c70cb3f1a 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -174,61 +174,62 @@ type ParticipantParams struct { PLIThrottleConfig sfu.PLIThrottleConfig CongestionControlConfig config.CongestionControlConfig // codecs that are enabled for this room - PublishEnabledCodecs []*livekit.Codec - SubscribeEnabledCodecs []*livekit.Codec - Logger logger.Logger - LoggerResolver logger.DeferredFieldResolver - Reporter roomobs.ParticipantSessionReporter - ReporterResolver roomobs.ParticipantReporterResolver - SimTracks map[uint32]interceptor.SimulcastTrackInfo - Grants *auth.ClaimGrants - TokenExpiresAt time.Time - InitialVersion uint32 - ClientConf *livekit.ClientConfiguration - ClientInfo ClientInfo - Region string - Migration bool - Reconnect bool - AdaptiveStream bool - AllowTCPFallback bool - TCPFallbackRTTThreshold int - AllowUDPUnstableFallback bool - TURNSEnabled bool - ParticipantListener types.LocalParticipantListener - ParticipantHelper types.LocalParticipantHelper - DisableSupervisor bool - ReconnectOnPublicationError bool - ReconnectOnSubscriptionError bool - ReconnectOnDataChannelError bool - VersionGenerator utils.TimedVersionGenerator - DisableDynacast bool - SubscriberAllowPause bool - SubscriptionLimitAudio int32 - SubscriptionLimitVideo int32 - PlayoutDelay *livekit.PlayoutDelay - SyncStreams bool - ForwardStats *sfu.ForwardStats - DisableSenderReportPassThrough bool - MetricConfig metric.MetricConfig - UseOneShotSignallingMode bool - EnableMetrics bool - DataChannelMaxBufferedAmount uint64 - DatachannelSlowThreshold int - DatachannelLossyTargetLatency time.Duration - FireOnTrackBySdp bool - DisableCodecRegression bool - LastPubReliableSeq uint32 - Country string - PreferVideoSizeFromMedia bool - UseSinglePeerConnection bool - EnableDataTracks bool - EnableRTPStreamRestartDetection bool - ForceBackupCodecPolicySimulcast bool - DisableTransceiverReuseForE2EE bool - EnableParticipantDataBlob bool - EnableStartAtDesiredQuality bool - MigrationWaitDuration time.Duration - ExcludeIPv6LocalCandidates bool + PublishEnabledCodecs []*livekit.Codec + SubscribeEnabledCodecs []*livekit.Codec + Logger logger.Logger + LoggerResolver logger.DeferredFieldResolver + Reporter roomobs.ParticipantSessionReporter + ReporterResolver roomobs.ParticipantReporterResolver + SimTracks map[uint32]interceptor.SimulcastTrackInfo + Grants *auth.ClaimGrants + TokenExpiresAt time.Time + InitialVersion uint32 + ClientConf *livekit.ClientConfiguration + ClientInfo ClientInfo + Region string + Migration bool + Reconnect bool + AdaptiveStream bool + AllowTCPFallback bool + TCPFallbackRTTThreshold int + AllowUDPUnstableFallback bool + TURNSEnabled bool + ParticipantListener types.LocalParticipantListener + ParticipantHelper types.LocalParticipantHelper + DisableSupervisor bool + ReconnectOnPublicationError bool + ReconnectOnSubscriptionError bool + ReconnectOnDataChannelError bool + VersionGenerator utils.TimedVersionGenerator + DisableDynacast bool + SubscriberAllowPause bool + SubscriptionLimitAudio int32 + SubscriptionLimitVideo int32 + PlayoutDelay *livekit.PlayoutDelay + SyncStreams bool + ForwardStats *sfu.ForwardStats + DisableSenderReportPassThrough bool + MetricConfig metric.MetricConfig + UseOneShotSignallingMode bool + EnableMetrics bool + DataChannelMaxBufferedAmount uint64 + DatachannelSlowThreshold int + DatachannelLossyTargetLatency time.Duration + DatachannelDataTrackTargetLatency time.Duration + FireOnTrackBySdp bool + DisableCodecRegression bool + LastPubReliableSeq uint32 + Country string + PreferVideoSizeFromMedia bool + UseSinglePeerConnection bool + EnableDataTracks bool + EnableRTPStreamRestartDetection bool + ForceBackupCodecPolicySimulcast bool + DisableTransceiverReuseForE2EE bool + EnableParticipantDataBlob bool + EnableStartAtDesiredQuality bool + MigrationWaitDuration time.Duration + ExcludeIPv6LocalCandidates bool } type ParticipantImpl struct { @@ -2035,33 +2036,34 @@ func (p *ParticipantImpl) setupTransportManager() error { params := TransportManagerParams{ // primary connection does not change, canSubscribe can change if permission was updated // after the participant has joined - SubscriberAsPrimary: subscriberAsPrimary, - UseSinglePeerConnection: p.params.UseSinglePeerConnection, - Config: p.params.Config, - Twcc: p.twcc, - ProtocolVersion: p.params.ProtocolVersion, - CongestionControlConfig: p.params.CongestionControlConfig, - EnabledPublishCodecs: p.enabledPublishCodecs, - EnabledSubscribeCodecs: p.enabledSubscribeCodecs, - SimTracks: p.params.SimTracks, - ClientInfo: p.params.ClientInfo, - Migration: p.params.Migration, - AllowTCPFallback: p.params.AllowTCPFallback, - TCPFallbackRTTThreshold: p.params.TCPFallbackRTTThreshold, - AllowUDPUnstableFallback: p.params.AllowUDPUnstableFallback, - TURNSEnabled: p.params.TURNSEnabled, - AllowPlayoutDelay: p.params.PlayoutDelay.GetEnabled(), - DataChannelMaxBufferedAmount: p.params.DataChannelMaxBufferedAmount, - DatachannelSlowThreshold: p.params.DatachannelSlowThreshold, - DatachannelLossyTargetLatency: p.params.DatachannelLossyTargetLatency, - Logger: p.params.Logger.WithComponent(sutils.ComponentTransport), - PublisherHandler: pth, - SubscriberHandler: sth, - DataChannelStats: p.dataChannelStats, - UseOneShotSignallingMode: p.params.UseOneShotSignallingMode, - FireOnTrackBySdp: p.params.FireOnTrackBySdp, - EnableDataTracks: p.params.EnableDataTracks, - ExcludeIPv6LocalCandidates: p.params.ExcludeIPv6LocalCandidates, + SubscriberAsPrimary: subscriberAsPrimary, + UseSinglePeerConnection: p.params.UseSinglePeerConnection, + Config: p.params.Config, + Twcc: p.twcc, + ProtocolVersion: p.params.ProtocolVersion, + CongestionControlConfig: p.params.CongestionControlConfig, + EnabledPublishCodecs: p.enabledPublishCodecs, + EnabledSubscribeCodecs: p.enabledSubscribeCodecs, + SimTracks: p.params.SimTracks, + ClientInfo: p.params.ClientInfo, + Migration: p.params.Migration, + AllowTCPFallback: p.params.AllowTCPFallback, + TCPFallbackRTTThreshold: p.params.TCPFallbackRTTThreshold, + AllowUDPUnstableFallback: p.params.AllowUDPUnstableFallback, + TURNSEnabled: p.params.TURNSEnabled, + AllowPlayoutDelay: p.params.PlayoutDelay.GetEnabled(), + DataChannelMaxBufferedAmount: p.params.DataChannelMaxBufferedAmount, + DatachannelSlowThreshold: p.params.DatachannelSlowThreshold, + DatachannelLossyTargetLatency: p.params.DatachannelLossyTargetLatency, + DatachannelDataTrackTargetLatency: p.params.DatachannelDataTrackTargetLatency, + Logger: p.params.Logger.WithComponent(sutils.ComponentTransport), + PublisherHandler: pth, + SubscriberHandler: sth, + DataChannelStats: p.dataChannelStats, + UseOneShotSignallingMode: p.params.UseOneShotSignallingMode, + FireOnTrackBySdp: p.params.FireOnTrackBySdp, + EnableDataTracks: p.params.EnableDataTracks, + ExcludeIPv6LocalCandidates: p.params.ExcludeIPv6LocalCandidates, } if p.params.SyncStreams && p.params.PlayoutDelay.GetEnabled() && p.params.ClientInfo.isFirefox() { // we will disable playout delay for Firefox if the user is expecting diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index 27ff8a3a3..3df0ba1fe 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -303,27 +303,28 @@ type PCTransport struct { } type TransportParams struct { - Handler transport.Handler - ProtocolVersion types.ProtocolVersion - Config *WebRTCConfig - Twcc *lktwcc.Responder - DirectionConfig DirectionConfig - CongestionControlConfig config.CongestionControlConfig - EnabledPublishCodecs []*livekit.Codec - EnabledSubscribeCodecs []*livekit.Codec - Logger logger.Logger - Transport livekit.SignalTarget - SimTracks map[uint32]sfuinterceptor.SimulcastTrackInfo - ClientInfo ClientInfo - IsOfferer bool - IsSendSide bool - AllowPlayoutDelay bool - UseOneShotSignallingMode bool - ExcludeIPv6LocalCandidates bool - FireOnTrackBySdp bool - DataChannelMaxBufferedAmount uint64 - DatachannelSlowThreshold int - DatachannelLossyTargetLatency time.Duration + Handler transport.Handler + ProtocolVersion types.ProtocolVersion + Config *WebRTCConfig + Twcc *lktwcc.Responder + DirectionConfig DirectionConfig + CongestionControlConfig config.CongestionControlConfig + EnabledPublishCodecs []*livekit.Codec + EnabledSubscribeCodecs []*livekit.Codec + Logger logger.Logger + Transport livekit.SignalTarget + SimTracks map[uint32]sfuinterceptor.SimulcastTrackInfo + ClientInfo ClientInfo + IsOfferer bool + IsSendSide bool + AllowPlayoutDelay bool + UseOneShotSignallingMode bool + ExcludeIPv6LocalCandidates bool + FireOnTrackBySdp bool + DataChannelMaxBufferedAmount uint64 + DatachannelSlowThreshold int + DatachannelLossyTargetLatency time.Duration + DatachannelDataTrackTargetLatency time.Duration // for development test DatachannelMaxReceiverBufferSize int @@ -914,7 +915,7 @@ func (t *PCTransport) onDataChannel(dc *webrtc.DataChannel) { if t.dataTrackDC != nil { t.dataTrackDC.Close() } - t.dataTrackDC = datachannel.NewDataChannelWriterUnreliable(dc, rawDC, 0, 0) + t.dataTrackDC = datachannel.NewDataChannelWriterUnreliable(dc, rawDC, t.params.DatachannelDataTrackTargetLatency, uint64(lossyDataChannelMinBufferedAmount)) } case kind == livekit.DataPacket_RELIABLE: @@ -1300,7 +1301,7 @@ func (t *PCTransport) CreateDataChannel(label string, dci *webrtc.DataChannelIni case dcPtr == &t.lossyDC: *dcPtr = datachannel.NewDataChannelWriterUnreliable(dc, rawDC, t.params.DatachannelLossyTargetLatency, uint64(lossyDataChannelMinBufferedAmount)) case dcPtr == &t.dataTrackDC: - *dcPtr = datachannel.NewDataChannelWriterUnreliable(dc, rawDC, 0, 0) + *dcPtr = datachannel.NewDataChannelWriterUnreliable(dc, rawDC, t.params.DatachannelDataTrackTargetLatency, uint64(lossyDataChannelMinBufferedAmount)) } if dcReady != nil { *dcReady = true diff --git a/pkg/rtc/transportmanager.go b/pkg/rtc/transportmanager.go index 6259d1a7c..308ad12b2 100644 --- a/pkg/rtc/transportmanager.go +++ b/pkg/rtc/transportmanager.go @@ -72,33 +72,34 @@ func (h TransportManagerTransportHandler) OnFailed(isShortLived bool, iceConnect // ------------------------------- type TransportManagerParams struct { - SubscriberAsPrimary bool - UseSinglePeerConnection bool - Config *WebRTCConfig - Twcc *twcc.Responder - ProtocolVersion types.ProtocolVersion - CongestionControlConfig config.CongestionControlConfig - EnabledSubscribeCodecs []*livekit.Codec - EnabledPublishCodecs []*livekit.Codec - SimTracks map[uint32]interceptor.SimulcastTrackInfo - ClientInfo ClientInfo - Migration bool - AllowTCPFallback bool - TCPFallbackRTTThreshold int - AllowUDPUnstableFallback bool - TURNSEnabled bool - AllowPlayoutDelay bool - DataChannelMaxBufferedAmount uint64 - DatachannelSlowThreshold int - DatachannelLossyTargetLatency time.Duration - Logger logger.Logger - PublisherHandler transport.Handler - SubscriberHandler transport.Handler - DataChannelStats *BytesTrackStats - UseOneShotSignallingMode bool - FireOnTrackBySdp bool - EnableDataTracks bool - ExcludeIPv6LocalCandidates bool + SubscriberAsPrimary bool + UseSinglePeerConnection bool + Config *WebRTCConfig + Twcc *twcc.Responder + ProtocolVersion types.ProtocolVersion + CongestionControlConfig config.CongestionControlConfig + EnabledSubscribeCodecs []*livekit.Codec + EnabledPublishCodecs []*livekit.Codec + SimTracks map[uint32]interceptor.SimulcastTrackInfo + ClientInfo ClientInfo + Migration bool + AllowTCPFallback bool + TCPFallbackRTTThreshold int + AllowUDPUnstableFallback bool + TURNSEnabled bool + AllowPlayoutDelay bool + DataChannelMaxBufferedAmount uint64 + DatachannelSlowThreshold int + DatachannelLossyTargetLatency time.Duration + DatachannelDataTrackTargetLatency time.Duration + Logger logger.Logger + PublisherHandler transport.Handler + SubscriberHandler transport.Handler + DataChannelStats *BytesTrackStats + UseOneShotSignallingMode bool + FireOnTrackBySdp bool + EnableDataTracks bool + ExcludeIPv6LocalCandidates bool } type TransportManager struct { @@ -149,27 +150,28 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro publisherSubscribeCodecs = params.EnabledSubscribeCodecs } publisher, err := NewPCTransport(TransportParams{ - ProtocolVersion: params.ProtocolVersion, - Config: params.Config, - Twcc: params.Twcc, - DirectionConfig: params.Config.Publisher, - CongestionControlConfig: params.CongestionControlConfig, - EnabledPublishCodecs: params.EnabledPublishCodecs, - EnabledSubscribeCodecs: publisherSubscribeCodecs, - Logger: lgr, - SimTracks: params.SimTracks, - ClientInfo: params.ClientInfo, - IsSendSide: params.UseOneShotSignallingMode || params.UseSinglePeerConnection, - AllowPlayoutDelay: params.AllowPlayoutDelay, - Transport: livekit.SignalTarget_PUBLISHER, - Handler: TransportManagerTransportHandler{params.PublisherHandler, t, lgr}, - UseOneShotSignallingMode: params.UseOneShotSignallingMode, - DataChannelMaxBufferedAmount: params.DataChannelMaxBufferedAmount, - DatachannelSlowThreshold: params.DatachannelSlowThreshold, - DatachannelLossyTargetLatency: params.DatachannelLossyTargetLatency, - FireOnTrackBySdp: params.FireOnTrackBySdp, - EnableDataTracks: params.EnableDataTracks, - ExcludeIPv6LocalCandidates: params.ExcludeIPv6LocalCandidates, + ProtocolVersion: params.ProtocolVersion, + Config: params.Config, + Twcc: params.Twcc, + DirectionConfig: params.Config.Publisher, + CongestionControlConfig: params.CongestionControlConfig, + EnabledPublishCodecs: params.EnabledPublishCodecs, + EnabledSubscribeCodecs: publisherSubscribeCodecs, + Logger: lgr, + SimTracks: params.SimTracks, + ClientInfo: params.ClientInfo, + IsSendSide: params.UseOneShotSignallingMode || params.UseSinglePeerConnection, + AllowPlayoutDelay: params.AllowPlayoutDelay, + Transport: livekit.SignalTarget_PUBLISHER, + Handler: TransportManagerTransportHandler{params.PublisherHandler, t, lgr}, + UseOneShotSignallingMode: params.UseOneShotSignallingMode, + DataChannelMaxBufferedAmount: params.DataChannelMaxBufferedAmount, + DatachannelSlowThreshold: params.DatachannelSlowThreshold, + DatachannelLossyTargetLatency: params.DatachannelLossyTargetLatency, + DatachannelDataTrackTargetLatency: params.DatachannelDataTrackTargetLatency, + FireOnTrackBySdp: params.FireOnTrackBySdp, + EnableDataTracks: params.EnableDataTracks, + ExcludeIPv6LocalCandidates: params.ExcludeIPv6LocalCandidates, }) if err != nil { return nil, err @@ -179,24 +181,25 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro if !t.params.UseOneShotSignallingMode && !t.params.UseSinglePeerConnection { lgr := LoggerWithPCTarget(params.Logger, livekit.SignalTarget_SUBSCRIBER) subscriber, err := NewPCTransport(TransportParams{ - ProtocolVersion: params.ProtocolVersion, - Config: params.Config, - DirectionConfig: params.Config.Subscriber, - CongestionControlConfig: params.CongestionControlConfig, - EnabledSubscribeCodecs: params.EnabledSubscribeCodecs, - Logger: lgr, - ClientInfo: params.ClientInfo, - IsOfferer: true, - IsSendSide: true, - AllowPlayoutDelay: params.AllowPlayoutDelay, - DataChannelMaxBufferedAmount: params.DataChannelMaxBufferedAmount, - DatachannelSlowThreshold: params.DatachannelSlowThreshold, - DatachannelLossyTargetLatency: params.DatachannelLossyTargetLatency, - Transport: livekit.SignalTarget_SUBSCRIBER, - Handler: TransportManagerTransportHandler{params.SubscriberHandler, t, lgr}, - FireOnTrackBySdp: params.FireOnTrackBySdp, - EnableDataTracks: params.EnableDataTracks, - ExcludeIPv6LocalCandidates: params.ExcludeIPv6LocalCandidates, + ProtocolVersion: params.ProtocolVersion, + Config: params.Config, + DirectionConfig: params.Config.Subscriber, + CongestionControlConfig: params.CongestionControlConfig, + EnabledSubscribeCodecs: params.EnabledSubscribeCodecs, + Logger: lgr, + ClientInfo: params.ClientInfo, + IsOfferer: true, + IsSendSide: true, + AllowPlayoutDelay: params.AllowPlayoutDelay, + DataChannelMaxBufferedAmount: params.DataChannelMaxBufferedAmount, + DatachannelSlowThreshold: params.DatachannelSlowThreshold, + DatachannelLossyTargetLatency: params.DatachannelLossyTargetLatency, + DatachannelDataTrackTargetLatency: params.DatachannelDataTrackTargetLatency, + Transport: livekit.SignalTarget_SUBSCRIBER, + Handler: TransportManagerTransportHandler{params.SubscriberHandler, t, lgr}, + FireOnTrackBySdp: params.FireOnTrackBySdp, + EnableDataTracks: params.EnableDataTracks, + ExcludeIPv6LocalCandidates: params.ExcludeIPv6LocalCandidates, }) if err != nil { return nil, err diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index ce850df6e..36656e126 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -520,26 +520,27 @@ func (r *RoomManager) StartSession( room: room, codecRegressionThreshold: r.config.Video.CodecRegressionThreshold, }, - ReconnectOnPublicationError: reconnectOnPublicationError, - ReconnectOnSubscriptionError: reconnectOnSubscriptionError, - ReconnectOnDataChannelError: reconnectOnDataChannelError, - VersionGenerator: r.versionGenerator, - SubscriberAllowPause: subscriberAllowPause, - SubscriptionLimitAudio: r.config.Limit.SubscriptionLimitAudio, - SubscriptionLimitVideo: r.config.Limit.SubscriptionLimitVideo, - PlayoutDelay: roomInternal.GetPlayoutDelay(), - SyncStreams: roomInternal.GetSyncStreams(), - ForwardStats: r.forwardStats, - MetricConfig: r.config.Metric, - UseOneShotSignallingMode: useOneShotSignallingMode, - DataChannelMaxBufferedAmount: r.config.RTC.DataChannelMaxBufferedAmount, - DatachannelSlowThreshold: r.config.RTC.DatachannelSlowThreshold, - DatachannelLossyTargetLatency: r.config.RTC.DatachannelLossyTargetLatency, - FireOnTrackBySdp: true, - UseSinglePeerConnection: pi.UseSinglePeerConnection, - EnableDataTracks: r.config.EnableDataTracks, - EnableParticipantDataBlob: r.config.EnableParticipantDataBlob, - EnableRTPStreamRestartDetection: r.config.RTC.EnableRTPStreamRestartDetection, + ReconnectOnPublicationError: reconnectOnPublicationError, + ReconnectOnSubscriptionError: reconnectOnSubscriptionError, + ReconnectOnDataChannelError: reconnectOnDataChannelError, + VersionGenerator: r.versionGenerator, + SubscriberAllowPause: subscriberAllowPause, + SubscriptionLimitAudio: r.config.Limit.SubscriptionLimitAudio, + SubscriptionLimitVideo: r.config.Limit.SubscriptionLimitVideo, + PlayoutDelay: roomInternal.GetPlayoutDelay(), + SyncStreams: roomInternal.GetSyncStreams(), + ForwardStats: r.forwardStats, + MetricConfig: r.config.Metric, + UseOneShotSignallingMode: useOneShotSignallingMode, + DataChannelMaxBufferedAmount: r.config.RTC.DataChannelMaxBufferedAmount, + DatachannelSlowThreshold: r.config.RTC.DatachannelSlowThreshold, + DatachannelLossyTargetLatency: r.config.RTC.DatachannelLossyTargetLatency, + DatachannelDataTrackTargetLatency: r.config.RTC.DatachannelDataTrackTargetLatency, + FireOnTrackBySdp: true, + UseSinglePeerConnection: pi.UseSinglePeerConnection, + EnableDataTracks: r.config.EnableDataTracks, + EnableParticipantDataBlob: r.config.EnableParticipantDataBlob, + EnableRTPStreamRestartDetection: r.config.RTC.EnableRTPStreamRestartDetection, }) if err != nil { prometheus.IncrementParticipantRtcCanceled(1) diff --git a/pkg/sfu/datachannel/datachannel_writer.go b/pkg/sfu/datachannel/datachannel_writer.go index 8fbbb3ccc..5a2ed73f7 100644 --- a/pkg/sfu/datachannel/datachannel_writer.go +++ b/pkg/sfu/datachannel/datachannel_writer.go @@ -130,9 +130,10 @@ func (w *DataChannelWriter[T]) writeUnreliable(p []byte) (n int, err error) { return 0, err } n, err = w.rawDC.Write(p) - if err != nil { - w.rate.AddBytes(n, int(w.bufferGetter.BufferedAmount()), mono.Now()) - } + // Track drained bytes on every write (not just failures) so the bitrate + // calculator has samples; otherwise Bitrate() has no data and the + // latency-based drop threshold above collapses to the static minBufferedAmount. + w.rate.AddBytes(n, int(w.bufferGetter.BufferedAmount()), mono.Now()) return }