diff --git a/go.mod b/go.mod index c3fce73ae..591c453a6 100644 --- a/go.mod +++ b/go.mod @@ -28,7 +28,7 @@ require ( github.com/olekukonko/tablewriter v0.0.5 github.com/ory/dockertest/v3 v3.11.0 github.com/pion/dtls/v2 v2.2.12 - github.com/pion/ice/v2 v2.3.36 + github.com/pion/ice/v2 v2.3.37 github.com/pion/interceptor v0.1.37 github.com/pion/rtcp v1.2.14 github.com/pion/rtp v1.8.9 diff --git a/go.sum b/go.sum index 6d0539605..b1ea92974 100644 --- a/go.sum +++ b/go.sum @@ -233,8 +233,8 @@ github.com/pion/datachannel v1.5.9/go.mod h1:kDUuk4CU4Uxp82NH4LQZbISULkX/HtzKa4P github.com/pion/dtls/v2 v2.2.7/go.mod h1:8WiMkebSHFD0T+dIU+UeBaoV7kDhOW5oDCzZ7WZ/F9s= github.com/pion/dtls/v2 v2.2.12 h1:KP7H5/c1EiVAAKUmXyCzPiQe5+bCJrpOeKg/L05dunk= github.com/pion/dtls/v2 v2.2.12/go.mod h1:d9SYc9fch0CqK90mRk1dC7AkzzpwJj6u2GU3u+9pqFE= -github.com/pion/ice/v2 v2.3.36 h1:SopeXiVbbcooUg2EIR8sq4b13RQ8gzrkkldOVg+bBsc= -github.com/pion/ice/v2 v2.3.36/go.mod h1:mBF7lnigdqgtB+YHkaY/Y6s6tsyRyo4u4rPGRuOjUBQ= +github.com/pion/ice/v2 v2.3.37 h1:ObIdaNDu1rCo7hObhs34YSBcO7fjslJMZV0ux+uZWh0= +github.com/pion/ice/v2 v2.3.37/go.mod h1:mBF7lnigdqgtB+YHkaY/Y6s6tsyRyo4u4rPGRuOjUBQ= github.com/pion/interceptor v0.1.37 h1:aRA8Zpab/wE7/c0O3fh1PqY0AJI3fCSEM5lRWJVorwI= github.com/pion/interceptor v0.1.37/go.mod h1:JzxbJ4umVTlZAf+/utHzNesY8tmRkM2lVmkS82TTj8Y= github.com/pion/logging v0.2.2 h1:M9+AIj/+pxNsDfAT64+MAVgJO0rsyLnoJKCqf//DoeY= diff --git a/pkg/config/configtest/checkyamltag.go b/pkg/config/configtest/checkyamltag.go index a8ee58371..2458a5b9e 100644 --- a/pkg/config/configtest/checkyamltag.go +++ b/pkg/config/configtest/checkyamltag.go @@ -31,6 +31,11 @@ func checkYAMLTags(t reflect.Type, seen map[reflect.Type]struct{}) error { for i := 0; i < t.NumField(); i++ { field := t.Field(i) + if !field.IsExported() { + // ignore unexported fields + continue + } + if field.Type.Kind() == reflect.Bool { // ignore boolean fields continue diff --git a/pkg/routing/roommanager.go b/pkg/routing/roommanager.go index 327c5806d..796ffe764 100644 --- a/pkg/routing/roommanager.go +++ b/pkg/routing/roommanager.go @@ -24,8 +24,6 @@ import ( "github.com/livekit/psrpc/pkg/middleware" ) -//go:generate go run github.com/maxbrunsfeld/counterfeiter/v6 -generate - //counterfeiter:generate . RoomManagerClient type RoomManagerClient interface { rpc.TypedRoomManagerClient diff --git a/pkg/routing/signal.go b/pkg/routing/signal.go index 858fab7c1..ce7c5f42d 100644 --- a/pkg/routing/signal.go +++ b/pkg/routing/signal.go @@ -36,8 +36,6 @@ import ( var ErrSignalWriteFailed = errors.New("signal write failed") var ErrSignalMessageDropped = errors.New("signal message dropped") -//go:generate go run github.com/maxbrunsfeld/counterfeiter/v6 -generate - //counterfeiter:generate . SignalClient type SignalClient interface { ActiveCount() int diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index fcdcf68b3..4a5a734d9 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -158,7 +158,6 @@ type ParticipantParams struct { ForwardStats *sfu.ForwardStats DisableSenderReportPassThrough bool MetricConfig metric.MetricConfig - DropRemoteICECandidates bool } type ParticipantImpl struct { @@ -964,7 +963,12 @@ func (p *ParticipantImpl) Close(sendLeave bool, reason types.ParticipantCloseRea p.clearMigrationTimer() if sendLeave { - p.sendLeaveRequest(reason, isExpectedToResume, false, false) + p.sendLeaveRequest( + reason, + isExpectedToResume, + false, // isExpectedToReconnect + false, // sendOnlyIfSupportingLeaveRequestWithAction + ) } if p.supervisor != nil { @@ -1071,7 +1075,12 @@ func (p *ParticipantImpl) MaybeStartMigration(force bool, onStart func()) bool { onStart() } - p.sendLeaveRequest(types.ParticipantCloseReasonMigrationRequested, true, false, true) + p.sendLeaveRequest( + types.ParticipantCloseReasonMigrationRequested, + true, // isExpectedToResume + false, // isExpectedToReconnect + true, // sendOnlyIfSupportingLeaveRequestWithAction + ) p.CloseSignalConnection(types.SignallingCloseReasonMigration) p.clearMigrationTimer() @@ -1449,7 +1458,6 @@ func (p *ParticipantImpl) setupTransportManager() error { PublisherHandler: pth, SubscriberHandler: sth, DataChannelStats: p.dataChannelStats, - DropRemoteICECandidates: p.params.DropRemoteICECandidates, } if p.params.SyncStreams && p.params.PlayoutDelay.GetEnabled() && p.params.ClientInfo.isFirefox() { // we will disable playout delay for Firefox if the user is expecting @@ -1820,8 +1828,14 @@ func (p *ParticipantImpl) setupDisconnectTimer() { } func (p *ParticipantImpl) onAnyTransportFailed() { - // clients support resuming of connections when websocket becomes disconnected - p.sendLeaveRequest(types.ParticipantCloseReasonPeerConnectionDisconnected, true, false, true) + p.sendLeaveRequest( + types.ParticipantCloseReasonPeerConnectionDisconnected, + true, // isExpectedToResume + false, // isExpectedToReconnect + true, // sendOnlyIfSupportingLeaveRequestWithAction + ) + + // clients support resuming of connections when signalling becomes disconnected p.CloseSignalConnection(types.SignallingCloseReasonTransportFailure) // detect when participant has actually left. @@ -2658,7 +2672,12 @@ func (p *ParticipantImpl) GetCachedDownTrack(trackID livekit.TrackID) (*webrtc.R } func (p *ParticipantImpl) IssueFullReconnect(reason types.ParticipantCloseReason) { - p.sendLeaveRequest(reason, false, true, false) + p.sendLeaveRequest( + reason, + false, // isExpectedToResume + true, // isExpectedToReconnect + false, // sendOnlyIfSupportingLeaveRequestWithAction + ) scr := types.SignallingCloseReasonUnknown switch reason { @@ -2807,7 +2826,14 @@ func (p *ParticipantImpl) setupEnabledCodecs(publishEnabledCodecs []*livekit.Cod } func (p *ParticipantImpl) GetEnabledPublishCodecs() []*livekit.Codec { - return p.enabledPublishCodecs + codecs := make([]*livekit.Codec, 0, len(p.enabledPublishCodecs)) + for _, c := range p.enabledPublishCodecs { + if c.Mime == "video/rtx" { + continue + } + codecs = append(codecs, c) + } + return codecs } func (p *ParticipantImpl) UpdateAudioTrack(update *livekit.UpdateLocalAudioTrack) error { diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 7da83559f..306d7b04a 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -18,7 +18,6 @@ import ( "context" "fmt" "math" - "net" "slices" "sort" "strings" @@ -55,8 +54,6 @@ const ( dataForwardLoadBalanceThreshold = 20 simulateDisconnectSignalTimeout = 5 * time.Second - - minIPTruncateLen = 8 ) var ( @@ -1840,25 +1837,11 @@ func connectionDetailsFields(infos []*types.ICEConnectionInfo) []interface{} { if c.Trickle { cStr += "[trickle]" } - remoteAddress := c.Remote.Address() - ipAddr := net.ParseIP(remoteAddress) - isPrivate := false - if ipAddr != nil { - isPrivate = ipAddr.IsPrivate() - } - if !isPrivate && len(remoteAddress) > minIPTruncateLen { - remoteAddress = remoteAddress[:len(remoteAddress)-3] + "..." - } - cStr += " " + fmt.Sprintf("%s %s %s:%d", c.Remote.NetworkType(), c.Remote.Type(), remoteAddress, c.Remote.Port()) + cStr += " " + fmt.Sprintf("%s %s %s:%d", c.Remote.NetworkType(), c.Remote.Type(), MaybeTruncateIP(c.Remote.Address()), c.Remote.Port()) if relatedAddress := c.Remote.RelatedAddress(); relatedAddress != nil { - ipAddr = net.ParseIP(relatedAddress.Address) - if ipAddr != nil { - isPrivate = ipAddr.IsPrivate() - relatedAddressAddress := relatedAddress.Address - if !isPrivate && len(relatedAddressAddress) > minIPTruncateLen { - relatedAddressAddress = relatedAddressAddress[:len(relatedAddressAddress)-3] + "..." - } - cStr += " " + fmt.Sprintf(" related %s:%d", relatedAddressAddress, relatedAddress.Port) + relatedAddr := MaybeTruncateIP(relatedAddress.Address) + if relatedAddr != "" { + cStr += " " + fmt.Sprintf(" related %s:%d", relatedAddr, relatedAddress.Port) } } candidates = append(candidates, cStr) diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index a3665dd46..fba7f6c51 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -150,10 +150,10 @@ func (w wrappedICECandidatePairLogger) MarshalLogObject(e zapcore.ObjectEncoder) if w.pair.Remote != nil { e.AddString("remoteProtocol", w.pair.Remote.Protocol.String()) e.AddString("remoteCandidateType", w.pair.Remote.Typ.String()) - e.AddString("remoteAdddress", w.pair.Remote.Address[:len(w.pair.Remote.Address)-3]+"...") + e.AddString("remoteAdddress", MaybeTruncateIP(w.pair.Remote.Address)) e.AddUint16("remotePort", w.pair.Remote.Port) if w.pair.Remote.RelatedAddress != "" { - e.AddString("relatedAdddress", w.pair.Remote.RelatedAddress[:len(w.pair.Remote.RelatedAddress)-3]+"...") + e.AddString("relatedAdddress", MaybeTruncateIP(w.pair.Remote.RelatedAddress)) e.AddUint16("relatedPort", w.pair.Remote.RelatedPort) } } @@ -254,7 +254,6 @@ type TransportParams struct { IsSendSide bool AllowPlayoutDelay bool DataChannelMaxBufferedAmount uint64 - DropRemoteICECandidates bool } func newPeerConnection(params TransportParams, onBandwidthEstimator func(estimator cc.BandwidthEstimator)) (*webrtc.PeerConnection, *webrtc.MediaEngine, error) { @@ -299,9 +298,12 @@ func newPeerConnection(params TransportParams, onBandwidthEstimator func(estimat se.DisableSRTCPReplayProtection(true) if !params.ProtocolVersion.SupportsICELite() || !params.ClientInfo.SupportPrflxOverRelay() { // if client don't support prflx over relay which is only Firefox, disable ICE Lite to ensure that - // dropping remote ICE candidates does not get enabled. Firefox does aggressive nomination and - // dropping remote ICE candidates means server would accept all switches and it could end up with - // the lower priority candidate. As Firefox does not support migration, ICE Lite can be disabled. + // aggressive nomination is handled properly. Firefox does aggressive nomination even if peer is + // ICE Lite (see comment as to historical reasons: https://github.com/pion/ice/pull/739#issuecomment-2452245066). + // pion/ice (as of v2.3.37) will accept all use-candidate switches when in ICE Lite mode. + // That combined with aggressive nomination from Firefox could potentially lead to the two ends + // ending up with different candidates. + // As Firefox does not support migration, ICE Lite can be disabled. se.SetLite(false) } se.SetDTLSRetransmissionInterval(dtlsRetransmissionInterval) @@ -1415,7 +1417,7 @@ func (t *PCTransport) handleRemoteICECandidate(e event) error { c := e.data.(*webrtc.ICECandidateInit) filtered := false - if t.preferTCP.Load() && !strings.Contains(c.Candidate, "tcp") { + if t.preferTCP.Load() && !strings.Contains(strings.ToLower(c.Candidate), "tcp") { t.params.Logger.Debugw("filtering out remote candidate", "candidate", c.Candidate) filtered = true } @@ -1435,12 +1437,6 @@ func (t *PCTransport) handleRemoteICECandidate(e event) error { return nil } - if t.params.DropRemoteICECandidates { - t.params.Logger.Debugw("dropping remote ICE candidate", "candidate", c.Candidate) - t.connectionDetails.AddRemoteCandidate(*c, true, true, true) - return nil - } - if err := t.pc.AddICECandidate(*c); err != nil { t.params.Logger.Warnw("failed to add cached ICE candidate", err, "candidate", c) return errors.Wrap(err, "add ice candidate failed") @@ -1465,25 +1461,6 @@ func (t *PCTransport) filterCandidates(sd webrtc.SessionDescription, preferTCP, return sd } - _, iceLite := parsed.Attribute("ice-lite") - var liteSet bool - if isLocal { - if t.localICEIsLite == nil { - t.localICEIsLite = &iceLite - liteSet = true - } - } else { - if t.remoteICEIsLite == nil { - t.remoteICEIsLite = &iceLite - liteSet = true - } - } - if liteSet && t.localICEIsLite != nil && t.remoteICEIsLite != nil { - // only drop remote candidates if local is lite and remote is not - t.params.DropRemoteICECandidates = t.params.DropRemoteICECandidates && (*t.localICEIsLite && !*t.remoteICEIsLite) - t.params.Logger.Debugw("setting DropRemoteICECandidates", "dropRemoteCandidate", t.params.DropRemoteICECandidates, "localICELite", *t.localICEIsLite, "remoteICELite", *t.remoteICEIsLite) - } - filterAttributes := func(attrs []sdp.Attribute) []sdp.Attribute { filteredAttrs := make([]sdp.Attribute, 0, len(attrs)) for _, a := range attrs { @@ -1494,7 +1471,7 @@ func (t *PCTransport) filterCandidates(sd webrtc.SessionDescription, preferTCP, filteredAttrs = append(filteredAttrs, a) continue } - excluded := (!isLocal && t.params.DropRemoteICECandidates) || (preferTCP && !c.NetworkType().IsTCP()) + excluded := preferTCP && !c.NetworkType().IsTCP() if !excluded { if !t.params.Config.UseMDNS && types.IsICECandidateMDNS(c) { excluded = true @@ -1712,10 +1689,6 @@ func (t *PCTransport) setRemoteDescription(sd webrtc.SessionDescription) error { } for _, c := range t.pendingRemoteCandidates { - if t.params.DropRemoteICECandidates { - t.connectionDetails.AddRemoteCandidate(*c, true, true, true) - continue - } if err := t.pc.AddICECandidate(*c); err != nil { t.params.Logger.Warnw("failed to add cached ICE candidate", err, "candidate", c) return errors.Wrap(err, "add ice candidate failed") diff --git a/pkg/rtc/transport_test.go b/pkg/rtc/transport_test.go index e475d32a3..c80b206ef 100644 --- a/pkg/rtc/transport_test.go +++ b/pkg/rtc/transport_test.go @@ -28,7 +28,6 @@ import ( "github.com/livekit/livekit-server/pkg/rtc/transport" "github.com/livekit/livekit-server/pkg/rtc/transport/transportfakes" - "github.com/livekit/livekit-server/pkg/rtc/types" "github.com/livekit/livekit-server/pkg/testutils" "github.com/livekit/protocol/livekit" ) @@ -504,115 +503,6 @@ func TestFilteringCandidates(t *testing.T) { transport.Close() } -func TestDropRemoteICECandidates(t *testing.T) { - cases := []struct { - name string - remoteLite bool - localLite bool - expecteLocalDrop bool - expecteRemoteDrop bool - }{ - { - name: "both not lite", - localLite: false, - remoteLite: false, - expecteLocalDrop: false, - expecteRemoteDrop: false, - }, - { - name: "remote lite", - localLite: false, - remoteLite: true, - expecteLocalDrop: false, - expecteRemoteDrop: true, - }, - { - name: "local lite", - localLite: true, - remoteLite: false, - expecteLocalDrop: true, - expecteRemoteDrop: false, - }, - { - name: "both lite", - localLite: true, - remoteLite: true, - expecteLocalDrop: false, - expecteRemoteDrop: false, - }, - } - - for _, c := range cases { - t.Run(c.name, func(t *testing.T) { - params := TransportParams{ - ParticipantID: "id", - ParticipantIdentity: "identity", - Config: &WebRTCConfig{}, - IsOfferer: true, - ProtocolVersion: types.CurrentProtocol, - DropRemoteICECandidates: true, - } - - paramsA := params - paramsA.Config.SettingEngine.SetLite(c.localLite) - handlerA := &transportfakes.FakeHandler{} - paramsA.Handler = handlerA - transportLocal, err := NewPCTransport(paramsA) - require.NoError(t, err) - _, err = transportLocal.pc.CreateDataChannel(LossyDataChannel, nil) - require.NoError(t, err) - - paramsB := params - paramsB.Config.SettingEngine.SetLite(c.remoteLite) - handlerB := &transportfakes.FakeHandler{} - paramsB.Handler = handlerB - paramsB.IsOfferer = false - transportRemote, err := NewPCTransport(paramsB) - require.NoError(t, err) - - require.False(t, transportLocal.IsEstablished()) - require.False(t, transportRemote.IsEstablished()) - - handleICEExchange(t, transportLocal, transportRemote, handlerA, handlerB) - var offer atomic.Pointer[webrtc.SessionDescription] - handlerA.OnOfferCalls(func(sd webrtc.SessionDescription) error { - parsed, err := sd.Unmarshal() - require.NoError(t, err) - _, lite := parsed.Attribute("ice-lite") - require.Equal(t, c.localLite, lite) - offer.Store(&sd) - return nil - }) - transportLocal.Negotiate(true) - require.Eventually(t, func() bool { - return offer.Load() != nil - }, 100*time.Millisecond, time.Millisecond*10, "offer not received") - - handlerB.OnAnswerCalls(func(sd webrtc.SessionDescription) error { - parsed, err := sd.Unmarshal() - require.NoError(t, err) - _, lite := parsed.Attribute("ice-lite") - require.Equal(t, c.remoteLite, lite, sd.SDP) - transportLocal.HandleRemoteDescription(sd) - return nil - }) - transportRemote.HandleRemoteDescription(*offer.Load()) - - require.Eventually(t, func() bool { - return transportLocal.IsEstablished() - }, 10*time.Second, time.Millisecond*10, "transportA is not established") - require.Eventually(t, func() bool { - return transportRemote.IsEstablished() - }, 10*time.Second, time.Millisecond*10, "transportB is not established") - - require.Equal(t, c.expecteLocalDrop, transportLocal.params.DropRemoteICECandidates) - require.Equal(t, c.expecteRemoteDrop, transportRemote.params.DropRemoteICECandidates) - - transportLocal.Close() - transportRemote.Close() - }) - } -} func handleICEExchange(t *testing.T, a, b *PCTransport, ah, bh *transportfakes.FakeHandler) { ah.OnICECandidateCalls(func(candidate *webrtc.ICECandidate, target livekit.SignalTarget) error { diff --git a/pkg/rtc/transportmanager.go b/pkg/rtc/transportmanager.go index bee94e700..cc69e514c 100644 --- a/pkg/rtc/transportmanager.go +++ b/pkg/rtc/transportmanager.go @@ -103,7 +103,6 @@ type TransportManagerParams struct { PublisherHandler transport.Handler SubscriberHandler transport.Handler DataChannelStats *telemetry.BytesTrackStats - DropRemoteICECandidates bool } type TransportManager struct { @@ -158,7 +157,6 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro ClientInfo: params.ClientInfo, Transport: livekit.SignalTarget_PUBLISHER, Handler: TransportManagerPublisherTransportHandler{TransportManagerTransportHandler{params.PublisherHandler, t, lgr}}, - DropRemoteICECandidates: params.DropRemoteICECandidates, }) if err != nil { return nil, err @@ -182,7 +180,6 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro DataChannelMaxBufferedAmount: params.DataChannelMaxBufferedAmount, Transport: livekit.SignalTarget_SUBSCRIBER, Handler: TransportManagerTransportHandler{params.SubscriberHandler, t, lgr}, - DropRemoteICECandidates: params.DropRemoteICECandidates, }) if err != nil { return nil, err diff --git a/pkg/rtc/utils.go b/pkg/rtc/utils.go index c5a54d6ac..21390b362 100644 --- a/pkg/rtc/utils.go +++ b/pkg/rtc/utils.go @@ -18,6 +18,7 @@ import ( "encoding/json" "errors" "io" + "net" "strings" "github.com/pion/webrtc/v3" @@ -28,6 +29,8 @@ import ( const ( trackIdSeparator = "|" + + cMinIPTruncateLen = 8 ) func UnpackStreamID(packed string) (participantID livekit.ParticipantID, trackID livekit.TrackID) { @@ -198,3 +201,16 @@ func LoggerWithCodecMime(l logger.Logger, mime string) logger.Logger { } return l } + +func MaybeTruncateIP(addr string) string { + ipAddr := net.ParseIP(addr) + if ipAddr == nil { + return "" + } + + if ipAddr.IsPrivate() || len(addr) <= cMinIPTruncateLen { + return addr + } + + return addr[:len(addr)-3] + "..." +} diff --git a/pkg/service/clients.go b/pkg/service/clients.go index 9fd887878..8a0eb39c1 100644 --- a/pkg/service/clients.go +++ b/pkg/service/clients.go @@ -27,8 +27,6 @@ import ( "github.com/livekit/protocol/utils/guid" ) -//go:generate go run github.com/maxbrunsfeld/counterfeiter/v6 -generate - //counterfeiter:generate . IOClient type IOClient interface { CreateEgress(ctx context.Context, info *livekit.EgressInfo) (*emptypb.Empty, error) diff --git a/pkg/service/signal.go b/pkg/service/signal.go index 742ce51d1..b636aa018 100644 --- a/pkg/service/signal.go +++ b/pkg/service/signal.go @@ -31,8 +31,6 @@ import ( "github.com/livekit/psrpc/pkg/middleware" ) -//go:generate go run github.com/maxbrunsfeld/counterfeiter/v6 -generate - //counterfeiter:generate . SessionHandler type SessionHandler interface { Logger(ctx context.Context) logger.Logger diff --git a/pkg/telemetry/analyticsservice.go b/pkg/telemetry/analyticsservice.go index 365e7ed05..57873acb0 100644 --- a/pkg/telemetry/analyticsservice.go +++ b/pkg/telemetry/analyticsservice.go @@ -29,7 +29,7 @@ import ( "github.com/livekit/livekit-server/pkg/routing" ) -//go:generate go run github.com/maxbrunsfeld/counterfeiter/v6 . AnalyticsService +//counterfeiter:generate . AnalyticsService type AnalyticsService interface { SendStats(ctx context.Context, stats []*livekit.AnalyticsStat) SendEvent(ctx context.Context, events *livekit.AnalyticsEvent) diff --git a/pkg/telemetry/telemetryservice.go b/pkg/telemetry/telemetryservice.go index 0079c5d5a..ba8ca50df 100644 --- a/pkg/telemetry/telemetryservice.go +++ b/pkg/telemetry/telemetryservice.go @@ -26,7 +26,9 @@ import ( "github.com/livekit/protocol/webhook" ) -//go:generate go run github.com/maxbrunsfeld/counterfeiter/v6 . TelemetryService +//go:generate go run github.com/maxbrunsfeld/counterfeiter/v6 -generate + +//counterfeiter:generate . TelemetryService type TelemetryService interface { // TrackStats is called periodically for each track in both directions (published/subscribed) TrackStats(key StatsKey, stat *livekit.AnalyticsStat)