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 <raja.gobi@tutanota.com>

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

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=

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

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,

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