diff --git a/config-sample.yaml b/config-sample.yaml index 71888d572..1cbf8a9e1 100644 --- a/config-sample.yaml +++ b/config-sample.yaml @@ -385,3 +385,7 @@ keys: # signal_message_size_limit: 2097152 # # same as above, but for agent worker WebSocket connections. defaults to 2 MiB, 0 for no limit # agent_signal_message_size_limit: 2097152 +# # max size (bytes) of an HTTP request body on the main API listener (Twirp room/egress/ingress/SIP +# # routes). requests larger than this are rejected before their body is decoded, so large messages +# # cannot exhaust memory. defaults to 10 MiB, 0 for no limit +# max_api_request_body_size: 10485760 diff --git a/go.mod b/go.mod index d72a345b6..b85d6d480 100644 --- a/go.mod +++ b/go.mod @@ -21,7 +21,7 @@ require ( github.com/jxskiss/base62 v1.1.0 github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 github.com/livekit/mediatransportutil v0.0.0-20260608063931-a3417d38cda0 - github.com/livekit/protocol v1.50.5-0.20260811022948-6ad5660c9f8f + github.com/livekit/protocol v1.50.5-0.20260814120900-8b1ab81c7d00 github.com/livekit/psrpc v0.7.3 github.com/mackerelio/go-osstat v0.2.8 github.com/magefile/mage v1.17.2 @@ -82,6 +82,7 @@ require ( github.com/olekukonko/cat v0.0.0-20250911104152-50322a0618f6 // indirect github.com/olekukonko/errors v1.3.0 // indirect github.com/olekukonko/ll v0.1.8 // indirect + github.com/petermattis/goid v0.0.0-20260725062400-500c67a39b75 // indirect github.com/puzpuzpuz/xsync/v4 v4.5.0 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.69.0 // indirect @@ -162,3 +163,9 @@ tool ( github.com/maxbrunsfeld/counterfeiter/v6 golang.org/x/tools/cmd/goimports ) + +replace github.com/pion/webrtc/v4 => github.com/livekit/webrtc-pion/v4 v4.2.18-warp.1 + +replace github.com/pion/dtls/v3 => github.com/livekit/dtls/v3 v3.1.5-warp.1 + +replace github.com/pion/ice/v4 => github.com/livekit/ice/v4 v4.4.0-warp.1 diff --git a/go.sum b/go.sum index fe6b2b28c..35dcb0413 100644 --- a/go.sum +++ b/go.sum @@ -158,14 +158,20 @@ github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0 github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= github.com/lithammer/shortuuid/v4 v4.2.0 h1:LMFOzVB3996a7b8aBuEXxqOBflbfPQAiVzkIcHO0h8c= github.com/lithammer/shortuuid/v4 v4.2.0/go.mod h1:D5noHZ2oFw/YaKCfGy0YxyE7M0wMbezmMjPdhyEFe6Y= +github.com/livekit/dtls/v3 v3.1.5-warp.1 h1:5jo2qQreDnUrgw6Al1F+9xVyXHvWQ2IaQkjF53FHEyU= +github.com/livekit/dtls/v3 v3.1.5-warp.1/go.mod h1:gz1K4jg6c+fq86oQMH4pilpCEOEPwmEr2jY+VcF/mkU= +github.com/livekit/ice/v4 v4.4.0-warp.1 h1:P9Y1f02VVx6kkRJxY3jzoS7HnblFbYkgMwiu6iOqCOU= +github.com/livekit/ice/v4 v4.4.0-warp.1/go.mod h1:obAyD+J+Hzs7QA7Y8YXHp5uIn6gb7z87pKedXZkrcFU= github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 h1:9x+U2HGLrSw5ATTo469PQPkqzdoU7be46ryiCDO3boc= github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= github.com/livekit/mediatransportutil v0.0.0-20260608063931-a3417d38cda0 h1:XHNNzebIKZRkLimla/hFGrAIX5EMWHctrgt3hLw7s+I= github.com/livekit/mediatransportutil v0.0.0-20260608063931-a3417d38cda0/go.mod h1:o8CFmAdrVwzJNOCsQCLUzXRjokkufNshnQHOe4fRaqU= -github.com/livekit/protocol v1.50.5-0.20260811022948-6ad5660c9f8f h1:1Tod/QfZbX2Y6qvmVm392uYvddWaY8bz4hbnSgpEcMc= -github.com/livekit/protocol v1.50.5-0.20260811022948-6ad5660c9f8f/go.mod h1:edX/q09IZsPPR6SRK1xqA2lcdskpXIZfVvZAPDVV4yw= +github.com/livekit/protocol v1.50.5-0.20260814120900-8b1ab81c7d00 h1:g4Rdg7gAqPX/CrV7ptAERPtDX1YbjpqAF67G/6pd2Zs= +github.com/livekit/protocol v1.50.5-0.20260814120900-8b1ab81c7d00/go.mod h1:/kYxa0dlTuH981LaBFHG/Swyr969d0+2+/6Lm7fFc34= github.com/livekit/psrpc v0.7.3 h1:bekuZt/ZQzg8+/M8G6G5jq7bvV9fAKdPHSOZeTwrIIc= github.com/livekit/psrpc v0.7.3/go.mod h1:rAI+m2+/cb4x9RXhLRtUx5ZwdfjjXOl4zi46IjEetaw= +github.com/livekit/webrtc-pion/v4 v4.2.18-warp.1 h1:fH+v4W+NFp9FfPzON6FaUFNmazGcctaAhb2P+Ksf+1s= +github.com/livekit/webrtc-pion/v4 v4.2.18-warp.1/go.mod h1:rbKGHo2OpNUImWTvRIV776/3xjjq/t47H3IZiTtwluc= github.com/mackerelio/go-osstat v0.2.8 h1:I2duicTaCGWoM53XwAwA9OIe1inu0xnVs8/pqOWWVr4= github.com/mackerelio/go-osstat v0.2.8/go.mod h1:SyS3XxKdoSKJnTGTkN5Yrh6VUQVuAURACfE6y+2DN4k= github.com/magefile/mage v1.17.2 h1:fyXVu1eadI8Ap1HCCNgEhJ5McIWiYhLR8uol64ZZc40= @@ -231,12 +237,10 @@ github.com/opencontainers/image-spec v1.1.1 h1:y0fUlFfIZhPF1W537XOLg0/fcx6zcHCJw github.com/opencontainers/image-spec v1.1.1/go.mod h1:qpqAh3Dmcf36wStyyWU+kCeDgrGnAve2nCC8+7h8Q0M= github.com/ory/dockertest/v4 v4.0.0 h1:i19aFsO/VXE0VrMk4ifnKW4G/KIJ93PCjLOslxXoPME= github.com/ory/dockertest/v4 v4.0.0/go.mod h1:b5Ofu8VIxWNhXFvQcLu17pRNQdoUBKtXBW74G4Ygzx8= +github.com/petermattis/goid v0.0.0-20260725062400-500c67a39b75 h1:VmZ6mKVkxavKEhEy4ZYyV7BwBYBFBP0TwIqmLk84fpU= +github.com/petermattis/goid v0.0.0-20260725062400-500c67a39b75/go.mod h1:pxMtw7cyUw6B2bRH0ZBANSPg+AoSud1I1iyJHI69jH4= github.com/pion/datachannel v1.6.2 h1:7EXQ8TH3vTouBUdRWYbcX2edSx9Yj6k5zl5P+qyxEPc= github.com/pion/datachannel v1.6.2/go.mod h1:pzbdAZvyGtXbcHM1hBbsFaOTf40lZizU/dNlvVOak6E= -github.com/pion/dtls/v3 v3.1.5 h1:9xJtVsHwMYeSjPp5Hh1FTis4DchnQWtnOa5o+6ygqfc= -github.com/pion/dtls/v3 v3.1.5/go.mod h1:gz1K4jg6c+fq86oQMH4pilpCEOEPwmEr2jY+VcF/mkU= -github.com/pion/ice/v4 v4.4.0 h1:wvHDDqimaC38Y7MVpD46Y63p246ChvXd87VKoLYS5b4= -github.com/pion/ice/v4 v4.4.0/go.mod h1:obAyD+J+Hzs7QA7Y8YXHp5uIn6gb7z87pKedXZkrcFU= github.com/pion/interceptor v0.1.47 h1:yw8t5pJ2f8t78NgU+8EmxhaqYLXS7uFCC/tAGOaSDBo= github.com/pion/interceptor v0.1.47/go.mod h1:7yoRBzaIDETPC6cIN8Zj9EyGqHv1ImOpcTFPha6MuOM= github.com/pion/logging v0.2.4 h1:tTew+7cmQ+Mc1pTBLKH2puKsOvhm32dROumOZ655zB8= @@ -263,8 +267,6 @@ github.com/pion/transport/v4 v4.1.0 h1:8S+nF2reM2cJuqC6g78OVy2BBgmbdns+acx3jA97B github.com/pion/transport/v4 v4.1.0/go.mod h1:06hFI+jCFcok2X2MekVufNZ/uzNZXivGBPfviSVcjgM= github.com/pion/turn/v5 v5.0.12 h1:6+b69ivQQXSlyfkp2AKripqD2k3W32qXK8QzCzpJWPI= github.com/pion/turn/v5 v5.0.12/go.mod h1:CQACsRDJtjQ+6RSrGHrS2PCIerLwbW3uqXRqOvtjAFg= -github.com/pion/webrtc/v4 v4.2.18 h1:smA/3g6Gy4RohM0VIZ5KKY/12TQbxv3XFgpUMyb2EUI= -github.com/pion/webrtc/v4 v4.2.18/go.mod h1:vmzi6s+rvhoIuT94DPqivB+0xJXs9rG4QRD+4MgBtlY= github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= diff --git a/pkg/config/config.go b/pkg/config/config.go index 71de31476..2ef6a3efe 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -122,6 +122,11 @@ type RTCConfig struct { TURNServers []TURNServer `yaml:"turn_servers,omitempty"` + // EnableWarp turns on WARP = SPED (DTLS-in-STUN, saves DTLS round-trip) + + // SNAP (SCTP INIT in SDP, saves SCTP/data-channel init round-trip). + // Off by default; experimental. + EnableWarp bool `yaml:"enable_warp,omitempty"` + // Deprecated StrictACKs bool `yaml:"strict_acks,omitempty"` @@ -345,6 +350,12 @@ type LimitConfig struct { SignalMessageSizeLimit int64 `yaml:"signal_message_size_limit,omitempty"` // same as SignalMessageSizeLimit, but for agent worker WebSocket connections. AgentSignalMessageSizeLimit int64 `yaml:"agent_signal_message_size_limit,omitempty"` + + // maximum size (in bytes) of an HTTP request body accepted on the main API + // listener (Twirp room/egress/ingress/SIP routes, etc). Requests larger than + // this are rejected before their body is decoded, so large messages cannot + // exhaust memory. A value of 0 disables the limit (unbounded). + MaxAPIRequestBodySize int64 `yaml:"max_api_request_body_size,omitempty"` } func (l LimitConfig) CheckRoomNameLength(name string) bool { @@ -551,8 +562,9 @@ var DefaultConfig = Config{ MaxDataBlobKeyLength: 256, MaxDataBlobSize: 64000, MaxDataTrackCustomEncodingLength: 32, - SignalMessageSizeLimit: 2 << 20, // 2 MiB - AgentSignalMessageSizeLimit: 2 << 20, // 2 MiB + SignalMessageSizeLimit: 2 << 20, // 2 MiB + AgentSignalMessageSizeLimit: 2 << 20, // 2 MiB + MaxAPIRequestBodySize: 10 << 20, // 10 MiB }, Logging: LoggingConfig{ PionLevel: "error", diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index e7f33097a..f35228f33 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -233,6 +233,7 @@ type ParticipantParams struct { EnableStartAtDesiredQuality bool MigrationWaitDuration time.Duration ExcludeIPv6LocalCandidates bool + EnableWarp bool } type ParticipantImpl struct { @@ -1453,9 +1454,9 @@ func (p *ParticipantImpl) recordRTCState(closeReason types.ParticipantCloseReaso } if p.IsConnectionCanceled(closeReason) { - prometheus.IncrementParticipantRtcCanceled(1) + prometheus.IncrementParticipantRtcCanceled(1, p.params.EnableWarp) } else { - prometheus.IncrementParticipantRtcFailure(1) + prometheus.IncrementParticipantRtcFailure(1, p.params.EnableWarp) } } @@ -2100,6 +2101,7 @@ func (p *ParticipantImpl) setupTransportManager() error { FireOnTrackBySdp: p.params.FireOnTrackBySdp, EnableDataTracks: p.params.EnableDataTracks, ExcludeIPv6LocalCandidates: p.params.ExcludeIPv6LocalCandidates, + EnableWarp: p.params.EnableWarp, } if p.params.SyncStreams && p.params.PlayoutDelay.GetEnabled() && p.params.ClientInfo.isFirefox() { // we will disable playout delay for Firefox if the user is expecting @@ -2683,8 +2685,8 @@ func (p *ParticipantImpl) onPrimaryTransportInitialConnected() { } if !p.sessionStartRecorded.Swap(true) { - prometheus.RecordSessionStartTime(int(p.ProtocolVersion()), time.Since(p.params.SessionStartTime)) - prometheus.IncrementParticipantRtcSuccess(1) + prometheus.RecordSessionStartTime(int(p.ProtocolVersion()), p.params.EnableWarp, time.Since(p.params.SessionStartTime)) + prometheus.IncrementParticipantRtcSuccess(1, p.params.EnableWarp) } p.updateState(livekit.ParticipantInfo_ACTIVE) } @@ -4442,3 +4444,7 @@ func (p *ParticipantImpl) PerformRpc(req *livekit.PerformRpcRequest, resultCh ch p.rpcLock.Unlock() }() } + +func (p *ParticipantImpl) IsWarpEnabled() bool { + return p.params.EnableWarp +} diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index b6e97a8b0..74f7f3059 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -1265,6 +1265,7 @@ func (r *Room) onStateChange(p types.LocalParticipant) { p.ToProto(), meta, false, + p.IsWarpEnabled(), p.TelemetryGuard(), ) diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index c1b5c0c01..cdd3eb6e1 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -330,6 +330,7 @@ type TransportParams struct { DatachannelMaxReceiverBufferSize int EnableDataTracks bool + EnableWarp bool } func newPeerConnection( @@ -382,6 +383,11 @@ func newPeerConnection( se.EnableSCTPZeroChecksum(true) } + if params.EnableWarp { + se.EnableSped(true) + se.EnableSctpSnap(true) + } + // // Disable SRTP replay protection (https://datatracker.ietf.org/doc/html/rfc3711#page-15). // Needed due to lack of RTX stream support in Pion. diff --git a/pkg/rtc/transportmanager.go b/pkg/rtc/transportmanager.go index 151af1bf1..9e8cf51b1 100644 --- a/pkg/rtc/transportmanager.go +++ b/pkg/rtc/transportmanager.go @@ -100,6 +100,7 @@ type TransportManagerParams struct { FireOnTrackBySdp bool EnableDataTracks bool ExcludeIPv6LocalCandidates bool + EnableWarp bool } type TransportManager struct { @@ -172,6 +173,7 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro FireOnTrackBySdp: params.FireOnTrackBySdp, EnableDataTracks: params.EnableDataTracks, ExcludeIPv6LocalCandidates: params.ExcludeIPv6LocalCandidates, + EnableWarp: params.EnableWarp, }) if err != nil { return nil, err @@ -200,6 +202,7 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro FireOnTrackBySdp: params.FireOnTrackBySdp, EnableDataTracks: params.EnableDataTracks, ExcludeIPv6LocalCandidates: params.ExcludeIPv6LocalCandidates, + EnableWarp: params.EnableWarp, }) if err != nil { return nil, err diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 225cd9861..ddc462e1f 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -430,6 +430,7 @@ type LocalParticipant interface { GetPublisherICESessionUfrag() (string, error) SupportsMoving() error GetLastReliableSequence(migrateOut bool) uint32 + IsWarpEnabled() bool SwapResponseSink(sink routing.MessageSink, reason SignallingCloseReason) GetResponseSink() routing.MessageSink diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index 6996a81e3..d4fcfd21b 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -965,6 +965,16 @@ type FakeLocalParticipant struct { isUsingSinglePeerConnectionReturnsOnCall map[int]struct { result1 bool } + IsWarpEnabledStub func() bool + isWarpEnabledMutex sync.RWMutex + isWarpEnabledArgsForCall []struct { + } + isWarpEnabledReturns struct { + result1 bool + } + isWarpEnabledReturnsOnCall map[int]struct { + result1 bool + } IssueFullReconnectStub func(types.ParticipantCloseReason) issueFullReconnectMutex sync.RWMutex issueFullReconnectArgsForCall []struct { @@ -6628,6 +6638,59 @@ func (fake *FakeLocalParticipant) IsUsingSinglePeerConnectionReturnsOnCall(i int }{result1} } +func (fake *FakeLocalParticipant) IsWarpEnabled() bool { + fake.isWarpEnabledMutex.Lock() + ret, specificReturn := fake.isWarpEnabledReturnsOnCall[len(fake.isWarpEnabledArgsForCall)] + fake.isWarpEnabledArgsForCall = append(fake.isWarpEnabledArgsForCall, struct { + }{}) + stub := fake.IsWarpEnabledStub + fakeReturns := fake.isWarpEnabledReturns + fake.recordInvocation("IsWarpEnabled", []interface{}{}) + fake.isWarpEnabledMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalParticipant) IsWarpEnabledCallCount() int { + fake.isWarpEnabledMutex.RLock() + defer fake.isWarpEnabledMutex.RUnlock() + return len(fake.isWarpEnabledArgsForCall) +} + +func (fake *FakeLocalParticipant) IsWarpEnabledCalls(stub func() bool) { + fake.isWarpEnabledMutex.Lock() + defer fake.isWarpEnabledMutex.Unlock() + fake.IsWarpEnabledStub = stub +} + +func (fake *FakeLocalParticipant) IsWarpEnabledReturns(result1 bool) { + fake.isWarpEnabledMutex.Lock() + defer fake.isWarpEnabledMutex.Unlock() + fake.IsWarpEnabledStub = nil + fake.isWarpEnabledReturns = struct { + result1 bool + }{result1} +} + +func (fake *FakeLocalParticipant) IsWarpEnabledReturnsOnCall(i int, result1 bool) { + fake.isWarpEnabledMutex.Lock() + defer fake.isWarpEnabledMutex.Unlock() + fake.IsWarpEnabledStub = nil + if fake.isWarpEnabledReturnsOnCall == nil { + fake.isWarpEnabledReturnsOnCall = make(map[int]struct { + result1 bool + }) + } + fake.isWarpEnabledReturnsOnCall[i] = struct { + result1 bool + }{result1} +} + func (fake *FakeLocalParticipant) IssueFullReconnect(arg1 types.ParticipantCloseReason) { fake.issueFullReconnectMutex.Lock() fake.issueFullReconnectArgsForCall = append(fake.issueFullReconnectArgsForCall, struct { diff --git a/pkg/service/auth.go b/pkg/service/auth.go index d019c4aae..9d8c1e921 100644 --- a/pkg/service/auth.go +++ b/pkg/service/auth.go @@ -17,6 +17,7 @@ package service import ( "context" "errors" + "fmt" "net/http" "strings" "time" @@ -88,13 +89,13 @@ func (m *APIKeyAuthMiddleware) ServeHTTP(w http.ResponseWriter, r *http.Request, secret := m.provider.GetSecret(v.APIKey()) if secret == "" { - HandleError(w, r, http.StatusUnauthorized, errors.New("invalid API key: "+v.APIKey())) + HandleError(w, r, http.StatusUnauthorized, ErrInvalidAPIKey, "apiKey", v.APIKey()) return } claims, grants, err := v.Verify(secret) if err != nil { - HandleError(w, r, http.StatusUnauthorized, errors.New("invalid token: "+authToken+", error: "+err.Error())) + HandleError(w, r, http.StatusUnauthorized, fmt.Errorf("%w: %s", ErrInvalidAuthorizationToken, err.Error())) return } diff --git a/pkg/service/egress.go b/pkg/service/egress.go index 697aecbb4..782d6cdc5 100644 --- a/pkg/service/egress.go +++ b/pkg/service/egress.go @@ -319,8 +319,16 @@ func (s *EgressService) UpdateLayout(ctx context.Context, req *livekit.UpdateLay return info, nil } +func redactedStreamUrls(urls []string) []string { + redacted := make([]string, len(urls)) + for i, u := range urls { + redacted[i], _ = utils.RedactStreamKey(u) + } + return redacted +} + func (s *EgressService) UpdateStream(ctx context.Context, req *livekit.UpdateStreamRequest) (*livekit.EgressInfo, error) { - AppendLogFields(ctx, "egressID", req.EgressId, "addUrls", req.AddOutputUrls, "removeUrls", req.RemoveOutputUrls) + AppendLogFields(ctx, "egressID", req.EgressId, "addUrls", redactedStreamUrls(req.AddOutputUrls), "removeUrls", redactedStreamUrls(req.RemoveOutputUrls)) if err := EnsureRecordPermission(ctx); err != nil { return nil, twirpAuthError(err) } diff --git a/pkg/service/requestlimit_test.go b/pkg/service/requestlimit_test.go new file mode 100644 index 000000000..7c7075301 --- /dev/null +++ b/pkg/service/requestlimit_test.go @@ -0,0 +1,135 @@ +// Copyright 2024 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package service_test + +import ( + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/livekit/livekit-server/pkg/service" +) + +// readAllHandler mimics the way a Twirp handler consumes the whole request body +// before doing anything else. It records how much it managed to read and whether +// the read failed (e.g. because the body limit was exceeded). +type readAllHandler struct { + bytesRead int + readErr error + called bool +} + +func (h *readAllHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + h.called = true + if r.Body == nil { + w.WriteHeader(http.StatusOK) + return + } + b, err := io.ReadAll(r.Body) + h.bytesRead = len(b) + h.readErr = err + if err != nil { + // a real decoder surfaces this as a 4xx/5xx; emulate that + w.WriteHeader(http.StatusBadRequest) + return + } + w.WriteHeader(http.StatusOK) +} + +func TestRequestBodyLimiter(t *testing.T) { + const limit = 1024 + + t.Run("rejects oversized declared Content-Length before decoding", func(t *testing.T) { + l := service.NewRequestBodyLimiter(limit) + handler := &readAllHandler{} + + body := strings.NewReader(strings.Repeat("a", limit*4)) + r := httptest.NewRequest(http.MethodPost, "/twirp/livekit.Egress/StartRoomCompositeEgress", body) + require.EqualValues(t, limit*4, r.ContentLength) + w := httptest.NewRecorder() + + l.ServeHTTP(w, r, handler.ServeHTTP) + + require.Equal(t, http.StatusRequestEntityTooLarge, w.Code) + // the body must never be handed to the decoder + require.False(t, handler.called) + }) + + t.Run("bounds body when Content-Length is absent/dishonest", func(t *testing.T) { + l := service.NewRequestBodyLimiter(limit) + handler := &readAllHandler{} + + body := strings.NewReader(strings.Repeat("a", limit*4)) + r := httptest.NewRequest(http.MethodPost, "/twirp/livekit.Egress/StartRoomCompositeEgress", body) + // simulate chunked encoding / unknown length + r.ContentLength = -1 + w := httptest.NewRecorder() + + l.ServeHTTP(w, r, handler.ServeHTTP) + + // the decoder was invoked but could not read more than the limit + require.True(t, handler.called) + require.Error(t, handler.readErr) + require.LessOrEqual(t, handler.bytesRead, limit) + }) + + t.Run("allows request within limit", func(t *testing.T) { + l := service.NewRequestBodyLimiter(limit) + handler := &readAllHandler{} + + payload := strings.Repeat("a", limit/2) + r := httptest.NewRequest(http.MethodPost, "/twirp/livekit.Egress/StartRoomCompositeEgress", strings.NewReader(payload)) + w := httptest.NewRecorder() + + l.ServeHTTP(w, r, handler.ServeHTTP) + + require.Equal(t, http.StatusOK, w.Code) + require.True(t, handler.called) + require.NoError(t, handler.readErr) + require.Equal(t, len(payload), handler.bytesRead) + }) + + t.Run("disabled when limit is non-positive", func(t *testing.T) { + l := service.NewRequestBodyLimiter(0) + handler := &readAllHandler{} + + payload := strings.Repeat("a", limit*8) + r := httptest.NewRequest(http.MethodPost, "/twirp/livekit.Egress/StartRoomCompositeEgress", strings.NewReader(payload)) + w := httptest.NewRecorder() + + l.ServeHTTP(w, r, handler.ServeHTTP) + + require.Equal(t, http.StatusOK, w.Code) + require.NoError(t, handler.readErr) + require.Equal(t, len(payload), handler.bytesRead) + }) + + t.Run("passes through nil body", func(t *testing.T) { + l := service.NewRequestBodyLimiter(limit) + handler := &readAllHandler{} + + r := httptest.NewRequest(http.MethodGet, "/", nil) + r.Body = nil + w := httptest.NewRecorder() + + l.ServeHTTP(w, r, handler.ServeHTTP) + + require.True(t, handler.called) + }) +} diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index 0447f14e9..57282b3c3 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -306,7 +306,7 @@ func (r *RoomManager) StartSession( room, err := r.getOrCreateRoom(ctx, createRoom) if err != nil { if pi.Identity != "" { - prometheus.IncrementParticipantRtcCanceled(1) + prometheus.IncrementParticipantRtcCanceled(1, false) } return err } @@ -359,7 +359,7 @@ func (r *RoomManager) StartSession( Leave: leave, }, }) - prometheus.IncrementParticipantRtcCanceled(1) + prometheus.IncrementParticipantRtcCanceled(1, participant.IsWarpEnabled()) return errors.New("could not restart closed participant") } @@ -383,11 +383,11 @@ func (r *RoomManager) StartSession( pi.ReconnectReason, ); err != nil { participant.GetLogger().Warnw("could not resume participant", err) - prometheus.IncrementParticipantRtcCanceled(1) + prometheus.IncrementParticipantRtcCanceled(1, participant.IsWarpEnabled()) return err } r.telemetry.ParticipantResumed(ctx, room.ToProto(), participant.ToProto(), r.currentNode.NodeID(), pi.ReconnectReason) - prometheus.IncrementParticipantRtcActive(1) + prometheus.IncrementParticipantRtcActive(1, participant.IsWarpEnabled()) go room.HandleSyncState(participant, pi.SyncState) @@ -419,7 +419,7 @@ func (r *RoomManager) StartSession( Leave: leave, }, }) - prometheus.IncrementParticipantRtcCanceled(1) + prometheus.IncrementParticipantRtcCanceled(1, false) return errors.New("could not restart participant") } @@ -484,6 +484,7 @@ func (r *RoomManager) StartSession( enabledCodecs = append(enabledCodecs, &livekit.Codec{Mime: mime.MimeTypeRTX.String()}) } + enableWarp := !useOneShotSignallingMode && r.config.RTC.EnableWarp participant, err = rtc.NewParticipant(rtc.ParticipantParams{ Identity: pi.Identity, Name: pi.Name, @@ -541,9 +542,10 @@ func (r *RoomManager) StartSession( EnableDataTracks: r.config.EnableDataTracks, EnableParticipantDataBlob: r.config.EnableParticipantDataBlob, EnableRTPStreamRestartDetection: r.config.RTC.EnableRTPStreamRestartDetection, + EnableWarp: enableWarp, }) if err != nil { - prometheus.IncrementParticipantRtcCanceled(1) + prometheus.IncrementParticipantRtcCanceled(1, enableWarp) return err } iceConfig := r.setIceConfig(room.Name(), participant) @@ -559,7 +561,7 @@ func (r *RoomManager) StartSession( if err = room.Join(participant, requestSource, &opts, iceServers); err != nil { pLogger.Errorw("could not join room", err) _ = participant.Close(true, types.ParticipantCloseReasonJoinFailed, false) - prometheus.IncrementParticipantRtcCanceled(1) + prometheus.IncrementParticipantRtcCanceled(1, enableWarp) return err } @@ -571,7 +573,7 @@ func (r *RoomManager) StartSession( participantServerClosers.Close() pLogger.Errorw("could not join register participant topic", err) _ = participant.Close(true, types.ParticipantCloseReasonMessageBusFailed, false) - prometheus.IncrementParticipantRtcCanceled(1) + prometheus.IncrementParticipantRtcCanceled(1, enableWarp) return err } @@ -582,7 +584,7 @@ func (r *RoomManager) StartSession( participantServerClosers.Close() pLogger.Errorw("could not join register participant topic for rtc rest participant server", err) _ = participant.Close(true, types.ParticipantCloseReasonMessageBusFailed, false) - prometheus.IncrementParticipantRtcCanceled(1) + prometheus.IncrementParticipantRtcCanceled(1, enableWarp) return err } } diff --git a/pkg/service/server.go b/pkg/service/server.go index 1cf16e45f..e26569804 100644 --- a/pkg/service/server.go +++ b/pkg/service/server.go @@ -111,6 +111,8 @@ func NewLivekitServer(conf *config.Config, MaxAge: 86400, }), negroni.HandlerFunc(RemoveDoubleSlashes), + // limit request body size so large messages cannot exhaust memory + NewRequestBodyLimiter(conf.Limit.MaxAPIRequestBodySize), } if keyProvider != nil { middlewares = append(middlewares, NewAPIKeyAuthMiddleware(keyProvider)) diff --git a/pkg/service/utils.go b/pkg/service/utils.go index be7c0dc99..cc31be859 100644 --- a/pkg/service/utils.go +++ b/pkg/service/utils.go @@ -44,8 +44,9 @@ import ( ) var ( - ErrGzipReadFailed = errors.New("cannot read decompressed data") - ErrGzipTooLarge = errors.New("decompressed data too large") + ErrGzipReadFailed = errors.New("cannot read decompressed data") + ErrGzipTooLarge = errors.New("decompressed data too large") + ErrRequestBodyTooLarge = errors.New("request body too large") ) var gzipReaderPool = sync.Pool{ @@ -106,6 +107,39 @@ func RemoveDoubleSlashes(w http.ResponseWriter, r *http.Request, next http.Handl next(w, r) } +// RequestBodyLimiter bounds the size of an incoming HTTP request body so that +// large messages cannot exhaust memory. The Twirp handlers decode the whole +// body before any grant check runs, so the limit is applied here, up front. +// +// It does not decode the body itself: a request whose Content-Length exceeds +// the limit is rejected with 413, and the body is wrapped with +// http.MaxBytesReader so a missing or dishonest Content-Length is still caught +// by the downstream decoder. +type RequestBodyLimiter struct { + maxBytes int64 +} + +func NewRequestBodyLimiter(maxBytes int64) *RequestBodyLimiter { + return &RequestBodyLimiter{maxBytes: maxBytes} +} + +func (l *RequestBodyLimiter) ServeHTTP(w http.ResponseWriter, r *http.Request, next http.HandlerFunc) { + if l.maxBytes <= 0 || r.Body == nil { + next(w, r) + return + } + + // reject early when the declared size already exceeds the limit + if r.ContentLength > l.maxBytes { + HandleError(w, r, http.StatusRequestEntityTooLarge, ErrRequestBodyTooLarge) + return + } + + // bound the read in case Content-Length is missing or wrong + r.Body = http.MaxBytesReader(w, r.Body, l.maxBytes) + next(w, r) +} + func IsValidDomain(domain string) bool { domainRegexp := regexp.MustCompile(`^(?i)[a-z0-9-]+(\.[a-z0-9-]+)+\.?$`) return domainRegexp.MatchString(domain) diff --git a/pkg/sfu/buffer/frameintegrity.go b/pkg/sfu/buffer/frameintegrity.go index 935263b7d..e2cbe8a43 100644 --- a/pkg/sfu/buffer/frameintegrity.go +++ b/pkg/sfu/buffer/frameintegrity.go @@ -100,7 +100,15 @@ func (ph *PacketHistory) AddPacket(extSeq uint64) { return } - for i := ph.last + 1; i < extSeq; i++ { + // A forward jump only needs at most packetCount slots cleared; anything older aliases + // into the same ring and would be overwritten anyway. Cap the loop so a single crafted + // sequence number jump (up to ~32k by the extension wrap-around heuristic) cannot force + // a large per-packet iteration count. + start := ph.last + 1 + if extSeq-start > uint64(ph.packetCount) { + start = extSeq - uint64(ph.packetCount) + } + for i := start; i < extSeq; i++ { ph.set(i, false) } @@ -208,8 +216,13 @@ func (fc *FrameIntegrityChecker) AddPacket(extSeq uint64, extFrameNum uint64, dd return } - // reset missing frames - for i := fc.last + 1; i <= extFrameNum; i++ { + // reset missing frames; cap to frameCount so a crafted frame-number jump cannot force a + // large loop (older frames alias into the same ring and get overwritten anyway). + start := fc.last + 1 + if extFrameNum-fc.last > uint64(fc.frameCount) { + start = extFrameNum - uint64(fc.frameCount) + 1 + } + for i := start; i <= extFrameNum; i++ { fc.frames[int(i-fc.base)%fc.frameCount].Reset() } fc.frames[int(extFrameNum-fc.base)%fc.frameCount].AddPacket(extSeq, ddVal) diff --git a/pkg/sfu/buffer/frameintegrity_test.go b/pkg/sfu/buffer/frameintegrity_test.go index 2815cf5e9..c492e3d91 100644 --- a/pkg/sfu/buffer/frameintegrity_test.go +++ b/pkg/sfu/buffer/frameintegrity_test.go @@ -15,6 +15,7 @@ package buffer import ( + "math/bits" "math/rand" "testing" @@ -72,8 +73,8 @@ func TestFrameIntegrityChecker(t *testing.T) { frames = append(frames, i) } require.False(t, fc.FrameIntegrity(frame)) - rand.Seed(int64(frame)) - rand.Shuffle(len(frames), func(i, j int) { frames[i], frames[j] = frames[j], frames[i] }) + rng := rand.New(rand.NewSource(int64(frame))) + rng.Shuffle(len(frames), func(i, j int) { frames[i], frames[j] = frames[j], frames[i] }) for i, f := range frames { fc.AddPacket(f, frame, &dd.DependencyDescriptor{ FirstPacketInFrame: f == firstFrame, @@ -84,3 +85,63 @@ func TestFrameIntegrityChecker(t *testing.T) { require.True(t, fc.FrameIntegrity(frame)) } } + +func countSetBits(ph *PacketHistory) int { + n := 0 + for _, w := range ph.bits { + n += bits.OnesCount64(w) + } + return n +} + +// A forward sequence-number jump much larger than the ring must clear the whole ring, +// leaving only the newly received sequence number set. +func TestPacketHistoryLargeForwardJump(t *testing.T) { + ph := NewPacketHistory(1000) // rounds up to a multiple of 64 + require.Equal(t, 1024, ph.packetCount) + + // Fill the entire ring so every slot holds a "received" bit. + base := uint64(100000) + ph.AddPacket(base) + for i := base + 1; i <= base+2000; i++ { + ph.AddPacket(i) + } + require.Equal(t, ph.packetCount, countSetBits(ph)) + last := base + 2000 + + // Forward jump well beyond both the ring and the ~32k extension wrap-around cap. The ring + // must end up fully cleared, with only newSeq marked received. + newSeq := last + 40000 + ph.AddPacket(newSeq) + + // If the cap under-cleared, stale bits from the pre-jump fill would survive here. + require.Equal(t, 1, countSetBits(ph)) + require.True(t, ph.PacketsConsecutive(newSeq, newSeq)) + require.False(t, ph.PacketsConsecutive(newSeq-5, newSeq)) + + // The window just below newSeq was cleared and can be refilled normally. + for i := newSeq - 5; i < newSeq; i++ { + ph.AddPacket(i) + } + require.True(t, ph.PacketsConsecutive(newSeq-5, newSeq)) +} + +// A forward frame-number jump much larger than frameCount must reset the whole frame ring, +// so no frame that aliases an old slot inherits stale integrity. +func TestFrameIntegrityCheckerLargeFrameJump(t *testing.T) { + fc := NewFrameIntegrityChecker(100, 1000) + + // Populate every ring slot with an integral single-packet frame. + for f := uint64(200); f <= 399; f++ { + fc.AddPacket(f, f, &dd.DependencyDescriptor{FirstPacketInFrame: true, LastPacketInFrame: true}) + } + require.True(t, fc.FrameIntegrity(399)) + + // Jump far beyond frameCount. The capped reset loop must clear the entire frame ring; if it + // under-cleared, some aliased slot would still report a stale frame's integrity. + newFrame := uint64(399 + 5000) + fc.AddPacket(50000, newFrame, &dd.DependencyDescriptor{}) // incomplete frame, no first/last + for f := newFrame - uint64(fc.frameCount) + 1; f <= newFrame; f++ { + require.False(t, fc.FrameIntegrity(f), "frame %d should not be integral after jump", f) + } +} diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index d30de32a4..b0326f4a9 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -65,6 +65,9 @@ const ( keyFrameIntervalMax = 1000 flushTimeout = 1 * time.Second + // upper bound on NACKed sequence numbers buffered between retransmit worker runs + maxPendingNACKs = 5000 + waitBeforeSendPaddingOnMute = 100 * time.Millisecond maxPaddingOnMuteDuration = 5 * time.Second paddingOnMuteInterval = 100 * time.Millisecond @@ -336,6 +339,12 @@ type DownTrack struct { keyFrameRequesterCh chan struct{} keyFrameRequesterChClosed bool + retransmitChMu sync.RWMutex + retransmitCh chan struct{} + retransmitChClosed bool + pendingNACKsMu sync.Mutex + pendingNACKs []uint16 + createdAt int64 lastUnmutedAt atomic.Time } @@ -363,6 +372,7 @@ func NewDownTrack(params DownTrackParams) (*DownTrack, error) { pacer: params.Pacer, maxLayerNotifierCh: make(chan string, 1), keyFrameRequesterCh: make(chan struct{}, 1), + retransmitCh: make(chan struct{}, 1), createdAt: time.Now().UnixNano(), receiver: params.Receiver, } @@ -422,6 +432,8 @@ func NewDownTrack(params DownTrackParams) (*DownTrack, error) { go d.keyFrameRequester() } + go d.retransmitWorker() + d.params.Receiver.AddOnReady(d.handleReceiverReady) d.rtxSequenceNumber.Store(uint64(rand.Intn(1<<14)) + uint64(1<<15)) // a random number in third quartile of sequence number space d.params.Logger.Debugw("downtrack created", "upstreamCodecs", d.upstreamCodecs) @@ -702,6 +714,7 @@ func (d *DownTrack) handleUpstreamCodecChange(mimeType string) { receiver := d.Receiver() d.forwarder.Restart() + d.flushSequencer() d.forwarder.DetermineCodec(codec.RTPCodecCapability, receiver.HeaderExtensions(), receiver.VideoLayerMode()) d.connectionStats.UpdateCodec(d.Mime(), isFECEnabled) @@ -1022,6 +1035,15 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) int32 { return 0 } + if tp.incomingHeaderSize > len(extPkt.Packet.Payload) { + d.params.Logger.Errorw( + "incoming header size overflow", errPayloadOverflow, + "incomingHeaderSize", tp.incomingHeaderSize, + "payloadSize", len(extPkt.Packet.Payload), + ) + return 0 + } + poolEntity := PacketFactory.Get().(*[]byte) payload := *poolEntity copy(payload, tp.codecBytes) @@ -1461,6 +1483,11 @@ func (d *DownTrack) CloseWithFlush(flush bool, isEnding bool) { close(d.keyFrameRequesterCh) d.keyFrameRequesterChMu.Unlock() + d.retransmitChMu.Lock() + d.retransmitChClosed = true + close(d.retransmitCh) + d.retransmitChMu.Unlock() + d.params.Listener.OnDownTrackClose(!flush) } @@ -1700,6 +1727,16 @@ func (d *DownTrack) Pause() VideoAllocation { func (d *DownTrack) Resync() { d.forwarder.Resync() + d.flushSequencer() +} + +// flushSequencer discards recorded packet metadata on a stream restart so that NACK +// retransmissions cannot use metadata that describes packets no longer in the receiver's +// (resynced) bucket. +func (d *DownTrack) flushSequencer() { + if d.sequencer != nil { + d.sequencer.flush() + } } func (d *DownTrack) ReceiverRestart(rcvr TrackReceiver) { @@ -1715,6 +1752,7 @@ func (d *DownTrack) ReceiverRestart(rcvr TrackReceiver) { receiver := d.Receiver() d.params.Logger.Infow("upstream receiver restart", "mime", receiver.Mime().String()) d.forwarder.Restart() + d.flushSequencer() d.forwarder.DetermineCodec(codec, receiver.HeaderExtensions(), receiver.VideoLayerMode()) } @@ -2032,7 +2070,7 @@ func (d *DownTrack) handleRTCP(bytes []byte) { numNACKs += uint32(len(packetList)) nacks = append(nacks, packetList...) } - go d.retransmitPackets(nacks) + d.postRetransmitPackets(nacks) } case *rtcp.TransportLayerCC: @@ -2140,6 +2178,18 @@ func (d *DownTrack) retransmitPacket(epm *extPacketMeta, sourcePkt []byte, isPro d.params.Logger.Errorw("could not unmarshal rtp packet to send via RTX", err) return 0, err } + // Defensive panic-safety net: the codec header size was recorded when the packet was first + // forwarded and is re-read here against a bucket packet. A stream restart flushes the + // sequencer (see flushSequencer), so metadata and payload should always agree; guard + // against a slice overflow regardless. + if int(epm.numCodecBytesIn) > len(pkt.Payload) { + d.params.Logger.Warnw( + "recorded codec header size overflows payload", errPayloadOverflow, + "numCodecBytesIn", epm.numCodecBytesIn, + "payloadSize", len(pkt.Payload), + ) + return 0, errPayloadOverflow + } hdr := RTPHeaderFactory.Get().(*rtp.Header) *hdr = rtp.Header{ Version: pkt.Header.Version, @@ -2253,6 +2303,49 @@ func (d *DownTrack) retransmitPacket(epm *extPacketMeta, sourcePkt []byte, isPro return headerSize + len(payload), nil } +// postRetransmitPackets buffers the NACKed sequence numbers, capped at +// maxPendingNACKs, and signals the retransmit worker without blocking. +func (d *DownTrack) postRetransmitPackets(nacks []uint16) { + if d.sequencer == nil || len(nacks) == 0 { + return + } + + d.pendingNACKsMu.Lock() + d.pendingNACKs = append(d.pendingNACKs, nacks...) + if len(d.pendingNACKs) > maxPendingNACKs { + // drop the oldest NACKs under a NACK flood + d.pendingNACKs = d.pendingNACKs[len(d.pendingNACKs)-maxPendingNACKs:] + } + d.pendingNACKsMu.Unlock() + + d.retransmitChMu.RLock() + if !d.retransmitChClosed { + select { + case d.retransmitCh <- struct{}{}: + default: + } + } + d.retransmitChMu.RUnlock() +} + +func (d *DownTrack) retransmitWorker() { + for { + _, more := <-d.retransmitCh + if !more { + return + } + + d.pendingNACKsMu.Lock() + nacks := d.pendingNACKs + d.pendingNACKs = nil + d.pendingNACKsMu.Unlock() + + if len(nacks) > 0 { + d.retransmitPackets(nacks) + } + } +} + func (d *DownTrack) retransmitPackets(nacks []uint16) { if d.sequencer == nil { return diff --git a/pkg/sfu/sequencer.go b/pkg/sfu/sequencer.go index 1e08ef9ce..0507120c1 100644 --- a/pkg/sfu/sequencer.go +++ b/pkg/sfu/sequencer.go @@ -148,6 +148,26 @@ func (s *sequencer) setRTT(rtt uint32) { } } +// flush discards all recorded packet metadata. It must be called on a stream restart: the +// metadata maps outgoing sequence numbers to source packets in the receiver's bucket, and a +// restart resyncs that bucket, so retransmitting against stale metadata would send the wrong +// packet (or read past the re-read payload). After flush, NACKs for pre-restart packets are +// ignored until the sequencer is re-initialized by the next push. +func (s *sequencer) flush() { + s.Lock() + defer s.Unlock() + + s.initialized = false + s.extStartSN = 0 + s.extHighestSN = 0 + s.extHighestTS = 0 + s.snOffset = 0 + clear(s.meta) + if s.snRangeMap != nil { + s.snRangeMap = utils.NewRangeMap[uint64, uint64]((s.size + 1) / 2) + } +} + func (s *sequencer) push( packetTime int64, extIncomingSN, extModifiedSN uint64, diff --git a/pkg/sfu/sequencer_test.go b/pkg/sfu/sequencer_test.go index 862ac531b..be91af8be 100644 --- a/pkg/sfu/sequencer_test.go +++ b/pkg/sfu/sequencer_test.go @@ -78,6 +78,39 @@ func Test_sequencer(t *testing.T) { require.Equal(t, 1, len(m)) } +func Test_sequencer_flush(t *testing.T) { + seq := newSequencer(500, false, logger.GetLogger()) + off := uint16(15) + + for i := uint64(1); i < 100; i++ { + seq.push(time.Now().UnixNano(), i, i+uint64(off), 123, true, 2, nil, 0, nil, nil) + } + preFlush := []uint16{57 + off, 58 + off} + + // flush discards all recorded metadata on a stream restart + seq.flush() + + // even after enough time elapses, a NACK for a pre-flush packet retransmits nothing + time.Sleep((ignoreRetransmission + 10) * time.Millisecond) + require.Equal(t, 0, len(seq.getExtPacketMetas(preFlush))) + + // the sequencer re-initializes on the next push and works normally for new packets + for i := uint64(200); i < 210; i++ { + seq.push(time.Now().UnixNano(), i, i+uint64(off), 456, true, 3, nil, 0, nil, nil) + } + postFlush := []uint16{205 + off} + require.Equal(t, 0, len(seq.getExtPacketMetas(postFlush))) // not enough time elapsed yet + time.Sleep((ignoreRetransmission + 10) * time.Millisecond) + res := seq.getExtPacketMetas(postFlush) + require.Equal(t, 1, len(res)) + require.Equal(t, uint16(205+off), res[0].targetSeqNo) + require.Equal(t, uint64(205), res[0].sourceSeqNo) + require.Equal(t, int8(3), res[0].layer) + + // pre-flush packets remain non-retransmittable + require.Equal(t, 0, len(seq.getExtPacketMetas(preFlush))) +} + func Test_sequencer_getNACKSeqNo_exclusion(t *testing.T) { type args struct { seqNo []uint16 diff --git a/pkg/telemetry/events.go b/pkg/telemetry/events.go index bf684156e..989dd870e 100644 --- a/pkg/telemetry/events.go +++ b/pkg/telemetry/events.go @@ -111,6 +111,7 @@ func (t *telemetryService) ParticipantActive( participant *livekit.ParticipantInfo, clientMeta *livekit.AnalyticsClientMeta, isMigration bool, + isWarp bool, guard *ReferenceGuard, ) { t.enqueue(func() { @@ -135,7 +136,7 @@ func (t *telemetryService) ParticipantActive( prometheus.AddParticipant() } worker.SetConnected() - prometheus.IncrementParticipantRtcActive(1) + prometheus.IncrementParticipantRtcActive(1, isWarp) ev := newParticipantEvent(livekit.AnalyticsEventType_PARTICIPANT_ACTIVE, room, participant) ev.ClientMeta = clientMeta diff --git a/pkg/telemetry/events_test.go b/pkg/telemetry/events_test.go index 3677791b4..03fca0e4a 100644 --- a/pkg/telemetry/events_test.go +++ b/pkg/telemetry/events_test.go @@ -86,7 +86,7 @@ func Test_OnParticipantLeft_EventIsSent(t *testing.T) { guard := &telemetry.ReferenceGuard{} // do - fixture.sut.ParticipantActive(context.Background(), room, participantInfo, &livekit.AnalyticsClientMeta{}, false, guard) + fixture.sut.ParticipantActive(context.Background(), room, participantInfo, &livekit.AnalyticsClientMeta{}, false, false, guard) fixture.sut.ParticipantLeft(context.Background(), room, participantInfo, true, guard) time.Sleep(time.Millisecond * 500) @@ -179,7 +179,7 @@ func Test_OnParticipantActive_EventIsSent(t *testing.T) { ClientConnectTime: 420, } - fixture.sut.ParticipantActive(context.Background(), room, participantInfo, clientMetaConnect, false, guard) + fixture.sut.ParticipantActive(context.Background(), room, participantInfo, clientMetaConnect, false, false, guard) time.Sleep(time.Millisecond * 500) require.Equal(t, 2, fixture.analytics.SendEventCallCount()) diff --git a/pkg/telemetry/prometheus/packets.go b/pkg/telemetry/prometheus/packets.go index 6ca9ce436..37e67e895 100644 --- a/pkg/telemetry/prometheus/packets.go +++ b/pkg/telemetry/prometheus/packets.go @@ -15,6 +15,8 @@ package prometheus import ( + "strconv" + "github.com/prometheus/client_golang/prometheus" "go.uber.org/atomic" @@ -149,7 +151,7 @@ func initPacketStats(nodeID string, nodeType livekit.NodeType) { Subsystem: "participant_join", Name: "total", ConstLabels: prometheus.Labels{"node_id": nodeID, "node_type": nodeType.String()}, - }, []string{"state"}) + }, []string{"state", "warp"}) promConnections = prometheus.NewGaugeVec(prometheus.GaugeOpts{ Namespace: livekitNamespace, Subsystem: "connection", @@ -297,71 +299,72 @@ func RecordRTT(country string, direction Direction, trackSource livekit.TrackSou func IncrementParticipantJoin(join uint32) { if join > 0 { participantSignalConnected.Add(uint64(join)) - promParticipantJoin.WithLabelValues("signal_connected").Add(float64(join)) + promParticipantJoin.WithLabelValues("signal_connected", "").Add(float64(join)) } } func IncrementParticipantJoinFail(fail uint32) { if fail > 0 { - promParticipantJoin.WithLabelValues("signal_failed").Add(float64(fail)) + promParticipantJoin.WithLabelValues("signal_failed", "").Add(float64(fail)) } } func IncrementParticipantJoinValidationFail(validationFail uint32) { if validationFail > 0 { - promParticipantJoin.WithLabelValues("signal_validation_failed").Add(float64(validationFail)) + promParticipantJoin.WithLabelValues("signal_validation_failed", "").Add(float64(validationFail)) } } func IncrementParticipantJoinUpgradeFail(upgradeFail uint32) { if upgradeFail > 0 { - promParticipantJoin.WithLabelValues("signal_upgrade_failed").Add(float64(upgradeFail)) + promParticipantJoin.WithLabelValues("signal_upgrade_failed", "").Add(float64(upgradeFail)) } } func IncrementParticipantJoinWriteInitialResponseFail(writeInitialResponseFail uint32) { if writeInitialResponseFail > 0 { - promParticipantJoin.WithLabelValues("signal_write_initial_response_failed").Add(float64(writeInitialResponseFail)) + promParticipantJoin.WithLabelValues("signal_write_initial_response_failed", "").Add(float64(writeInitialResponseFail)) } } func IncrementParticipantRtcInit(init uint32) { if init > 0 { participantRTCInit.Add(uint64(init)) - promParticipantJoin.WithLabelValues("rtc_init").Add(float64(init)) + promParticipantJoin.WithLabelValues("rtc_init", "").Add(float64(init)) } } func IncrementParticipantRtcConnected(connected uint32) { if connected > 0 { participantRTCConnected.Add(uint64(connected)) - promParticipantJoin.WithLabelValues("rtc_connected").Add(float64(connected)) + promParticipantJoin.WithLabelValues("rtc_connected", "").Add(float64(connected)) } } -func IncrementParticipantRtcActive(active uint32) { +func IncrementParticipantRtcActive(active uint32, warp bool) { if active > 0 { participantRTCActive.Add(uint64(active)) - promParticipantJoin.WithLabelValues("rtc_active").Add(float64(active)) + promParticipantJoin.WithLabelValues("rtc_active", strconv.FormatBool(warp)).Add(float64(active)) } } -func IncrementParticipantRtcCanceled(canceled uint64) { +func IncrementParticipantRtcCanceled(canceled uint64, warp bool) { if canceled > 0 { participantRTCCanceled.Add(canceled) - promParticipantJoin.WithLabelValues("rtc_canceled").Add(float64(canceled)) + promParticipantJoin.WithLabelValues("rtc_canceled", strconv.FormatBool(warp)).Add(float64(canceled)) } } -func IncrementParticipantRtcSuccess(success uint64) { +// todo: check if need to record warp to rtcSucc/Failure +func IncrementParticipantRtcSuccess(success uint64, warp bool) { if success > 0 { - promParticipantJoin.WithLabelValues("rtc_success").Add(float64(success)) + promParticipantJoin.WithLabelValues("rtc_success", strconv.FormatBool(warp)).Add(float64(success)) } } -func IncrementParticipantRtcFailure(failure uint64) { +func IncrementParticipantRtcFailure(failure uint64, warp bool) { if failure > 0 { - promParticipantJoin.WithLabelValues("rtc_failure").Add(float64(failure)) + promParticipantJoin.WithLabelValues("rtc_failure", strconv.FormatBool(warp)).Add(float64(failure)) } } diff --git a/pkg/telemetry/prometheus/rooms.go b/pkg/telemetry/prometheus/rooms.go index 0733413a3..9a7a9c35b 100644 --- a/pkg/telemetry/prometheus/rooms.go +++ b/pkg/telemetry/prometheus/rooms.go @@ -113,7 +113,7 @@ func initRoomStats(nodeID string, nodeType livekit.NodeType) { Name: "start_time_ms", ConstLabels: prometheus.Labels{"node_id": nodeID, "node_type": nodeType.String()}, Buckets: prometheus.ExponentialBucketsRange(100, 10000, 15), - }, []string{"protocol_version"}) + }, []string{"protocol_version", "warp"}) promSessionDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{ Namespace: livekitNamespace, Subsystem: "session", @@ -291,8 +291,8 @@ func RecordSessionJoinLatency(protocolVersion int, d time.Duration) { promSessionJoinLatency.WithLabelValues(strconv.Itoa(protocolVersion)).Observe(float64(d.Milliseconds())) } -func RecordSessionStartTime(protocolVersion int, d time.Duration) { - promSessionStartTime.WithLabelValues(strconv.Itoa(protocolVersion)).Observe(float64(d.Milliseconds())) +func RecordSessionStartTime(protocolVersion int, warp bool, d time.Duration) { + promSessionStartTime.WithLabelValues(strconv.Itoa(protocolVersion), strconv.FormatBool(warp)).Observe(float64(d.Milliseconds())) } func RecordSessionDuration(protocolVersion int, d time.Duration) { diff --git a/pkg/telemetry/telemetryfakes/fake_telemetry_service.go b/pkg/telemetry/telemetryfakes/fake_telemetry_service.go index 5c0f28881..25a241bef 100644 --- a/pkg/telemetry/telemetryfakes/fake_telemetry_service.go +++ b/pkg/telemetry/telemetryfakes/fake_telemetry_service.go @@ -83,7 +83,7 @@ type FakeTelemetryService struct { arg2 string arg3 *livekit.EgressInfo } - ParticipantActiveStub func(context.Context, *livekit.Room, *livekit.ParticipantInfo, *livekit.AnalyticsClientMeta, bool, *telemetry.ReferenceGuard) + ParticipantActiveStub func(context.Context, *livekit.Room, *livekit.ParticipantInfo, *livekit.AnalyticsClientMeta, bool, bool, *telemetry.ReferenceGuard) participantActiveMutex sync.RWMutex participantActiveArgsForCall []struct { arg1 context.Context @@ -91,7 +91,8 @@ type FakeTelemetryService struct { arg3 *livekit.ParticipantInfo arg4 *livekit.AnalyticsClientMeta arg5 bool - arg6 *telemetry.ReferenceGuard + arg6 bool + arg7 *telemetry.ReferenceGuard } ParticipantJoinedStub func(context.Context, *livekit.Room, *livekit.ParticipantInfo, *livekit.ClientInfo, *livekit.AnalyticsClientMeta, bool, *telemetry.ReferenceGuard) participantJoinedMutex sync.RWMutex @@ -698,7 +699,7 @@ func (fake *FakeTelemetryService) NotifyEgressEventArgsForCall(i int) (context.C return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3 } -func (fake *FakeTelemetryService) ParticipantActive(arg1 context.Context, arg2 *livekit.Room, arg3 *livekit.ParticipantInfo, arg4 *livekit.AnalyticsClientMeta, arg5 bool, arg6 *telemetry.ReferenceGuard) { +func (fake *FakeTelemetryService) ParticipantActive(arg1 context.Context, arg2 *livekit.Room, arg3 *livekit.ParticipantInfo, arg4 *livekit.AnalyticsClientMeta, arg5 bool, arg6 bool, arg7 *telemetry.ReferenceGuard) { fake.participantActiveMutex.Lock() fake.participantActiveArgsForCall = append(fake.participantActiveArgsForCall, struct { arg1 context.Context @@ -706,13 +707,14 @@ func (fake *FakeTelemetryService) ParticipantActive(arg1 context.Context, arg2 * arg3 *livekit.ParticipantInfo arg4 *livekit.AnalyticsClientMeta arg5 bool - arg6 *telemetry.ReferenceGuard - }{arg1, arg2, arg3, arg4, arg5, arg6}) + arg6 bool + arg7 *telemetry.ReferenceGuard + }{arg1, arg2, arg3, arg4, arg5, arg6, arg7}) stub := fake.ParticipantActiveStub - fake.recordInvocation("ParticipantActive", []interface{}{arg1, arg2, arg3, arg4, arg5, arg6}) + fake.recordInvocation("ParticipantActive", []interface{}{arg1, arg2, arg3, arg4, arg5, arg6, arg7}) fake.participantActiveMutex.Unlock() if stub != nil { - fake.ParticipantActiveStub(arg1, arg2, arg3, arg4, arg5, arg6) + fake.ParticipantActiveStub(arg1, arg2, arg3, arg4, arg5, arg6, arg7) } } @@ -722,17 +724,17 @@ func (fake *FakeTelemetryService) ParticipantActiveCallCount() int { return len(fake.participantActiveArgsForCall) } -func (fake *FakeTelemetryService) ParticipantActiveCalls(stub func(context.Context, *livekit.Room, *livekit.ParticipantInfo, *livekit.AnalyticsClientMeta, bool, *telemetry.ReferenceGuard)) { +func (fake *FakeTelemetryService) ParticipantActiveCalls(stub func(context.Context, *livekit.Room, *livekit.ParticipantInfo, *livekit.AnalyticsClientMeta, bool, bool, *telemetry.ReferenceGuard)) { fake.participantActiveMutex.Lock() defer fake.participantActiveMutex.Unlock() fake.ParticipantActiveStub = stub } -func (fake *FakeTelemetryService) ParticipantActiveArgsForCall(i int) (context.Context, *livekit.Room, *livekit.ParticipantInfo, *livekit.AnalyticsClientMeta, bool, *telemetry.ReferenceGuard) { +func (fake *FakeTelemetryService) ParticipantActiveArgsForCall(i int) (context.Context, *livekit.Room, *livekit.ParticipantInfo, *livekit.AnalyticsClientMeta, bool, bool, *telemetry.ReferenceGuard) { fake.participantActiveMutex.RLock() defer fake.participantActiveMutex.RUnlock() argsForCall := fake.participantActiveArgsForCall[i] - return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5, argsForCall.arg6 + return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5, argsForCall.arg6, argsForCall.arg7 } func (fake *FakeTelemetryService) ParticipantJoined(arg1 context.Context, arg2 *livekit.Room, arg3 *livekit.ParticipantInfo, arg4 *livekit.ClientInfo, arg5 *livekit.AnalyticsClientMeta, arg6 bool, arg7 *telemetry.ReferenceGuard) { diff --git a/pkg/telemetry/telemetryservice.go b/pkg/telemetry/telemetryservice.go index 28c90400b..bdf9c820c 100644 --- a/pkg/telemetry/telemetryservice.go +++ b/pkg/telemetry/telemetryservice.go @@ -40,7 +40,7 @@ type TelemetryService interface { // ParticipantJoined - a participant establishes signal connection to a room ParticipantJoined(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, clientInfo *livekit.ClientInfo, clientMeta *livekit.AnalyticsClientMeta, shouldSendEvent bool, guard *ReferenceGuard) // ParticipantActive - a participant establishes media connection - ParticipantActive(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, clientMeta *livekit.AnalyticsClientMeta, isMigration bool, guard *ReferenceGuard) + ParticipantActive(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, clientMeta *livekit.AnalyticsClientMeta, isMigration bool, isWarp bool, guard *ReferenceGuard) // ParticipantResumed - there has been an ICE restart or connection resume attempt, and we've received their signal connection ParticipantResumed(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, nodeID livekit.NodeID, reason livekit.ReconnectReason) // ParticipantLeft - the participant leaves the room, only sent if ParticipantActive has been called before @@ -108,7 +108,7 @@ func (n NullTelemetryService) RoomStarted(ctx context.Context, room *livekit.Roo func (n NullTelemetryService) RoomEnded(ctx context.Context, room *livekit.Room) {} func (n NullTelemetryService) ParticipantJoined(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, clientInfo *livekit.ClientInfo, clientMeta *livekit.AnalyticsClientMeta, shouldSendEvent bool, guard *ReferenceGuard) { } -func (n NullTelemetryService) ParticipantActive(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, clientMeta *livekit.AnalyticsClientMeta, isMigration bool, guard *ReferenceGuard) { +func (n NullTelemetryService) ParticipantActive(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, clientMeta *livekit.AnalyticsClientMeta, isMigration bool, isWarp bool, guard *ReferenceGuard) { } func (n NullTelemetryService) ParticipantResumed(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, nodeID livekit.NodeID, reason livekit.ReconnectReason) { }