From a47e21b6cb945aabee88c98650366cef8cbf7a99 Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Mon, 29 Jun 2026 10:44:08 -0700 Subject: [PATCH] Data track schema metadata (#4622) * Async attributes on participant. How it is different from existing participant attributes? 1. Async attribute can be added one at a time. 2. These are not included in `ParticipantInfo`. 3. Get an attribute bt participant identity and async attribute ID as and when needed. * clean up * get full definitions, not just ids * listener OnDataTrackSchema * name length config * data blob * deps * static check * Add missing request ID * Update protocol commit * Wire up StoreDataBlobResponse * Pass request ID through in GetDataBlobResponse * Pin protocol for schema metadata * Pass through schema and frame encoding * Support custom encoding identifiers * Rename config key * Increase default length to 32 * Make log messages more generic * Use getters with built-in null check * Do not bump deps * Rename function * Use protocol v1.48.1 release --------- Co-authored-by: boks1971 --- go.mod | 2 +- go.sum | 4 +-- pkg/config/config.go | 37 +++++++++++++++++++----- pkg/rtc/participant_data_blob_handler.go | 10 +++++++ pkg/rtc/participant_data_track.go | 15 ++++++++++ 5 files changed, 58 insertions(+), 10 deletions(-) diff --git a/go.mod b/go.mod index 84378d367..06e261c39 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.48.1-0.20260623210753-2e1bfd81dd63 + github.com/livekit/protocol v1.48.1 github.com/livekit/psrpc v0.7.2 github.com/mackerelio/go-osstat v0.2.7 github.com/magefile/mage v1.17.2 diff --git a/go.sum b/go.sum index 3a0b13d74..4e0188b7a 100644 --- a/go.sum +++ b/go.sum @@ -160,8 +160,8 @@ github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 h1:9x+U2HGLrSw5AT 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.48.1-0.20260623210753-2e1bfd81dd63 h1:Rj9/54oztXeioAwUkukXBwcPE/GxL97WylFd1m7V2pQ= -github.com/livekit/protocol v1.48.1-0.20260623210753-2e1bfd81dd63/go.mod h1:jO+y05AU9Ec4JswDyuzKCZ4bhziOS0CzMqgnbj60Dzs= +github.com/livekit/protocol v1.48.1 h1:vFnMqGeknJo6jDwW4wU+x5KuxgA95fPuz7MriAnLzCc= +github.com/livekit/protocol v1.48.1/go.mod h1:jO+y05AU9Ec4JswDyuzKCZ4bhziOS0CzMqgnbj60Dzs= github.com/livekit/psrpc v0.7.2 h1:6oZ+NODJ2pLyaT6VqDq1F4Qc/3TpDUSpyphj/P9MhQc= github.com/livekit/psrpc v0.7.2/go.mod h1:rAI+m2+/cb4x9RXhLRtUx5ZwdfjjXOl4zi46IjEetaw= github.com/mackerelio/go-osstat v0.2.7 h1:TCavZi10wF49bT6iQZ9eT2keGZQpC69MTDfdJej5e94= diff --git a/pkg/config/config.go b/pkg/config/config.go index f85f6a841..90135c781 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -298,6 +298,8 @@ type LimitConfig struct { MaxDataBlobKeyLength int `yaml:"max_data_blob_key_length,omitempty"` MaxDataBlobSize uint32 `yaml:"max_data_blobs_size,omitempty"` + + MaxDataTrackCustomEncodingLength int `yaml:"max_data_track_custom_encoding_length,omitempty"` } func (l LimitConfig) CheckRoomNameLength(name string) bool { @@ -332,6 +334,26 @@ func (l LimitConfig) CheckDataBlobKeyLength(key string) bool { return l.MaxDataBlobKeyLength == 0 || len(key) <= l.MaxDataBlobKeyLength } +func (l LimitConfig) CheckDataTrackCustomEncodingLength(identifier string) bool { + return l.MaxDataTrackCustomEncodingLength == 0 || len(identifier) <= l.MaxDataTrackCustomEncodingLength +} + +func (l LimitConfig) CheckDataTrackFrameEncoding(encoding *livekit.DataTrackFrameEncoding) bool { + custom, ok := encoding.GetValue().(*livekit.DataTrackFrameEncoding_Custom) + if !ok { + return true + } + return len(custom.Custom) != 0 && l.CheckDataTrackCustomEncodingLength(custom.Custom) +} + +func (l LimitConfig) CheckDataTrackSchemaID(schema *livekit.DataTrackSchemaId) bool { + custom, ok := schema.GetEncoding().GetValue().(*livekit.DataTrackSchemaEncoding_Custom) + if !ok { + return true + } + return len(custom.Custom) != 0 && l.CheckDataTrackCustomEncodingLength(custom.Custom) +} + func (l LimitConfig) CheckDataBlobsSize(dataBlobs []*livekit.DataBlob) bool { if l.MaxDataBlobSize == 0 { return true @@ -475,13 +497,14 @@ var DefaultConfig = Config{ UpdateBatchTargetSize: 128 * 1024, }, Limit: LimitConfig{ - MaxMetadataSize: 512 * 1024, - MaxAttributesSize: 64 * 1024, - MaxRoomNameLength: 256, - MaxParticipantIdentityLength: 256, - MaxParticipantNameLength: 256, - MaxDataBlobKeyLength: 256, - MaxDataBlobSize: 64000, + MaxMetadataSize: 512 * 1024, + MaxAttributesSize: 64 * 1024, + MaxRoomNameLength: 256, + MaxParticipantIdentityLength: 256, + MaxParticipantNameLength: 256, + MaxDataBlobKeyLength: 256, + MaxDataBlobSize: 64000, + MaxDataTrackCustomEncodingLength: 32, }, Logging: LoggingConfig{ PionLevel: "error", diff --git a/pkg/rtc/participant_data_blob_handler.go b/pkg/rtc/participant_data_blob_handler.go index e6ebcc8ed..8c19ee765 100644 --- a/pkg/rtc/participant_data_blob_handler.go +++ b/pkg/rtc/participant_data_blob_handler.go @@ -41,6 +41,16 @@ func (p *ParticipantImpl) HandleStoreDataBlobRequest(req *livekit.StoreDataBlobR return } + if !p.params.LimitConfig.CheckDataTrackSchemaID(req.Blob.Key.GetSchemaId()) { + p.pubLogger.Warnw("data blob key schema id is invalid", nil, "req", logger.Proto(req)) + p.sendRequestResponse(&livekit.RequestResponse{ + RequestId: req.RequestId, + Reason: livekit.RequestResponse_INVALID_REQUEST, + Message: "encoding identifier is empty or exceeds the maximum length", + }) + return + } + if len(req.Blob.Contents) == 0 { p.sendRequestResponse(&livekit.RequestResponse{ RequestId: req.RequestId, diff --git a/pkg/rtc/participant_data_track.go b/pkg/rtc/participant_data_track.go index dbda1ede0..df9854744 100644 --- a/pkg/rtc/participant_data_track.go +++ b/pkg/rtc/participant_data_track.go @@ -60,6 +60,19 @@ func (p *ParticipantImpl) HandlePublishDataTrackRequest(req *livekit.PublishData return } + if !p.params.LimitConfig.CheckDataTrackFrameEncoding(req.FrameEncoding) || + !p.params.LimitConfig.CheckDataTrackSchemaID(req.Schema) { + p.pubLogger.Warnw("invalid encoding identifier", nil, "req", logger.Proto(req)) + p.sendRequestResponse(&livekit.RequestResponse{ + Reason: livekit.RequestResponse_INVALID_REQUEST, + Message: "encoding identifier is empty or exceeds the maximum length", + Request: &livekit.RequestResponse_PublishDataTrack{ + PublishDataTrack: utils.CloneProto(req), + }, + }) + return + } + publishedDataTracks := p.UpDataTrackManager.GetPublishedDataTracks() for _, dt := range publishedDataTracks { message := "" @@ -95,6 +108,8 @@ func (p *ParticipantImpl) HandlePublishDataTrackRequest(req *livekit.PublishData Name: req.Name, Encryption: req.Encryption, } + dti.FrameEncoding = utils.CloneProto(req.GetFrameEncoding()) + dti.Schema = utils.CloneProto(req.GetSchema()) dt := NewDataTrack( DataTrackParams{ Logger: p.params.Logger.WithValues("trackID", dti.Sid),