Merge remote-tracking branch 'origin/master' into renovate-pinning

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

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

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=

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",

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
+}

pkg/rtc/room.go

@@ -1265,6 +1265,7 @@ func (r *Room) onStateChange(p types.LocalParticipant) {
 			p.ToProto(),
 			meta,
 			false,
+			p.IsWarpEnabled(),
 			p.TelemetryGuard(),
 		)
 

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.

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

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

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 {

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
 		}
 

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)
 	}

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)
+	})
+}

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
 		}
 	}

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))

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)

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)

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)
+	}
+}

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

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,

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

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

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())

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))
 	}
 }
 

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) {

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) {

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) {
 }