Data track schema metadata (#1159)

Adds support for associating schema metadata with a published data track
and storing/retrieving schema definitions.

Schema storage is built on-top of data blobs, a general purpose
mechanism for storing large (in the order of KBs), arbitrary data blobs
in a room:
- Protocol: livekit/protocol#1595
- SFU implementation:
https://github.com/livekit/livekit/tree/raja_async_attributes

Protocol additions for schema metadata:
- livekit/protocol#1553

Closes BOT-368

---------

Co-authored-by: github-actions <41898282+github-actions[bot]@users.noreply.github.com>

Author: ladvoc

.changeset/data_tracks_schema_metadata_support.md

@@ -0,0 +1,10 @@
+---
+livekit-api: patch
+livekit-datatrack: minor
+livekit-ffi: patch
+livekit-protocol: patch
+livekit-uniffi: patch
+livekit: patch
+---
+
+Data tracks schema metadata support.

.github/workflows/tests.yml

@@ -151,7 +151,9 @@ jobs:
         uses: livekit/dev-server-action@61e2b4dcb170dd3591e0c9b0db3c3fe5db93b500
         with:
           github-token: ${{ github.token }}
-
+          config: |
+            # TODO: Remove once this is enabled by default.
+            enable_participant_data_blob: true
       - name: Test
         env:
           RUST_LOG: info

livekit-datatrack/src/lib.rs

@@ -17,6 +17,9 @@
 /// Common types for local and remote tracks.
 mod track;
 
+/// Schema and frame encoding metadata for typed tracks.
+mod schema;
+
 /// Local track publication.
 mod local;
 
@@ -40,7 +43,7 @@ mod error;
 
 /// Public APIs re-exported by client SDKs.
 pub mod api {
-    pub use crate::{error::*, frame::*, local::*, remote::*, track::*};
+    pub use crate::{error::*, frame::*, local::*, remote::*, schema::*, track::*};
 }
 
 /// Internal APIs used within client SDKs to power data tracks functionality.

livekit-datatrack/src/local/events.rs

@@ -15,6 +15,7 @@
 use crate::{
     api::{DataTrackInfo, DataTrackOptions, LocalDataTrack, PublishError},
     packet::Handle,
+    schema::{DataTrackFrameEncoding, DataTrackSchemaId},
 };
 use bytes::Bytes;
 use from_variants::FromVariants;
@@ -124,6 +125,8 @@ pub struct SfuPublishRequest {
     pub handle: Handle,
     pub name: String,
     pub uses_e2ee: bool,
+    pub schema: Option<DataTrackSchemaId>,
+    pub frame_encoding: Option<DataTrackFrameEncoding>,
 }
 
 /// Request sent to the SFU to unpublish a track.

livekit-datatrack/src/local/manager.rs

@@ -108,6 +108,14 @@ impl Manager {
     }
 
     async fn on_publish_request(&mut self, event: PublishRequest) {
+        if let Err(error) = crate::schema::validate_schema(
+            event.options.frame_encoding.as_ref(),
+            event.options.schema.as_ref().map(|id| id.encoding()),
+        ) {
+            _ = event.result_tx.send(Err(PublishError::InvalidSchema(error)));
+            return;
+        }
+
         let Some(handle) = self.handle_allocator.get() else {
             _ = event.result_tx.send(Err(PublishError::LimitReached));
             return;
@@ -134,6 +142,8 @@ impl Manager {
             handle,
             name: event.options.name,
             uses_e2ee: self.encryption_provider.is_some(),
+            schema: event.options.schema,
+            frame_encoding: event.options.frame_encoding,
         };
         _ = self.event_out_tx.send(event.into()).await;
     }
@@ -286,6 +296,8 @@ impl Manager {
                         handle: info.pub_handle,
                         name: info.name.clone(),
                         uses_e2ee: info.uses_e2ee,
+                        schema: info.schema.clone(),
+                        frame_encoding: info.frame_encoding.clone(),
                     };
                     _ = state_tx.send(PublishState::Republishing);
                     _ = self.event_out_tx.send(event.into()).await;
@@ -543,6 +555,8 @@ mod tests {
                             pub_handle,
                             name: event.name,
                             uses_e2ee: event.uses_e2ee,
+                            schema: None,
+                            frame_encoding: None,
                         };
                         let event = SfuPublishResponse { handle: event.handle, result: Ok(info) };
                         _ = input.send(event.into());
@@ -622,6 +636,8 @@ mod tests {
             pub_handle: handle,
             name: "test".into(),
             uses_e2ee: false,
+            schema: None,
+            frame_encoding: None,
         };
         let event = SfuPublishResponse { handle, result: Ok(info) };
         input.send(event.into()).unwrap();
@@ -652,6 +668,8 @@ mod tests {
             pub_handle: event.handle,
             name: "secure".into(),
             uses_e2ee: true,
+            schema: None,
+            frame_encoding: None,
         };
         let event = SfuPublishResponse { handle: event.handle, result: Ok(info) };
         input.send(event.into()).unwrap();
@@ -692,6 +710,8 @@ mod tests {
             pub_handle: handle,
             name: track_name.clone(),
             uses_e2ee: false,
+            schema: None,
+            frame_encoding: None,
         };
         let event = SfuPublishResponse { handle, result: Ok(info) };
         input.send(event.into()).unwrap();
@@ -717,6 +737,8 @@ mod tests {
             pub_handle: handle,
             name: track_name.clone(),
             uses_e2ee: false,
+            schema: None,
+            frame_encoding: None,
         };
         let event = SfuPublishResponse { handle, result: Ok(info) };
         input.send(event.into()).unwrap();
@@ -746,6 +768,8 @@ mod tests {
                 pub_handle: event.handle,
                 name: name.into(),
                 uses_e2ee: false,
+                schema: None,
+                frame_encoding: None,
             };
             let event = SfuPublishResponse { handle: event.handle, result: Ok(info) };
             input.send(event.into()).unwrap();
@@ -785,6 +809,8 @@ mod tests {
             pub_handle: event.handle,
             name: "active".into(),
             uses_e2ee: false,
+            schema: None,
+            frame_encoding: None,
         };
         let event = SfuPublishResponse { handle: event.handle, result: Ok(info) };
         input.send(event.into()).unwrap();

livekit-datatrack/src/local/mod.rs

@@ -13,7 +13,8 @@
 // limitations under the License.
 
 use crate::{
-    api::{DataTrack, DataTrackFrame, DataTrackInfo, InternalError},
+    api::{DataTrack, DataTrackFrame, DataTrackInfo, DataTrackSchemaError, InternalError},
+    schema::{DataTrackFrameEncoding, DataTrackSchemaId},
     track::DataTrackInner,
 };
 use std::{fmt, marker::PhantomData, sync::Arc};
@@ -153,6 +154,8 @@ impl Drop for LocalTrackInner {
 #[derive(Clone, Debug)]
 pub struct DataTrackOptions {
     pub(crate) name: String,
+    pub(crate) schema: Option<DataTrackSchemaId>,
+    pub(crate) frame_encoding: Option<DataTrackFrameEncoding>,
 }
 
 impl DataTrackOptions {
@@ -165,7 +168,17 @@ impl DataTrackOptions {
     /// - Must be unique per publisher
     ///
     pub fn new(name: impl Into<String>) -> Self {
-        Self { name: name.into() }
+        Self { name: name.into(), schema: None, frame_encoding: None }
+    }
+
+    /// Sets the schema associated with frames sent on the track.
+    pub fn with_schema(self, schema: DataTrackSchemaId) -> Self {
+        Self { schema: Some(schema), ..self }
+    }
+
+    /// Sets the encoding of frames sent on the track.
+    pub fn with_frame_encoding(self, encoding: DataTrackFrameEncoding) -> Self {
+        Self { frame_encoding: Some(encoding), ..self }
     }
 }
 
@@ -216,6 +229,10 @@ pub enum PublishError {
     #[error("Room disconnected")]
     Disconnected,
 
+    /// Schema metadata is invalid.
+    #[error(transparent)]
+    InvalidSchema(DataTrackSchemaError),
+
     /// Internal error, please report on GitHub.
     #[error(transparent)]
     Internal(#[from] InternalError),

livekit-datatrack/src/local/proto.rs

@@ -33,7 +33,15 @@ impl From<SfuPublishRequest> for proto::PublishDataTrackRequest {
     fn from(event: SfuPublishRequest) -> Self {
         use proto::encryption::Type;
         let encryption = if event.uses_e2ee { Type::Gcm } else { Type::None }.into();
-        Self { pub_handle: event.handle.into(), name: event.name, encryption }
+        let schema = event.schema.map(Into::into);
+        let frame_encoding = event.frame_encoding.map(Into::into);
+        Self {
+            pub_handle: event.handle.into(),
+            name: event.name,
+            encryption,
+            schema,
+            frame_encoding,
+        }
     }
 }
 
@@ -74,8 +82,18 @@ impl TryFrom<proto::DataTrackInfo> for DataTrackInfo {
             proto::encryption::Type::Gcm => true,
             other => Err(anyhow!("Unsupported E2EE type: {:?}", other))?,
         };
+        let frame_encoding = msg.frame_encoding.map(Into::into);
         let sid: DataTrackSid = msg.sid.try_into().map_err(anyhow::Error::from)?;
-        Ok(Self { pub_handle: handle, sid: RwLock::new(sid).into(), name: msg.name, uses_e2ee })
+        let schema = msg.schema.map(|schema| schema.into());
+
+        Ok(Self {
+            pub_handle: handle,
+            sid: RwLock::new(sid).into(),
+            name: msg.name,
+            uses_e2ee,
+            schema,
+            frame_encoding,
+        })
     }
 }
 
@@ -106,12 +124,17 @@ impl From<DataTrackInfo> for proto::DataTrackInfo {
             proto::encryption::Type::Gcm
         } else {
             proto::encryption::Type::None
-        };
+        } as i32;
+        let sid = info.sid().to_string();
+        let schema = info.schema.map(|schema| schema.into());
+        let frame_encoding = info.frame_encoding.map(Into::into);
         Self {
             pub_handle: info.pub_handle.into(),
-            sid: info.sid().to_string(),
+            sid,
             name: info.name,
-            encryption: encryption as i32,
+            encryption,
+            schema,
+            frame_encoding,
         }
     }
 }
@@ -128,6 +151,8 @@ pub fn publish_responses_for_sync_state(
 
 #[cfg(test)]
 mod tests {
+    use crate::schema::{DataTrackFrameEncoding, DataTrackSchemaEncoding, DataTrackSchemaId};
+
     use super::*;
     use fake::{Fake, Faker};
 
@@ -137,6 +162,8 @@ mod tests {
             handle: 1u32.try_into().unwrap(),
             name: "track".into(),
             uses_e2ee: true,
+            schema: None,
+            frame_encoding: None,
         };
         let request: proto::PublishDataTrackRequest = event.into();
         assert_eq!(request.pub_handle, 1);
@@ -159,6 +186,12 @@ mod tests {
                 sid: "DTR_1234".into(),
                 name: "track".into(),
                 encryption: proto::encryption::Type::Gcm.into(),
+                schema: proto::DataTrackSchemaId {
+                    name: "schema".into(),
+                    encoding: Some(DataTrackSchemaEncoding::JsonSchema.into()),
+                }
+                .into(),
+                frame_encoding: Some(DataTrackFrameEncoding::Json.into()),
             }
             .into(),
         };
@@ -169,9 +202,43 @@ mod tests {
         assert_eq!(info.pub_handle, 1u32.try_into().unwrap());
         assert_eq!(*info.sid.read().unwrap(), "DTR_1234".to_string().try_into().unwrap());
         assert_eq!(info.name, "track");
+        assert_eq!(
+            info.schema,
+            Some(DataTrackSchemaId::new("schema", DataTrackSchemaEncoding::JsonSchema))
+        );
+        assert_eq!(info.frame_encoding, Some(DataTrackFrameEncoding::Json));
         assert!(info.uses_e2ee);
     }
 
+    #[test]
+    fn test_frame_encoding_mapping() {
+        let base = proto::DataTrackInfo {
+            pub_handle: 1,
+            sid: "DTR_1234".into(),
+            name: "track".into(),
+            encryption: proto::encryption::Type::None.into(),
+            schema: None,
+            frame_encoding: None,
+        };
+
+        let info: DataTrackInfo = base.clone().try_into().unwrap();
+        assert_eq!(info.frame_encoding, None);
+
+        let unspecified = proto::DataTrackInfo {
+            frame_encoding: Some(DataTrackFrameEncoding::Other.into()),
+            ..base.clone()
+        };
+        let info: DataTrackInfo = unspecified.try_into().unwrap();
+        assert_eq!(info.frame_encoding, Some(DataTrackFrameEncoding::Other));
+
+        let custom = proto::DataTrackInfo {
+            frame_encoding: Some(DataTrackFrameEncoding::Custom("my_encoding".into()).into()),
+            ..base
+        };
+        let info: DataTrackInfo = custom.try_into().unwrap();
+        assert_eq!(info.frame_encoding, Some(DataTrackFrameEncoding::Custom("my_encoding".into())));
+    }
+
     #[test]
     fn test_from_request_response() {
         use proto::request_response::{Reason, Request};